diff --git a/core/management/__init__.py b/core/management/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/core/management/commands/__init__.py b/core/management/commands/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/core/management/commands/moonraker_worker.py b/core/management/commands/moonraker_worker.py new file mode 100644 index 0000000..60daad3 --- /dev/null +++ b/core/management/commands/moonraker_worker.py @@ -0,0 +1,227 @@ +"""Long-running Moonraker WebSocket worker. + +Run with ``python manage.py moonraker_worker``. The worker: + +1. Enables SQLite WAL mode and a write busy-timeout so it can coexist + with Gunicorn. +2. Starts an asyncio ``TaskGroup`` that holds one task per configured + printer and a reconciler task that refreshes the printer list + periodically (live reload — no container restart on printer + config changes). +3. For each printer, opens a persistent WebSocket to Moonraker, + subscribes to print status objects, and feeds incoming + notifications into :func:`core.services.printer_status_sync.apply_status_event`. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import logging +import os +import signal +from dataclasses import dataclass +from typing import Any + +from asgiref.sync import sync_to_async +from django.core.management.base import BaseCommand +from django.db import connection + +from core.models import PrinterProfile, PrintQueue +from core.services.moonraker_ws import MoonrakerWebSocketClient +from core.services.printer_status_sync import apply_status_event + +logger = logging.getLogger(__name__) + +DEFAULT_RELOAD_INTERVAL = 10.0 + + +@dataclass(frozen=True) +class PrinterSnapshot: + """Minimal view of a PrinterProfile for diffing.""" + + pk: int + moonraker_url: str + moonraker_api_key: str + updated_at: Any # datetime + + +@sync_to_async +def _enable_sqlite_wal() -> None: + if connection.vendor != "sqlite": + return + with connection.cursor() as cursor: + cursor.execute("PRAGMA journal_mode=WAL;") + cursor.execute("PRAGMA busy_timeout=5000;") + logger.info("SQLite WAL mode + busy_timeout enabled") + + +@sync_to_async +def _load_active_printer_snapshots() -> dict[int, PrinterSnapshot]: + rows = PrinterProfile.objects.exclude(moonraker_url="").values( + "pk", "moonraker_url", "moonraker_api_key", "updated_at" + ) + return { + row["pk"]: PrinterSnapshot( + pk=row["pk"], + moonraker_url=row["moonraker_url"], + moonraker_api_key=row["moonraker_api_key"], + updated_at=row["updated_at"], + ) + for row in rows + } + + +@sync_to_async +def _load_printer(pk: int) -> PrinterProfile | None: + return PrinterProfile.objects.filter(pk=pk).first() + + +@sync_to_async +def _load_active_queue_entry(printer_pk: int) -> PrintQueue | None: + return ( + PrintQueue.objects.filter( + printer_id=printer_pk, + status=PrintQueue.STATUS_PRINTING, + ) + .select_related("plate__print_job") + .order_by("started_at") + .first() + ) + + +@sync_to_async +def _apply_event_sync(entry: PrintQueue, event: dict[str, Any]) -> None: + apply_status_event(entry, event) + + +async def _handle_event_for_printer(printer: PrinterProfile, event: dict[str, Any]) -> None: + entry = await _load_active_queue_entry(printer.pk) + if entry is None: + return + await _apply_event_sync(entry, event) + + +async def _run_printer(printer: PrinterProfile) -> None: + async def on_event(event: dict[str, Any]) -> None: + await _handle_event_for_printer(printer, event) + + client = MoonrakerWebSocketClient(printer, on_event) + try: + await client.run() + except asyncio.CancelledError: + logger.info("Worker task for printer %s cancelled", printer.name) + raise + + +class TaskGroupLike: + """Minimal task-group interface we use, for testability.""" + + def create_task(self, coro: Any, *, name: str | None = None) -> asyncio.Task: # pragma: no cover + raise NotImplementedError + + +async def _reconcile_printers( + active: dict[int, tuple[asyncio.Task, PrinterSnapshot]], + task_group: Any, +) -> dict[int, tuple[asyncio.Task, PrinterSnapshot]]: + """Reconcile active WebSocket tasks against the current DB state. + + Spawns new tasks for newly added printers, cancels tasks for + removed printers, and restarts tasks whose configuration changed. + + Returns the updated ``active`` map. The function mutates the + given map in-place **and** returns it for convenience. + """ + snapshots = await _load_active_printer_snapshots() + + # Cancel removed / changed printers. + for pk in list(active.keys()): + task, prev = active[pk] + snap = snapshots.get(pk) + if snap is None: + logger.info("Printer %s removed, cancelling task", pk) + task.cancel() + del active[pk] + continue + if ( + snap.moonraker_url != prev.moonraker_url + or snap.moonraker_api_key != prev.moonraker_api_key + or snap.updated_at != prev.updated_at + ): + logger.info("Printer %s changed, restarting task", pk) + task.cancel() + del active[pk] + + # Spawn missing printers. + for pk, snap in snapshots.items(): + if pk in active: + continue + printer = await _load_printer(pk) + if printer is None: + continue + logger.info("Starting worker task for printer %s (%s)", printer.name, snap.moonraker_url) + task = task_group.create_task(_run_printer(printer), name=f"moonraker_ws_{pk}") + active[pk] = (task, snap) + + return active + + +async def _reconcile_loop( + task_group: Any, + active: dict[int, tuple[asyncio.Task, PrinterSnapshot]], + interval: float, + stop_event: asyncio.Event, +) -> None: + while not stop_event.is_set(): + try: + await _reconcile_printers(active, task_group) + except Exception: # noqa: BLE001 + logger.exception("Reconcile loop iteration failed") + with contextlib.suppress(asyncio.TimeoutError): + await asyncio.wait_for(stop_event.wait(), timeout=interval) + + +async def _main(reload_interval: float) -> None: + await _enable_sqlite_wal() + + stop_event = asyncio.Event() + loop = asyncio.get_running_loop() + for sig in (signal.SIGTERM, signal.SIGINT): + loop.add_signal_handler(sig, stop_event.set) + + # Shared between _main and the reconciler so we can cancel all + # per-printer tasks on shutdown. Without this, exiting the + # TaskGroup context would hang waiting for long-running printer + # tasks that never observe stop_event. + active: dict[int, tuple[asyncio.Task, PrinterSnapshot]] = {} + + async with asyncio.TaskGroup() as tg: + reconciler = tg.create_task( + _reconcile_loop(tg, active, reload_interval, stop_event), + name="moonraker_reconciler", + ) + await stop_event.wait() + logger.info("Shutdown signal received, stopping worker") + reconciler.cancel() + for pk, (task, _snap) in list(active.items()): + logger.debug("Cancelling worker task for printer %s", pk) + task.cancel() + + +class Command(BaseCommand): + help = "Run the long-running Moonraker WebSocket status worker." + + def add_arguments(self, parser: Any) -> None: + parser.add_argument( + "--reload-interval", + type=float, + default=float(os.environ.get("WORKER_RELOAD_INTERVAL", DEFAULT_RELOAD_INTERVAL)), + help="Seconds between printer-list reconciliations (default: 10).", + ) + + def handle(self, *args: Any, **options: Any) -> None: + reload_interval = options["reload_interval"] + logger.info("Starting Moonraker worker (reload_interval=%.1fs)", reload_interval) + with contextlib.suppress(KeyboardInterrupt): + asyncio.run(_main(reload_interval)) diff --git a/core/migrations/0016_printqueue_progress_fields.py b/core/migrations/0016_printqueue_progress_fields.py new file mode 100644 index 0000000..9f16f38 --- /dev/null +++ b/core/migrations/0016_printqueue_progress_fields.py @@ -0,0 +1,28 @@ +# Generated by Django 6.0.3 on 2026-04-08 20:17 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('core', '0015_rename_remote_job_id'), + ] + + operations = [ + migrations.AddField( + model_name='printqueue', + name='last_error', + field=models.TextField(blank=True, default='', help_text='Last error reported by the printer backend (if any)'), + ), + migrations.AddField( + model_name='printqueue', + name='progress', + field=models.FloatField(blank=True, help_text='Live print progress 0..1 from the printer backend', null=True), + ), + migrations.AddField( + model_name='printqueue', + name='status_updated_at', + field=models.DateTimeField(blank=True, help_text='Last time the status was updated from the printer backend', null=True), + ), + ] diff --git a/core/models/queue.py b/core/models/queue.py index 02c6621..445cd5b 100644 --- a/core/models/queue.py +++ b/core/models/queue.py @@ -71,6 +71,23 @@ class PrintQueue(models.Model): started_at = models.DateTimeField(null=True, blank=True, help_text="When printing started") completed_at = models.DateTimeField(null=True, blank=True, help_text="When the printer finished (before review)") + # Live status from the printer backend (populated by moonraker_worker) --- + progress = models.FloatField( + null=True, + blank=True, + help_text="Live print progress 0..1 from the printer backend", + ) + status_updated_at = models.DateTimeField( + null=True, + blank=True, + help_text="Last time the status was updated from the printer backend", + ) + last_error = models.TextField( + blank=True, + default="", + help_text="Last error reported by the printer backend (if any)", + ) + class Meta: ordering = ["-priority", "position", "added_at"] indexes = [ diff --git a/core/services/moonraker.py b/core/services/moonraker.py index ee621bc..0aba483 100644 --- a/core/services/moonraker.py +++ b/core/services/moonraker.py @@ -13,6 +13,24 @@ DEFAULT_TIMEOUT = 15 # seconds +_KLIPPER_STATE_MAP = { + "printing": NormalizedJobStatus.STATE_PRINTING, + "complete": NormalizedJobStatus.STATE_COMPLETE, + "error": NormalizedJobStatus.STATE_ERROR, + "cancelled": NormalizedJobStatus.STATE_CANCELLED, + "standby": NormalizedJobStatus.STATE_STANDBY, + "paused": NormalizedJobStatus.STATE_PRINTING, +} + + +def map_klipper_state(raw_state: str) -> str: + """Map a raw Klipper/Moonraker ``print_stats.state`` to a normalized state. + + Unknown states fall back to :attr:`NormalizedJobStatus.STATE_IDLE`. + """ + return _KLIPPER_STATE_MAP.get(raw_state, NormalizedJobStatus.STATE_IDLE) + + class MoonrakerError(PrinterError): """Raised when a Moonraker API operation fails.""" @@ -161,19 +179,8 @@ def get_job_status(self) -> NormalizedJobStatus: print_stats = status_data.get("print_stats", {}) virtual_sd = status_data.get("virtual_sdcard", {}) - raw_state = print_stats.get("state", "unknown") - state_map = { - "printing": NormalizedJobStatus.STATE_PRINTING, - "complete": NormalizedJobStatus.STATE_COMPLETE, - "error": NormalizedJobStatus.STATE_ERROR, - "cancelled": NormalizedJobStatus.STATE_CANCELLED, - "standby": NormalizedJobStatus.STATE_STANDBY, - "paused": NormalizedJobStatus.STATE_PRINTING, - } - state = state_map.get(raw_state, NormalizedJobStatus.STATE_IDLE) - return NormalizedJobStatus( - state=state, + state=map_klipper_state(print_stats.get("state", "unknown")), progress=virtual_sd.get("progress", 0.0), filename=print_stats.get("filename", ""), ) diff --git a/core/services/moonraker_ws.py b/core/services/moonraker_ws.py new file mode 100644 index 0000000..3612645 --- /dev/null +++ b/core/services/moonraker_ws.py @@ -0,0 +1,176 @@ +"""Moonraker JSON-RPC WebSocket client for live printer status. + +The :class:`MoonrakerWebSocketClient` holds one persistent WebSocket +connection to a Moonraker instance, subscribes to the objects we care +about (``print_stats``, ``virtual_sdcard``, …) and invokes a caller +supplied async callback for every JSON-RPC notification it receives. + +The client is responsible for: + +* Building the ws(s):// URL from a Moonraker REST URL. +* Optionally obtaining a one-shot token when an API key is configured. +* Reconnecting with exponential backoff on network failures or + unexpected disconnects. + +The client is **not** responsible for interpreting events — that is +the job of :mod:`core.services.printer_status_sync`. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +from collections.abc import Awaitable, Callable +from typing import Any +from urllib.parse import urlparse, urlunparse + +import requests +import websockets +from websockets.exceptions import ConnectionClosed, WebSocketException + +from core.models import PrinterProfile +from core.services.moonraker import DEFAULT_TIMEOUT, MoonrakerError + +logger = logging.getLogger(__name__) + +#: Objects we subscribe to for live status updates. +SUBSCRIBE_OBJECTS: dict[str, None] = { + "print_stats": None, + "virtual_sdcard": None, + "display_status": None, + "heater_bed": None, + "extruder": None, +} + +#: Reconnect backoff bounds in seconds. +RECONNECT_INITIAL_DELAY = 1.0 +RECONNECT_MAX_DELAY = 30.0 + + +EventHandler = Callable[[dict[str, Any]], Awaitable[None]] + + +def build_ws_url(moonraker_url: str) -> str: + """Translate a Moonraker REST URL into its ``/websocket`` WS URL. + + Examples: + >>> build_ws_url("http://printer.local:7125") + 'ws://printer.local:7125/websocket' + >>> build_ws_url("https://printer.local/moonraker/") + 'wss://printer.local/moonraker/websocket' + """ + parsed = urlparse(moonraker_url) + scheme = {"http": "ws", "https": "wss"}.get(parsed.scheme, "ws") + path = (parsed.path or "").rstrip("/") + "/websocket" + return urlunparse((scheme, parsed.netloc, path, "", "", "")) + + +def fetch_oneshot_token(printer: PrinterProfile) -> str | None: + """Request a one-shot WebSocket token from Moonraker. + + Returns ``None`` if the printer has no API key configured, or if + the ``/access/oneshot_token`` endpoint is unavailable (e.g. no + ``[authorization]`` section in moonraker.conf → Moonraker happily + accepts unauthenticated websocket connections). + """ + if not printer.moonraker_api_key: + return None + + url = printer.moonraker_url.rstrip("/") + "/access/oneshot_token" + try: + response = requests.get( + url, + headers={"X-Api-Key": printer.moonraker_api_key}, + timeout=DEFAULT_TIMEOUT, + ) + response.raise_for_status() + data = response.json() + except requests.RequestException as exc: + logger.warning( + "Oneshot token request failed for %s: %s — connecting without token", + printer.name, + exc, + ) + return None + + token = data.get("result") + if not isinstance(token, str): + logger.warning("Unexpected oneshot token payload for %s: %r", printer.name, data) + return None + return token + + +class MoonrakerWebSocketClient: + """One persistent WebSocket connection to a Moonraker instance.""" + + def __init__(self, printer: PrinterProfile, on_event: EventHandler): + self.printer = printer + self.on_event = on_event + self._ws_url = build_ws_url(printer.moonraker_url) + self._reconnect_delay = RECONNECT_INITIAL_DELAY + + async def run(self) -> None: + """Connect, subscribe, and pump events until cancelled. + + Reconnects on unexpected disconnects with exponential backoff. + The backoff is reset to :data:`RECONNECT_INITIAL_DELAY` after + every successful subscription so that a transient outage + doesn't permanently increase reconnect sleeps. Cancellation + (``asyncio.CancelledError``) propagates out cleanly. + """ + while True: + try: + await self._connect_and_listen() + # Clean close from the server -> retry with backoff. + logger.info("%s: websocket closed cleanly, reconnecting", self.printer.name) + except asyncio.CancelledError: + raise + except (ConnectionClosed, WebSocketException, OSError, MoonrakerError) as exc: + logger.warning("%s: websocket error: %s", self.printer.name, exc) + except Exception: # noqa: BLE001 + logger.exception("%s: unexpected worker error", self.printer.name) + + await asyncio.sleep(self._reconnect_delay) + self._reconnect_delay = min(self._reconnect_delay * 2, RECONNECT_MAX_DELAY) + + async def _connect_and_listen(self) -> None: + # Run the blocking token fetch in the default executor so we + # don't stall the event loop. + loop = asyncio.get_running_loop() + token = await loop.run_in_executor(None, fetch_oneshot_token, self.printer) + url = self._ws_url + (f"?token={token}" if token else "") + + logger.info("%s: connecting to %s", self.printer.name, self._ws_url) + async with websockets.connect(url, ping_interval=20, ping_timeout=20) as ws: + await self._subscribe(ws) + logger.info("%s: subscribed, listening for events", self.printer.name) + # A successful subscribe means the connection is healthy; + # reset the backoff so the next outage starts fresh. + self._reconnect_delay = RECONNECT_INITIAL_DELAY + async for raw in ws: + try: + event = json.loads(raw) + except json.JSONDecodeError: + logger.warning("%s: non-JSON frame received", self.printer.name) + continue + if "method" not in event: + # JSON-RPC response to our subscribe call etc. — ignore. + continue + try: + await self.on_event(event) + except Exception: # noqa: BLE001 + logger.exception( + "%s: event handler failed for %s", + self.printer.name, + event.get("method"), + ) + + async def _subscribe(self, ws: Any) -> None: + request = { + "jsonrpc": "2.0", + "method": "printer.objects.subscribe", + "params": {"objects": SUBSCRIBE_OBJECTS}, + "id": 1, + } + await ws.send(json.dumps(request)) diff --git a/core/services/printer_status_sync.py b/core/services/printer_status_sync.py new file mode 100644 index 0000000..270281d --- /dev/null +++ b/core/services/printer_status_sync.py @@ -0,0 +1,155 @@ +"""Synchronous event -> PrintQueue update logic for the Moonraker worker. + +The WebSocket worker (``core.services.moonraker_ws``) receives raw +Moonraker JSON-RPC notifications and hands them to +:func:`apply_status_event` together with the matching +:class:`~core.models.PrintQueue` entry. All business logic around +"is this a terminal state?", "should we write to the DB now?" and +"which fields to update" lives here, keeping the async I/O layer +dumb and this module fully unit-testable. + +The module is intentionally sync and free of Django async helpers. +The caller (worker command) is responsible for using +``asgiref.sync.sync_to_async`` when invoking these functions from +an event loop. +""" + +from __future__ import annotations + +import logging +from datetime import timedelta +from typing import Any + +from django.utils import timezone + +from core.models import PrintQueue +from core.services.moonraker import map_klipper_state +from core.services.printer_backend import NormalizedJobStatus + +logger = logging.getLogger(__name__) + +#: Minimum interval between consecutive progress writes per queue entry. +PROGRESS_WRITE_INTERVAL = timedelta(seconds=5) + +#: Map Moonraker ``notify_history_changed`` job statuses -> normalized state. +_HISTORY_STATUS_MAP = { + "completed": NormalizedJobStatus.STATE_COMPLETE, + "cancelled": NormalizedJobStatus.STATE_CANCELLED, + "error": NormalizedJobStatus.STATE_ERROR, + "interrupted": NormalizedJobStatus.STATE_ERROR, + "klippy_shutdown": NormalizedJobStatus.STATE_ERROR, + "server_exit": NormalizedJobStatus.STATE_ERROR, +} + + +def apply_status_event(entry: PrintQueue, event: dict[str, Any]) -> bool: + """Apply a Moonraker JSON-RPC notification to a ``PrintQueue`` entry. + + Handles two Moonraker notifications: + + * ``notify_history_changed`` with ``action="finished"``: transitions + the entry to ``awaiting_review`` and sets ``completed_at``. + * ``notify_status_update``: reads ``print_stats.state`` and + ``virtual_sdcard.progress`` and either updates the progress + (throttled) or, if the state is terminal, triggers the same + transition as above. + + Args: + entry: The PrintQueue entry currently associated with the + printer that produced the event. Must have ``status == + STATUS_PRINTING`` for any transition to happen. + event: The raw Moonraker notification payload (a JSON-RPC + request dict with ``method`` and ``params``). + + Returns: + ``True`` if the entry was modified and saved, ``False`` if the + event was ignored (throttled progress, wrong status, unknown + method, …). + """ + method = event.get("method") + params = event.get("params") or [] + + if method == "notify_history_changed": + return _handle_history_changed(entry, params) + if method == "notify_status_update": + return _handle_status_update(entry, params) + + logger.debug("Ignoring Moonraker event: %s", method) + return False + + +def _handle_history_changed(entry: PrintQueue, params: list[Any]) -> bool: + if entry.status != PrintQueue.STATUS_PRINTING: + return False + if not params: + return False + payload = params[0] if isinstance(params[0], dict) else {} + if payload.get("action") != "finished": + return False + + job = payload.get("job") or {} + raw_status = job.get("status", "completed") + normalized = _HISTORY_STATUS_MAP.get(raw_status, NormalizedJobStatus.STATE_COMPLETE) + + return _transition_to_review(entry, normalized, raw_status) + + +def _handle_status_update(entry: PrintQueue, params: list[Any]) -> bool: + if entry.status != PrintQueue.STATUS_PRINTING: + return False + if not params: + return False + status_block = params[0] if isinstance(params[0], dict) else {} + + # Terminal state via print_stats -> transition immediately. + print_stats = status_block.get("print_stats") or {} + raw_state = print_stats.get("state") + if raw_state: + normalized = map_klipper_state(raw_state) + if normalized in NormalizedJobStatus.TERMINAL_STATES: + return _transition_to_review(entry, normalized, raw_state) + + # Otherwise update progress (throttled). + virtual_sd = status_block.get("virtual_sdcard") or {} + new_progress = virtual_sd.get("progress") + if new_progress is None: + return False + + now = timezone.now() + if entry.status_updated_at and (now - entry.status_updated_at) < PROGRESS_WRITE_INTERVAL: + return False + + entry.progress = float(new_progress) + entry.status_updated_at = now + entry.save(update_fields=["progress", "status_updated_at"]) + return True + + +def _transition_to_review(entry: PrintQueue, normalized_state: str, raw_state: str) -> bool: + """Mark the entry as awaiting review with the given terminal state.""" + now = timezone.now() + entry.status = PrintQueue.STATUS_AWAITING_REVIEW + entry.completed_at = now + entry.status_updated_at = now + if normalized_state in ( + NormalizedJobStatus.STATE_ERROR, + NormalizedJobStatus.STATE_CANCELLED, + ): + entry.last_error = f"Printer reported: {raw_state}" + else: + # Clear any leftover error from a previous retry. + entry.last_error = "" + entry.save( + update_fields=[ + "status", + "completed_at", + "status_updated_at", + "last_error", + ] + ) + logger.info( + "Queue entry %s transitioned to awaiting_review (state=%s)", + entry.pk, + raw_state, + ) + return True diff --git a/core/tests/test_management_moonraker_worker.py b/core/tests/test_management_moonraker_worker.py new file mode 100644 index 0000000..5c3f24d --- /dev/null +++ b/core/tests/test_management_moonraker_worker.py @@ -0,0 +1,121 @@ +"""Tests for the moonraker_worker management command reconciler.""" + +from __future__ import annotations + +import asyncio +from unittest import IsolatedAsyncioTestCase + +from django.contrib.auth.models import User +from django.test import TransactionTestCase +from django.utils import timezone + +from core.management.commands.moonraker_worker import ( + _reconcile_printers, +) +from core.models import PrinterProfile + + +class FakeTaskGroup: + """Collects create_task calls without actually running coroutines.""" + + def __init__(self): + self.created: list[tuple[str, object]] = [] + + def create_task(self, coro, *, name=None): + self.created.append((name or "", coro)) + # Close the coroutine to avoid "never awaited" warnings. + coro.close() + task = asyncio.get_event_loop().create_future() + task.set_result(None) + # We need something with .cancel() for the reconciler map. + return _FakeTask(name) + + +class _FakeTask: + def __init__(self, name: str | None): + self.name = name + self.cancelled = False + + def cancel(self) -> None: + self.cancelled = True + + +class ReconcilePrintersTests(IsolatedAsyncioTestCase, TransactionTestCase): + """Async tests for the reconciler. Use TransactionTestCase so that + ``sync_to_async`` ORM calls see committed data.""" + + def setUp(self): + super().setUp() + self.user = User.objects.create_user(username="wt", password="x") + + async def _create_printer(self, name: str, url: str, key: str = "") -> PrinterProfile: + from asgiref.sync import sync_to_async + + return await sync_to_async(PrinterProfile.objects.create)( + name=name, + moonraker_url=url, + moonraker_api_key=key, + created_by=self.user, + ) + + async def test_spawns_task_for_new_printer(self): + await self._create_printer("P1", "http://p1:7125") + tg = FakeTaskGroup() + active: dict = {} + + await _reconcile_printers(active, tg) + + self.assertEqual(len(tg.created), 1) + self.assertEqual(len(active), 1) + + async def test_ignores_printer_without_url(self): + await self._create_printer("P1", "") + tg = FakeTaskGroup() + active: dict = {} + + await _reconcile_printers(active, tg) + + self.assertEqual(len(tg.created), 0) + self.assertEqual(active, {}) + + async def test_removed_printer_is_cancelled(self): + from asgiref.sync import sync_to_async + + p1 = await self._create_printer("P1", "http://p1:7125") + tg = FakeTaskGroup() + active: dict = {} + await _reconcile_printers(active, tg) + (task, _snap) = active[p1.pk] + + # Clear moonraker_url (= effectively removed from active set). + await sync_to_async(PrinterProfile.objects.filter(pk=p1.pk).update)(moonraker_url="") + + await _reconcile_printers(active, tg) + + self.assertTrue(task.cancelled) + self.assertNotIn(p1.pk, active) + + async def test_changed_printer_restarts_task(self): + from asgiref.sync import sync_to_async + + p1 = await self._create_printer("P1", "http://p1:7125") + tg = FakeTaskGroup() + active: dict = {} + await _reconcile_printers(active, tg) + (old_task, _snap) = active[p1.pk] + self.assertEqual(len(tg.created), 1) + + # Touch updated_at by saving a new URL. + await sync_to_async(PrinterProfile.objects.filter(pk=p1.pk).update)( + moonraker_url="http://p1-new:7125", + updated_at=timezone.now(), + ) + + await _reconcile_printers(active, tg) + + self.assertTrue(old_task.cancelled) + self.assertEqual(len(tg.created), 2) # one new spawn + self.assertIn(p1.pk, active) + (new_task, new_snap) = active[p1.pk] + self.assertIsNot(new_task, old_task) + self.assertEqual(new_snap.moonraker_url, "http://p1-new:7125") diff --git a/core/tests/test_services_moonraker_ws.py b/core/tests/test_services_moonraker_ws.py new file mode 100644 index 0000000..9a4f2f9 --- /dev/null +++ b/core/tests/test_services_moonraker_ws.py @@ -0,0 +1,77 @@ +"""Tests for the Moonraker WebSocket helpers.""" + +from unittest.mock import MagicMock, patch + +import requests +from django.test import TestCase + +from core.services.moonraker_ws import build_ws_url, fetch_oneshot_token +from core.tests.mixins import TestDataMixin + + +class BuildWsUrlTests(TestCase): + def test_plain_http(self): + self.assertEqual( + build_ws_url("http://printer.local:7125"), + "ws://printer.local:7125/websocket", + ) + + def test_https(self): + self.assertEqual( + build_ws_url("https://printer.local"), + "wss://printer.local/websocket", + ) + + def test_subpath(self): + self.assertEqual( + build_ws_url("http://host/moonraker/"), + "ws://host/moonraker/websocket", + ) + + def test_trailing_slash(self): + self.assertEqual( + build_ws_url("http://host:7125/"), + "ws://host:7125/websocket", + ) + + +class FetchOneshotTokenTests(TestDataMixin, TestCase): + def setUp(self): + super().setUp() + self.printer.moonraker_url = "http://printer:7125" + + def test_no_api_key_returns_none(self): + self.printer.moonraker_api_key = "" + self.assertIsNone(fetch_oneshot_token(self.printer)) + + @patch("core.services.moonraker_ws.requests.get") + def test_with_api_key_returns_token(self, mock_get): + self.printer.moonraker_api_key = "secret" + mock_resp = MagicMock() + mock_resp.json.return_value = {"result": "abc123"} + mock_resp.raise_for_status.return_value = None + mock_get.return_value = mock_resp + + token = fetch_oneshot_token(self.printer) + + self.assertEqual(token, "abc123") + mock_get.assert_called_once() + kwargs = mock_get.call_args.kwargs + self.assertEqual(kwargs["headers"]["X-Api-Key"], "secret") + + @patch("core.services.moonraker_ws.requests.get") + def test_request_error_returns_none(self, mock_get): + self.printer.moonraker_api_key = "secret" + mock_get.side_effect = requests.ConnectionError("down") + + self.assertIsNone(fetch_oneshot_token(self.printer)) + + @patch("core.services.moonraker_ws.requests.get") + def test_unexpected_payload_returns_none(self, mock_get): + self.printer.moonraker_api_key = "secret" + mock_resp = MagicMock() + mock_resp.json.return_value = {"result": {"nested": "nope"}} + mock_resp.raise_for_status.return_value = None + mock_get.return_value = mock_resp + + self.assertIsNone(fetch_oneshot_token(self.printer)) diff --git a/core/tests/test_services_printer_status_sync.py b/core/tests/test_services_printer_status_sync.py new file mode 100644 index 0000000..b21d725 --- /dev/null +++ b/core/tests/test_services_printer_status_sync.py @@ -0,0 +1,178 @@ +"""Tests for the printer status event -> PrintQueue sync logic.""" + +from datetime import timedelta + +from django.test import TestCase +from django.utils import timezone + +from core.models import PrintJob, PrintJobPlate, PrintQueue +from core.services.printer_status_sync import ( + PROGRESS_WRITE_INTERVAL, + apply_status_event, +) +from core.tests.mixins import TestDataMixin + + +class ApplyStatusEventTests(TestDataMixin, TestCase): + """Tests for ``apply_status_event``.""" + + def setUp(self): + super().setUp() + self.printer.moonraker_url = "http://printer:7125" + self.printer.save() + + self.job = PrintJob.objects.create( + name="Test Job", + status=PrintJob.STATUS_SLICED, + created_by=self.user, + ) + self.plate = PrintJobPlate.objects.create( + print_job=self.job, + plate_number=1, + status=PrintJobPlate.STATUS_PRINTING, + ) + self.entry = PrintQueue.objects.create( + plate=self.plate, + printer=self.printer, + status=PrintQueue.STATUS_PRINTING, + started_at=timezone.now(), + ) + + # ----- notify_history_changed ------------------------------------------ + + def test_history_finished_completed_transitions_to_review(self): + event = { + "method": "notify_history_changed", + "params": [{"action": "finished", "job": {"status": "completed"}}], + } + changed = apply_status_event(self.entry, event) + self.entry.refresh_from_db() + self.assertTrue(changed) + self.assertEqual(self.entry.status, PrintQueue.STATUS_AWAITING_REVIEW) + self.assertIsNotNone(self.entry.completed_at) + self.assertEqual(self.entry.last_error, "") + + def test_history_finished_cancelled_sets_last_error(self): + event = { + "method": "notify_history_changed", + "params": [{"action": "finished", "job": {"status": "cancelled"}}], + } + apply_status_event(self.entry, event) + self.entry.refresh_from_db() + self.assertEqual(self.entry.status, PrintQueue.STATUS_AWAITING_REVIEW) + self.assertIn("cancelled", self.entry.last_error) + + def test_history_finished_completed_clears_previous_last_error(self): + self.entry.last_error = "old failure from previous retry" + self.entry.save(update_fields=["last_error"]) + event = { + "method": "notify_history_changed", + "params": [{"action": "finished", "job": {"status": "completed"}}], + } + apply_status_event(self.entry, event) + self.entry.refresh_from_db() + self.assertEqual(self.entry.status, PrintQueue.STATUS_AWAITING_REVIEW) + self.assertEqual(self.entry.last_error, "") + + def test_history_finished_error_sets_last_error(self): + event = { + "method": "notify_history_changed", + "params": [{"action": "finished", "job": {"status": "error"}}], + } + apply_status_event(self.entry, event) + self.entry.refresh_from_db() + self.assertEqual(self.entry.status, PrintQueue.STATUS_AWAITING_REVIEW) + self.assertIn("error", self.entry.last_error) + + def test_history_action_added_is_ignored(self): + event = { + "method": "notify_history_changed", + "params": [{"action": "added", "job": {"status": "in_progress"}}], + } + changed = apply_status_event(self.entry, event) + self.entry.refresh_from_db() + self.assertFalse(changed) + self.assertEqual(self.entry.status, PrintQueue.STATUS_PRINTING) + + def test_history_on_non_printing_entry_is_ignored(self): + self.entry.status = PrintQueue.STATUS_WAITING + self.entry.save(update_fields=["status"]) + event = { + "method": "notify_history_changed", + "params": [{"action": "finished", "job": {"status": "completed"}}], + } + changed = apply_status_event(self.entry, event) + self.assertFalse(changed) + + # ----- notify_status_update: terminal state ---------------------------- + + def test_status_update_terminal_state_triggers_transition(self): + event = { + "method": "notify_status_update", + "params": [{"print_stats": {"state": "complete"}}], + } + changed = apply_status_event(self.entry, event) + self.entry.refresh_from_db() + self.assertTrue(changed) + self.assertEqual(self.entry.status, PrintQueue.STATUS_AWAITING_REVIEW) + + def test_status_update_error_state_sets_last_error(self): + event = { + "method": "notify_status_update", + "params": [{"print_stats": {"state": "error"}}], + } + apply_status_event(self.entry, event) + self.entry.refresh_from_db() + self.assertEqual(self.entry.status, PrintQueue.STATUS_AWAITING_REVIEW) + self.assertIn("error", self.entry.last_error) + + # ----- notify_status_update: progress throttling ----------------------- + + def test_status_update_progress_first_write(self): + event = { + "method": "notify_status_update", + "params": [{"virtual_sdcard": {"progress": 0.25}}], + } + changed = apply_status_event(self.entry, event) + self.entry.refresh_from_db() + self.assertTrue(changed) + self.assertAlmostEqual(self.entry.progress, 0.25) + self.assertIsNotNone(self.entry.status_updated_at) + + def test_status_update_progress_within_throttle_window_ignored(self): + self.entry.progress = 0.10 + self.entry.status_updated_at = timezone.now() + self.entry.save(update_fields=["progress", "status_updated_at"]) + + event = { + "method": "notify_status_update", + "params": [{"virtual_sdcard": {"progress": 0.12}}], + } + changed = apply_status_event(self.entry, event) + self.entry.refresh_from_db() + self.assertFalse(changed) + self.assertAlmostEqual(self.entry.progress, 0.10) + + def test_status_update_progress_after_throttle_window_written(self): + self.entry.progress = 0.10 + self.entry.status_updated_at = timezone.now() - PROGRESS_WRITE_INTERVAL - timedelta(seconds=1) + self.entry.save(update_fields=["progress", "status_updated_at"]) + + event = { + "method": "notify_status_update", + "params": [{"virtual_sdcard": {"progress": 0.55}}], + } + changed = apply_status_event(self.entry, event) + self.entry.refresh_from_db() + self.assertTrue(changed) + self.assertAlmostEqual(self.entry.progress, 0.55) + + # ----- unknown events --------------------------------------------------- + + def test_unknown_method_ignored(self): + event = {"method": "notify_whatever", "params": []} + self.assertFalse(apply_status_event(self.entry, event)) + + def test_missing_params_ignored(self): + event = {"method": "notify_status_update"} + self.assertFalse(apply_status_event(self.entry, event)) diff --git a/core/views/queue.py b/core/views/queue.py index 1b4486b..08785c3 100644 --- a/core/views/queue.py +++ b/core/views/queue.py @@ -22,7 +22,6 @@ ) from core.services.printer_backend import ( PrinterError, - PrinterNotConfiguredError, get_printer_backend, ) from core.services.queue import PrintStartError, start_print_for_queue_entry @@ -393,12 +392,19 @@ def post(self, request: HttpRequest, pk: int) -> HttpResponse: return redirect("core:printqueue_list") +#: If ``status_updated_at`` is older than this, the worker is +#: considered stale and the client is told so via ``worker_stale``. +WORKER_STALE_THRESHOLD_SECONDS = 60 + + class QueueCheckPrinterStatusView(LoginRequiredMixin, View): - """JSON endpoint: poll Moonraker for a printing queue entry. + """JSON endpoint: return the current status of a printing queue entry. - Returns ``{"status": "printing"|"awaiting_review"|"error", ...}``. - If the printer reports the job as complete the queue entry is moved - to *awaiting_review*. + Data is served entirely from the database. The ``moonraker_worker`` + management command is responsible for pushing live progress and + terminal-state transitions into the DB via its WebSocket + connection to Moonraker, so this endpoint no longer talks to the + printer directly. """ def get(self, request: HttpRequest, pk: int) -> JsonResponse: @@ -408,43 +414,24 @@ def get(self, request: HttpRequest, pk: int) -> JsonResponse: ) if entry.status != PrintQueue.STATUS_PRINTING: - return JsonResponse({"status": entry.status}) - - printer = entry.printer - try: - backend = get_printer_backend(printer) - job_status = backend.get_job_status() - state = job_status.state - progress = job_status.progress - - if job_status.is_terminal: - # complete / error / cancelled / standby all mean the - # print is no longer running -> operator review. - entry.status = PrintQueue.STATUS_AWAITING_REVIEW - entry.completed_at = timezone.now() - entry.save(update_fields=["status", "completed_at"]) - message = "Print finished — awaiting review." if state == "complete" else f"Printer reported: {state}" - return JsonResponse( - { - "status": "awaiting_review", - "message": message, - } - ) - - return JsonResponse( - { - "status": "printing", - "printer_state": state, - "progress": progress, - } - ) - - except PrinterNotConfiguredError as exc: - return JsonResponse({"status": "error", "message": str(exc)}) - except PrinterError as exc: - logger.warning("Printer poll failed for entry %s: %s", pk, exc) - msg = "Could not communicate with printer. Check server logs for details." - return JsonResponse({"status": "error", "message": msg}) + payload: dict = {"status": entry.status} + if entry.last_error: + payload["message"] = entry.last_error + return JsonResponse(payload) + + payload = { + "status": "printing", + "progress": entry.progress, + } + if entry.status_updated_at is not None: + payload["status_updated_at"] = entry.status_updated_at.isoformat() + age = (timezone.now() - entry.status_updated_at).total_seconds() + if age > WORKER_STALE_THRESHOLD_SECONDS: + payload["worker_stale"] = True + else: + # No update yet -> worker either hasn't connected or just started. + payload["worker_stale"] = True + return JsonResponse(payload) class CancelQueueEntryView(PrinterControlMixin, View): diff --git a/docker-compose.yml b/docker-compose.yml index 07f653c..10703d2 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -18,6 +18,24 @@ services: - orcaslicer - spoolman + worker: + build: + context: . + target: release + entrypoint: [] + command: ["python", "manage.py", "moonraker_worker"] + volumes: + - media_data:/app/media + - db_data:/app/data + environment: + - DJANGO_SECRET_KEY=${DJANGO_SECRET_KEY:-change-me-in-production} + - DEBUG=${DEBUG:-0} + - LOG_LEVEL=${LOG_LEVEL:-INFO} + - WORKER_RELOAD_INTERVAL=${WORKER_RELOAD_INTERVAL:-10} + restart: unless-stopped + depends_on: + - web + orcaslicer: image: ghcr.io/afkfelix/orca-slicer-api:latest-orca2.3.1 ports: diff --git a/requirements.txt b/requirements.txt index 816fec7..f5ac80a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,3 +4,4 @@ pillow>=10.0 markdown>=3.5 bleach>=6.0 whitenoise>=6.0 +websockets>=14,<16