viewit
cdf11f6bfe
First stage of splitting the 6368-line kobrax_moonraker_bridge.py monolith.
The low-coupling, self-contained pieces move into their own modules;
kobrax_moonraker_bridge.py re-exports them so every caller (12 test files,
the PyInstaller spec) keeps working unchanged - not a single test or the
spec needed editing.
Extracted:
- spoolman_client.py <- SpoolmanClient (zero coupling)
- gcode_store.py <- GCodeStore (stdlib only)
- gcode_meta.py <- _parse_gcode_*/_extract_* metadata helpers
- camera.py <- CameraCache + _find_ffmpeg
- bridge_logging.py <- _BrowserLogHandler, the log ring buffer + SSE
queues, _set_verbose_http_log (the shared mutable
buffer/queues are re-imported so the log endpoints
still operate on the same objects the handler writes)
Facade shrinks from 6368 to 5566 lines. All 184 tests green after each
extraction. Verified the re-exported names resolve and the shared log
objects are identical by reference across modules. No behavior change.
46 lines
1.7 KiB
Python
46 lines
1.7 KiB
Python
"""
|
|
spoolman_client.py - thin synchronous HTTP client for Spoolman filament tracking.
|
|
|
|
Extracted from kobrax_moonraker_bridge.py as part of splitting that module up;
|
|
re-exported from there so existing imports keep working.
|
|
|
|
────────────────────────────────────────────────────────────────────────────
|
|
Copyright (C) 2026 viewit (KX-Bridge contributors)
|
|
|
|
Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md.
|
|
"""
|
|
|
|
|
|
class SpoolmanClient:
|
|
"""Thin synchronous HTTP client for Spoolman filament tracking.
|
|
|
|
Designed to be called from daemon threads (poll loop, _on_print callbacks).
|
|
Uses requests (already in requirements) so no event-loop dependency.
|
|
"""
|
|
|
|
def __init__(self, server_url: str, sync_rate: int = 0):
|
|
self.server_url = server_url.rstrip("/")
|
|
self.sync_rate = sync_rate
|
|
|
|
def _req(self, method: str, path: str, **kwargs):
|
|
import requests
|
|
r = requests.request(method, f"{self.server_url}{path}", timeout=5, **kwargs)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def health_check(self) -> bool:
|
|
try:
|
|
self._req("GET", "/api/v1/health")
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def list_spools(self) -> list:
|
|
return self._req("GET", "/api/v1/spool")
|
|
|
|
def use_filament(self, spool_id: int, use_length_mm: float) -> None:
|
|
"""Report consumed filament length in mm. Spoolman converts to weight
|
|
using the spool's filament profile density."""
|
|
self._req("PUT", f"/api/v1/spool/{spool_id}/use",
|
|
json={"use_length": round(use_length_mm, 2)})
|