dune-weaver/modules/core/execution.py

"""
Firmware-delegated execution: the board runs everything, the host proxies
commands and observes status.

The FluidNC firmware owns kinematics, `.thr` playback, playlist sequencing,
pre-execution clears, quiet hours and auto-home. This module is the host's
entire runtime layer on top of that:

  - Commands: each user action is one or a few HTTP calls to the board
    ($Sand/Run, $Playlist/Run, /sand_stop, /sand_pause, /sand_resume,
    $Playlist/Skip, /sand_feed, /sand_goto).
  - Truth: /sand_status is the single source of runtime state. One observer
    task polls it, translates it into the frontend's /ws/status contract,
    detects edges (file transitions -> play history, Hold -> pause accounting,
    clearing -> clear-speed shim, run end -> state reset) and broadcasts.
  - Context: the host remembers what run *it* started (RunContext) to fill the
    contract fields the board doesn't report (playlist files, run mode).

No sequencing happens on the host. If the board reboots or the backend
restarts mid-run, the observer resynchronizes from /sand_status alone.
"""

import asyncio
import logging
import os
import time
from dataclasses import dataclass, field
from typing import Callable, Literal, Optional

from modules.connection import connection_manager
from modules.core.state import state

logger = logging.getLogger(__name__)

# Firmware clear modes; legacy host values are mapped onto them.
CLEAR_MODES = ("none", "adaptive", "in", "out", "sideway", "random")
_LEGACY_CLEAR = {
    "clear_from_in": "in",
    "clear_from_out": "out",
    "clear_sideway": "sideway",
    "clear_in": "in",
    "clear_out": "out",
}


class ExecutionError(Exception):
    """Raised when the board rejects or times out on an execution command."""


@dataclass
class RunContext:
    """Host-side knowledge about the run we started (not board truth)."""
    kind: Literal["pattern", "playlist"]
    playlist_name: Optional[str] = None
    files: Optional[list] = None  # host paths, unshuffled mirror order
    run_mode: str = "single"      # frontend value: 'single' | 'indefinite'
    shuffle: bool = False
    clear_pattern: str = "none"
    started_at: float = field(default_factory=time.time)


current_run: Optional[RunContext] = None


def map_clear_mode(value) -> str:
    """Map a frontend/legacy clear value onto the firmware's clear modes."""
    if not value:
        return "none"
    value = str(value).lower()
    if value in CLEAR_MODES:
        return value
    if value in _LEGACY_CLEAR:
        return _LEGACY_CLEAR[value]
    logger.warning(f"Unknown clear mode '{value}', using none")
    return "none"


def _state(st: Optional[dict]) -> str:
    """Machine state without the GRBL substate suffix ('Hold:0' -> 'Hold')."""
    return ((st or {}).get("state") or "").split(":", 1)[0]


def _from_sd_path(sd_path: str) -> Optional[str]:
    """Map a board SD path ('/patterns/x.thr', '/sd/patterns/x.thr') to the
    host-relative path ('./patterns/x.thr') the frontend/history expect."""
    if not sd_path:
        return None
    p = sd_path.replace("\\", "/")
    if p.startswith("/sd/"):
        p = p[3:]
    if not p.startswith("/"):
        p = "/" + p
    return "." + p


# ---------------------------------------------------------------------------
# Commands
# ---------------------------------------------------------------------------

def _require_conn():
    if not state.conn or not state.conn.is_connected():
        raise ExecutionError("Connection not established")
    return state.conn


async def _wait_for_idle(timeout: float = 15.0) -> bool:
    """Poll the board until it reports Idle (used before idle-gated NVS writes)."""
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            st = await asyncio.to_thread(state.conn.get_status)
            if _state(st) == "Idle" and not st.get("running"):
                return True
        except Exception as e:
            logger.debug(f"Idle wait poll failed: {e}")
        await asyncio.sleep(0.5)
    return False


async def run_pattern(file_path: str, clear_pattern: str = "none") -> None:
    """Run one pattern via $Sand/Run (firmware sequences clear -> pattern and
    aborts any current job first)."""
    global current_run
    conn = _require_conn()
    from modules.core.pattern_manager import _ensure_on_board, _to_sd_path

    sd_path = _to_sd_path(file_path)
    await asyncio.to_thread(_ensure_on_board, file_path, sd_path)
    try:
        await asyncio.to_thread(conn.set_feed, int(state.speed))
    except Exception as e:
        logger.warning(f"Could not set feed before run: {e}")
    mode = map_clear_mode(clear_pattern)
    await asyncio.to_thread(conn.run_pattern, sd_path, mode)

    current_run = RunContext(kind="pattern", clear_pattern=mode)
    state.current_playlist = None
    state.current_playlist_name = None
    # Optimistic; the observer confirms/corrects from /sand_status.
    state.current_playing_file = file_path
    observer.on_run_started()


async def start_playlist(playlist_name: str, run_mode: str = "single",
                         pause_time: float = 0, clear_pattern: str = "none",
                         shuffle: bool = False) -> None:
    """Run a playlist on the board via $Playlist/Run.

    The run options are the firmware's NVS $Playlist/* globals; NVS writes are
    idle-gated, so a run-while-running stops the board first.
    """
    global current_run
    conn = _require_conn()
    from modules.core import playlist_manager, board_settings
    from modules.core.pattern_manager import THETA_RHO_DIR, _ensure_on_board, _to_sd_path

    playlist = playlist_manager.get_playlist(playlist_name)
    if not playlist:
        raise ExecutionError(f"Playlist '{playlist_name}' not found")
    files = playlist["files"]
    if not files:
        raise ExecutionError(f"Playlist '{playlist_name}' is empty")
    host_paths = [os.path.join(THETA_RHO_DIR, f) for f in files]

    # Idle-gate: NVS settings writes are rejected mid-motion.
    st = None
    try:
        st = await asyncio.to_thread(conn.get_status)
    except Exception:
        pass
    if st and (st.get("running") or _state(st) not in ("Idle", "Alarm")):
        await asyncio.to_thread(conn.stop)
        if not await _wait_for_idle(15.0):
            raise ExecutionError("Table is busy and did not stop in time")

    mode = map_clear_mode(clear_pattern)
    for key, value in (
        ("Playlist/Mode", "loop" if run_mode == "indefinite" else "single"),
        ("Playlist/Shuffle", "ON" if shuffle else "OFF"),
        ("Playlist/PauseTime", max(0, int(pause_time or 0))),
        ("Playlist/ClearPattern", mode),
    ):
        await asyncio.to_thread(conn.set_setting, key, value)

    # The board needs the playlist file before Run; patterns can trail behind
    # (each one plays for minutes) except the first, which is ensured now.
    await asyncio.to_thread(board_settings.mirror_playlist, playlist_name, files, conn)
    await asyncio.to_thread(_ensure_on_board, host_paths[0], _to_sd_path(host_paths[0]))

    def _mirror_rest():
        for p in host_paths[1:]:
            _ensure_on_board(p, _to_sd_path(p))
    import threading
    threading.Thread(target=_mirror_rest, daemon=True).start()

    try:
        await asyncio.to_thread(conn.set_feed, int(state.speed))
    except Exception as e:
        logger.warning(f"Could not set feed before playlist: {e}")
    await asyncio.to_thread(conn.run_command, f"$Playlist/Run={playlist_name}")

    current_run = RunContext(
        kind="playlist", playlist_name=playlist_name, files=host_paths,
        run_mode=run_mode, shuffle=shuffle, clear_pattern=mode,
    )
    state.current_playlist = host_paths
    state.current_playlist_name = playlist_name
    state.playlist_mode = run_mode
    observer.on_run_started()
    logger.info(f"Started playlist '{playlist_name}' on the board "
                f"(mode={run_mode}, shuffle={shuffle}, pause={pause_time}s, clear={mode})")


async def stop(force: bool = False) -> bool:
    """Clean stop. Returns True once the board is Idle (always True for force)."""
    try:
        conn = _require_conn()
        await asyncio.to_thread(conn.stop)
    except Exception as e:
        if not force:
            raise ExecutionError(f"Stop failed: {e}")
        logger.warning(f"Force stop: board stop failed ({e}), resetting host state anyway")
    if force:
        _reset_run_state()
        observer.reset()
        return True
    return await _wait_for_idle(10.0)


async def pause() -> None:
    conn = _require_conn()
    await asyncio.to_thread(conn.pause)


async def resume() -> None:
    conn = _require_conn()
    await asyncio.to_thread(conn.resume)


async def skip() -> bool:
    """Skip the current pattern: $Playlist/Skip for playlists (also overrides
    quiet hours for one pattern); stopping is the 'skip' of a single pattern."""
    conn = _require_conn()
    st = observer.last_raw or {}
    pl = st.get("playlist") or {}
    if pl.get("active"):
        await asyncio.to_thread(conn.skip)
        return True
    if st.get("running"):
        await asyncio.to_thread(conn.stop)
        return True
    return False


async def set_speed(speed: float) -> None:
    conn = _require_conn()
    state.speed = speed
    await asyncio.to_thread(conn.set_feed, int(speed))


def _reset_run_state() -> None:
    global current_run
    current_run = None
    state.current_playing_file = None
    state.current_playlist = None
    state.current_playlist_name = None
    state.pause_requested = False
    state.pause_time_remaining = 0
    state.execution_progress = None


# ---------------------------------------------------------------------------
# Status translation: /sand_status -> the frontend's /ws/status contract
# ---------------------------------------------------------------------------

def translate_status(st: Optional[dict], obs: Optional["BoardObserver"] = None,
                     now: Optional[float] = None) -> dict:
    """Translate a raw board status into the /ws/status data object.

    Pure given (st, observer timing, state); unit-testable with fixtures.
    """
    obs = obs or observer
    now = now if now is not None else time.time()
    connected = bool(state.conn and state.conn.is_connected())

    if st is None:
        return {
            "connection_status": connected,
            "current_file": None,
            "is_running": False,
            "is_paused": False,
            "is_homing": state.is_homing,
            "sensor_homing_failed": state.sensor_homing_failed,
            "is_clearing": False,
            "speed": state.speed,
            "pause_time_remaining": 0,
            "original_pause_time": None,
            "progress": None,
            "playlist": None,
            "current_theta": state.current_theta,
            "current_rho": state.current_rho,
            "firmware_version": state.firmware_version,
            "table_type": None,
        }

    pl = st.get("playlist") or {}
    running = bool(st.get("running"))
    # GRBL states can carry a substate suffix (e.g. "Hold:0", "Door:1").
    machine_state = _state(st)
    clearing = bool(pl.get("clearing"))
    pause_remaining = pl.get("pause_remaining", -1)
    pause_total = pl.get("pause_total", -1)

    current_file = _from_sd_path(st.get("file")) if running else None

    # --- progress ---
    progress_obj = None
    if running:
        fraction = st.get("progress", -1)
        fraction = fraction if isinstance(fraction, (int, float)) and fraction >= 0 else 0
        elapsed = max(0.0, now - obs.file_started_at - obs.hold_accumulated
                      - (now - obs.hold_started_at if obs.hold_started_at else 0))
        if fraction > 0.02:
            remaining = max(0.0, elapsed / fraction - elapsed)
        elif obs.cached_history and obs.cached_history.get("actual_time_seconds"):
            remaining = max(0.0, obs.cached_history["actual_time_seconds"] - elapsed)
        else:
            remaining = None
        progress_obj = {
            "current": int(fraction * 1000),
            "total": 1000,
            "percentage": round(fraction * 100, 1),
            "elapsed_time": elapsed,
            "remaining_time": remaining,
        }
        if obs.cached_history:
            progress_obj["last_completed_time"] = obs.cached_history

    # --- playlist ---
    playlist_obj = None
    playlist_active = bool(pl.get("active"))
    ctx = current_run
    if (playlist_active and pl.get("total", 0) > 0) or (ctx and ctx.kind == "playlist"):
        files = [f for f in (state.current_playlist or [])]
        index = int(pl.get("index", 0) or 0)
        total = int(pl.get("total", 0) or 0) or len(files)
        shuffled = bool(ctx.shuffle) if ctx else False
        next_file = None
        if files and not shuffled:
            # While clearing, the "next" thing is the pattern the clear precedes.
            next_idx = index if clearing else index + 1
            if 0 <= next_idx < len(files):
                next_file = files[next_idx]
        playlist_obj = {
            "name": pl.get("name") or (ctx.playlist_name if ctx else None),
            "current_index": index,
            "total_files": total,
            "mode": (ctx.run_mode if ctx else "indefinite"),
            "files": files,
            "next_file": next_file,
            "shuffled": shuffled,
        }

    return {
        "connection_status": connected,
        "current_file": current_file,
        "is_running": running,
        "is_paused": machine_state == "Hold",
        "is_homing": machine_state == "Home" or state.is_homing,
        "sensor_homing_failed": state.sensor_homing_failed,
        "is_clearing": clearing,
        "speed": state.speed,
        "pause_time_remaining": pause_remaining if pause_remaining >= 0 else 0,
        "original_pause_time": pause_total if pause_total >= 0 else None,
        "progress": progress_obj,
        "playlist": playlist_obj,
        "current_theta": st.get("theta", state.current_theta),
        "current_rho": st.get("rho", state.current_rho),
        "firmware_version": st.get("fw") or state.firmware_version,
        "table_type": None,
    }


# ---------------------------------------------------------------------------
# Observer: single poll loop — edges, history, shims, broadcast
# ---------------------------------------------------------------------------

class BoardObserver:
    """Polls /sand_status and turns transitions into host behavior."""

    POLL_ACTIVE = 1.0
    POLL_IDLE = 2.0
    OFFLINE_GRACE = 3  # consecutive failures before a run is declared over

    def __init__(self):
        self.prev: Optional[dict] = None
        self.last_raw: Optional[dict] = None
        self.last_translated: dict = {}
        self.file_started_at: float = 0.0
        self.hold_started_at: Optional[float] = None
        self.hold_accumulated: float = 0.0
        self.last_progress: float = -1.0
        self.cached_history: Optional[dict] = None
        self._clear_speed_saved: Optional[float] = None
        self._was_quiet_wled_off = False
        self._fail_count = 0
        self._tick = 0
        self._task: Optional[asyncio.Task] = None
        # Set by main.py to fan out to /ws/status clients (avoids circular import).
        self.on_status: Optional[Callable] = None

    def reset(self) -> None:
        self.prev = None
        self.file_started_at = 0.0
        self.hold_started_at = None
        self.hold_accumulated = 0.0
        self.last_progress = -1.0
        self.cached_history = None
        self._clear_speed_saved = None

    def on_run_started(self) -> None:
        """Called by the command layer so the first poll attributes time correctly."""
        self.file_started_at = time.time()
        self.hold_accumulated = 0.0
        self.hold_started_at = None
        self.last_progress = -1.0

    # -- lifecycle ---------------------------------------------------------

    def start(self) -> None:
        self._task = asyncio.create_task(self._run())

    async def astop(self) -> None:
        if self._task:
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass

    async def _run(self) -> None:
        while True:
            interval = self.POLL_IDLE
            try:
                interval = await self._tick_once()
            except asyncio.CancelledError:
                raise
            except Exception as e:
                logger.warning(f"Status observer tick failed: {e}")
            await asyncio.sleep(interval)

    async def _tick_once(self) -> float:
        st = None
        if state.conn and state.conn.is_connected():
            try:
                st = await asyncio.to_thread(connection_manager.poll_status_once)
            except Exception as e:
                logger.debug(f"Status poll failed: {e}")
        await self.process(st)
        active = bool(st and (st.get("running") or (st.get("playlist") or {}).get("active")
                              or _state(st) in ("Hold", "Home", "Jog")))
        return self.POLL_ACTIVE if active else self.POLL_IDLE

    # -- core (sync-friendly for tests; only I/O bits are awaited) ----------

    async def process(self, st: Optional[dict], now: Optional[float] = None) -> dict:
        """One observation step: edge detection + translation + broadcast."""
        now = now if now is not None else time.time()

        if st is None:
            self._fail_count += 1
            if self._fail_count == self.OFFLINE_GRACE and self.prev is not None:
                # Board gone: close out the run without claiming completion.
                logger.warning("Board unreachable — closing out the observed run")
                self._close_file(self.prev, now, aborted=True)
                _reset_run_state()
                self.reset()
        else:
            self._fail_count = 0
            if self.prev is not None and st.get("uptime", 0) < self.prev.get("uptime", 0):
                logger.warning("Board rebooted mid-observation — resetting run context")
                _reset_run_state()
                self.reset()
            self._detect_edges(st, now)
            self.prev = st
            self.last_raw = st

        await self._quiet_hours_wled(now)

        self._tick += 1
        if self._tick % 30 == 0 and state.conn and state.conn.is_connected():
            try:
                from modules.core import board_settings
                settings_map = await asyncio.to_thread(state.conn.get_settings)
                board_settings.adopt_still_sands(settings_map)
            except Exception as e:
                logger.debug(f"Board settings adopt failed: {e}")

        self.last_translated = translate_status(st, self, now)
        # Mirror the progress 4-tuple for the MQTT handler, which unpacks it.
        prog = self.last_translated.get("progress")
        state.execution_progress = (
            (prog["current"], prog["total"], prog["remaining_time"], prog["elapsed_time"])
            if prog else None
        )
        if self.on_status:
            try:
                await self.on_status(self.last_translated)
            except Exception as e:
                logger.debug(f"Status broadcast failed: {e}")
        return self.last_translated

    def _detect_edges(self, st: dict, now: float) -> None:
        prev = self.prev or {}
        prev_pl = prev.get("playlist") or {}
        pl = st.get("playlist") or {}
        prev_file = prev.get("file") or ""
        cur_file = st.get("file") or ""
        prev_running = bool(prev.get("running"))
        running = bool(st.get("running"))

        # File end: the file changed while running, or playback stopped.
        if prev_running and prev_file and (cur_file != prev_file or not running):
            self._close_file(prev, now, aborted=False)

        # File start.
        if running and cur_file and (not prev_running or cur_file != prev_file):
            self.file_started_at = now
            self.hold_accumulated = 0.0
            self.hold_started_at = None
            self.last_progress = -1.0
            host_path = _from_sd_path(cur_file)
            state.current_playing_file = host_path
            self._cache_history(host_path)
            self._on_playing_leds()

        if running and isinstance(st.get("progress"), (int, float)) and st["progress"] >= 0:
            self.last_progress = st["progress"]

        # Hold (pause) edges — for pause accounting and the MQTT mirror.
        prev_hold = _state(prev) == "Hold"
        hold = _state(st) == "Hold"
        if hold and not prev_hold:
            self.hold_started_at = now
            state.pause_requested = True
        elif prev_hold and not hold:
            if self.hold_started_at:
                self.hold_accumulated += now - self.hold_started_at
            self.hold_started_at = None
            state.pause_requested = False

        # Clearing edges — clear-speed shim.
        prev_clearing = bool(prev_pl.get("clearing"))
        clearing = bool(pl.get("clearing"))
        if clearing and not prev_clearing and state.clear_pattern_speed:
            self._clear_speed_saved = state.speed
            self._set_feed_safe(state.clear_pattern_speed)
        elif prev_clearing and not clearing and self._clear_speed_saved:
            self._set_feed_safe(self._clear_speed_saved)
            self._clear_speed_saved = None

        # Run end: nothing running, no active playlist, machine idle.
        was_active = prev_running or bool(prev_pl.get("active"))
        is_active = running or bool(pl.get("active"))
        if was_active and not is_active and _state(st) in ("Idle", "Alarm"):
            logger.info("Observed run end on the board")
            _reset_run_state()
            state.save()
            self._on_idle_leds()

    def _close_file(self, prev_st: dict, now: float, aborted: bool) -> None:
        """Log history for the file that just finished/stopped."""
        prev_pl = prev_st.get("playlist") or {}
        prev_file = prev_st.get("file") or ""
        if not prev_file or prev_pl.get("clearing"):
            return  # clears are not history-worthy (matches old semantics)
        from modules.core.pattern_manager import log_execution_time
        hold_extra = (now - self.hold_started_at) if self.hold_started_at else 0.0
        actual = max(0.0, now - self.file_started_at - self.hold_accumulated - hold_extra)
        completed = (not aborted) and self.last_progress >= 0.98
        try:
            log_execution_time(
                pattern_name=os.path.basename(prev_file),
                table_type="fluidnc",
                speed=int(state.speed or 0),
                actual_time=actual,
                total_coordinates=0,
                was_completed=completed,
            )
        except Exception as e:
            logger.warning(f"Could not log execution history: {e}")

    def _cache_history(self, host_path: Optional[str]) -> None:
        self.cached_history = None
        if not host_path:
            return
        try:
            from modules.core.pattern_manager import get_last_completed_execution_time
            self.cached_history = get_last_completed_execution_time(
                os.path.basename(host_path), state.speed)
        except Exception as e:
            logger.debug(f"History lookup failed: {e}")

    def _set_feed_safe(self, mm: float) -> None:
        try:
            if state.conn:
                state.conn.set_feed(int(mm))
        except Exception as e:
            logger.warning(f"Clear-speed feed change failed: {e}")

    def _on_playing_leds(self) -> None:
        if state.led_controller and state.led_automation_enabled:
            try:
                asyncio.get_running_loop().create_task(
                    state.led_controller.effect_playing_async(None))
            except RuntimeError:
                pass

    def _on_idle_leds(self) -> None:
        if state.led_controller and state.led_automation_enabled:
            try:
                asyncio.get_running_loop().create_task(
                    state.led_controller.effect_idle_async(None))
            except RuntimeError:
                pass

    async def _quiet_hours_wled(self, now: float) -> None:
        """The one surviving host Still Sands behavior: switch WLED off during
        quiet hours (the board handles its own ring via $Sands/LedOff)."""
        if state.led_provider != "wled" or not state.led_controller:
            return
        from modules.core.pattern_manager import is_in_scheduled_pause_period
        in_quiet = bool(state.scheduled_pause_control_wled and is_in_scheduled_pause_period())
        if in_quiet and not self._was_quiet_wled_off:
            self._was_quiet_wled_off = True
            await state.led_controller.set_power_async(0)
            logger.info("Still Sands: WLED off")
        elif not in_quiet and self._was_quiet_wled_off:
            self._was_quiet_wled_off = False
            await state.led_controller.set_power_async(1)
            await state.led_controller.effect_idle_async(None)
            logger.info("Still Sands: WLED restored")


observer = BoardObserver()


def get_cached_status() -> dict:
    """Last translated status (what /ws/status clients get on connect)."""
    if observer.last_translated:
        return observer.last_translated
    return translate_status(observer.last_raw, observer)