commit bf32667f77786295f37fbe759dbd4b850773464f Author: Jonas Menzel Date: Sun Oct 25 03:01:17 2026 +0100 CC2 ↔ Spoolman bridge with NFC tag generator - Reads /opt/usr/canvas_record_info.json over SSH (pinned host key, restricted key), decodes Spoolman spool ids from the tag weight - Mirrors spools into CC2 Control's spool library and assigns slots, keeps its own slot mapping, sets Spoolman locations - Books the drop of CC2 Control's measured remaining weight to Spoolman after each print, idempotent across restarts and interrupted bookings - Dry-run by default; web UI for settings, SSH key, status and log - Tag generator for NFC Tools command chains (NTAG213/215/216) - Tests, mock environment, Dockerfile, Gitea Actions CI Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_014GcEk5tkfFSYD6NTF66NkC diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..3f5d7de --- /dev/null +++ b/.dockerignore @@ -0,0 +1,11 @@ +.git +.gitea +**/__pycache__ +**/*.pyc +.pytest_cache +data +.env +*.md +!README.md +docker-compose.yml +tools diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..0a9582a --- /dev/null +++ b/.env.example @@ -0,0 +1,17 @@ +# Startwerte. Nach dem ersten Start gelten die Einstellungen aus der Oberfläche (/data/settings.json). +PRINTER_HOST=10.10.2.202 +SSH_PORT=22 +SSH_USER=root +CC2_URL=http://10.10.2.202:8081 +SPOOLMAN_URL=http://10.10.6.10:7912 +ID_OFFSET=10000 +POLL_INTERVAL=5 +# Dry-Run ist Standard; ausschalten nur über die Oberfläche. +DRY_RUN=true + +# Optionaler Basic-Auth-Schutz der Oberfläche (beide setzen, um ihn zu aktivieren): +# BRIDGE_USER=admin +# BRIDGE_PASSWORD=ein-langes-passwort + +# PORT=5066 +# LOG_LEVEL=INFO diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml new file mode 100644 index 0000000..0c67085 --- /dev/null +++ b/.gitea/workflows/ci.yml @@ -0,0 +1,97 @@ +name: CI + +# Jeder Push und jeder Pull Request: Tests, dann Image-Build. +# Veröffentlicht wird nur auf main und nur, wenn die Tests grün sind (`needs:`): +# +# Push auf main → :latest, :sha- +# +# Actions auf Commit-SHAs festgelegt (Kommentar = Version), wie in den anderen Repos. +on: + push: + pull_request: + +concurrency: + group: ci-${{ github.ref }} + cancel-in-progress: true + +env: + # Gitea hat keinen Artefakt-Dienst für Build-Records → Upload/Summary abschalten. + DOCKER_BUILD_RECORD_UPLOAD: "false" + DOCKER_BUILD_SUMMARY: "false" + IMAGE: git.menzel.center/menzeljonas/cc2-spoolman-bridge + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0 + + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0 + with: + python-version: "3.12" # wie im Dockerfile + + - name: Abhängigkeiten installieren + run: pip install -r requirements-dev.txt + + - name: pytest + run: python -m pytest + + image: + needs: test + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0 + + - uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f # v3.12.0 + + - name: Veröffentlichen? + id: publish + run: | + if [[ "$GITHUB_EVENT_NAME" == push && "$GITHUB_REF" == refs/heads/main ]]; then + echo "push=true" >> "$GITHUB_OUTPUT" + else + echo "push=false" >> "$GITHUB_OUTPUT" + fi + echo "short=${GITHUB_SHA::12}" >> "$GITHUB_OUTPUT" + + - name: An der Registry anmelden + if: steps.publish.outputs.push == 'true' + uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3.7.0 + with: + registry: git.menzel.center + username: ${{ gitea.actor }} + # Bevorzugt das Repo-Secret REGISTRY_TOKEN (Token mit package: write); + # fehlt es, wird der automatische Actions-Token verwendet. + password: ${{ secrets.REGISTRY_TOKEN || secrets.GITEA_TOKEN }} + + - name: Bauen (und auf main hochladen) + uses: docker/build-push-action@10e90e3645eae34f1e60eeb005ba3a3d33f178e8 # v6.19.2 + with: + context: . + target: runtime + platforms: linux/amd64 + push: ${{ steps.publish.outputs.push == 'true' }} + tags: | + ${{ env.IMAGE }}:latest + ${{ env.IMAGE }}:sha-${{ steps.publish.outputs.short }} + build-args: | + APP_VERSION=${{ steps.publish.outputs.short }} + labels: | + org.opencontainers.image.title=cc2-spoolman-bridge + org.opencontainers.image.source=https://git.menzel.center/menzeljonas/cc2-spoolman-bridge + org.opencontainers.image.revision=${{ github.sha }} + # Registry-Cache: der Gitea-Runner hat keinen Actions-Cache-Dienst. + cache-from: type=registry,ref=${{ env.IMAGE }}:buildcache + cache-to: ${{ steps.publish.outputs.push == 'true' && format('type=registry,ref={0}:buildcache,mode=max', env.IMAGE) || '' }} + provenance: false + + - name: Hinweis bei fehlgeschlagener Anmeldung + if: failure() && steps.publish.outputs.push == 'true' + run: | + { + echo "Build oder Push nach git.menzel.center fehlgeschlagen." + echo "" + echo "Bei 'unauthorized'/'denied' ein Repo-Secret **REGISTRY_TOKEN** anlegen:" + echo "1. Benutzereinstellungen → Anwendungen → Neuen Token erzeugen, *package*: Lesen und Schreiben" + echo "2. Repository → Einstellungen → Actions → Secrets → REGISTRY_TOKEN" + } >> "$GITHUB_STEP_SUMMARY" diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0c38e41 --- /dev/null +++ b/.gitignore @@ -0,0 +1,9 @@ +__pycache__/ +*.pyc +.pytest_cache/ +.venv/ +.env +data/ +*.key +id_ed25519* +id_rsa* diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..4c4ecf8 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,25 @@ +# syntax=docker/dockerfile:1 +FROM python:3.12-slim AS base +ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 PIP_NO_CACHE_DIR=1 PIP_DISABLE_PIP_VERSION_CHECK=1 +WORKDIR /app +COPY requirements.txt . +RUN pip install -r requirements.txt + +FROM base AS test +COPY requirements-dev.txt pytest.ini ./ +RUN pip install -r requirements-dev.txt +COPY app ./app +COPY tests ./tests +RUN python -m pytest + +FROM base AS runtime +ARG APP_VERSION=dev +ENV APP_VERSION=${APP_VERSION} DATA_DIR=/data PORT=5066 +COPY app ./app +RUN useradd --uid 1000 --create-home bridge && mkdir -p /data && chown bridge:bridge /data +USER bridge +VOLUME ["/data"] +EXPOSE 5066 +HEALTHCHECK --interval=30s --timeout=5s --start-period=10s \ + CMD python -c "import os,urllib.request; urllib.request.urlopen('http://127.0.0.1:%s/healthz' % os.environ.get('PORT','5066'), timeout=4)" +CMD ["python", "-m", "app.web"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..e7c8d7d --- /dev/null +++ b/README.md @@ -0,0 +1,155 @@ +# cc2-spoolman-bridge + +Bridge zwischen der Canvas des **Elegoo Centauri Carbon 2** (Community-Firmware, CC2 Control) und **Spoolman**, +mit Weboberfläche und **NFC-Tag-Generator**. + +- Liegt eine Spule mit eigenem Tag in der Canvas, erkennt die Bridge die Spoolman-Spule am Tag, legt sie in + CC2 Control an (bzw. findet sie wieder), weist sie dem Slot zu und setzt in Spoolman den Ort („CC2 Slot 1–4“, + beim Entnehmen „Lager“). +- Den von CC2 Control gemessenen Verbrauch bucht sie nach jedem Druck auf genau diese Spoolman-Spule. +- Der Tag-Generator gibt für jede Spoolman-Spule die vollständige Befehlskette zum Beschreiben eines leeren NTAG213 aus. + +## Was die Bridge darf und was nicht + +| System | Zugriff | +|---|---| +| Drucker | **nur lesen**: `cat /opt/usr/canvas_record_info.json` per SSH. Kein MQTT, keine Druckerbefehle, keine Registrierung als MQTT-Client. | +| CC2 Control | Spulenverwaltung: `GET /api/spools`, `POST /api/spools/save`, `/assign`, `/enable` (nur nach Bestätigung) | +| Spoolman | `GET /spool`, `PATCH /spool/{id}` (`location`), `PUT /spool/{id}/use` | + +**Dry-Run ist Standard.** Solange er aktiv ist, werden alle Schreibaktionen nur als „würde …“ protokolliert. +Ausschalten unter *Einstellungen* (mit Rückfrage). + +## Einrichtung + +### 1. SSH-Schlüssel auf dem Drucker einschränken + +Eigenen Schlüssel erzeugen (ohne Passphrase): + +```bash +ssh-keygen -t ed25519 -N "" -C cc2-spoolman-bridge -f cc2_bridge_key +``` + +Auf dem Drucker in `/root/.ssh/authorized_keys` (bzw. dem Pfad, den dein SSH-Server nutzt) eintragen. Jede +Ausführung liefert dann nur noch den Canvas-Datensatz, egal welcher Befehl gesendet wird, und nur von `10.10.6.10`: + +``` +command="cat /opt/usr/canvas_record_info.json",from="10.10.6.10",no-pty,no-port-forwarding,no-agent-forwarding,no-X11-forwarding ssh-ed25519 AAAA… cc2-spoolman-bridge +``` + +Die Oberfläche zeigt diese Zeile mit dem passenden öffentlichen Schlüssel nach dem Hochladen an. Container auf +`10.10.6.10` verbinden sich über Docker-NAT mit der Host-IP, `from="10.10.6.10"` passt also. + +### 2. Container starten (Docker-Host 10.10.6.10) + +```bash +mkdir -p /opt/cc2-spoolman-bridge && cd /opt/cc2-spoolman-bridge +# docker-compose.yml und .env.example aus dem Repo hierher kopieren +cp .env.example .env +mkdir -p data && sudo chown 1000:1000 data +# privates Image: einmalig mit einem Gitea-Token (package: Lesen) anmelden +docker login git.menzel.center -u menzeljonas +docker compose pull && docker compose up -d +``` + +Oberfläche: `http://10.10.6.10:5066` + +### 3. In der Oberfläche + +1. *Einstellungen* prüfen (Drucker-IP, CC2-Control- und Spoolman-URL, ID-Offset 10000). +2. Privaten Schlüssel einfügen oder hochladen und speichern. Er liegt dann mit Rechten 600 in + `/data/ssh/id_key`; die Oberfläche zeigt nur Fingerprint und öffentlichen Schlüssel. +3. **Verbindung testen**: zeigt die 4 Slots und den Host-Key-Fingerprint des Druckers. Fingerprint vergleichen und + bestätigen. Danach wird er bei jeder Verbindung geprüft; ein geänderter Host-Key wird abgelehnt. +4. *Status* im Dry-Run beobachten. Wenn die „würde …“-Einträge stimmen, Dry-Run ausschalten. + +### Umgebungsvariablen + +Siehe [`.env.example`](.env.example). `PRINTER_HOST`, `SSH_PORT`, `SSH_USER`, `CC2_URL`, `SPOOLMAN_URL`, +`ID_OFFSET`, `POLL_INTERVAL` und `DRY_RUN` sind nur Startwerte, bis die Einstellungen einmal in der Oberfläche +gespeichert wurden (`/data/settings.json`). `BRIDGE_USER` und `BRIDGE_PASSWORD` schalten Basic-Auth ein +(`/healthz` bleibt offen). `PORT` (Standard 5066), `LOG_LEVEL`. + +## Funktionsweise + +**Tag erkennen.** Die Bridge liest alle ~5 s über eine stehende SSH-Verbindung den Canvas-Datensatz. +`channels0`–`channels3` sind Slot 1–4. Ist `weight` ≥ `ID_OFFSET`, ist `weight − ID_OFFSET` die Spoolman-Spulen-ID; +kleinere Werte sind echte Gewichte von Original- oder Fremdtags und werden ignoriert. Neu ist ein Tag, wenn sich die ID +im Slot gegenüber dem letzten Datensatz ändert. + +**Zuordnung.** Die Bridge speichert Slot ↔ Spule selbst (`/data/state.json`). Fällt `weight` nach einer +Bearbeitung am Display auf 1000 zurück, bleibt die Zuordnung. Aufgehoben wird sie nur, wenn +- der Slot länger als 20 s leer ist (Tray-Status 0), +- CC2 Control die Spule ins Lager legt (Slot ohne Spule und ohne offene Rückfrage), +- im Slot ein Tag mit einer anderen ID gescannt wird oder +- die Spule in einem anderen Slot gescannt wird (Umzug). + +Weist du eine gespiegelte Spule in CC2 Control von Hand einem belegten Slot zu, übernimmt die Bridge das. + +**CC2 Control.** Gespiegelte Spulen erkennt die Bridge an `note` = `spoolman:`. Fehlt die Spule, legt sie sie +mit `name` = Label (`lot_nr`, sonst Hersteller + ID dreistellig), `brand`, `material`, `color`, `diameter`, +`density`, `net`, `remaining`, `tare` und `price` aus Spoolman an und weist sie dem Slot zu. Damit ist auch die +Rückfrage beantwortet. Passt das Spoolman-Material nicht zum Tag (CC2-Regel: gleich Typ/Name oder Produktlinie wie +`PLA+` zu `PLA`), nimmt sie den Untertyp vom Tag, damit CC2 Control nicht erneut fragt. + +**Verbrauch.** CC2 Control zieht den gemessenen Verbrauch vom `remaining` der zugewiesenen Spule ab. (`used` ist in +CC2 Control der Zeitstempel der letzten Nutzung, keine Menge.) Die Bridge bucht für jede gespiegelte Spule die +Abnahme von `remaining` seit der letzten Buchung per `PUT /spool/{id}/use` nach Spoolman. Das geschieht erst, wenn kein +Druck läuft (`job.active`) und `remaining` 60 s unverändert ist. Der gebuchte Stand wird persistent gespeichert. +Vor jeder Buchung schreibt die Bridge die Absicht in den Zustand. Bricht eine Buchung ab, prüft sie nach dem +Neustart an `used_weight` in Spoolman, ob sie schon angekommen ist, statt doppelt zu buchen. Steigt `remaining` +(z. B. durch Nachwiegen in CC2 Control), wird nichts gebucht, nur protokolliert, und der neue Stand gilt als +Ausgangspunkt. + +**Spulenverwaltung aus.** Ist sie in CC2 Control ausgeschaltet, zählt CC2 Control keinen Verbrauch. Die Oberfläche +zeigt das deutlich an und bietet einen Button zum Einschalten (mit Rückfrage, im Dry-Run nur protokolliert). + +## Tag-Format (Canvas, NTAG213/215/216) + +NFC Tools → *Other* → *Advanced NFC commands*, Befehle `A2:PP:XXXXXXXX` (PP = Seite hex), kommagetrennt. +Seiten 0–3 werden nie geschrieben. **Nie „Lock tag“ oder „Set password“ verwenden.** + +| Seite | Inhalt | +|---|---| +| 04–09 | NDEF-URL (`https://www.elegoo.com`; der Drucker ignoriert sie) | +| 10 | `36` + Herstellercode `EEEEEE` | +| 11 | `EE` + Filament-Code (2 Byte) + `00` | +| 12 | Material (4 Byte) | +| 13 | Untertyp (2 Byte) + `0000`; danach wählt der Elegoo Slicer das Profil | +| 14 | Farbe RRGGBB + Finish-Byte (`FF`) | +| 15 | Düse min / max (je 2 Byte, °C) | +| 16 | `00000000` | +| 17 | Durchmesser in 1/100 mm + **Gewicht = Spulen-ID + 10000** | +| 18 | `0036C800` | + +Die Farbe wird auf die nächste Farbe der Display-Palette vorgeschlagen (Abstand in CIELAB; gesättigtes Blau wird auf +`2850DF` abgebildet). Material und Untertyp werden aus dem Spoolman-Material vorgeschlagen; alles ist änderbar. +Optional kann statt der Elegoo-URL ein Link auf die Spule in Spoolman auf den Tag (noch nicht am Drucker getestet). + +## Entwicklung + +```bash +docker build --target test . # Tests (pytest) im Container +docker build -t cc2-spoolman-bridge:dev . # Image +python tools/mock_env.py --host-header mock:8081 # Mock für Drucker (SSH :2222), CC2 Control, Spoolman +``` + +Die Tests decken ab: Tag-Kodierung gegen die verifizierte Vorlage, ID-Dekodierung, Palette und Materialvorschlag, +Zuordnung bei verschwindender ID, Verbrauchs-Delta und Idempotenz (auch nach Abbruch), Dry-Run, Request-Format für +CC2 Control (Host-Header, kein Origin, `X-CC2-Request`, `key=value`-Body, `enable` als nacktes `on`/`off`) und SSH +mit Host-Key-Prüfung gegen einen lokalen SSH-Server. + +CI (Gitea Actions, `.gitea/workflows/ci.yml`): pytest → Build `linux/amd64` → auf `main` Push nach +`git.menzel.center/menzeljonas/cc2-spoolman-bridge` mit den Tags `latest` und `sha-`. Für den Push braucht +das Repo ggf. ein Actions-Secret `REGISTRY_TOKEN` (Token mit *package*: Lesen und Schreiben). + +## Quellen und Lizenzen + +- Material- und Untertyp-Codes (`app/materials.py`, Lizenztext in [THIRD_PARTY_NOTICES.md](THIRD_PARTY_NOTICES.md)) aus + [cjtallman/canvas-rfid-tool](https://github.com/cjtallman/canvas-rfid-tool) (`index.html`), **MIT License**, + Copyright (c) cjtallman. Dort wiederum nach + [DnG-Crafts/ELG-RFID](https://github.com/DnG-Crafts/ELG-RFID) und + [Savion/elegoo-rfid-editor](https://github.com/Savion/elegoo-rfid-editor). +- Spool-API und HTTP-Guard von CC2 Control: + [Neme77/centauri-carbon-2-community](https://github.com/Neme77/centauri-carbon-2-community) + (`cc2-control/docs/API.md`, Abschnitt „Spool tracking“, und `src/spools.h`, Branch `develop`). diff --git a/THIRD_PARTY_NOTICES.md b/THIRD_PARTY_NOTICES.md new file mode 100644 index 0000000..b68b305 --- /dev/null +++ b/THIRD_PARTY_NOTICES.md @@ -0,0 +1,30 @@ +# Third-party notices + +## cjtallman/canvas-rfid-tool + +`app/materials.py` enthält die Tabellen `MATERIALS` und `SUBTYPES` aus +https://github.com/cjtallman/canvas-rfid-tool (`index.html`). + +``` +MIT License + +Copyright (c) 2026 cjtallman + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. +``` diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/bridge.py b/app/bridge.py new file mode 100644 index 0000000..7b0726a --- /dev/null +++ b/app/bridge.py @@ -0,0 +1,614 @@ +"""Bridge between the Canvas record (SSH), CC2 Control's spool library and Spoolman. + +Rules: +- The printer is only read (SSH `cat` of the canvas record). No MQTT, no printer commands. +- Writes go to CC2 Control's spool endpoints and to Spoolman only, and only when dry-run is off. +- Slot <-> spool mappings are kept by the bridge; a tag id vanishing from the record + (weight falls back to 1000 after a touchscreen edit) does not drop a mapping. +""" + +from __future__ import annotations + +import json +import logging +import queue +import threading +import time +from collections import deque +from pathlib import Path + +from . import tagcodec +from .cc2control import CC2Control, CC2Error, slot_spool_id, spoolman_id_of, tray_statuses +from .spoolman import Spoolman, SpoolmanError, SpoolmanNotFound, color_hex, label, summary, vendor_name +from .sshsource import CanvasRecordReader, KeyStore, SSHError +from .storage import Settings, SettingsStore, StateStore + +log = logging.getLogger("bridge") + +MIN_BOOKING = 0.05 # grams; smaller changes accumulate until they reach this +SPOOL_CACHE_SECONDS = 300 +LOG_KEEP = 1000 +SOURCE_NAMES = {"ssh": "Drucker (SSH)", "cc2": "CC2 Control", "spoolman": "Spoolman"} + + +class ActionLog: + def __init__(self, data_dir: Path, clock=time.time): + self.path = data_dir / "actions.log" + self.clock = clock + self.entries: deque[dict] = deque(maxlen=300) + self._lock = threading.Lock() + try: + lines = self.path.read_text().splitlines()[-LOG_KEEP:] + self.path.write_text("".join(line + "\n" for line in lines)) + for line in lines[-300:]: + try: + self.entries.append(json.loads(line)) + except ValueError: + pass + except FileNotFoundError: + pass + + def add(self, level: str, msg: str) -> None: + entry = {"time": self.clock(), "level": level, "msg": msg} + getattr(log, {"error": "error", "warn": "warning"}.get(level, "info"))("%s", msg) + with self._lock: + self.entries.append(entry) + try: + self.path.parent.mkdir(parents=True, exist_ok=True) + with self.path.open("a") as f: + f.write(json.dumps(entry, ensure_ascii=False) + "\n") + except OSError: + pass + + def recent(self, n: int = 200) -> list[dict]: + with self._lock: + return list(self.entries)[-n:][::-1] + + +def material_fits(material: str, tray_type: str, tray_name: str) -> bool: + """CC2 Control's rule: same as the tray's type or name, or a product line of the type.""" + m, t, n = material.upper().strip(), tray_type.upper().strip(), tray_name.upper().strip() + if not m: + return False + if m in (t, n): + return True + return bool(t) and m.startswith(t) and len(m) > len(t) and m[len(t)] in " -+" + + +def cc2_spool_fields(sm: dict, channel: dict | None) -> tuple[dict, list[str]]: + """Fields for /api/spools/save from a Spoolman spool (and the tray it sits in).""" + notes = [] + fil = sm.get("filament") or {} + channel = channel or {} + tray_type = str(channel.get("type") or "") + tray_name = str(channel.get("detailed_type") or "") + material = (fil.get("material") or "").strip() + if tray_type and not material_fits(material, tray_type, tray_name): + new = tray_name or tray_type + notes.append(f"Material {material or '(leer)'} passt nicht zum Tag ({tray_type}); verwende {new}") + material = new + if not material: + material = "PLA" + color = color_hex(sm) + if color is None: + color = tagcodec.normalize_hex_color(str(channel.get("color") or "").replace("0x", "")) or "000000" + notes.append(f"Spoolman-Spule hat keine Farbe; verwende #{color}") + fields: dict = { + "name": label(sm), + "brand": vendor_name(sm), + "material": material, + "color": "#" + color, + "note": f"spoolman:{sm['id']}", + } + + def num(key, value, lo, hi): + try: + v = float(value) + except (TypeError, ValueError): + return + if lo <= v <= hi: + fields[key] = round(v, 3) + else: + notes.append(f"{key}={v} außerhalb {lo}-{hi}, weggelassen") + + num("diameter", fil.get("diameter"), 1, 3.5) + num("density", fil.get("density"), 0.5, 3) + num("net", sm.get("initial_weight") or fil.get("weight"), 1, 20000) + if sm.get("remaining_weight") is not None: + num("remaining", max(0.0, float(sm["remaining_weight"])), 0, 20000) + tare = sm.get("spool_weight") or fil.get("spool_weight") + if tare: + num("tare", tare, 0, 5000) + price = sm.get("price") or fil.get("price") + if price: + num("price", price, 0, 1e7) + return fields, notes + + +class Bridge: + def __init__(self, data_dir: Path, settings: SettingsStore, keys: KeyStore, *, + reader_factory=None, cc2_factory=None, spoolman_factory=None, clock=time.time): + self.data_dir = data_dir + self.settings_store = settings + self.keys = keys + self.clock = clock + self.state = StateStore(data_dir) + self.log = ActionLog(data_dir, clock) + self.reader_factory = reader_factory or ( + lambda s: CanvasRecordReader(keys, s.printer_host, s.ssh_port, s.ssh_user)) + self.cc2_factory = cc2_factory or (lambda s: CC2Control(s.cc2_url)) + self.spoolman_factory = spoolman_factory or (lambda s: Spoolman(s.spoolman_url)) + self._clients_for: dict | None = None + self.reader = self.cc2 = self.spoolman = None + self.lock = threading.RLock() + self.commands: queue.Queue = queue.Queue() + self.wake = threading.Event() + self.errors: dict[str, str | None] = {"ssh": None, "cc2": None, "spoolman": None} + self.empty_since: dict[int, float] = {} + self.remaining_seen: dict[str, tuple[float, float]] = {} # cc2 id -> (value, since) + self.dry_logged: dict[str, str] = {} + self.spool_cache: dict[int, tuple[float, dict]] = {} + self.record_slots: list[dict] = [] + self.tray_status: dict[int, int] = {} + self.snapshot: dict = {"last_poll": None} + self.settings: Settings = settings.get() + + # ---------- infrastructure ---------- + + def _clients(self) -> None: + s = self.settings + key = {k: getattr(s, k) for k in ("printer_host", "ssh_port", "ssh_user", "cc2_url", "spoolman_url")} + if key == self._clients_for: + return + for c in (self.reader, self.cc2, self.spoolman): + if c is not None: + try: + c.close() + except Exception: + pass + self.reader = self.reader_factory(s) + self.cc2 = self.cc2_factory(s) + self.spoolman = self.spoolman_factory(s) + self._clients_for = key + + def _error(self, source: str, msg: str | None) -> None: + if msg != self.errors.get(source): + if msg: + self.log.add("error", msg) + elif self.errors.get(source): + self.log.add("info", f"{SOURCE_NAMES[source]} wieder erreichbar") + self.errors[source] = msg + + def _write(self, key: str, msg: str, fn): + """Run a write, or in dry-run only log it once per key/message.""" + if self.settings.dry_run: + if self.dry_logged.get(key) != msg: + self.dry_logged[key] = msg + self.log.add("dry", f"würde {msg}") + return None, False + self.dry_logged.pop(key, None) + result = fn() + self.log.add("info", f"erledigt: {msg}") + return result, True + + def run(self, stop: threading.Event) -> None: + while not stop.is_set(): + started = self.clock() + try: + self.poll_once() + except Exception: # never let the loop die + log.exception("poll failed") + self.log.add("error", "Interner Fehler im Abfragezyklus (Details im Container-Log)") + interval = self.settings.poll_interval + self.wake.wait(max(0.5, interval - (self.clock() - started))) + self.wake.clear() + + # ---------- one cycle ---------- + + def poll_once(self) -> None: + with self.lock: + self.settings = self.settings_store.get() + self._clients() + self._run_commands() + record = cc2 = None + try: + record = self.reader.read() + self._error("ssh", None) + except SSHError as e: + self._error("ssh", str(e)) + try: + cc2 = self.cc2.spools() + self._error("cc2", None) + except CC2Error as e: + self._error("cc2", str(e)) + + if record is not None: + self._process_record(record) + self._process_scans() + if cc2 is not None: + self._process_trays(cc2) + self._process_cc2_slots(cc2) + self._reconcile_cc2(cc2) + self._consumption(cc2) + self._reconcile_locations() + self._refresh_spool_cache() + self.state.save() + self._build_snapshot(record, cc2) + + def _run_commands(self) -> None: + while True: + try: + cmd, arg = self.commands.get_nowait() + except queue.Empty: + return + try: + if cmd == "enable": + self._write("enable", "Spulenverwaltung in CC2 Control einschalten", lambda: self.cc2.enable(True)) + elif cmd == "reassign": + m = self.state.data["slots"].get(str(arg)) + if m: + m["assigned"] = False + self.dry_logged.pop(f"assign:{arg}", None) + self.log.add("info", f"Slot {arg + 1}: Zuordnung wird erneut an CC2 Control gesendet") + elif cmd == "release": + if str(arg) in self.state.data["slots"]: + self._release(arg, "manuell aufgehoben") + except (CC2Error, SpoolmanError) as e: + self.log.add("error", str(e)) + + # ---------- record / scans ---------- + + def _process_record(self, record: dict) -> None: + offset = self.settings.id_offset + slots = tagcodec.parse_record(record, offset) + prev_record = self.state.data.get("last_record") + prev = tagcodec.parse_record(prev_record, offset) if prev_record else None + mappings = self.state.data["slots"] + for s in slots: + n, tid = s["slot"], s["spool_id"] + if tid is None: + continue + if prev is not None: + is_new = tid != prev[n]["spool_id"] + else: + m = mappings.get(str(n)) + is_new = m is None or m["spool"] != tid + if is_new: + self.state.data.setdefault("scans", []).append({"slot": n, "spool": tid, "time": self.clock()}) + self.log.add("info", f"Slot {n + 1}: Tag mit Spoolman-ID {tid} erkannt") + self.state.data["last_record"] = record + self.record_slots = slots + + def _process_scans(self) -> None: + pending = [] + for scan in self.state.data.get("scans", []): + n, tid = scan["slot"], scan["spool"] + try: + sm = self.spoolman.spool(tid) + self._error("spoolman", None) + except SpoolmanNotFound: + self.log.add("warn", f"Slot {n + 1}: Spoolman-Spule {tid} existiert nicht; nichts geändert") + continue + except SpoolmanError as e: + self._error("spoolman", str(e)) + pending.append(scan) # retry next cycle + continue + self.spool_cache[tid] = (self.clock(), sm) + self._map(n, tid, label(sm), reason="Tag gescannt") + self.state.data["scans"] = pending + + def _map(self, n: int, tid: int, name: str, reason: str, cc2_id: str | None = None) -> None: + mappings = self.state.data["slots"] + cur = mappings.get(str(n)) + if cur and cur["spool"] == tid: + if cc2_id is None: + cur["assigned"] = False # rescan: make sure CC2 Control has it in the slot + self.dry_logged.pop(f"assign:{n}", None) + self._want_location(tid, self.settings.slot_location(n)) + return + for k, m in list(mappings.items()): + if k != str(n) and m["spool"] == tid: + del mappings[k] + self.log.add("info", f"Spule {name} von Slot {int(k) + 1} nach Slot {n + 1} umgezogen") + if cur: + self._release(n, f"andere Spule ({name})") + mappings[str(n)] = { + "spool": tid, "name": name, "since": self.clock(), + "cc2": cc2_id, "assigned": cc2_id is not None, "cc2_seen": cc2_id is not None, + } + self.empty_since.pop(n, None) + self.log.add("info", f"Slot {n + 1}: {name} (Spoolman {tid}) zugeordnet ({reason})") + self._want_location(tid, self.settings.slot_location(n)) + + def _release(self, n: int, reason: str) -> None: + m = self.state.data["slots"].pop(str(n), None) + if not m: + return + self.dry_logged.pop(f"assign:{n}", None) + self.log.add("info", f"Slot {n + 1}: Zuordnung zu {m.get('name') or m['spool']} aufgehoben ({reason})") + if not any(o["spool"] == m["spool"] for o in self.state.data["slots"].values()): + self._want_location(m["spool"], self.settings.location_storage) + + # ---------- CC2 Control ---------- + + def _process_trays(self, cc2: dict) -> None: + statuses = {} + for slot in cc2.get("slots") or []: + printer = slot.get("printer") + if isinstance(printer, dict) and slot.get("slot") in (0, 1, 2, 3): + try: + statuses[slot["slot"]] = int(printer.get("status")) + except (TypeError, ValueError): + pass + if not statuses: + try: + statuses = tray_statuses(self.cc2.canvas()) + except CC2Error: + statuses = {} + self.tray_status = statuses + now = self.clock() + for n in range(4): + st = statuses.get(n) + if st is None: + continue + if st == 0: + since = self.empty_since.setdefault(n, now) + if now - since >= self.settings.empty_grace and str(n) in self.state.data["slots"]: + self._release(n, f"Slot seit {int(now - since)} s leer") + else: + self.empty_since.pop(n, None) + + def _process_cc2_slots(self, cc2: dict) -> None: + by_id = {str(s.get("id")): s for s in cc2.get("spools") or []} + mappings = self.state.data["slots"] + for slot in cc2.get("slots") or []: + n = slot.get("slot") + if n not in (0, 1, 2, 3): + continue + sid = slot_spool_id(slot) + m = mappings.get(str(n)) + mirrored = spoolman_id_of(by_id[sid]) if sid in by_id else None + # A spool we already map to another slot (fresh scan) wins over CC2 Control's older binding. + elsewhere = mirrored is not None and any( + o["spool"] == mirrored for k, o in mappings.items() if k != str(n)) + if m is None: + if mirrored is not None and not elsewhere and self.tray_status.get(n) in (1, 2): + # assigned in CC2 Control by hand + self._map(n, mirrored, by_id[sid].get("name") or str(mirrored), "in CC2 Control zugewiesen", sid) + continue + if m.get("cc2") and sid == m["cc2"]: + m["cc2_seen"] = m["assigned"] = True + continue + if not m.get("cc2_seen"): + continue + if sid: + if elsewhere: + continue + if mirrored is not None and mirrored != m["spool"]: + self._map(n, mirrored, by_id[sid].get("name") or str(mirrored), "in CC2 Control zugewiesen", sid) + elif mirrored is None: + self._release(n, "in CC2 Control mit anderer Spule belegt") + elif not slot.get("question"): + self._release(n, "CC2 Control hat die Spule ins Lager gelegt") + + def _reconcile_cc2(self, cc2: dict) -> None: + mirrors: dict[int, dict] = {} + for sp in cc2.get("spools") or []: + smid = spoolman_id_of(sp) + if smid is not None and (smid not in mirrors or mirrors[smid].get("archived")): + mirrors[smid] = sp + slots = {s.get("slot"): s for s in cc2.get("slots") or []} + record_slots = {s["slot"]: s for s in self.record_slots} + for k, m in list(self.state.data["slots"].items()): + n = int(k) + if m.get("assigned"): + continue + try: + mirror = mirrors.get(m["spool"]) + if mirror is None: + sm = self._spool(m["spool"]) + if sm is None: + continue + channel = (record_slots.get(n) or {}).get("channel") + fields, notes = cc2_spool_fields(sm, channel) + for note in notes: + self.log.add("warn", f"{fields['name']}: {note}") + resp, done = self._write( + f"create:{m['spool']}", f"in CC2 Control Spule {fields['name']} anlegen " + f"({fields['material']}, {fields['color']}, Rest {fields.get('remaining', '?')} g)", + lambda: self.cc2.save_spool(fields)) + if not done: + self._write(f"assign:{n}", f"in CC2 Control {fields['name']} in Slot {n + 1} legen", lambda: None) + continue + cc2_id = str((resp or {}).get("id") or "") + if not cc2_id: + self.log.add("error", "CC2 Control hat keine Spulen-ID zurückgegeben") + continue + self.state.data["booked"][cc2_id] = { + "spoolman": m["spool"], "remaining": float(fields.get("remaining", fields.get("net", 1000))), + "pending": None, "last": None, + } + m["cc2"] = cc2_id + else: + m["cc2"] = str(mirror["id"]) + if mirror.get("archived"): + _, done = self._write(f"unarchive:{m['cc2']}", f"in CC2 Control {mirror.get('name')} aus dem Archiv holen", + lambda: self.cc2.save_spool({"id": m["cc2"], "archived": "0"})) + if not done: + continue + if slot_spool_id(slots.get(n) or {}) == m["cc2"]: + m["assigned"] = m["cc2_seen"] = True + continue + _, done = self._write(f"assign:{n}", f"in CC2 Control {m.get('name')} in Slot {n + 1} legen", + lambda: self.cc2.assign(n, m["cc2"])) + if done: + m["assigned"] = m["cc2_seen"] = True + except CC2Error as e: + self.log.add("error", f"Slot {n + 1}: {e}") + + # ---------- consumption ---------- + + def _consumption(self, cc2: dict) -> None: + job_active = bool((cc2.get("job") or {}).get("active")) + booked = self.state.data["booked"] + now = self.clock() + for sp in cc2.get("spools") or []: + smid = spoolman_id_of(sp) + if smid is None: + continue + cid = str(sp.get("id")) + try: + rem = float(sp.get("remaining")) + except (TypeError, ValueError): + continue + seen = self.remaining_seen.get(cid) + if seen is None or abs(seen[0] - rem) > 1e-6: + self.remaining_seen[cid] = (rem, now) + b = booked.get(cid) + if b is None or b.get("spoolman") != smid: + booked[cid] = {"spoolman": smid, "remaining": rem, "pending": None, "last": None} + self.log.add("info", f"{sp.get('name')}: Ausgangsstand {rem:.1f} g übernommen (Spoolman {smid})") + continue + try: + if b.get("pending") and not self._resolve_pending(cid, b): + continue + delta = b["remaining"] - rem + if delta < -MIN_BOOKING: + self.log.add("warn", f"{sp.get('name')}: Rest in CC2 Control um {-delta:.2f} g gestiegen; " + "nicht gebucht, neuer Ausgangsstand") + b["remaining"] = rem + continue + if delta < MIN_BOOKING or job_active: + continue + if now - self.remaining_seen[cid][1] < self.settings.settle_seconds: + continue + self._book(cid, b, sp.get("name") or cid, smid, delta, rem) + except SpoolmanNotFound: + self._write_error(f"book:{cid}", f"{sp.get('name')}: Spoolman-Spule {smid} existiert nicht; Verbrauch nicht gebucht") + except SpoolmanError as e: + self._error("spoolman", str(e)) + + def _write_error(self, key: str, msg: str) -> None: + if self.dry_logged.get(key) != msg: + self.dry_logged[key] = msg + self.log.add("error", msg) + + def _book(self, cid: str, b: dict, name: str, smid: int, delta: float, rem: float) -> None: + if self.settings.dry_run: + self._write(f"book:{cid}", f"{delta:.2f} g von {name} auf Spoolman-Spule {smid} buchen", lambda: None) + return + before = float(self.spoolman.spool(smid).get("used_weight") or 0) + b["pending"] = {"grams": round(delta, 3), "target": rem, "before": before, "time": self.clock()} + self.state.save() # intent first: a crash after the PUT must not book twice + self.spoolman.use(smid, delta) + self._booked(b, delta, rem, f"{delta:.2f} g von {name} auf Spoolman-Spule {smid} gebucht") + + def _booked(self, b: dict, grams: float, target: float, msg: str) -> None: + b["remaining"] = target + b["pending"] = None + b["last"] = {"time": self.clock(), "grams": round(grams, 3)} + self.state.save() + self.log.add("info", msg) + + def _resolve_pending(self, cid: str, b: dict) -> bool: + """A booking was interrupted: check whether Spoolman already has it.""" + p = b["pending"] + smid = b["spoolman"] + used = float(self.spoolman.spool(smid).get("used_weight") or 0) + if used >= p["before"] + p["grams"] - 0.01: + self._booked(b, p["grams"], p["target"], f"Unterbrochene Buchung über {p['grams']:.2f} g war bereits in Spoolman") + return True + if self.settings.dry_run: + self._write(f"book:{cid}", f"unterbrochene Buchung über {p['grams']:.2f} g auf Spoolman-Spule {smid} wiederholen", + lambda: None) + return False + self.spoolman.use(smid, p["grams"]) + self._booked(b, p["grams"], p["target"], f"Unterbrochene Buchung über {p['grams']:.2f} g auf Spoolman-Spule {smid} nachgeholt") + return True + + # ---------- Spoolman locations ---------- + + def _want_location(self, smid: int, location: str) -> None: + locs = self.state.data["locations"] + entry = locs.setdefault(str(smid), {"want": None, "done": None}) + entry["want"] = location + + def _reconcile_locations(self) -> None: + for k, entry in self.state.data["locations"].items(): + if entry["want"] == entry["done"]: + continue + smid = int(k) + name = self._name(smid) + try: + _, done = self._write(f"location:{k}", f"in Spoolman den Ort von {name} auf „{entry['want']}“ setzen", + lambda: self.spoolman.set_location(smid, entry["want"])) + if done: + entry["done"] = entry["want"] + self._error("spoolman", None) + except SpoolmanNotFound: + self.log.add("warn", f"Spoolman-Spule {smid} existiert nicht mehr; Ort nicht gesetzt") + entry["done"] = entry["want"] + except SpoolmanError as e: + self._error("spoolman", str(e)) + + # ---------- Spoolman cache ---------- + + def _spool(self, smid: int) -> dict | None: + cached = self.spool_cache.get(smid) + if cached and self.clock() - cached[0] < SPOOL_CACHE_SECONDS: + return cached[1] + try: + sm = self.spoolman.spool(smid) + except SpoolmanNotFound: + self._write_error(f"missing:{smid}", f"Spoolman-Spule {smid} existiert nicht") + return None + except SpoolmanError as e: + self._error("spoolman", str(e)) + return cached[1] if cached else None + self.spool_cache[smid] = (self.clock(), sm) + return sm + + def _name(self, smid: int) -> str: + cached = self.spool_cache.get(smid) + return label(cached[1]) if cached else f"Spoolman-Spule {smid}" + + def _refresh_spool_cache(self) -> None: + for m in self.state.data["slots"].values(): + self._spool(m["spool"]) + + # ---------- UI snapshot ---------- + + def _build_snapshot(self, record: dict | None, cc2: dict | None) -> None: + record_slots = tagcodec.parse_record(record or self.state.data.get("last_record") or {}, self.settings.id_offset) + cc2_slots = {s.get("slot"): s for s in (cc2 or {}).get("slots") or []} + cc2_spools = {str(s.get("id")): s for s in (cc2 or {}).get("spools") or []} + booked = self.state.data["booked"] + slots = [] + for n in range(4): + m = self.state.data["slots"].get(str(n)) + cs = cc2_slots.get(n) or {} + sid = slot_spool_id(cs) + cached = self.spool_cache.get(m["spool"]) if m else None + b = booked.get(m["cc2"]) if m and m.get("cc2") else None + slots.append({ + "slot": n, + "record": record_slots[n], + "tray_status": self.tray_status.get(n), + "mapping": m, + "spoolman": summary(cached[1]) if cached else None, + "cc2_spool": cc2_spools.get(sid) if sid else None, + "cc2_question": cs.get("question") or "", + "cc2_matches": bool(m and m.get("cc2") and sid == m["cc2"]), + "last_booking": (b or {}).get("last"), + }) + self.snapshot = { + "last_poll": self.clock(), + "dry_run": self.settings.dry_run, + "errors": dict(self.errors), + "cc2_enabled": None if cc2 is None else bool(cc2.get("enabled")), + "job_active": None if cc2 is None else bool((cc2.get("job") or {}).get("active")), + "slots": slots, + "ssh_key": self.keys.has_key(), + "host_key": self.keys.host_key() is not None, + } + diff --git a/app/cc2control.py b/app/cc2control.py new file mode 100644 index 0000000..90aca85 --- /dev/null +++ b/app/cc2control.py @@ -0,0 +1,156 @@ +"""CC2 Control HTTP client (spool management only). + +Request rules from CC2 Control's HTTP guard: the Host header must be exactly the +printer's host:port, no Origin header, and writes need `X-CC2-Request: 1` with a +plain-text body of `key=value` lines. +""" + +from __future__ import annotations + +import re +from urllib.parse import urlsplit + +import httpx + +TEXT_LIMITS = {"name": 96, "brand": 96, "note": 96, "material": 32} +_FORBIDDEN = re.compile(r'["\\\x00-\x1f\x7f]') + +NOTE_PREFIX = "spoolman:" + + +class CC2Error(RuntimeError): + pass + + +def clean_text(value, limit: int | None = None) -> str: + """UTF-8 without quotes, backslashes and control characters, cut to `limit` bytes.""" + text = _FORBIDDEN.sub(" ", str(value)).strip() + text = re.sub(r" {2,}", " ", text) + if limit is not None: + raw = text.encode("utf-8")[:limit] + text = raw.decode("utf-8", "ignore").rstrip() + return text + + +def fmt_number(value) -> str: + v = float(value) + if v == int(v): + return str(int(v)) + return f"{v:.3f}".rstrip("0").rstrip(".") + + +def encode_body(fields: dict) -> str: + lines = [] + for key, value in fields.items(): + if value is None: + continue + if not re.fullmatch(r"[a-z_]+", key): + raise ValueError(f"invalid key {key!r}") + if isinstance(value, bool): + value = "1" if value else "0" + elif isinstance(value, (int, float)): + value = fmt_number(value) + else: + value = clean_text(value, TEXT_LIMITS.get(key)) + lines.append(f"{key}={value}") + return "\n".join(lines) + "\n" + + +def spoolman_id_of(cc2_spool: dict) -> int | None: + note = str(cc2_spool.get("note") or "") + m = re.search(r"\bspoolman:(\d+)\b", note) + return int(m.group(1)) if m else None + + +def slot_spool_id(slot: dict) -> str | None: + """CC2 Control spool ids are strings; an empty string means the slot holds no spool.""" + s = slot.get("spool") + if isinstance(s, dict): + s = s.get("id") + s = "" if s is None else str(s) + return s or None + + +class CC2Control: + def __init__(self, base_url: str, timeout: float = 8.0, transport: httpx.BaseTransport | None = None): + parts = urlsplit(base_url) + self.host_header = parts.netloc + self.client = httpx.Client( + base_url=base_url.rstrip("/"), timeout=timeout, transport=transport, + headers={"Host": self.host_header, "Accept": "application/json"}, + ) + # httpx never adds Origin by itself; make sure nothing else does either. + self.client.headers.pop("Origin", None) + + def close(self): + self.client.close() + + def _get(self, path: str) -> dict: + try: + r = self.client.get(path) + except httpx.HTTPError as e: + raise CC2Error(f"CC2 Control nicht erreichbar ({type(e).__name__})") from None + if r.status_code != 200: + raise CC2Error(f"CC2 Control {path}: HTTP {r.status_code}") + try: + return r.json() + except ValueError: + raise CC2Error(f"CC2 Control {path}: keine JSON-Antwort") from None + + def _post(self, path: str, fields: dict | str) -> dict: + # /api/spools/enable takes the bare word "on"/"off" as its body. + body = fields if isinstance(fields, str) else encode_body(fields) + try: + r = self.client.post( + path, content=body.encode("utf-8"), + headers={"X-CC2-Request": "1", "Content-Type": "text/plain; charset=utf-8"}, + ) + except httpx.HTTPError as e: + raise CC2Error(f"CC2 Control nicht erreichbar ({type(e).__name__})") from None + if r.status_code not in (200, 201): + detail = "" + try: + detail = str(r.json().get("error") or "")[:200] + except ValueError: + pass + raise CC2Error(f"CC2 Control {path}: HTTP {r.status_code} {detail}".strip()) + try: + return r.json() + except ValueError: + return {} + + def canvas(self) -> dict: + return self._get("/api/canvas") + + def printer(self) -> dict: + return self._get("/api/printer") + + def spools(self) -> dict: + return self._get("/api/spools") + + def save_spool(self, fields: dict) -> dict: + return self._post("/api/spools/save", fields) + + def assign(self, slot: int, spool_id: int | None) -> dict: + return self._post("/api/spools/assign", {"slot": slot, "spool": "" if spool_id is None else spool_id}) + + def enable(self, on: bool) -> dict: + return self._post("/api/spools/enable", "on" if on else "off") + + +def tray_statuses(canvas: dict) -> dict[int, int]: + """Slot -> tray status (0 = no filament tip at the feeder, 1 = loaded, 2 = feeding).""" + try: + trays = canvas["telemetry"]["result"]["canvas_info"]["canvas_list"][0]["tray_list"] + except (KeyError, IndexError, TypeError): + return {} + out = {} + for i, tray in enumerate(trays or []): + if not isinstance(tray, dict): + continue + slot = tray.get("tray_id", i) + try: + out[int(slot)] = int(tray.get("status")) + except (TypeError, ValueError): + continue + return out diff --git a/app/materials.py b/app/materials.py new file mode 100644 index 0000000..5c1c0ae --- /dev/null +++ b/app/materials.py @@ -0,0 +1,136 @@ +"""Elegoo Canvas material and subtype codes. + +Taken verbatim from cjtallman/canvas-rfid-tool (index.html, MIT License, +Copyright (c) 2026 cjtallman, see THIRD_PARTY_NOTICES.md), which in turn credits +DnG-Crafts/ELG-RFID and Savion/elegoo-rfid-editor. +""" + +from __future__ import annotations + +import re + +# name -> (4-byte material code, subtype family byte, suggested nozzle min/max) +MATERIALS: dict[str, dict] = { + "PLA": {"code": 0x00807665, "family": 0x00, "temps": (190, 230)}, + "PETG": {"code": 0x80698471, "family": 0x01, "temps": (230, 260)}, + "ABS": {"code": 0x00656683, "family": 0x02, "temps": (240, 270)}, + "TPU": {"code": 0x00848085, "family": 0x03, "temps": (210, 240)}, + "PA": {"code": 0x00008065, "family": 0x04, "temps": (260, 290)}, + "CPE": {"code": 0x00678069, "family": 0x05, "temps": (240, 270)}, + "PC": {"code": 0x00008067, "family": 0x06, "temps": (260, 290)}, + "PVA": {"code": 0x00808665, "family": 0x07, "temps": (190, 220)}, + "ASA": {"code": 0x00658365, "family": 0x08, "temps": (240, 270)}, + "BVOH": {"code": 0x42564F48, "family": 0x09, "temps": (190, 220)}, + "EVA": {"code": 0x00455641, "family": 0x0A, "temps": (190, 220)}, + "HIPS": {"code": 0x48495053, "family": 0x0B, "temps": (230, 250)}, + "PP": {"code": 0x00005050, "family": 0x0C, "temps": (220, 250)}, + "PPA": {"code": 0x00505041, "family": 0x0D, "temps": (280, 310)}, + "PPS": {"code": 0x00505053, "family": 0x0E, "temps": (300, 330)}, +} + +# (code, name, hidden on the CC2 display) +SUBTYPES: list[tuple[int, str, bool]] = [ + (0x0000, "PLA", False), (0x0001, "PLA+", False), (0x0002, "PLA Pro", False), + (0x0003, "PLA Silk", False), (0x0004, "PLA-CF", False), (0x0005, "PLA Carbon", True), + (0x0006, "PLA Matte", False), (0x0007, "PLA Fluo", False), (0x0008, "PLA Wood", False), + (0x0009, "PLA Basic", False), (0x000A, "RAPID PLA+", False), (0x000B, "PLA Marble", False), + (0x000C, "PLA Galaxy", False), (0x000D, "PLA Red Copper", False), (0x000E, "PLA Sparkle", True), + (0x0100, "PETG", False), (0x0101, "PETG-CF", False), (0x0102, "PETG-GF", False), + (0x0103, "PETG Pro", True), (0x0104, "PETG Translucent", False), (0x0105, "RAPID PETG", False), + (0x0200, "ABS", False), (0x0201, "ABS-GF", True), + (0x0300, "TPU", False), (0x0301, "TPU 95A", False), (0x0302, "RAPID TPU 95A", False), + (0x0400, "PA", False), (0x0401, "PA-CF", True), (0x0403, "PAHT-CF", False), + (0x0404, "PA6", True), (0x0405, "PA6-CF", True), (0x0406, "PA12", True), (0x0407, "PA12-CF", True), + (0x0500, "CPE", False), + (0x0600, "PC", False), (0x0601, "PCTG", True), (0x0602, "PC-FR", False), + (0x0700, "PVA", False), (0x0800, "ASA", False), (0x0900, "BVOH", False), (0x0A00, "EVA", False), + (0x0B00, "HIPS", False), + (0x0C00, "PP", False), (0x0C01, "PP-CF", True), (0x0C02, "PP-GF", True), + (0x0D00, "PPA", False), (0x0D01, "PPA-CF", True), (0x0D02, "PPA-GF", True), + (0x0E00, "PPS", False), (0x0E02, "PPS-CF", True), +] + +SUBTYPE_BY_CODE = {code: name for code, name, _ in SUBTYPES} +MATERIAL_BY_CODE = {m["code"]: name for name, m in MATERIALS.items()} + + +def material_of_subtype(code: int) -> str | None: + family = code >> 8 + for name, m in MATERIALS.items(): + if m["family"] == family: + return name + return None + + +def subtypes_for(material: str) -> list[tuple[int, str, bool]]: + family = MATERIALS[material]["family"] + return [s for s in SUBTYPES if s[0] >> 8 == family] + + +def _tokens(text: str) -> set[str]: + text = text.upper().replace("+", " PLUS ") + return set(re.findall(r"[A-Z0-9]+", text)) + + +# Spoolman material spellings that do not name the Elegoo base material directly. +_ALIASES = {"PCTG": "PC", "PA6": "PA", "PA12": "PA", "PAHT": "PA", "NYLON": "PA", "PET": "PETG"} + +_MODIFIERS = [ + # (tokens that must all be present, subtype name suffix) + ({"RAPID", "PLUS"}, "RAPID {m}+"), + ({"RAPID"}, "RAPID {m}"), + ({"95A"}, "{m} 95A"), + ({"SILK"}, "{m} Silk"), + ({"MATTE"}, "{m} Matte"), + ({"MATT"}, "{m} Matte"), + ({"CF"}, "{m}-CF"), + ({"CARBON"}, "{m}-CF"), + ({"GF"}, "{m}-GF"), + ({"WOOD"}, "{m} Wood"), + ({"MARBLE"}, "{m} Marble"), + ({"GALAXY"}, "{m} Galaxy"), + ({"SPARKLE"}, "{m} Sparkle"), + ({"FLUO"}, "{m} Fluo"), + ({"TRANSLUCENT"}, "{m} Translucent"), + ({"TRANSPARENT"}, "{m} Translucent"), + ({"PRO"}, "{m} Pro"), + ({"BASIC"}, "{m} Basic"), + ({"PLUS"}, "{m}+"), +] + + +def suggest(material: str | None, name: str | None = "") -> dict: + """Suggest Elegoo material, subtype and nozzle temps from Spoolman text.""" + material = material or "" + tokens = _tokens(f"{material} {name or ''}") + mat_tokens = _tokens(material) + base = None + # Prefer the Spoolman material field; fall back to the filament name. + for pool in (mat_tokens, tokens): + for tok in sorted(pool, key=len, reverse=True): + cand = _ALIASES.get(tok, tok) + if cand in MATERIALS: + base = cand + break + if tok.startswith("PLA"): # e.g. "PLAPLUS" + base = "PLA" + break + if base: + break + if base is None: + base = "PLA" + subtype = MATERIALS[base]["family"] << 8 + names = {n.upper(): c for c, n, _ in subtypes_for(base)} + # Exact subtype names first (e.g. "PCTG", "PAHT-CF"). + for tok in tokens: + if tok in names and tok != base: + subtype = names[tok] + break + else: + for needed, pattern in _MODIFIERS: + if needed <= tokens: + cand = pattern.format(m=base).upper() + if cand in names: + subtype = names[cand] + break + return {"material": base, "subtype": subtype, "temps": MATERIALS[base]["temps"]} diff --git a/app/spoolman.py b/app/spoolman.py new file mode 100644 index 0000000..2fd1333 --- /dev/null +++ b/app/spoolman.py @@ -0,0 +1,94 @@ +"""Minimal Spoolman REST client (/api/v1).""" + +from __future__ import annotations + +import re + +import httpx + + +class SpoolmanError(RuntimeError): + pass + + +class SpoolmanNotFound(SpoolmanError): + pass + + +class Spoolman: + def __init__(self, base_url: str, timeout: float = 8.0, transport: httpx.BaseTransport | None = None): + self.client = httpx.Client(base_url=base_url.rstrip("/") + "/api/v1", timeout=timeout, transport=transport) + + def close(self): + self.client.close() + + def _request(self, method: str, path: str, **kw): + try: + r = self.client.request(method, path, **kw) + except httpx.HTTPError as e: + raise SpoolmanError(f"Spoolman nicht erreichbar ({type(e).__name__})") from None + if r.status_code == 404: + raise SpoolmanNotFound(f"Spoolman {path}: nicht gefunden") + if r.status_code >= 400: + raise SpoolmanError(f"Spoolman {method} {path}: HTTP {r.status_code}") + try: + return r.json() + except ValueError: + raise SpoolmanError(f"Spoolman {path}: keine JSON-Antwort") from None + + def spool(self, spool_id: int) -> dict: + return self._request("GET", f"/spool/{int(spool_id)}") + + def spools(self, archived: bool = False) -> list[dict]: + return self._request("GET", "/spool", params={"allow_archived": str(archived).lower()}) + + def set_location(self, spool_id: int, location: str) -> dict: + return self._request("PATCH", f"/spool/{int(spool_id)}", json={"location": location}) + + def use(self, spool_id: int, grams: float) -> dict: + return self._request("PUT", f"/spool/{int(spool_id)}/use", json={"use_weight": round(float(grams), 3)}) + + +def vendor_name(spool: dict) -> str: + return ((spool.get("filament") or {}).get("vendor") or {}).get("name") or "" + + +def label(spool: dict) -> str: + """lot_nr like SUNLU001; otherwise built from vendor + id the same way.""" + lot = (spool.get("lot_nr") or "").strip() + if lot: + return lot + vendor = re.sub(r"[^A-Z0-9]", "", vendor_name(spool).upper()) or "SPOOL" + return f"{vendor}{int(spool['id']):03d}" + + +def color_hex(spool: dict) -> str | None: + fil = spool.get("filament") or {} + color = fil.get("color_hex") + if not color and fil.get("multi_color_hexes"): + color = str(fil["multi_color_hexes"]).split(",")[0] + if not color: + return None + color = color.strip().lstrip("#").upper()[:6] + return color if re.fullmatch(r"[0-9A-F]{6}", color) else None + + +def summary(spool: dict) -> dict: + fil = spool.get("filament") or {} + return { + "id": spool.get("id"), + "label": label(spool), + "vendor": vendor_name(spool), + "name": fil.get("name") or "", + "material": fil.get("material") or "", + "color": color_hex(spool), + "diameter": fil.get("diameter"), + "density": fil.get("density"), + "remaining": spool.get("remaining_weight"), + "initial": spool.get("initial_weight") or fil.get("weight"), + "spool_weight": spool.get("spool_weight") or fil.get("spool_weight"), + "extruder_temp": fil.get("settings_extruder_temp"), + "location": spool.get("location") or "", + "archived": bool(spool.get("archived")), + "used_weight": spool.get("used_weight"), + } diff --git a/app/sshsource.py b/app/sshsource.py new file mode 100644 index 0000000..1eeff2c --- /dev/null +++ b/app/sshsource.py @@ -0,0 +1,245 @@ +"""Read /opt/usr/canvas_record_info.json from the printer over SSH. + +The printer's authorized_keys entry forces `cat /opt/usr/canvas_record_info.json` +for this key, so every exec returns that file whatever command is sent. One +transport is kept open and a new channel is opened per read; nothing keeps +running on the printer between reads. +""" + +from __future__ import annotations + +import base64 +import hashlib +import io +import json +import os +import socket +import threading +from pathlib import Path + +import paramiko + +from .storage import atomic_write, atomic_write_json, read_json + +REMOTE_COMMAND = "cat /opt/usr/canvas_record_info.json" +MAX_RECORD_BYTES = 256 * 1024 + + +class SSHError(RuntimeError): + pass + + +class HostKeyUnknown(SSHError): + def __init__(self, key: paramiko.PKey): + self.key = key + super().__init__(f"Host-Key des Druckers noch nicht bestätigt ({fingerprint(key)})") + + +class HostKeyMismatch(SSHError): + def __init__(self, key: paramiko.PKey, expected: str): + self.key = key + super().__init__( + f"Host-Key des Druckers hat sich geändert: {fingerprint(key)}, erwartet {expected}. " + "Verbindung abgelehnt." + ) + + +def fingerprint(key: paramiko.PKey) -> str: + digest = hashlib.sha256(key.asbytes()).digest() + return "SHA256:" + base64.b64encode(digest).decode().rstrip("=") + + +def load_private_key(text: str) -> paramiko.PKey: + """Parse an OpenSSH/PEM private key (no passphrase).""" + text = text.replace("\r\n", "\n").strip() + "\n" + errors = [] + for cls in (paramiko.Ed25519Key, paramiko.ECDSAKey, paramiko.RSAKey): + try: + return cls.from_private_key(io.StringIO(text)) + except paramiko.PasswordRequiredException: + raise SSHError("Der Schlüssel ist passwortgeschützt; bitte einen Schlüssel ohne Passphrase verwenden") from None + except (paramiko.SSHException, ValueError, IndexError, TypeError) as e: + errors.append(type(e).__name__) + raise SSHError("Kein gültiger privater SSH-Schlüssel (ed25519, ecdsa oder rsa)") + + +class KeyStore: + """Private key and pinned printer host key in /ssh.""" + + def __init__(self, data_dir: Path): + self.dir = data_dir / "ssh" + self.key_path = self.dir / "id_key" + self.host_path = self.dir / "host_key.json" + + def has_key(self) -> bool: + return self.key_path.exists() + + def save_key(self, text: str) -> paramiko.PKey: + text = text.replace("\r\n", "\n").strip() + "\n" + key = load_private_key(text) + self.dir.mkdir(parents=True, exist_ok=True) + os.chmod(self.dir, 0o700) + atomic_write(self.key_path, text.encode(), mode=0o600) + return key + + def delete_key(self) -> None: + try: + self.key_path.unlink() + except FileNotFoundError: + pass + + def load_key(self) -> paramiko.PKey: + try: + os.chmod(self.key_path, 0o600) + return load_private_key(self.key_path.read_text()) + except FileNotFoundError: + raise SSHError("Kein SSH-Schlüssel hinterlegt") from None + + def key_info(self) -> dict | None: + if not self.has_key(): + return None + try: + key = self.load_key() + except SSHError as e: + return {"error": str(e)} + return { + "type": key.get_name(), + "fingerprint": fingerprint(key), + "public": f"{key.get_name()} {key.get_base64()} cc2-spoolman-bridge", + } + + def host_key(self) -> dict | None: + return read_json(self.host_path, None) + + def pin_host_key(self, host: str, key: paramiko.PKey) -> None: + self.dir.mkdir(parents=True, exist_ok=True) + atomic_write_json(self.host_path, { + "host": host, "type": key.get_name(), "key": key.get_base64(), "fingerprint": fingerprint(key), + }, mode=0o600) + + def forget_host_key(self) -> None: + try: + self.host_path.unlink() + except FileNotFoundError: + pass + + +class CanvasRecordReader: + def __init__(self, keys: KeyStore, host: str, port: int = 22, user: str = "root", timeout: float = 10.0): + self.keys = keys + self.host, self.port, self.user, self.timeout = host, port, user, timeout + self._transport: paramiko.Transport | None = None + self._lock = threading.Lock() + + def close(self) -> None: + with self._lock: + self._close() + + def _close(self) -> None: + if self._transport is not None: + try: + self._transport.close() + except Exception: + pass + self._transport = None + + def _connect(self, accept_unknown: bool = False) -> paramiko.PKey: + """Open and authenticate a transport; returns the server's host key.""" + key = self.keys.load_key() + try: + sock = socket.create_connection((self.host, self.port), timeout=self.timeout) + except OSError as e: + raise SSHError(f"Drucker {self.host}:{self.port} nicht erreichbar ({e.strerror or type(e).__name__})") from None + t = paramiko.Transport(sock) + t.banner_timeout = t.auth_timeout = self.timeout + try: + t.start_client(timeout=self.timeout) + server_key = t.get_remote_server_key() + pinned = self.keys.host_key() + if pinned is None: + if not accept_unknown: + raise HostKeyUnknown(server_key) + elif pinned.get("type") != server_key.get_name() or pinned.get("key") != server_key.get_base64(): + raise HostKeyMismatch(server_key, pinned.get("fingerprint", "?")) + t.auth_publickey(self.user, key) + if not t.is_authenticated(): + raise SSHError("SSH-Anmeldung fehlgeschlagen") + t.set_keepalive(30) + except SSHError: + t.close() + raise + except paramiko.AuthenticationException: + t.close() + raise SSHError("SSH-Anmeldung mit dem hinterlegten Schlüssel abgelehnt") from None + except (paramiko.SSHException, OSError, EOFError) as e: + t.close() + raise SSHError(f"SSH-Fehler ({type(e).__name__})") from None + self._transport = t + return server_key + + def _exec(self) -> bytes: + assert self._transport is not None + chan = self._transport.open_session(timeout=self.timeout) + try: + chan.settimeout(self.timeout) + chan.exec_command(REMOTE_COMMAND) + chunks, size = [], 0 + while True: + data = chan.recv(32768) + if not data: + break + size += len(data) + if size > MAX_RECORD_BYTES: + raise SSHError("Antwort des Druckers ist zu groß") + chunks.append(data) + status = chan.recv_exit_status() + if status not in (0, -1): + raise SSHError(f"Befehl auf dem Drucker endete mit Status {status}") + return b"".join(chunks) + finally: + chan.close() + + def read(self) -> dict: + """Read and parse the canvas record; reconnects once on a dead transport.""" + with self._lock: + for attempt in (0, 1): + if self._transport is None or not self._transport.is_active(): + self._close() + self._connect() + try: + raw = self._exec() + break + except (paramiko.SSHException, OSError, EOFError, socket.timeout) as e: + self._close() + if attempt: + raise SSHError(f"SSH-Lesefehler ({type(e).__name__})") from None + return parse_record_json(raw) + + def test(self) -> dict: + """Fresh connection; an unknown host key is reported, not trusted.""" + with self._lock: + self._close() + try: + server_key = self._connect(accept_unknown=True) + raw = self._exec() + except HostKeyMismatch: + raise + finally: + self._close() + pinned = self.keys.host_key() + return { + "host_key": {"type": server_key.get_name(), "key": server_key.get_base64(), + "fingerprint": fingerprint(server_key)}, + "host_key_pinned": pinned is not None, + "record": parse_record_json(raw), + } + + +def parse_record_json(raw: bytes) -> dict: + try: + data = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, ValueError): + raise SSHError("canvas_record_info.json ist kein gültiges JSON") from None + if not isinstance(data, dict): + raise SSHError("canvas_record_info.json hat ein unerwartetes Format") + return data diff --git a/app/storage.py b/app/storage.py new file mode 100644 index 0000000..aadcbfd --- /dev/null +++ b/app/storage.py @@ -0,0 +1,161 @@ +"""Settings and state persistence in the data volume (atomic JSON writes).""" + +from __future__ import annotations + +import json +import os +import tempfile +import threading +from dataclasses import asdict, dataclass, fields +from pathlib import Path + + +def atomic_write(path: Path, data: bytes, mode: int = 0o644) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.") + try: + os.fchmod(fd, mode) + with os.fdopen(fd, "wb") as f: + f.write(data) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp, path) + except BaseException: + try: + os.unlink(tmp) + except FileNotFoundError: + pass + raise + try: + dfd = os.open(path.parent, os.O_DIRECTORY) + try: + os.fsync(dfd) + finally: + os.close(dfd) + except OSError: + pass + + +def atomic_write_json(path: Path, obj, mode: int = 0o644) -> None: + atomic_write(path, json.dumps(obj, indent=2, sort_keys=True, ensure_ascii=False).encode(), mode) + + +def read_json(path: Path, default): + try: + return json.loads(path.read_text()) + except FileNotFoundError: + return default + + +@dataclass +class Settings: + printer_host: str = "10.10.2.202" + ssh_port: int = 22 + ssh_user: str = "root" + cc2_url: str = "http://10.10.2.202:8081" + spoolman_url: str = "http://10.10.6.10:7912" + id_offset: int = 10000 + poll_interval: float = 5.0 + location_slot: str = "CC2 Slot {n}" + location_storage: str = "Lager" + empty_grace: float = 20.0 # seconds with tray status 0 before a slot counts as empty + settle_seconds: float = 60.0 # `used` must be unchanged this long before booking + dry_run: bool = True + + # environment variable -> field (only used as defaults before settings.json exists) + ENV = { + "PRINTER_HOST": "printer_host", "SSH_PORT": "ssh_port", "SSH_USER": "ssh_user", + "CC2_URL": "cc2_url", "SPOOLMAN_URL": "spoolman_url", "ID_OFFSET": "id_offset", + "POLL_INTERVAL": "poll_interval", "DRY_RUN": "dry_run", + } + + @classmethod + def from_dict(cls, d: dict) -> "Settings": + s = cls() + s.update(d) + return s + + def update(self, d: dict) -> None: + for f in fields(self): + if f.name in d and d[f.name] is not None: + setattr(self, f.name, _coerce(f.type, d[f.name], f.name)) + self.validate() + + def validate(self) -> None: + if not 1 <= self.ssh_port <= 65535: + raise ValueError("SSH-Port ungültig") + if not 0 <= self.id_offset <= 65000: + raise ValueError("ID-Offset muss zwischen 0 und 65000 liegen") + if not 1 <= self.poll_interval <= 300: + raise ValueError("Abfrage-Intervall muss zwischen 1 und 300 s liegen") + if "{n}" not in self.location_slot: + raise ValueError("Ortsname für Slots muss {n} enthalten") + for name in ("cc2_url", "spoolman_url"): + v = getattr(self, name) + if not v.startswith(("http://", "https://")): + raise ValueError(f"{name} muss mit http:// oder https:// beginnen") + setattr(self, name, v.rstrip("/")) + if not self.location_storage.strip(): + raise ValueError("Ortsname für das Lager fehlt") + + def to_dict(self) -> dict: + return asdict(self) + + def slot_location(self, slot: int) -> str: + return self.location_slot.replace("{n}", str(slot + 1)) + + +def _coerce(typ, value, name): + typ = typ if isinstance(typ, str) else typ.__name__ + try: + if typ == "bool": + if isinstance(value, str): + return value.strip().lower() in ("1", "true", "yes", "on", "ja") + return bool(value) + if typ == "int": + return int(value) + if typ == "float": + return float(value) + return str(value).strip() + except (TypeError, ValueError): + raise ValueError(f"Ungültiger Wert für {name}") from None + + +class SettingsStore: + def __init__(self, data_dir: Path, env: dict | None = None): + self.path = data_dir / "settings.json" + self._lock = threading.Lock() + stored = read_json(self.path, None) + if stored is None: + env = os.environ if env is None else env + defaults = {f: env[k] for k, f in Settings.ENV.items() if env.get(k)} + self.settings = Settings.from_dict(defaults) + else: + self.settings = Settings.from_dict(stored) + + def get(self) -> Settings: + with self._lock: + return Settings.from_dict(self.settings.to_dict()) + + def save(self, changes: dict) -> Settings: + with self._lock: + new = Settings.from_dict({**self.settings.to_dict(), **changes}) + atomic_write_json(self.path, new.to_dict()) + self.settings = new + return Settings.from_dict(new.to_dict()) + + +class StateStore: + """Bridge state: slot mappings, booked consumption, last record.""" + + def __init__(self, data_dir: Path): + self.path = data_dir / "state.json" + self.data = read_json(self.path, {}) + self.data.setdefault("version", 1) + self.data.setdefault("slots", {}) + self.data.setdefault("booked", {}) + self.data.setdefault("locations", {}) + self.data.setdefault("last_record", None) + + def save(self) -> None: + atomic_write_json(self.path, self.data) diff --git a/app/tagcodec.py b/app/tagcodec.py new file mode 100644 index 0000000..1509bb8 --- /dev/null +++ b/app/tagcodec.py @@ -0,0 +1,267 @@ +"""Encode/decode Elegoo Canvas NFC tags (NTAG213/215/216) and canvas records. + +Page numbers are hex, as used by NFC Tools ("A2:PP:XXXXXXXX"). All values are +big-endian. Pages 0-3 (UID, lock bits, capability container) are never written. +""" + +from __future__ import annotations + +import colorsys +import math +import re +from dataclasses import dataclass, field + +from .materials import MATERIAL_BY_CODE, MATERIALS, SUBTYPE_BY_CODE, material_of_subtype + +DEFAULT_URL = "https://www.elegoo.com" +DEFAULT_ID_OFFSET = 10000 +MAX_WEIGHT = 0xFFFF + +NDEF_FIRST_PAGE = 0x04 +NDEF_LAST_PAGE = 0x0F # canvas data starts at page 0x10 +CANVAS_PAGES = range(0x10, 0x19) + +PAGE_18_DEFAULT = 0x0036C800 # copied from original Elegoo spools; meaning unknown + +# Display palette from /opt/inst/filament_color_set.json +PALETTE = [ + "FFFFFF", "FFF242", "DBF47A", "09CC3A", "077747", "0B6283", "0BE2A0", "74D9F3", + "48A7FA", "2850DF", "433089", "A03BF7", "F32FF8", "D4B1DD", "F95D77", "F72221", + "7C4C00", "F88D36", "FCEBD7", "D2C5A3", "AF7832", "898989", "BCBCBC", "000000", +] +BLUE = "2850DF" + +_URI_PREFIXES = [ + ("https://www.", 0x02), ("http://www.", 0x01), ("https://", 0x04), ("http://", 0x03), +] + + +class TagError(ValueError): + pass + + +@dataclass +class TagData: + spool_id: int + color: str # RRGGBB + material: str = "PLA" + subtype: int = 0x0001 + temp_min: int = 190 + temp_max: int = 230 + diameter: float = 1.75 + finish: int = 0xFF + filament_code: int = 0x0000 + id_offset: int = DEFAULT_ID_OFFSET + url: str = DEFAULT_URL + warnings: list[str] = field(default_factory=list) + + +def normalize_hex_color(value: str | None) -> str | None: + if not value: + return None + v = value.strip().lstrip("#").upper() + if re.fullmatch(r"[0-9A-F]{8}", v): # RRGGBBAA + v = v[:6] + if re.fullmatch(r"[0-9A-F]{3}", v): + v = "".join(c * 2 for c in v) + return v if re.fullmatch(r"[0-9A-F]{6}", v) else None + + +# ---------- NDEF ---------- + +def encode_ndef_url(url: str) -> bytes: + """Lock-control TLV + NDEF URI record + terminator, as on the original tags.""" + prefix_code, rest = 0x00, url + for prefix, code in _URI_PREFIXES: + if url.startswith(prefix): + prefix_code, rest = code, url[len(prefix):] + break + body = rest.encode("utf-8") + payload = bytes([prefix_code]) + body + if len(payload) > 255: + raise TagError("URL zu lang") + record = bytes([0xD1, 0x01, len(payload), 0x55]) + payload + data = bytes([0x01, 0x03, 0xA0, 0x0C, 0x34, 0x03, len(record)]) + record + b"\xFE" + max_len = (NDEF_LAST_PAGE - NDEF_FIRST_PAGE + 1) * 4 + if len(data) > max_len: + raise TagError(f"URL zu lang: {len(data)} Byte, maximal {max_len} (Seiten 04-0F)") + pad = (-len(data)) % 4 + return data + b"\x00" * pad + + +def decode_ndef_url(data: bytes) -> str | None: + i = 0 + while i < len(data): + t = data[i] + if t == 0x00: + i += 1 + continue + if t == 0xFE or i + 1 >= len(data): + return None + length = data[i + 1] + value = data[i + 2:i + 2 + length] + if t == 0x03 and len(value) >= 5 and value[3] == 0x55: + plen = value[2] + payload = value[4:4 + plen] + prefix = {code: p for p, code in _URI_PREFIXES}.get(payload[0], "") + return prefix + payload[1:].decode("utf-8", "replace") + i += 2 + length + return None + + +# ---------- canvas pages ---------- + +def encode_pages(tag: TagData) -> dict[int, bytes]: + if tag.material not in MATERIALS: + raise TagError(f"Unbekanntes Material {tag.material}") + if tag.subtype not in SUBTYPE_BY_CODE: + raise TagError(f"Unbekannter Untertyp 0x{tag.subtype:04X}") + if material_of_subtype(tag.subtype) != tag.material: + raise TagError(f"Untertyp {SUBTYPE_BY_CODE[tag.subtype]} passt nicht zu {tag.material}") + color = normalize_hex_color(tag.color) + if color is None: + raise TagError("Farbe muss RRGGBB sein") + weight = tag.spool_id + tag.id_offset + if tag.spool_id < 0 or weight > MAX_WEIGHT: + raise TagError(f"Spulen-ID {tag.spool_id} passt nicht in 2 Byte (max. {MAX_WEIGHT - tag.id_offset})") + for t in (tag.temp_min, tag.temp_max): + if not 0 <= t <= 500: + raise TagError("Düsentemperatur außerhalb 0-500 °C") + if tag.temp_min > tag.temp_max: + raise TagError("Düse min ist größer als max") + diameter = round(tag.diameter * 100) + if not 0 < diameter <= 0xFFFF: + raise TagError("Ungültiger Durchmesser") + + pages: dict[int, bytes] = {} + ndef = encode_ndef_url(tag.url) + for i in range(0, len(ndef), 4): + pages[NDEF_FIRST_PAGE + i // 4] = ndef[i:i + 4] + pages[0x10] = bytes([0x36, 0xEE, 0xEE, 0xEE]) + pages[0x11] = bytes([0xEE]) + tag.filament_code.to_bytes(2, "big") + b"\x00" + pages[0x12] = MATERIALS[tag.material]["code"].to_bytes(4, "big") + pages[0x13] = tag.subtype.to_bytes(2, "big") + b"\x00\x00" + pages[0x14] = bytes.fromhex(color) + bytes([tag.finish & 0xFF]) + pages[0x15] = tag.temp_min.to_bytes(2, "big") + tag.temp_max.to_bytes(2, "big") + pages[0x16] = b"\x00\x00\x00\x00" + pages[0x17] = diameter.to_bytes(2, "big") + weight.to_bytes(2, "big") + pages[0x18] = PAGE_18_DEFAULT.to_bytes(4, "big") + return pages + + +def pages_to_commands(pages: dict[int, bytes]) -> str: + for p in pages: + if p < 4: + raise TagError("Seiten 0-3 dürfen nicht geschrieben werden") + return ",".join(f"A2:{p:02X}:{pages[p].hex().upper()}" for p in sorted(pages)) + + +def encode_commands(tag: TagData) -> str: + return pages_to_commands(encode_pages(tag)) + + +def parse_commands(text: str) -> dict[int, bytes]: + pages: dict[int, bytes] = {} + for part in re.split(r"[,\s]+", text.strip()): + if not part: + continue + m = re.fullmatch(r"A2:([0-9A-Fa-f]{2}):([0-9A-Fa-f]{8})", part) + if not m: + raise TagError(f"Ungültiger Befehl: {part}") + pages[int(m.group(1), 16)] = bytes.fromhex(m.group(2)) + return pages + + +def decode_pages(pages: dict[int, bytes], id_offset: int = DEFAULT_ID_OFFSET) -> dict: + def page(n: int) -> bytes: + return pages.get(n, b"\x00\x00\x00\x00") + + ndef = b"".join(page(p) for p in range(NDEF_FIRST_PAGE, NDEF_LAST_PAGE + 1)) + mat_code = int.from_bytes(page(0x12), "big") + subtype = int.from_bytes(page(0x13)[:2], "big") + weight = int.from_bytes(page(0x17)[2:], "big") + return { + "url": decode_ndef_url(ndef), + "header": page(0x10)[0], + "manufacturer": page(0x10)[1:].hex().upper() + page(0x11)[:1].hex().upper(), + "filament_code": int.from_bytes(page(0x11)[1:3], "big"), + "material": MATERIAL_BY_CODE.get(mat_code, f"0x{mat_code:08X}"), + "subtype": subtype, + "subtype_name": SUBTYPE_BY_CODE.get(subtype, "?"), + "color": page(0x14)[:3].hex().upper(), + "finish": page(0x14)[3], + "temp_min": int.from_bytes(page(0x15)[:2], "big"), + "temp_max": int.from_bytes(page(0x15)[2:], "big"), + "diameter": int.from_bytes(page(0x17)[:2], "big") / 100, + "weight": weight, + "spool_id": spool_id_from_weight(weight, id_offset), + } + + +# ---------- canvas record (/opt/usr/canvas_record_info.json) ---------- + +def spool_id_from_weight(weight, id_offset: int = DEFAULT_ID_OFFSET) -> int | None: + """weight >= offset encodes a Spoolman spool id; smaller values are real weights.""" + try: + w = int(float(str(weight).strip())) + except (TypeError, ValueError): + return None + if w < id_offset or w > MAX_WEIGHT: + return None + return w - id_offset + + +def parse_record(record: dict, id_offset: int = DEFAULT_ID_OFFSET) -> list[dict]: + """Return 4 slot dicts (slot 0-3) with the raw channel and the decoded spool id.""" + slots = [] + for n in range(4): + ch = record.get(f"channels{n}") or {} + if not isinstance(ch, dict): + ch = {} + occupied = any(str(v).strip() for v in ch.values()) + slots.append({ + "slot": n, + "channel": ch, + "occupied": occupied, + "spool_id": spool_id_from_weight(ch.get("weight"), id_offset) if occupied else None, + }) + return slots + + +# ---------- colour suggestion ---------- + +def _srgb_to_lab(hexcolor: str) -> tuple[float, float, float]: + rgb = [int(hexcolor[i:i + 2], 16) / 255 for i in (0, 2, 4)] + lin = [c / 12.92 if c <= 0.04045 else ((c + 0.055) / 1.055) ** 2.4 for c in rgb] + r, g, b = lin + x = (0.4124 * r + 0.3576 * g + 0.1805 * b) / 0.95047 + y = 0.2126 * r + 0.7152 * g + 0.0722 * b + z = (0.0193 * r + 0.1192 * g + 0.9505 * b) / 1.08883 + + def f(t): + return t ** (1 / 3) if t > 216 / 24389 else (24389 / 27 * t + 16) / 116 + + fx, fy, fz = f(x), f(y), f(z) + return 116 * fy - 16, 500 * (fx - fy), 200 * (fy - fz) + + +def _lab_distance(a: str, b: str) -> float: + return math.dist(_srgb_to_lab(a), _srgb_to_lab(b)) + + +def _is_saturated_blue(hexcolor: str) -> bool: + r, g, b = (int(hexcolor[i:i + 2], 16) / 255 for i in (0, 2, 4)) + h, s, v = colorsys.rgb_to_hsv(r, g, b) + return 215 <= h * 360 <= 245 and s >= 0.6 and v >= 0.35 + + +def nearest_palette_color(hexcolor: str | None) -> str: + color = normalize_hex_color(hexcolor) + if color is None: + return "000000" + if color in PALETTE: + return color + if _is_saturated_blue(color): + # Lab distance lands on the violet entries for saturated blues. + return BLUE + return min(PALETTE, key=lambda p: _lab_distance(color, p)) diff --git a/app/templates/base.html b/app/templates/base.html new file mode 100644 index 0000000..ab8e2cf --- /dev/null +++ b/app/templates/base.html @@ -0,0 +1,131 @@ + + + + + +{% block title %}CC2 Bridge{% endblock %} + + + +
+

CC2 ↔ Spoolman

+ + +
+
+{% block content %}{% endblock %} +
+
+ +{% block script %}{% endblock %} + + diff --git a/app/templates/settings.html b/app/templates/settings.html new file mode 100644 index 0000000..2f5d698 --- /dev/null +++ b/app/templates/settings.html @@ -0,0 +1,156 @@ +{% extends "base.html" %} +{% block title %}CC2 Bridge – Einstellungen{% endblock %} +{% block content %} +

Einstellungen

+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ +
+
+
+ +

SSH-Schlüssel für den Drucker

+
+
Lade …
+
+ + +
+
+ + + +
+

Der Schlüssel wird mit Dateirechten 600 im Daten-Volume gespeichert und nie wieder angezeigt.

+
+ +

Verbindung testen

+
+
+
+
+
+
+{% endblock %} +{% block script %} + +{% endblock %} diff --git a/app/templates/status.html b/app/templates/status.html new file mode 100644 index 0000000..f42095f --- /dev/null +++ b/app/templates/status.html @@ -0,0 +1,97 @@ +{% extends "base.html" %} +{% block title %}CC2 Bridge – Status{% endblock %} +{% block content %} +
+

Canvas-Slots

+
Lade …
+

+

Protokoll

+
+ +
ZeitAktion
+
+{% endblock %} +{% block script %} + +{% endblock %} diff --git a/app/templates/tags.html b/app/templates/tags.html new file mode 100644 index 0000000..a0b62f9 --- /dev/null +++ b/app/templates/tags.html @@ -0,0 +1,171 @@ +{% extends "base.html" %} +{% block title %}CC2 Bridge – Tag-Generator{% endblock %} +{% block content %} + +

Spoolman-Spulen

+
+
+
+ +
+
+ +
IDLabelHerstellerMaterialFarbeRestOrt
Lade …
+
+
+ + +{% endblock %} +{% block script %} + +{% endblock %} diff --git a/app/web.py b/app/web.py new file mode 100644 index 0000000..8f1a573 --- /dev/null +++ b/app/web.py @@ -0,0 +1,305 @@ +"""Web UI and JSON API.""" + +from __future__ import annotations + +import base64 +import hmac +import logging +import os +import threading +from contextlib import asynccontextmanager +from pathlib import Path + +import paramiko +import segno +from fastapi import FastAPI, HTTPException, Request +from fastapi.responses import HTMLResponse, JSONResponse, Response +from fastapi.templating import Jinja2Templates + +from . import materials, tagcodec +from .bridge import Bridge +from .sshsource import CanvasRecordReader, HostKeyMismatch, KeyStore, SSHError, fingerprint +from .spoolman import SpoolmanError, summary +from .storage import SettingsStore + +log = logging.getLogger("web") +HERE = Path(__file__).parent +VERSION = os.environ.get("APP_VERSION", "dev") + + +def create_app(data_dir: Path | None = None, *, bridge: Bridge | None = None, start_loop: bool = True, + env: dict | None = None) -> FastAPI: + env = os.environ if env is None else env + data_dir = Path(data_dir or env.get("DATA_DIR", "/data")) + data_dir.mkdir(parents=True, exist_ok=True) + settings = bridge.settings_store if bridge else SettingsStore(data_dir, env) + keys = bridge.keys if bridge else KeyStore(data_dir) + bridge = bridge or Bridge(data_dir, settings, keys) + stop = threading.Event() + tested_host_keys: dict[str, object] = {} # fingerprint -> key from the last connection test + + @asynccontextmanager + async def lifespan(app): + thread = None + if start_loop: + thread = threading.Thread(target=bridge.run, args=(stop,), name="bridge", daemon=True) + thread.start() + yield + stop.set() + bridge.wake.set() + if thread: + thread.join(timeout=15) + + app = FastAPI(title="CC2 Spoolman Bridge", lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None) + app.state.bridge = bridge + templates = Jinja2Templates(directory=str(HERE / "templates")) + + user, password = env.get("BRIDGE_USER", ""), env.get("BRIDGE_PASSWORD", "") + + @app.middleware("http") + async def guard(request: Request, call_next): + if user and password and request.url.path != "/healthz": + ok = False + auth = request.headers.get("authorization", "") + if auth.lower().startswith("basic "): + try: + u, _, p = base64.b64decode(auth[6:]).decode("utf-8").partition(":") + ok = hmac.compare_digest(u, user) & hmac.compare_digest(p, password) + except (ValueError, UnicodeDecodeError): + ok = False + if not ok: + return Response(status_code=401, headers={"WWW-Authenticate": 'Basic realm="CC2 Bridge"'}) + # Writes only from our own pages: a custom header cannot be sent cross-site without CORS. + if request.method not in ("GET", "HEAD") and request.headers.get("x-bridge-request") != "1": + return JSONResponse({"error": "Fehlender X-Bridge-Request-Header"}, status_code=403) + response = await call_next(request) + response.headers["X-Content-Type-Options"] = "nosniff" + response.headers["Referrer-Policy"] = "no-referrer" + response.headers["X-Frame-Options"] = "DENY" + return response + + def page(request: Request, name: str, **ctx): + return templates.TemplateResponse(request, name, {"version": VERSION, "active": name, **ctx}) + + # ---------- pages ---------- + + @app.get("/", response_class=HTMLResponse) + def index(request: Request): + return page(request, "status.html") + + @app.get("/tags", response_class=HTMLResponse) + def tags(request: Request): + return page(request, "tags.html") + + @app.get("/settings", response_class=HTMLResponse) + def settings_page(request: Request): + return page(request, "settings.html") + + @app.get("/healthz") + def healthz(): + return {"ok": True, "version": VERSION, "last_poll": bridge.snapshot.get("last_poll")} + + # ---------- status ---------- + + @app.get("/api/status") + def status(): + return {**bridge.snapshot, "dry_run": settings.get().dry_run, "log": bridge.log.recent(200)} + + @app.post("/api/cc2/enable") + async def cc2_enable(request: Request): + body = await _json(request) + if body.get("confirm") is not True: + raise HTTPException(400, "Bestätigung fehlt") + bridge.commands.put(("enable", None)) + bridge.wake.set() + return {"ok": True, "dry_run": settings.get().dry_run} + + @app.post("/api/slots/{slot}/{action}") + def slot_action(slot: int, action: str): + if slot not in range(4) or action not in ("reassign", "release"): + raise HTTPException(404) + bridge.commands.put((action, slot)) + bridge.wake.set() + return {"ok": True} + + # ---------- settings ---------- + + @app.get("/api/settings") + def get_settings(): + return settings.get().to_dict() + + @app.post("/api/settings") + async def post_settings(request: Request): + body = await _json(request) + allowed = set(settings.get().to_dict()) + try: + s = settings.save({k: v for k, v in body.items() if k in allowed}) + except ValueError as e: + raise HTTPException(400, str(e)) from None + bridge.log.add("info", "Einstellungen gespeichert" + (" (Dry-Run aktiv)" if s.dry_run else " (Schreiben AKTIV)")) + bridge.wake.set() + return s.to_dict() + + # ---------- SSH ---------- + + @app.get("/api/ssh") + def ssh_info(): + return {"key": keys.key_info(), "host_key": keys.host_key()} + + @app.post("/api/ssh/key") + async def ssh_key(request: Request): + body = await _json(request) + text = body.get("key") + if not isinstance(text, str) or len(text) > 20000: + raise HTTPException(400, "Schlüssel fehlt") + try: + keys.save_key(text) + except SSHError as e: + raise HTTPException(400, str(e)) from None + bridge.log.add("info", "SSH-Schlüssel gespeichert") + if bridge.reader: + bridge.reader.close() + return {"key": keys.key_info()} + + @app.delete("/api/ssh/key") + def ssh_key_delete(): + keys.delete_key() + if bridge.reader: + bridge.reader.close() + bridge.log.add("info", "SSH-Schlüssel gelöscht") + return {"key": None} + + @app.post("/api/ssh/test") + def ssh_test(): + s = settings.get() + reader = CanvasRecordReader(keys, s.printer_host, s.ssh_port, s.ssh_user) + try: + result = reader.test() + except HostKeyMismatch as e: + return JSONResponse({"ok": False, "error": str(e), "host_key": {"fingerprint": fingerprint(e.key)}}, + status_code=200) + except SSHError as e: + return {"ok": False, "error": str(e)} + hk = result["host_key"] + if not result["host_key_pinned"]: + tested_host_keys.clear() + tested_host_keys[hk["fingerprint"]] = hk + slots = tagcodec.parse_record(result["record"], s.id_offset) + return {"ok": True, "host_key": {"fingerprint": hk["fingerprint"], "type": hk["type"]}, + "host_key_pinned": result["host_key_pinned"], "slots": slots} + + @app.post("/api/ssh/hostkey") + async def ssh_hostkey(request: Request): + body = await _json(request) + hk = tested_host_keys.get(body.get("fingerprint")) + if not hk: + raise HTTPException(400, "Bitte zuerst „Verbindung testen“ und dann diesen Fingerprint bestätigen") + key = paramiko.PKey.from_type_string(hk["type"], base64.b64decode(hk["key"])) + keys.pin_host_key(settings.get().printer_host, key) + tested_host_keys.clear() + bridge.log.add("info", f"Host-Key des Druckers bestätigt ({hk['fingerprint']})") + bridge.wake.set() + return {"host_key": keys.host_key()} + + @app.delete("/api/ssh/hostkey") + def ssh_hostkey_delete(): + keys.forget_host_key() + if bridge.reader: + bridge.reader.close() + bridge.log.add("info", "Gespeicherter Host-Key des Druckers gelöscht") + return {"host_key": None} + + # ---------- tag generator ---------- + + @app.get("/api/materials") + def get_materials(): + return { + "materials": {k: {"temps": v["temps"], "code": f"{v['code']:08X}"} for k, v in materials.MATERIALS.items()}, + "subtypes": [{"code": c, "name": n, "hidden": h, "material": materials.material_of_subtype(c)} + for c, n, h in materials.SUBTYPES], + "palette": tagcodec.PALETTE, + } + + @app.get("/api/spoolman/spools") + def spoolman_spools(archived: bool = False): + s = settings.get() + client = bridge.spoolman_factory(s) + try: + spools = client.spools(archived=archived) + except SpoolmanError as e: + raise HTTPException(502, str(e)) from None + finally: + client.close() + out = [] + for sp in spools: + info = summary(sp) + sug = materials.suggest(info["material"], info["name"]) + info["suggest"] = { + "material": sug["material"], "subtype": sug["subtype"], + "temp_min": sug["temps"][0], "temp_max": sug["temps"][1], + "color": tagcodec.nearest_palette_color(info["color"]), + } + info["spoolman_url"] = f"{s.spoolman_url}/spool/show/{info['id']}" + out.append(info) + out.sort(key=lambda x: x["id"]) + return {"spools": out, "id_offset": s.id_offset, "max_id": tagcodec.MAX_WEIGHT - s.id_offset} + + @app.post("/api/tag") + async def tag(request: Request): + body = await _json(request) + s = settings.get() + try: + data = tagcodec.TagData( + spool_id=int(body["spool_id"]), + color=str(body["color"]), + material=str(body["material"]), + subtype=int(body["subtype"]), + temp_min=int(body["temp_min"]), + temp_max=int(body["temp_max"]), + diameter=float(body.get("diameter") or 1.75), + finish=int(body.get("finish", 0xFF)), + id_offset=s.id_offset, + url=str(body.get("url") or tagcodec.DEFAULT_URL), + ) + commands = tagcodec.encode_commands(data) + except (KeyError, TypeError, ValueError) as e: + raise HTTPException(400, str(e) if isinstance(e, tagcodec.TagError) else f"Ungültige Eingabe ({e})") from None + decoded = tagcodec.decode_pages(tagcodec.parse_commands(commands), s.id_offset) + qr = segno.make(commands, error="l").svg_inline(scale=3, border=2, dark="#000", light="#fff") + return {"commands": commands, "decoded": decoded, "qr_svg": qr} + + @app.post("/api/tag/decode") + async def tag_decode(request: Request): + body = await _json(request) + try: + pages = tagcodec.parse_commands(str(body.get("commands", ""))) + except tagcodec.TagError as e: + raise HTTPException(400, str(e)) from None + return tagcodec.decode_pages(pages, settings.get().id_offset) + + return app + + +async def _json(request: Request) -> dict: + try: + body = await request.json() + except ValueError: + raise HTTPException(400, "Ungültiges JSON") from None + if not isinstance(body, dict): + raise HTTPException(400, "Ungültiges JSON") + return body + + +def main() -> None: + import uvicorn + + logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s: %(message)s") + logging.getLogger("paramiko").setLevel(logging.WARNING) + logging.getLogger("httpx").setLevel(logging.WARNING) + uvicorn.run(create_app(), host="0.0.0.0", port=int(os.environ.get("PORT", "5066")), + log_level="warning", proxy_headers=False) + + +if __name__ == "__main__": + main() diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..cdc5fb9 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,11 @@ +# Beispiel für den Docker-Host 10.10.6.10 (neben Spoolman). +services: + cc2-spoolman-bridge: + image: ${BRIDGE_IMAGE:-git.menzel.center/menzeljonas/cc2-spoolman-bridge:latest} + container_name: cc2-spoolman-bridge + restart: unless-stopped + env_file: .env + ports: + - "5066:5066" + volumes: + - ./data:/data diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..5cdbf2a --- /dev/null +++ b/pytest.ini @@ -0,0 +1,6 @@ +[pytest] +testpaths = tests +addopts = -q +filterwarnings = + ignore::DeprecationWarning:fastapi.testclient + ignore:Using `httpx` diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..1441c92 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,2 @@ +-r requirements.txt +pytest==9.1.1 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..33442d1 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,6 @@ +fastapi==0.162.0 +uvicorn==0.54.0 +jinja2==3.1.6 +httpx==0.28.1 +paramiko==5.1.0 +segno==1.6.6 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/fakes.py b/tests/fakes.py new file mode 100644 index 0000000..d4d49e9 --- /dev/null +++ b/tests/fakes.py @@ -0,0 +1,159 @@ +"""In-memory stand-ins for CC2 Control, Spoolman and the SSH record reader.""" + +from __future__ import annotations + +import copy +import itertools + +from app.cc2control import CC2Error +from app.spoolman import SpoolmanError, SpoolmanNotFound + + +def channel(weight="", type_="PLA", detailed="PLA+", color="0x000000"): + if weight == "": + return {k: "" for k in ("brand", "code", "color", "date", "detailed_type", "diameter", "manufacturer", + "nozzle_max_temp", "nozzle_min_temp", "type", "weight")} + return {"brand": "0xEEEEEEEE", "code": "0x0001", "color": color, "date": "2000/00", "detailed_type": detailed, + "diameter": "1.750000", "manufacturer": "ELEGOO", "nozzle_max_temp": "230", "nozzle_min_temp": "190", + "type": type_, "weight": str(weight)} + + +def record(*weights): + rec = {"auto_plug_in_enable": 0, "color_table_size": 0, "extern_channel": {}} + for n in range(4): + w = weights[n] if n < len(weights) else "" + rec[f"channels{n}"] = w if isinstance(w, dict) else channel(w) + return rec + + +class FakeReader: + def __init__(self, rec=None): + self.record = rec or record() + self.fail = False + + def read(self): + from app.sshsource import SSHError + if self.fail: + raise SSHError("Drucker nicht erreichbar") + return copy.deepcopy(self.record) + + def close(self): + pass + + +class FakeCC2: + def __init__(self, enabled=True): + self.enabled = enabled + self.slots = [{"slot": i, "spool": "", "last": "", "question": "", "printer": {"status": 0}} for i in range(5)] + self.slots[4]["printer"] = None + self.library: list[dict] = [] + self.job = {"active": False} + self.calls: list[tuple] = [] + self.fail = False + self._ids = (f"s{i:04d}" for i in itertools.count(1)) + + def _check(self): + if self.fail: + raise CC2Error("CC2 Control nicht erreichbar") + + def spools_json(self): + return {"enabled": self.enabled, "slots": self.slots, "spools": self.library, "job": self.job, "log": []} + + def spools(self): + self._check() + return copy.deepcopy(self.spools_json()) + + def canvas(self): + self._check() + return {} + + def save_spool(self, fields): + self._check() + self.calls.append(("save", dict(fields))) + if fields.get("id"): + sp = next(s for s in self.library if s["id"] == fields["id"]) + if "archived" in fields: + sp["archived"] = fields["archived"] in ("1", 1, True) + return {"ok": True, "id": sp["id"]} + sp = {"id": next(self._ids), "name": fields.get("name", ""), "brand": fields.get("brand", ""), + "material": fields["material"], "color": fields["color"], "diameter": fields.get("diameter", 1.75), + "density": fields.get("density", 1.24), "net": fields.get("net", 1000), + "remaining": fields.get("remaining", fields.get("net", 1000)), "tare": fields.get("tare", 0), + "low": 0, "price": 0, "note": fields.get("note", ""), "created": 0, "used": 0, "archived": False} + self.library.append(sp) + return {"ok": True, "id": sp["id"]} + + def assign(self, slot, spool_id): + self._check() + self.calls.append(("assign", slot, spool_id)) + for s in self.slots: + if spool_id and s["spool"] == spool_id: + s["spool"] = "" + self.slots[slot]["spool"] = spool_id or "" + self.slots[slot]["question"] = "" + return {"ok": True} + + def enable(self, on): + self._check() + self.calls.append(("enable", on)) + self.enabled = on + return {"ok": True, "enabled": on} + + def consume(self, cc2_id, grams): + sp = next(s for s in self.library if s["id"] == cc2_id) + sp["remaining"] -= grams + + def close(self): + pass + + +class FakeSpoolman: + def __init__(self): + self.db: dict[int, dict] = {} + self.calls: list[tuple] = [] + self.fail = False + self.fail_use_after_apply = False + + def add(self, sid, vendor="SUNLU", material="PLA+", color="000000", remaining=800.0, lot=None, **extra): + self.db[sid] = { + "id": sid, "lot_nr": lot if lot is not None else f"{vendor.upper()}{sid:03d}", + "remaining_weight": remaining, "initial_weight": 1000.0, "spool_weight": 150.0, "used_weight": 0.0, + "location": "", "archived": False, + "filament": {"id": sid, "name": f"{material} {color}", "material": material, "color_hex": color, + "diameter": 1.75, "density": 1.24, "weight": 1000.0, "vendor": {"name": vendor}}, + **extra, + } + return self.db[sid] + + def _check(self): + if self.fail: + raise SpoolmanError("Spoolman nicht erreichbar") + + def spool(self, sid): + self._check() + if sid not in self.db: + raise SpoolmanNotFound("nicht gefunden") + return copy.deepcopy(self.db[sid]) + + def spools(self, archived=False): + self._check() + return [copy.deepcopy(s) for s in self.db.values() if archived or not s["archived"]] + + def set_location(self, sid, location): + self._check() + self.calls.append(("location", sid, location)) + self.db[sid]["location"] = location + return self.db[sid] + + def use(self, sid, grams): + self._check() + self.calls.append(("use", sid, round(grams, 3))) + sp = self.db[sid] + sp["used_weight"] += grams + sp["remaining_weight"] -= grams + if self.fail_use_after_apply: + raise SpoolmanError("Timeout") + return sp + + def close(self): + pass diff --git a/tests/test_bridge.py b/tests/test_bridge.py new file mode 100644 index 0000000..7d3d153 --- /dev/null +++ b/tests/test_bridge.py @@ -0,0 +1,340 @@ +import json + +import pytest + +from app.bridge import Bridge, cc2_spool_fields, material_fits +from app.sshsource import KeyStore +from app.storage import SettingsStore + +from .fakes import FakeCC2, FakeReader, FakeSpoolman, channel, record + + +class Clock: + def __init__(self): + self.t = 1_800_000_000.0 + + def __call__(self): + return self.t + + def tick(self, s): + self.t += s + + +@pytest.fixture +def env(tmp_path): + return make_env(tmp_path) + + +def make_env(tmp_path, dry_run=False, reader=None, cc2=None, spoolman=None, clock=None): + settings = SettingsStore(tmp_path, env={}) + settings.save({"dry_run": dry_run, "settle_seconds": 30, "empty_grace": 20}) + reader = reader or FakeReader() + cc2 = cc2 or FakeCC2() + sm = spoolman or FakeSpoolman() + clock = clock or Clock() + bridge = Bridge(tmp_path, settings, KeyStore(tmp_path), reader_factory=lambda s: reader, + cc2_factory=lambda s: cc2, spoolman_factory=lambda s: sm, clock=clock) + return {"bridge": bridge, "reader": reader, "cc2": cc2, "sm": sm, "clock": clock, "settings": settings, + "path": tmp_path} + + +def insert(e, slot, spool_id, **kw): + """Put a tagged spool into a slot: tray loaded and the record carries the id.""" + e["cc2"].slots[slot]["printer"] = {"status": 1} + e["cc2"].slots[slot]["question"] = "inserted" + e["reader"].record[f"channels{slot}"] = channel(10000 + spool_id, **kw) + + +def poll(e, n=1, step=5): + for _ in range(n): + e["bridge"].poll_once() + e["clock"].tick(step) + + +def test_scan_creates_assigns_and_sets_location(env): + env["sm"].add(5, vendor="Sunlu", material="PLA+", color="1E40AF", remaining=812.5) + insert(env, 1, 5) + poll(env) + cc2 = env["cc2"] + saves = [c for c in cc2.calls if c[0] == "save"] + assert len(saves) == 1 + f = saves[0][1] + assert f["name"] == "SUNLU005" and f["brand"] == "Sunlu" and f["material"] == "PLA+" + assert f["color"] == "#1E40AF" and f["note"] == "spoolman:5" and f["remaining"] == 812.5 + assert f["net"] == 1000 and f["tare"] == 150 and f["diameter"] == 1.75 and f["density"] == 1.24 + assert ("assign", 1, "s0001") in cc2.calls + assert cc2.slots[1]["spool"] == "s0001" and cc2.slots[1]["question"] == "" + assert ("location", 5, "CC2 Slot 2") in env["sm"].calls + poll(env, 3) + # nothing repeated + assert len([c for c in cc2.calls if c[0] in ("save", "assign")]) == 2 + assert len([c for c in env["sm"].calls if c[0] == "location"]) == 1 + + +def test_existing_mirror_is_reused(env): + env["sm"].add(5) + env["cc2"].save_spool({"name": "SUNLU005", "material": "PLA+", "color": "#000000", "note": "spoolman:5"}) + env["cc2"].calls.clear() + insert(env, 0, 5) + poll(env) + assert [c for c in env["cc2"].calls if c[0] == "save"] == [] + assert ("assign", 0, "s0001") in env["cc2"].calls + + +def test_unknown_spool_id_is_ignored(env): + insert(env, 0, 42) + poll(env) + assert env["cc2"].calls == [] + assert env["bridge"].state.data["slots"] == {} + assert any("42 existiert nicht" in e["msg"] for e in env["bridge"].log.recent()) + + +def test_real_weight_tags_are_ignored(env): + env["reader"].record["channels0"] = channel(1000) + env["cc2"].slots[0]["printer"] = {"status": 1} + poll(env, 2) + assert env["cc2"].calls == [] and env["bridge"].state.data["slots"] == {} + + +def test_mapping_survives_weight_falling_back_to_1000(env): + env["sm"].add(5) + insert(env, 2, 5) + poll(env) + # touchscreen edit: weight falls back, CC2 asks again ("changed") with no spool in the slot + env["reader"].record["channels2"] = channel(1000) + env["cc2"].slots[2]["spool"] = "" + env["cc2"].slots[2]["question"] = "changed" + poll(env, 10) + m = env["bridge"].state.data["slots"]["2"] + assert m["spool"] == 5 + assert env["sm"].db[5]["location"] == "CC2 Slot 3" + + +def test_mapping_survives_restart(env): + env["sm"].add(5) + insert(env, 0, 5) + poll(env) + env["reader"].record["channels0"] = channel(1000) + poll(env) + e2 = make_env(env["path"], reader=env["reader"], cc2=env["cc2"], spoolman=env["sm"], clock=env["clock"]) + poll(e2, 3) + assert e2["bridge"].state.data["slots"]["0"]["spool"] == 5 + assert len([c for c in env["cc2"].calls if c[0] == "save"]) == 1 + + +def test_empty_slot_releases_to_storage_after_grace(env): + env["sm"].add(5) + insert(env, 0, 5) + poll(env) + env["cc2"].slots[0]["printer"] = {"status": 0} + poll(env, 3, step=5) # 0, 5, 10 s + assert "0" in env["bridge"].state.data["slots"] + poll(env, 3, step=5) # >= 20 s + assert "0" not in env["bridge"].state.data["slots"] + assert env["sm"].db[5]["location"] == "Lager" + + +def test_cc2_returning_spool_to_storage_releases(env): + env["sm"].add(5) + insert(env, 0, 5) + poll(env, 2) + env["cc2"].slots[0]["spool"] = "" # CC2 gave it back to storage (no question) + poll(env) + assert "0" not in env["bridge"].state.data["slots"] + assert env["sm"].db[5]["location"] == "Lager" + + +def test_new_scan_with_other_id_replaces_mapping(env): + env["sm"].add(5) + env["sm"].add(8, color="F88D36") + insert(env, 0, 5) + poll(env, 2) + insert(env, 0, 8, color="0xF88D36") + poll(env, 2) + assert env["bridge"].state.data["slots"]["0"]["spool"] == 8 + assert env["sm"].db[5]["location"] == "Lager" + assert env["sm"].db[8]["location"] == "CC2 Slot 1" + assert env["cc2"].slots[0]["spool"] == "s0002" + + +def test_spool_moved_to_other_slot(env): + env["sm"].add(5) + insert(env, 0, 5) + poll(env, 2) + env["reader"].record["channels0"] = channel("") + insert(env, 3, 5) + poll(env, 2) + slots = env["bridge"].state.data["slots"] + assert "0" not in slots and slots["3"]["spool"] == 5 + assert env["sm"].db[5]["location"] == "CC2 Slot 4" + + +def test_consumption_booked_after_print_once(env): + env["sm"].add(5, remaining=800) + insert(env, 0, 5) + poll(env) + cc2 = env["cc2"] + cc2.job = {"active": True} + for _ in range(10): + cc2.consume("s0001", 1.25) + poll(env) + assert not [c for c in env["sm"].calls if c[0] == "use"] # not during the print + cc2.job = {"active": False} + poll(env, 3, step=5) + assert not [c for c in env["sm"].calls if c[0] == "use"] # waits for settle_seconds + poll(env, 6, step=5) + uses = [c for c in env["sm"].calls if c[0] == "use"] + assert uses == [("use", 5, 12.5)] + poll(env, 5) + assert [c for c in env["sm"].calls if c[0] == "use"] == uses + + +def test_no_double_booking_after_restart(env): + env["sm"].add(5, remaining=800) + insert(env, 0, 5) + poll(env) + env["cc2"].consume("s0001", 3.25) + poll(env, 10) + assert [c for c in env["sm"].calls if c[0] == "use"] == [("use", 5, 3.25)] + e2 = make_env(env["path"], reader=env["reader"], cc2=env["cc2"], spoolman=env["sm"], clock=env["clock"]) + poll(e2, 10) + assert [c for c in env["sm"].calls if c[0] == "use"] == [("use", 5, 3.25)] + env["cc2"].consume("s0001", 1.0) + poll(e2, 10) + assert [c for c in env["sm"].calls if c[0] == "use"] == [("use", 5, 3.25), ("use", 5, 1.0)] + + +def test_interrupted_booking_is_not_repeated(env): + env["sm"].add(5, remaining=800) + insert(env, 0, 5) + poll(env) + env["cc2"].consume("s0001", 4.0) + env["sm"].fail_use_after_apply = True # PUT reached Spoolman, answer lost + poll(env, 10) + env["sm"].fail_use_after_apply = False + e2 = make_env(env["path"], reader=env["reader"], cc2=env["cc2"], spoolman=env["sm"], clock=env["clock"]) + poll(e2, 10) + assert [c for c in env["sm"].calls if c[0] == "use"] == [("use", 5, 4.0)] + assert env["sm"].db[5]["used_weight"] == 4.0 + b = e2["bridge"].state.data["booked"]["s0001"] + assert b["pending"] is None and b["remaining"] == 796.0 + + +def test_interrupted_booking_retried_when_not_applied(env): + env["sm"].add(5, remaining=800) + insert(env, 0, 5) + poll(env) + env["cc2"].consume("s0001", 4.0) + poll(env, 7) # settles, but Spoolman is down when booking + env["sm"].fail = True + poll(env, 3) + env["sm"].fail = False + poll(env, 3) + assert [c for c in env["sm"].calls if c[0] == "use"] == [("use", 5, 4.0)] + + +def test_negative_change_logged_not_booked(env): + env["sm"].add(5, remaining=800) + insert(env, 0, 5) + poll(env) + env["cc2"].consume("s0001", -20) # weigh-in raised the remaining weight + poll(env, 10) + assert not [c for c in env["sm"].calls if c[0] == "use"] + assert any("gestiegen" in e["msg"] for e in env["bridge"].log.recent()) + env["cc2"].consume("s0001", 5) + poll(env, 10) + assert [c for c in env["sm"].calls if c[0] == "use"] == [("use", 5, 5.0)] + + +def test_dry_run_writes_nothing(tmp_path): + e = make_env(tmp_path, dry_run=True) + e["sm"].add(5) + e["cc2"].save_spool({"name": "X", "material": "PLA", "color": "#000000", "note": "spoolman:7"}) + e["sm"].add(7) + e["cc2"].calls.clear() + insert(e, 0, 5) + poll(e, 3) + e["cc2"].consume("s0001", 10) + poll(e, 20) + assert e["cc2"].calls == [] + assert e["sm"].calls == [] + msgs = [x["msg"] for x in e["bridge"].log.recent(500) if x["level"] == "dry"] + assert any("Spule SUNLU005 anlegen" in m for m in msgs) + assert any("Ort von" in m for m in msgs) + assert any("10.00 g" in m for m in msgs) + # logged once, not every cycle + assert len([m for m in msgs if "10.00 g" in m]) == 1 + # switching dry-run off applies what is pending + e["settings"].save({"dry_run": False}) + poll(e, 10) + assert ("use", 7, 10.0) in e["sm"].calls + assert e["sm"].db[5]["location"] == "CC2 Slot 1" + assert e["cc2"].slots[0]["spool"] == "s0002" + + +def test_outages_do_not_lose_scans(env): + env["sm"].add(5) + env["sm"].fail = True + insert(env, 0, 5) + poll(env, 2) + assert env["bridge"].state.data["slots"] == {} + assert env["bridge"].state.data["scans"] + env["sm"].fail = False + poll(env) + assert env["bridge"].state.data["slots"]["0"]["spool"] == 5 + + +def test_cc2_outage_retries_assignment(env): + env["sm"].add(5) + insert(env, 0, 5) + env["cc2"].fail = True + poll(env, 2) + env["cc2"].fail = False + poll(env, 2) + assert env["cc2"].slots[0]["spool"] == "s0001" + + +def test_manual_assignment_in_cc2_is_adopted(env): + env["sm"].add(9) + env["cc2"].save_spool({"name": "SUNLU009", "material": "PLA", "color": "#000000", "note": "spoolman:9"}) + env["cc2"].assign(3, "s0001") + env["cc2"].slots[3]["printer"] = {"status": 1} + poll(env) + assert env["bridge"].state.data["slots"]["3"]["spool"] == 9 + assert env["sm"].db[9]["location"] == "CC2 Slot 4" + + +def test_enable_command_respects_dry_run(tmp_path): + e = make_env(tmp_path, dry_run=True, cc2=FakeCC2(enabled=False)) + e["bridge"].commands.put(("enable", None)) + poll(e) + assert e["cc2"].calls == [] and e["bridge"].snapshot["cc2_enabled"] is False + e["settings"].save({"dry_run": False}) + e["bridge"].commands.put(("enable", None)) + poll(e) + assert e["cc2"].calls == [("enable", True)] + + +def test_state_file_is_valid_json(env): + env["sm"].add(5) + insert(env, 0, 5) + poll(env) + data = json.loads((env["path"] / "state.json").read_text()) + assert data["slots"]["0"]["spool"] == 5 + assert not list(env["path"].glob(".state.json.*")) + + +def test_material_fits(): + assert material_fits("PLA+", "PLA", "PLA+") + assert material_fits("PLA Matte", "PLA", "PLA") + assert material_fits("PLA-CF", "PLA", "") + assert not material_fits("Silk PLA", "PLA", "PLA Silk") + assert material_fits("PLA Silk", "PLA", "PLA Silk") + assert not material_fits("PETG", "PLA", "PLA") + + +def test_fields_use_tag_material_when_spoolman_material_does_not_fit(): + sm = FakeSpoolman().add(3, material="Silk PLA", color="FFD700") + fields, notes = cc2_spool_fields(sm, channel(10003, type_="PLA", detailed="PLA Silk")) + assert fields["material"] == "PLA Silk" and notes + assert fields["color"] == "#FFD700" diff --git a/tests/test_cc2control.py b/tests/test_cc2control.py new file mode 100644 index 0000000..2e6565b --- /dev/null +++ b/tests/test_cc2control.py @@ -0,0 +1,115 @@ +import json + +import httpx +import pytest + +from app.cc2control import CC2Control, CC2Error, clean_text, encode_body, spoolman_id_of, tray_statuses +from app.spoolman import Spoolman + + +class Recorder: + def __init__(self, status=200, body=None): + self.requests: list[httpx.Request] = [] + self.status = status + self.body = body if body is not None else {"ok": True} + + def __call__(self, request: httpx.Request): + self.requests.append(request) + return httpx.Response(self.status, json=self.body) + + +def client(rec): + return CC2Control("http://10.10.2.202:8081", transport=httpx.MockTransport(rec)) + + +def test_write_request_format(): + rec = Recorder(201, {"ok": True, "id": "a1b2"}) + resp = client(rec).save_spool({"name": "SUNLU005", "brand": "Sunlu", "material": "PLA+", "color": "#1E40AF", + "diameter": 1.75, "net": 1000.0, "remaining": 812.5, "note": "spoolman:5"}) + assert resp == {"ok": True, "id": "a1b2"} + r = rec.requests[0] + assert r.method == "POST" and r.url.path == "/api/spools/save" + assert r.headers["host"] == "10.10.2.202:8081" + assert "origin" not in r.headers and "sec-fetch-site" not in r.headers + assert r.headers["x-cc2-request"] == "1" + assert "transfer-encoding" not in r.headers + assert r.headers["content-type"].startswith("text/plain") + assert r.content.decode() == ("name=SUNLU005\nbrand=Sunlu\nmaterial=PLA+\ncolor=#1E40AF\ndiameter=1.75\n" + "net=1000\nremaining=812.5\nnote=spoolman:5\n") + # every header only once (CC2 Control fails closed on duplicates) + names = [k.lower() for k, _ in r.headers.raw] + assert len(names) == len(set(names)) + + +def test_get_request_has_no_write_marker_or_origin(): + rec = Recorder(body={"enabled": True, "slots": [], "spools": []}) + assert client(rec).spools()["enabled"] is True + r = rec.requests[0] + assert r.method == "GET" and r.url.path == "/api/spools" + assert r.headers["host"] == "10.10.2.202:8081" and "origin" not in r.headers + + +def test_assign_and_empty_slot(): + rec = Recorder() + c = client(rec) + c.assign(2, "a1b2") + c.assign(2, None) + assert rec.requests[0].content == b"slot=2\nspool=a1b2\n" + assert rec.requests[1].content == b"slot=2\nspool=\n" + + +def test_enable_sends_bare_word(): + rec = Recorder(body={"ok": True, "enabled": True}) + client(rec).enable(True) + client(rec).enable(False) + assert rec.requests[0].url.path == "/api/spools/enable" + assert rec.requests[0].content == b"on" and rec.requests[1].content == b"off" + assert rec.requests[0].headers["x-cc2-request"] == "1" + + +def test_errors_are_reported_without_crash(): + rec = Recorder(400, {"ok": False, "error": "Invalid spool"}) + with pytest.raises(CC2Error, match="400 Invalid spool"): + client(rec).save_spool({"material": "PLA", "color": "#000000"}) + + def boom(request): + raise httpx.ConnectError("refused") + with pytest.raises(CC2Error, match="nicht erreichbar"): + CC2Control("http://10.10.2.202:8081", transport=httpx.MockTransport(boom)).spools() + + +def test_text_cleaning(): + assert clean_text('Sun"lu\\ \x01PLA\n') == "Sun lu PLA" + long = "ä" * 60 # 120 bytes + out = clean_text(long, 96) + assert len(out.encode()) <= 96 and out == "ä" * 48 + body = encode_body({"name": "x" * 200, "material": "M" * 50}) + name, material = body.splitlines() + assert len(name) == len("name=") + 96 and len(material) == len("material=") + 32 + assert encode_body({"note": "a=b", "archived": "0", "skip": None}) == "note=a=b\narchived=0\n" + + +def test_spoolman_note_and_trays(): + assert spoolman_id_of({"note": "spoolman:17"}) == 17 + assert spoolman_id_of({"note": "gekauft 2025 spoolman:3 rest"}) == 3 + assert spoolman_id_of({"note": "spoolman:x"}) is None + canvas = {"telemetry": {"result": {"canvas_info": {"canvas_list": [{"tray_list": [ + {"tray_id": 0, "status": 1}, {"tray_id": 1, "status": 0}, {"tray_id": 2, "status": 2}, {"tray_id": 3}]}]}}}} + assert tray_statuses(canvas) == {0: 1, 1: 0, 2: 2} + assert tray_statuses({}) == {} + + +def test_spoolman_requests(): + seen = [] + + def handler(request): + seen.append(request) + return httpx.Response(200, json={"id": 5}) + sm = Spoolman("http://10.10.6.10:7912", transport=httpx.MockTransport(handler)) + sm.spool(5) + sm.set_location(5, "CC2 Slot 1") + sm.use(5, 12.3456) + assert [(r.method, r.url.path) for r in seen] == [ + ("GET", "/api/v1/spool/5"), ("PATCH", "/api/v1/spool/5"), ("PUT", "/api/v1/spool/5/use")] + assert json.loads(seen[1].content) == {"location": "CC2 Slot 1"} + assert json.loads(seen[2].content) == {"use_weight": 12.346} diff --git a/tests/test_ssh.py b/tests/test_ssh.py new file mode 100644 index 0000000..d1a09c3 --- /dev/null +++ b/tests/test_ssh.py @@ -0,0 +1,159 @@ +"""SSH reader against a local paramiko server that behaves like the restricted printer key.""" + +import io +import json +import os +import socket +import stat +import threading + +import paramiko +import pytest + +from app.sshsource import CanvasRecordReader, HostKeyMismatch, HostKeyUnknown, KeyStore, SSHError + +from .fakes import record + +RECORD = record(10001, "", 1000, 10008) + + +class Server(paramiko.ServerInterface): + def __init__(self, allowed: paramiko.PKey): + self.allowed = allowed + self.commands = [] + + def check_channel_request(self, kind, chanid): + return paramiko.OPEN_SUCCEEDED if kind == "session" else paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED + + def get_allowed_auths(self, username): + return "publickey" + + def check_auth_publickey(self, username, key): + ok = username == "root" and key.get_base64() == self.allowed.get_base64() + return paramiko.AUTH_SUCCESSFUL if ok else paramiko.AUTH_FAILED + + def check_channel_exec_request(self, channel, command): + self.commands.append(command) + + def reply(): # forced command: always the canvas record + channel.sendall(json.dumps(RECORD).encode()) + channel.send_exit_status(0) + channel.close() + threading.Timer(0.05, reply).start() # after the exec request is acknowledged + return True + + +@pytest.fixture +def printer(): + host_key = paramiko.Ed25519Key.from_private_key(io.StringIO(_new_key_text())) + client_key_text = _new_key_text() + client_key = paramiko.Ed25519Key.from_private_key(io.StringIO(client_key_text)) + sock = socket.socket() + sock.bind(("127.0.0.1", 0)) + sock.listen(5) + servers = [] + stop = threading.Event() + + def accept(): + while not stop.is_set(): + try: + conn, _ = sock.accept() + except OSError: + return + t = paramiko.Transport(conn) + t.add_server_key(host_key) + srv = Server(client_key) + servers.append((t, srv)) + try: + t.start_server(server=srv) + except Exception: + pass + threading.Thread(target=accept, daemon=True).start() + yield {"port": sock.getsockname()[1], "host_key": host_key, "client_key_text": client_key_text, + "servers": servers} + stop.set() + sock.close() + for t, _ in servers: + t.close() + + +def _new_key_text(): + from cryptography.hazmat.primitives import serialization + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + return Ed25519PrivateKey.generate().private_bytes( + serialization.Encoding.PEM, serialization.PrivateFormat.OpenSSH, serialization.NoEncryption()).decode() + + +def test_key_store_permissions_and_no_secret_in_info(tmp_path, printer): + keys = KeyStore(tmp_path) + keys.save_key(printer["client_key_text"]) + mode = stat.S_IMODE(os.stat(keys.key_path).st_mode) + assert mode == 0o600 + info = keys.key_info() + assert info["fingerprint"].startswith("SHA256:") and info["type"] == "ssh-ed25519" + assert "PRIVATE" not in json.dumps(info) + + +def test_rejects_garbage_key(tmp_path): + with pytest.raises(SSHError): + KeyStore(tmp_path).save_key("not a key") + + +def test_first_contact_requires_confirmation(tmp_path, printer): + keys = KeyStore(tmp_path) + keys.save_key(printer["client_key_text"]) + reader = CanvasRecordReader(keys, "127.0.0.1", printer["port"]) + with pytest.raises(HostKeyUnknown): + reader.read() + result = reader.test() + assert result["host_key_pinned"] is False + assert result["record"]["channels0"]["weight"] == "10001" + keys.pin_host_key("127.0.0.1", printer["host_key"]) + rec = reader.read() + assert rec["channels3"]["weight"] == "10008" + # transport is reused between reads + reader.read() + assert len(printer["servers"]) == 3 # refused read, test(), then one transport for both reads + reader.close() + + +def test_changed_host_key_is_rejected(tmp_path, printer): + keys = KeyStore(tmp_path) + keys.save_key(printer["client_key_text"]) + other = paramiko.Ed25519Key.from_private_key(io.StringIO(_new_key_text())) + keys.pin_host_key("127.0.0.1", other) + reader = CanvasRecordReader(keys, "127.0.0.1", printer["port"]) + with pytest.raises(HostKeyMismatch): + reader.read() + with pytest.raises(HostKeyMismatch): + reader.test() + + +def test_wrong_client_key_fails_cleanly(tmp_path, printer): + keys = KeyStore(tmp_path) + keys.save_key(_new_key_text()) + keys.pin_host_key("127.0.0.1", printer["host_key"]) + with pytest.raises(SSHError, match="abgelehnt|fehlgeschlagen"): + CanvasRecordReader(keys, "127.0.0.1", printer["port"]).read() + + +def test_reconnects_after_drop(tmp_path, printer): + keys = KeyStore(tmp_path) + keys.save_key(printer["client_key_text"]) + keys.pin_host_key("127.0.0.1", printer["host_key"]) + reader = CanvasRecordReader(keys, "127.0.0.1", printer["port"]) + reader.read() + for t, _ in printer["servers"]: + t.close() + assert reader.read()["channels0"]["weight"] == "10001" + + +def test_unreachable_printer(tmp_path, printer): + keys = KeyStore(tmp_path) + keys.save_key(printer["client_key_text"]) + s = socket.socket() + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + s.close() + with pytest.raises(SSHError, match="nicht erreichbar"): + CanvasRecordReader(keys, "127.0.0.1", port, timeout=2).read() diff --git a/tests/test_tagcodec.py b/tests/test_tagcodec.py new file mode 100644 index 0000000..4848000 --- /dev/null +++ b/tests/test_tagcodec.py @@ -0,0 +1,118 @@ +import pytest + +from app import materials, tagcodec +from app.tagcodec import TagData, encode_commands, parse_commands + +TEMPLATE = ("A2:04:0103A00C,A2:05:34030FD1,A2:06:010B5502,A2:07:656C6567,A2:08:6F6F2E63,A2:09:6F6DFE00," + "A2:10:36EEEEEE,A2:11:EE000000,A2:12:00807665,A2:13:00010000,A2:14:000000FF,A2:15:00BE00E6," + "A2:16:00000000,A2:17:00AF2711,A2:18:0036C800") + + +def pla_plus(spool_id, color): + return TagData(spool_id=spool_id, color=color, material="PLA", subtype=0x0001, temp_min=190, temp_max=230) + + +def test_template_reproduced_exactly(): + assert encode_commands(pla_plus(1, "000000")) == TEMPLATE + + +@pytest.mark.parametrize("spool_id,color,p14,p17", [ + (1, "000000", "A2:14:000000FF", "A2:17:00AF2711"), + (5, "2850DF", "A2:14:2850DFFF", "A2:17:00AF2715"), + (8, "F88D36", "A2:14:F88D36FF", "A2:17:00AF2718"), +]) +def test_expected_pages(spool_id, color, p14, p17): + cmds = encode_commands(pla_plus(spool_id, color)).split(",") + assert p14 in cmds and p17 in cmds + # only pages 14 and 17 differ from the template + diff = set(cmds) ^ set(TEMPLATE.split(",")) + assert {c[:5] for c in diff} <= {"A2:14", "A2:17"} + + +def test_never_writes_pages_0_to_3(): + pages = tagcodec.encode_pages(pla_plus(1, "000000")) + assert min(pages) == 4 + with pytest.raises(tagcodec.TagError): + tagcodec.pages_to_commands({3: b"\0\0\0\0"}) + + +def test_decode_roundtrip(): + d = tagcodec.decode_pages(parse_commands(TEMPLATE)) + assert d["url"] == "https://www.elegoo.com" + assert d["material"] == "PLA" and d["subtype_name"] == "PLA+" + assert (d["temp_min"], d["temp_max"]) == (190, 230) + assert d["diameter"] == 1.75 and d["weight"] == 10001 and d["spool_id"] == 1 + assert d["manufacturer"] == "EEEEEEEE" + + +def test_rapid_petg_codes(): + tag = TagData(spool_id=12, color="FFFFFF", material="PETG", subtype=0x0105, temp_min=240, temp_max=255) + pages = tagcodec.encode_pages(tag) + assert pages[0x12].hex().upper() == "80698471" + assert pages[0x13].hex().upper() == "01050000" + assert pages[0x15].hex().upper() == "00F000FF" + + +def test_validation(): + with pytest.raises(tagcodec.TagError): + encode_commands(TagData(spool_id=1, color="000000", material="PLA", subtype=0x0100)) # PETG subtype + with pytest.raises(tagcodec.TagError): + encode_commands(pla_plus(60000, "000000")) # does not fit 2 bytes + with pytest.raises(tagcodec.TagError): + encode_commands(pla_plus(1, "zzzzzz")) + with pytest.raises(tagcodec.TagError): + encode_commands(TagData(spool_id=1, color="000000", temp_min=240, temp_max=200)) + + +def test_custom_url_fits_before_canvas_pages(): + tag = pla_plus(123, "000000") + tag.url = "http://10.10.6.10:7912/spool/show/123" + pages = tagcodec.encode_pages(tag) + assert max(p for p in pages if p < 0x10) <= 0x0F + assert tagcodec.decode_pages(pages)["url"] == tag.url + tag.url = "https://example.com/" + "x" * 40 + with pytest.raises(tagcodec.TagError): + tagcodec.encode_pages(tag) + + +@pytest.mark.parametrize("weight,expected", [ + ("10001", 1), ("10005", 5), ("10000", 0), ("9999", None), ("1000", None), ("", None), + ("abc", None), ("70000", None), ("10008.000", 8), +]) +def test_spool_id_from_weight(weight, expected): + assert tagcodec.spool_id_from_weight(weight) == expected + + +def test_parse_record(): + rec = {"channels0": {"weight": "10001", "type": "PLA"}, "channels1": {"weight": "", "type": ""}, + "channels3": {"weight": "1000", "type": "PETG"}} + slots = tagcodec.parse_record(rec) + assert [s["spool_id"] for s in slots] == [1, None, None, None] + assert [s["occupied"] for s in slots] == [True, False, False, True] + + +@pytest.mark.parametrize("color,expected", [ + ("000000", "000000"), ("2850DF", "2850DF"), ("0000FF", "2850DF"), ("0047AB", "2850DF"), + ("FF8000", "F88D36"), ("FF0000", "F72221"), ("FFFFFF", "FFFFFF"), ("808080", "898989"), + ("1E90FF", "48A7FA"), ("800080", "433089"), (None, "000000"), ("#ffa500", "F88D36"), +]) +def test_palette_suggestion(color, expected): + assert tagcodec.nearest_palette_color(color) == expected + + +@pytest.mark.parametrize("material,name,mat,sub", [ + ("PLA+", "", "PLA", 0x0001), ("PLA", "Silk Gold", "PLA", 0x0003), ("PETG", "", "PETG", 0x0100), + ("PETG", "Rapid PETG", "PETG", 0x0105), ("PLA", "Matte Black", "PLA", 0x0006), ("PETG-CF", "", "PETG", 0x0101), + ("ABS", "", "ABS", 0x0200), ("TPU", "95A", "TPU", 0x0301), ("PCTG", "", "PC", 0x0601), + ("ASA", "", "ASA", 0x0800), ("", "", "PLA", 0x0000), ("PLA", "Rapid PLA+", "PLA", 0x000A), +]) +def test_material_suggestion(material, name, mat, sub): + s = materials.suggest(material, name) + assert (s["material"], s["subtype"]) == (mat, sub) + + +def test_tables_complete(): + assert len(materials.MATERIALS) == 15 + assert len(materials.SUBTYPES) == 50 + for code, name, _ in materials.SUBTYPES: + assert materials.material_of_subtype(code) is not None, name diff --git a/tests/test_web.py b/tests/test_web.py new file mode 100644 index 0000000..8c73214 --- /dev/null +++ b/tests/test_web.py @@ -0,0 +1,110 @@ +import base64 +import os +import stat + +import pytest +from fastapi.testclient import TestClient + +from app.bridge import Bridge +from app.sshsource import KeyStore +from app.storage import SettingsStore +from app.web import create_app + +from .fakes import FakeCC2, FakeReader, FakeSpoolman +from .test_ssh import _new_key_text + +H = {"X-Bridge-Request": "1"} + + +def make_client(tmp_path, env=None): + settings = SettingsStore(tmp_path, env={}) + sm = FakeSpoolman() + sm.add(5, material="PLA+", color="1E40AF") + bridge = Bridge(tmp_path, settings, KeyStore(tmp_path), reader_factory=lambda s: FakeReader(), + cc2_factory=lambda s: FakeCC2(), spoolman_factory=lambda s: sm) + app = create_app(tmp_path, bridge=bridge, start_loop=False, env=env or {}) + return TestClient(app), bridge + + +def test_pages_render(tmp_path): + c, _ = make_client(tmp_path) + for path in ("/", "/tags", "/settings"): + r = c.get(path) + assert r.status_code == 200 and "CC2" in r.text + + +def test_writes_need_marker_header(tmp_path): + c, _ = make_client(tmp_path) + assert c.post("/api/settings", json={"dry_run": False}).status_code == 403 + assert c.post("/api/settings", json={"dry_run": True}, headers=H).status_code == 200 + + +def test_settings_roundtrip_and_validation(tmp_path): + c, _ = make_client(tmp_path) + s = c.get("/api/settings").json() + assert s["dry_run"] is True and s["id_offset"] == 10000 + r = c.post("/api/settings", json={"poll_interval": "7", "location_storage": "Regal"}, headers=H) + assert r.json()["poll_interval"] == 7.0 + assert c.get("/api/settings").json()["location_storage"] == "Regal" + assert c.post("/api/settings", json={"location_slot": "Slot"}, headers=H).status_code == 400 + assert SettingsStore(tmp_path).get().location_storage == "Regal" + + +def test_ssh_key_upload_never_returned(tmp_path): + c, _ = make_client(tmp_path) + text = _new_key_text() + r = c.post("/api/ssh/key", json={"key": text}, headers=H) + assert r.status_code == 200 + body = r.text + c.get("/api/ssh").text + c.get("/api/status").text + assert "PRIVATE KEY" not in body and text.splitlines()[1] not in body + assert r.json()["key"]["fingerprint"].startswith("SHA256:") + path = tmp_path / "ssh" / "id_key" + assert stat.S_IMODE(os.stat(path).st_mode) == 0o600 + assert c.post("/api/ssh/key", json={"key": "junk"}, headers=H).status_code == 400 + assert c.delete("/api/ssh/key", headers=H).json() == {"key": None} + assert not path.exists() + + +def test_hostkey_only_from_last_test(tmp_path): + c, _ = make_client(tmp_path) + r = c.post("/api/ssh/hostkey", json={"fingerprint": "SHA256:forged"}, headers=H) + assert r.status_code == 400 + + +def test_tag_endpoint(tmp_path): + c, _ = make_client(tmp_path) + r = c.post("/api/tag", json={"spool_id": 5, "color": "2850DF", "material": "PLA", "subtype": 1, + "temp_min": 190, "temp_max": 230}, headers=H) + assert r.status_code == 200 + d = r.json() + assert "A2:14:2850DFFF" in d["commands"] and "A2:17:00AF2715" in d["commands"] + assert d["decoded"]["spool_id"] == 5 and d["qr_svg"].startswith(" FastAPI: + app = FastAPI() + + @app.middleware("http") + async def guard(request: Request, call_next): + raw = [k.decode().lower() for k, _ in request.scope["headers"]] + if len(raw) != len(set(raw)): + return JSONResponse({"error": "duplicate header"}, 403) + if request.headers.get("host") != host_header or "origin" in request.headers \ + or "transfer-encoding" in request.headers: + return JSONResponse({"error": "guard"}, 403) + if request.method == "POST" and request.headers.get("x-cc2-request") != "1": + return JSONResponse({"error": "marker"}, 403) + return await call_next(request) + + def form(body: bytes) -> dict: + text = body.decode("utf-8").rstrip("\r\n") + out = {} + for line in text.split("\n"): + if "=" not in line: + raise HTTPException(400, "Invalid form") + k, v = line.split("=", 1) + if not k or k in out or len(k) >= 16 or len(v) > 96: + raise HTTPException(400, "Invalid form") + out[k] = v + return out + + @app.get("/api/spools") + def spools(): + with LOCK: + STATE["requests"].append("GET /api/spools") + return {"available": True, "enabled": STATE["enabled"], "slots": STATE["slots"], "spools": STATE["spools"], + "job": STATE["job"], "log": []} + + @app.get("/api/canvas") + def canvas(): + return {"telemetry": {"result": {"canvas_info": {"canvas_list": [{"tray_list": [ + {"tray_id": s["slot"], "status": s["printer"]["status"]} for s in STATE["slots"][:4]]}]}}}} + + @app.post("/api/spools/enable") + async def enable(request: Request): + body = await request.body() + if body not in (b"on", b"off"): + raise HTTPException(400, "Invalid switch") + with LOCK: + STATE["enabled"] = body == b"on" + STATE["requests"].append(f"POST enable {body.decode()}") + return {"ok": True, "enabled": STATE["enabled"]} + + @app.post("/api/spools/save") + async def save(request: Request): + f = form(await request.body()) + with LOCK: + STATE["requests"].append("POST save " + json.dumps(f, ensure_ascii=False)) + if f.get("id"): + sp = next((s for s in STATE["spools"] if s["id"] == f["id"]), None) + if not sp: + return JSONResponse({"ok": False, "error": "Unknown spool"}, 404) + if "remaining" in f: + return JSONResponse({"ok": False, "error": "Correct the remaining weight with a weigh-in"}, 400) + for k in ("name", "brand", "material", "color", "note"): + if k in f: + sp[k] = f[k] + if "archived" in f: + sp["archived"] = f["archived"] == "1" + return {"ok": True, "id": sp["id"]} + if not f.get("material") or not f.get("color"): + return JSONResponse({"ok": False, "error": "A spool needs a material and a colour"}, 400) + sp = {"id": next(IDS), "name": f.get("name", ""), "brand": f.get("brand", ""), "material": f["material"], + "color": f["color"], "diameter": float(f.get("diameter", 1.75)), "density": float(f.get("density", 1.24)), + "net": float(f.get("net", 1000)), "remaining": float(f.get("remaining", f.get("net", 1000))), + "tare": float(f.get("tare", 0)), "low": 0.0, "price": float(f.get("price", 0)), + "note": f.get("note", ""), "created": 0, "used": 0, "archived": False} + STATE["spools"].append(sp) + return JSONResponse({"ok": True, "id": sp["id"]}, 201) + + @app.post("/api/spools/assign") + async def assign(request: Request): + f = form(await request.body()) + with LOCK: + STATE["requests"].append("POST assign " + json.dumps(f)) + slot = int(f["slot"]) + for s in STATE["slots"]: + if f["spool"] and s["spool"] == f["spool"]: + s["spool"] = "" + STATE["slots"][slot]["spool"] = f["spool"] + STATE["slots"][slot]["question"] = "" + return {"ok": True} + + return app + + +# ---------- Spoolman ---------- + +def make_spoolman() -> FastAPI: + app = FastAPI() + + def get(sid: int): + if sid not in STATE["spoolman"]: + raise HTTPException(404, "not found") + return STATE["spoolman"][sid] + + @app.get("/api/v1/spool") + def spools(allow_archived: bool = False): + return [s for s in STATE["spoolman"].values() if allow_archived or not s["archived"]] + + @app.get("/api/v1/spool/{sid}") + def spool(sid: int): + return get(sid) + + @app.patch("/api/v1/spool/{sid}") + async def patch(sid: int, request: Request): + body = await request.json() + with LOCK: + sp = get(sid) + sp.update({k: v for k, v in body.items() if k in ("location",)}) + STATE["requests"].append(f"PATCH spoolman {sid} {json.dumps(body, ensure_ascii=False)}") + return sp + + @app.put("/api/v1/spool/{sid}/use") + async def use(sid: int, request: Request): + body = await request.json() + with LOCK: + sp = get(sid) + sp["used_weight"] += body["use_weight"] + sp["remaining_weight"] -= body["use_weight"] + STATE["requests"].append(f"PUT spoolman {sid} use {body['use_weight']}") + return sp + + return app + + +# ---------- control ---------- + +def make_control() -> FastAPI: + app = FastAPI() + + @app.get("/state") + def state(): + return STATE + + @app.post("/insert") + def insert(slot: int, spool: int, color: str = "0x000000"): + with LOCK: + STATE["slots"][slot]["printer"] = {"status": 1} + STATE["slots"][slot]["question"] = "inserted" + STATE["slots"][slot]["spool"] = "" + STATE["record"][f"channels{slot}"] = channel(10000 + spool, color=color) + return {"ok": True} + + @app.post("/remove") + def remove(slot: int): + with LOCK: + STATE["slots"][slot]["printer"] = {"status": 0} + STATE["record"][f"channels{slot}"] = channel("") + return {"ok": True} + + @app.post("/touch") + def touch(slot: int): + with LOCK: + STATE["record"][f"channels{slot}"]["weight"] = "1000" + STATE["slots"][slot]["spool"] = "" + STATE["slots"][slot]["question"] = "changed" + return {"ok": True} + + @app.post("/job") + def job(active: int): + with LOCK: + STATE["job"]["active"] = bool(active) + return {"ok": True} + + @app.post("/consume") + def consume(slot: int, grams: float): + with LOCK: + sid = STATE["slots"][slot]["spool"] + sp = next(s for s in STATE["spools"] if s["id"] == sid) + sp["remaining"] -= grams + return {"ok": True, "remaining": sp["remaining"]} + + @app.post("/authorize") + async def authorize(request: Request): + AUTHORIZED.add((await request.body()).decode().split()[1]) + return {"ok": True} + + return app + + +# ---------- SSH ---------- + +AUTHORIZED: set[str] = set() + + +class SSHServer(paramiko.ServerInterface): + def check_channel_request(self, kind, chanid): + return paramiko.OPEN_SUCCEEDED if kind == "session" else paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED + + def get_allowed_auths(self, username): + return "publickey" + + def check_auth_publickey(self, username, key): + return paramiko.AUTH_SUCCESSFUL if username == "root" and key.get_base64() in AUTHORIZED else paramiko.AUTH_FAILED + + def check_channel_exec_request(self, channel_, command): + def reply(): # forced command, whatever was sent + with LOCK: + data = json.dumps(STATE["record"]).encode() + channel_.sendall(data) + channel_.send_exit_status(0) + channel_.close() + threading.Timer(0.05, reply).start() # after the exec request is acknowledged + return True + + +def ssh_server(port: int): + host_key = paramiko.Ed25519Key.generate() if hasattr(paramiko.Ed25519Key, "generate") else None + if host_key is None: + from io import StringIO + + from cryptography.hazmat.primitives import serialization + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + pem = Ed25519PrivateKey.generate().private_bytes( + serialization.Encoding.PEM, serialization.PrivateFormat.OpenSSH, serialization.NoEncryption()).decode() + host_key = paramiko.Ed25519Key.from_private_key(StringIO(pem)) + sock = socket.socket() + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind(("0.0.0.0", port)) + sock.listen(10) + print(f"mock SSH on {port}, host key {host_key.fingerprint}", flush=True) + while True: + conn, _ = sock.accept() + t = paramiko.Transport(conn) + t.add_server_key(host_key) + try: + t.start_server(server=SSHServer()) + except Exception: + pass + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--host-header", default="127.0.0.1:8081") + args = ap.parse_args() + seed_spoolman() + threading.Thread(target=ssh_server, args=(2222,), daemon=True).start() + servers = [ + uvicorn.Server(uvicorn.Config(make_cc2(args.host_header), host="0.0.0.0", port=8081, log_level="warning")), + uvicorn.Server(uvicorn.Config(make_spoolman(), host="0.0.0.0", port=7912, log_level="warning")), + uvicorn.Server(uvicorn.Config(make_control(), host="0.0.0.0", port=8099, log_level="warning")), + ] + threads = [threading.Thread(target=s.run, daemon=True) for s in servers[1:]] + for t in threads: + t.start() + servers[0].run() + + +if __name__ == "__main__": + main()