Compare commits

..

16 Commits

Author SHA1 Message Date
6ab5c31035 feat(kxgauge): add KXGauge round-display integration
All checks were successful
Testing Build / build (push) Successful in 8m44s
Adds an optional per-printer integration with KXGauge
(https://gitea.it-drui.de/viewit/kxgauge), a small ESP32 round-face
display that shows printer status as an emotion and hotend temperature
as a color ring. KXGauge only exposes a GET-only HTTP API with no
push/websocket, so the bridge actively pushes to it from the existing
MQTT callbacks (_on_temp for the heat ring, _on_print + offline
transitions for the emotion) whenever state actually changes, with a
built-in dedupe so it doesn't spam the device every poll tick.

- kxgauge_client.py: thin synchronous HTTP client (mirrors
  spoolman_client.py's shape - called from the MQTT reader thread).
- New [kxgauge]/[kxgauge_mapping] config.ini sections; the mapping
  (kobra_state -> KXGauge emotion) is user-editable with a sane
  default and falls back per-key if only partially configured.
- Settings UI: new card under Integrations with enable/URL/target-temp
  fields, a per-state emotion mapping list, and a connection-test
  button (/api/kxgauge/test).
- Multi-printer aware: kxgauge_url/enabled/heat_peak merge per
  [printer_N] like the existing power-switch settings.

Also fixes a real bug found while testing this: _find_config_path()
resolves to the live project config/config.ini, not a sandboxed path,
so any test hitting /api/settings POST without stubbing it out will
silently overwrite the real printer config. test_settings.py already
guards against this - test_kxgauge.py now does too.
2026-08-05 15:13:43 +02:00
55acd323f4 fix(docker): copy all refactored bridge modules into the image
All checks were successful
Testing Build / build (push) Successful in 8m40s
The Dockerfile still only COPYed the pre-refactor module list, so the
testing image built from the split-module branch crashed immediately
with ModuleNotFoundError: No module named 'spoolman_client'. Add the
new bridge_*.py mixins plus spoolman_client/gcode_store/gcode_meta/
camera/credentials. Verified with a local docker build + run against
the real printer.
2026-08-05 10:25:34 +02:00
c22e0d0919 feat(update): treat the testing channel as docker-only in the update check
All checks were successful
Testing Build / build (push) Successful in 8m47s
A testing-<sha> build has no Gitea releases (the testing workflow builds only
a Docker image), so the in-app update check must not fall through to the
stable path - which would wrongly offer a stable "update". handle_api_update_check
now short-circuits for a "testing" version with docker_only:true and nothing
to update (no Gitea round-trip at all), and handle_api_update_apply refuses
self-update on the testing channel the same way it already does for nightly.

Two new tests cover both.
2026-08-04 21:22:34 +02:00
2e4dbf0da1 ci: add isolated testing-branch Docker build workflow
Adds .gitea/workflows/testing.yml, which builds and pushes only the
gitea.it-drui.de/viewit/kx-bridge:testing (+ :testing-<shortsha>) image on a
push to the `testing` branch.

Strictly isolated from nightly/master by construction:
- triggers only on push to `testing` (plus workflow_dispatch), no cron
- contains no git push / tag / release step, so it never writes back to any
  branch (unlike nightly.yml, which resets NIGHTLY_CHANGELOG.md back to
  nightly)
- uses its own :testing* image tags, never overwriting :nightly / :latest

Conversely, the existing workflows don't react to testing pushes:
nightly.yml is bound to the nightly branch, release.yml to v* tags,
pr-check.yml to PRs against nightly. So a testing push can't trigger any of
them either.
2026-08-04 21:16:21 +02:00
701ef0d516 refactor(bridge): drop now-unused imports from the facade (stage 3 cleanup)
After extracting everything into modules/mixins, several stdlib imports in
kobrax_moonraker_bridge.py (sqlite3, uuid, hashlib, tempfile, subprocess,
pathlib, html, urllib.parse.quote) are no longer referenced there - they
travelled with GCodeStore/credentials/camera/endpoints. Removed.

Facade is now 1315 lines (down from 6368). All 184 tests green; live-verified
against a real printer that every mixin's endpoints resolve and serve
correctly (kobra_state, AMS slots flowing end to end). PyInstaller spec needs
no change - all 12 new modules are statically importable from the facade.
2026-08-04 21:07:10 +02:00
f76c059fca refactor(bridge): extract EndpointsMixin (stage 3, mixin 5/6) 2026-08-04 20:25:25 +02:00
49c2fe5a7a refactor(bridge): extract MoonrakerCompatMixin (stage 3, mixin 4/6) 2026-08-04 20:20:57 +02:00
cda0e984ec refactor(bridge): extract AmsFilamentMixin (stage 3, mixin 3/6) 2026-08-04 20:18:20 +02:00
5311d74fd0 refactor(bridge): extract MqttCallbacksMixin + bridge_constants (stage 3, mixin 2/6) 2026-08-04 20:16:17 +02:00
1c18a85021 refactor(bridge): extract SpoolmanMixin (stage 3, mixin 1/6) 2026-08-04 20:10:48 +02:00
ebb48aac82 refactor(bridge): extract printer credential fetch/decrypt into credentials.py (stage 2)
Moves the coherent printer-credential unit (_kx_generate_signature,
_kx_decrypt_info, _kx_fetch_credentials plus the pycryptodome import guard)
out of the facade into credentials.py, re-exported for the "add printer"
flow. The remaining free functions (build_app, main, _build_per_printer_args,
_default_data_dir, _mqtt_error_msg) stay in the facade - they're entrypoint
logic or travel more naturally with the core mixin in the next stage.

All 184 tests green, no behavior change.
2026-08-04 19:43:25 +02:00
cdf11f6bfe refactor(bridge): extract self-contained classes/helpers into modules (stage 1)
First stage of splitting the 6368-line kobrax_moonraker_bridge.py monolith.
The low-coupling, self-contained pieces move into their own modules;
kobrax_moonraker_bridge.py re-exports them so every caller (12 test files,
the PyInstaller spec) keeps working unchanged - not a single test or the
spec needed editing.

Extracted:
- spoolman_client.py  <- SpoolmanClient (zero coupling)
- gcode_store.py      <- GCodeStore (stdlib only)
- gcode_meta.py       <- _parse_gcode_*/_extract_* metadata helpers
- camera.py           <- CameraCache + _find_ffmpeg
- bridge_logging.py   <- _BrowserLogHandler, the log ring buffer + SSE
                         queues, _set_verbose_http_log (the shared mutable
                         buffer/queues are re-imported so the log endpoints
                         still operate on the same objects the handler writes)

Facade shrinks from 6368 to 5566 lines. All 184 tests green after each
extraction. Verified the re-exported names resolve and the shared log
objects are identical by reference across modules. No behavior change.
2026-08-04 19:41:17 +02:00
2b2d5ee0a7 fix(filaments): guard against non-string profile names, log name collisions
Found during a targeted code review, not from a user report.

parse_profile()'s `name` field went straight through clean_name() without
routing through first_str() first, unlike filament_vendor/filament_type/
default_filament_colour right below it - all of which handle the
documented case where OrcaSlicer stores a field as ["value"] instead of a
plain string. If `data["name"]` was ever a list, clean_name()'s re.sub()
raised TypeError since it requires a string argument. Fixed by routing it
through first_str() like its neighbors.

Also added a debug log when sys_by_name (the system-profile lookup index)
overwrites an entry due to a name collision - clean_name() deliberately
collapses variant-suffixed profile names (e.g. "...@base" vs
"...@Anycubic Kobra X 0.4 nozzle") onto the same cleaned name, so a
collision is expected, but the resulting last-write-wins overwrite was
previously silent, making an unexpected inherits-parent resolution hard to
debug.

New tests in tests/test_orca_filaments_parser.py add the first dedicated
coverage for parse_profile()/parse_profile_bytes()/clean_name() - previous
tests only used pre-parsed profile dicts as fixtures and never exercised
the parsing logic itself.
2026-08-04 15:31:57 +02:00
ce670f5f93 fix(config): don't crash the whole bridge on a typo'd numeric config value
Found during a targeted code review, not from a user report.

MQTT_PORT, POLL_INTERVAL, and the other numeric module-level shortcuts in
config_loader.py ran int(get(...)) unguarded at import time. A hand-edited
config.ini with a typo (e.g. "mqtt_port = 98833x") raised an uncaught
ValueError before the bridge even started, with a raw traceback instead of
a usable diagnostic - list_printers() already guarded this exact class of
input the same way, but the module-level constants didn't. Added
_safe_int() (same try/except-with-fallback pattern) and applied it
everywhere int() was called unguarded on a config value.

Also wrapped migrate_env_to_config()'s filesystem writes (which also run
at import time during first-run .env migration) in try/except, so a
permission or disk-full error logs a clear message before re-raising
instead of surfacing as a bare traceback pointing into configparser.

New tests in tests/test_config_loader_robustness.py cover both, including
an end-to-end subprocess test that imports config_loader against a
malformed config.ini (a plain re-import wouldn't re-exercise the
import-time code path due to Python's module caching).
2026-08-04 15:31:46 +02:00
5eeed97514 fix(bridge): camera process-race and unhandled JSON errors in two handlers
Found during a targeted code review, not from a user report:

- _run_jpeg_loop()/_run_h264_loop() operated directly on the shared
  self._proc_jpeg/self._proc_h264 instance attributes in their cleanup,
  unlike _run_mjpeg_loop() (already fixed for exactly this) which uses a
  local `proc` reference. If a loop's task is cancelled - e.g. by
  CameraCache.reset() after the printer rotates its stream URL on reboot -
  while a new task has already started and assigned its own process to the
  shared attribute, the cancelled task's cleanup killed the NEWER process
  instead of its own, leaking its own ffmpeg child as an orphan. Applied
  the same local-variable + identity-check pattern already used by
  _run_mjpeg_loop.

- handle_api_settings_post and handle_api_update_apply were the only two
  of ~84 handlers that called `await request.json()` without a try/except -
  every other handler follows the established pattern of returning a clean
  400 for a malformed body. A trivial malformed request to either endpoint
  produced an unhandled 500 with a full traceback instead.

New tests in tests/test_camera_process_race.py,
tests/test_settings.py, and tests/test_update_check.py cover both.
2026-08-04 15:31:36 +02:00
ce71299896 fix(mqtt): harden kobrax_client.py against malformed data and concurrent requests
Found during a targeted code review, not from a user report:

- _drain() could get permanently stuck: valid JSON that isn't an object
  (e.g. a bare number or list, which json.loads() happily accepts) crashed
  _dispatch()'s dict-oriented logic, and the exception escaped before
  self._buf was advanced past the bad packet - so the same malformed bytes
  sat at the front of the buffer and re-crashed every subsequent _drain()
  call, forcing a reconnect each time. Now the buffer always advances (via
  try/finally) and _dispatch() rejects non-dict payloads with a warning log
  instead of crashing.

- _pending_msgid/_pending_report were mutated without synchronization
  while the reader thread reads/resolves them in _dispatch() - a
  check-then-set race on the shared report_key slot meant two concurrent
  publish() calls for the same msg_type could have one silently miss its
  reply. Added a dedicated lock around all registration/cleanup.

- The report-topic resolution path never verified a reply's msgid matched
  the waiter it was about to resolve, so a late reply for an
  already-timed-out request could be delivered to an unrelated, newer
  caller waiting on the same report_key. Entries now carry their own msgid
  for this comparison; replies without a msgid still resolve normally
  (most printer push-reports don't carry one).

- upload_gcode() silently sent a request with an empty session token when
  the upload URL was missing "?s=", instead of raising a clear error at
  the actual point of failure. It also leaked the upload socket's file
  descriptor on any send/recv failure other than a timeout, since there
  was no try/finally around its lifetime.

New tests in tests/test_client_robustness.py cover all of the above.
2026-08-04 15:31:24 +02:00
29 changed files with 6155 additions and 5097 deletions

View File

@@ -0,0 +1,96 @@
name: Testing Build
# Isolierter Test-Kanal: baut ausschließlich das Docker-Image
# gitea.it-drui.de/viewit/kx-bridge:testing (+ :testing-<shortsha>).
#
# Bewusst KEIN Gitea-Release, KEIN Tag, KEIN Rück-Push in irgendeinen Branch -
# damit ein Push nach `testing` niemals nightly oder master berührt (und
# umgekehrt: nightly.yml/release.yml/pr-check.yml triggern nicht auf
# `testing`-Pushes, da sie an nightly / v*-Tags / PRs-gegen-nightly gebunden
# sind). Der Workflow schreibt nie ins Repo zurück.
on:
push:
branches:
- testing
paths:
- '**.py'
- 'Dockerfile'
- 'requirements.txt'
- 'web/**'
- 'data/**'
- '.gitea/workflows/testing.yml'
workflow_dispatch:
jobs:
build:
runs-on: server-runner
steps:
- name: Checkout
run: |
if [ -d .git ]; then
git fetch origin testing
git reset --hard origin/testing
git clean -fd
else
git clone --branch testing https://gitea.it-drui.de/viewit/KX-Bridge-Release.git .
fi
- name: Install Docker CLI
run: |
if ! command -v docker >/dev/null 2>&1; then
ARCH=$(uname -m)
if [ "$ARCH" = "x86_64" ]; then
DARCH="x86_64"
BARCH="amd64"
else
DARCH="aarch64"
BARCH="arm64"
fi
wget -qO- "https://download.docker.com/linux/static/stable/${DARCH}/docker-27.5.1.tgz" \
| tar xz --strip-components=1 -C /usr/local/bin docker/docker
chmod +x /usr/local/bin/docker
mkdir -p /usr/local/lib/docker/cli-plugins
wget -qO /usr/local/lib/docker/cli-plugins/docker-buildx \
"https://github.com/docker/buildx/releases/download/v0.23.0/buildx-v0.23.0.linux-${BARCH}"
chmod +x /usr/local/lib/docker/cli-plugins/docker-buildx
fi
docker version --format '{{.Client.Version}}'
- name: Set up QEMU
run: |
docker run --rm --privileged tonistiigi/binfmt:latest --install all
- name: Set up buildx
run: |
docker buildx inspect kxbuilder 2>/dev/null || \
docker buildx create --name kxbuilder --use
docker buildx use kxbuilder
- name: Login to Gitea registry
run: |
echo "${{ secrets.REGISTRY_TOKEN }}" | \
docker login gitea.it-drui.de -u "${{ secrets.REGISTRY_USER }}" --password-stdin
- name: Compute testing version
run: |
# Reiner Commit-SHA-Suffix - keine Tag-Zähllogik, keine Stable-Tag-
# Abhängigkeit, kein Commit. Nur die VERSION-Datei im Arbeitsverzeichnis.
VERSION="testing-$(git rev-parse --short HEAD)"
echo "VERSION=${VERSION}" > /tmp/testing_version.env
echo "Computed testing version: ${VERSION}"
- name: Build & push (amd64 + arm64)
run: |
. /tmp/testing_version.env
# VERSION-Datei nur im Arbeitsverzeichnis für den Docker-Build setzen
# (KEIN Commit, KEIN Push).
echo "$VERSION" > VERSION
docker buildx build \
--platform linux/amd64,linux/arm64,linux/arm/v7 \
--push \
--provenance=false \
--no-cache \
-t "gitea.it-drui.de/viewit/kx-bridge:testing" \
-t "gitea.it-drui.de/viewit/kx-bridge:${VERSION}" \
.

View File

@@ -9,6 +9,7 @@ RUN pip install --no-cache-dir -r requirements.txt && \
apt-get purge -y gcc python3-dev && apt-get autoremove -y && rm -rf /var/lib/apt/lists/*
COPY kobrax_moonraker_bridge.py .
COPY bridge_*.py .
COPY web/ ./web/
# Statische Daten (orca_filaments.json etc.) liegen in /app/static/, NICHT in
# /app/data/ — letzteres wird vom User als Volume gemountet (Runtime-State).
@@ -17,6 +18,12 @@ COPY config_loader.py .
COPY env_loader.py .
COPY kobrax_client.py .
COPY orca_filaments.py .
COPY spoolman_client.py .
COPY kxgauge_client.py .
COPY gcode_store.py .
COPY gcode_meta.py .
COPY camera.py .
COPY credentials.py .
COPY VERSION .
COPY anycubic_slicer.crt .
COPY anycubic_slicer.key .

View File

@@ -1,2 +1,6 @@
## Changes in this build
- Fix: **combined ACE-RFID filament tags (e.g. "GEEETECH PLA Bas" from third-party RFID tools) still weren't auto-matching to imported OrcaSlicer profiles**, even after the first attempt at this in a previous nightly — the matching logic was only ever wired into the OrcaSlicer slicer-sync endpoint, never into the actual MQTT status path that feeds the dashboard and Happy-Hare gate data, so the dashboard kept showing the raw unmatched RFID string. Centralized the matching so all three places that resolve a slot's filament profile benefit identically. Also added variant-token disambiguation (e.g. "Bas" vs. "Matte") for when a vendor has multiple profiles of the same material, and fixed a related edge case where a manual per-slot override could be incorrectly treated as stale on an RFID-tagged spool (Issue #101, thanks @Blaim for the extensive debugging that pinpointed this).
- Feat: new setting under Settings → Print — "Delete file from printer after successful print" — automatically removes a GCode file from the printer's own storage once it finishes printing successfully, keeping only the copy in the bridge's own GCode store. Off by default, and only ever applies to files that were uploaded through the bridge itself (so there's always a backup); files started directly from the printer or Anycubic Slicer are never touched.
- Fix: **the dashboard could stay stuck showing a printer as online/"ready" indefinitely after it was physically switched off or unplugged**, discovered while testing the new smart-plug power-switch feature. Root cause was two-fold: the MQTT socket had no TCP keepalive, so a connection killed without a clean close (unplugged, not a graceful shutdown) could look alive to the OS for 15+ minutes; and even once the dead connection was detected, the status poll loop could get stuck waiting on a reconnect attempt that was already running elsewhere, so it never reached the code that flips the dashboard to "offline". Live-tested against a real printer, including that no sockets, threads, or file descriptors are left behind across repeated disconnect/reconnect cycles — a disconnected printer is now detected and reflected on the dashboard within about 15 seconds.
- Fix: a range of smaller robustness issues found in an internal code review — a single malformed MQTT message from the printer could get permanently stuck at the front of the receive buffer and force a reconnect on every subsequent poll; concurrent requests of the same type could occasionally have their responses mixed up; the camera stream could leak an orphaned ffmpeg process after a printer reboot rotated its stream URL while a new stream was already starting; `/api/settings` and `/api/update/apply` returned an unhandled server error instead of a clean "invalid request" for a malformed request body; a typo'd numeric value in `config.ini` (e.g. a stray character in the port number) could prevent the bridge from starting at all instead of falling back to the default; and a rare filament-profile-name collision during import is now logged instead of silently resolved. None of these were reported as user-facing bugs — added as defense-in-depth after a targeted review, with new tests covering each case.

862
bridge_ams.py Normal file
View File

@@ -0,0 +1,862 @@
"""
bridge_ams.py - AmsFilamentMixin for KobraXBridge.
AMS/ACE slot topology + aggregation, per-slot filament-profile resolution
(material normalization, RFID vendor matching, effective slot profile),
ACE dryer presets, the lane-data builder for OrcaSlicer sync, and the
_on_multicolor_box / _on_light MQTT callbacks (which live here because they
are all about AMS state). Mixed into KobraXBridge.
────────────────────────────────────────────────────────────────────────────
Copyright (C) 2026 viewit (KX-Bridge contributors)
Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md.
"""
import re
import time
import logging
import threading
log = logging.getLogger("bridge")
class AmsFilamentMixin:
def _default_ace_dry_presets(self) -> dict[str, dict]:
return {
"pla": {"temp": 45, "duration_sec": 4 * 3600},
"pla_plus": {"temp": 45, "duration_sec": 4 * 3600},
"petg": {"temp": 50, "duration_sec": 4 * 3600},
"tpu": {"temp": 55, "duration_sec": 4 * 3600},
"abs_asa": {"temp": 45, "duration_sec": 8 * 3600},
"pa_pc": {"temp": 55, "duration_sec": 12 * 3600},
"custom_1": {"name": "Custom 1", "temp": 45, "duration_sec": 4 * 3600},
"custom_2": {"name": "Custom 2", "temp": 45, "duration_sec": 4 * 3600},
"custom_3": {"name": "Custom 3", "temp": 45, "duration_sec": 4 * 3600},
}
def _sanitize_ace_dry_presets(self, presets: dict) -> dict[str, dict]:
out = self._default_ace_dry_presets()
for key in list(out.keys()):
src = presets.get(key) if isinstance(presets, dict) else None
if not isinstance(src, dict):
continue
try:
t = int(src.get("temp", out[key]["temp"]))
except Exception:
t = out[key]["temp"]
try:
d = int(src.get("duration_sec", out[key]["duration_sec"]))
except Exception:
d = out[key]["duration_sec"]
out[key]["temp"] = max(30, min(80, t))
out[key]["duration_sec"] = max(10 * 60, min(24 * 3600, d))
if key.startswith("custom_"):
name = str(src.get("name", out[key].get("name", key.replace("_", " ").title()))).strip()
out[key]["name"] = name or out[key].get("name", "Custom")
return out
def _load_ace_dry_presets_config(self) -> dict[str, dict]:
import configparser
defaults = self._default_ace_dry_presets()
cfg_path = self._find_config_path()
if not cfg_path.is_file():
return defaults
cfg = configparser.ConfigParser(interpolation=None)
cfg.read(cfg_path, encoding="utf-8")
sec = "ace_dry_presets"
if not cfg.has_section(sec):
return defaults
out = {}
for key, d in defaults.items():
temp_k = f"{key}_temp"
dur_k = f"{key}_duration_sec"
try:
temp = int(cfg.get(sec, temp_k, fallback=str(d["temp"])))
except Exception:
temp = d["temp"]
try:
dur = int(cfg.get(sec, dur_k, fallback=str(d["duration_sec"])))
except Exception:
dur = d["duration_sec"]
out[key] = {
"temp": max(30, min(80, temp)),
"duration_sec": max(10 * 60, min(24 * 3600, dur)),
}
if key.startswith("custom_"):
name_k = f"{key}_name"
name = cfg.get(sec, name_k, fallback=str(d.get("name", key.replace("_", " ").title()))).strip()
out[key]["name"] = name or str(d.get("name", "Custom"))
return out
@staticmethod
def _detect_filament_mode(boxes: list, head_tools_model: int = -1) -> str:
"""Detect active filament topology mode.
Modes:
- toolhead: only toolhead slots
- ace_direct: ACE channels directly mapped, no toolhead box present.
Covers one unit (Kobra X) as well as multiple daisy-chained units
(Kobra S1 with 2+ ACE Pro, Issue #95) — each unit contributes a
block of 4 global slots at box_id * 4.
- ace_hub: toolhead + ACE via hub (slot 4 as hub path)
"""
toolhead = any(b.get("id") == -1 for b in boxes)
ace = any(b.get("id", -1) >= 0 for b in boxes)
if ace and toolhead:
return "ace_hub"
if ace:
return "ace_direct"
return "toolhead"
@staticmethod
def _aggregate_slots(boxes: list, mode: str = "toolhead") -> tuple:
"""Aggregate multi_color_box list into a flat global slot list."""
toolhead = next((b for b in boxes if b.get("id") == -1), None)
ace_boxes = sorted(
[b for b in boxes if b.get("id", -1) >= 0],
key=lambda b: b["id"]
)
global_slots: list = []
global_loaded: int = -1
if mode == "toolhead":
if toolhead:
for local_idx, s in enumerate(toolhead.get("slots") or []):
s = dict(s)
s["global_index"] = local_idx
s["box_id"] = -1
global_slots.append(s)
loaded = toolhead.get("loaded_slot", -1)
if loaded >= 0:
global_loaded = loaded
return global_slots, global_loaded
if mode == "ace_direct":
# One or more ACE units, no toolhead buffer (Kobra X: 1 unit,
# Kobra S1: up to 2+ units, Issue #95). Global index =
# box_id * 4 + local slot, so the numbering matches
# _global_to_box_slot's //4-%4 fallback and stays stable
# regardless of report order.
for ace in ace_boxes:
ace_id = int(ace["id"])
base = ace_id * 4
for local_idx, s in enumerate((ace.get("slots") or [])[:4]):
s = dict(s)
s["global_index"] = base + local_idx
s["box_id"] = ace_id
global_slots.append(s)
ace_loaded = ace.get("loaded_slot", -1)
if 0 <= ace_loaded < 4:
global_loaded = base + ace_loaded
return global_slots, global_loaded
# ace_hub
if toolhead:
for local_idx, s in enumerate((toolhead.get("slots") or [])[:3]):
s = dict(s)
s["global_index"] = local_idx
s["box_id"] = -1
global_slots.append(s)
th_loaded = toolhead.get("loaded_slot", -1)
if 0 <= th_loaded <= 2:
global_loaded = th_loaded
for ace in ace_boxes:
ace_id = ace["id"]
base = 3 + ace_id * 4
for local_idx, s in enumerate(ace.get("slots") or []):
s = dict(s)
s["global_index"] = base + local_idx
s["box_id"] = ace_id
global_slots.append(s)
ace_loaded = ace.get("loaded_slot", -1)
if ace_loaded >= 0:
global_loaded = base + ace_loaded
return global_slots, global_loaded
def _global_to_box_slot(self, global_index: int) -> tuple:
"""Convert a global slot index to (box_id, local_slot_index)."""
for s in self._ams_slots:
if s.get("global_index") == global_index:
return s.get("box_id", -1), s.get("index", global_index)
ace_present = any(s.get("box_id", -1) >= 0 for s in self._ams_slots)
if self._filament_mode == "ace_direct" and ace_present:
return global_index // 4, global_index % 4
if not ace_present or global_index < 3:
return -1, global_index
offset = global_index - 3
return offset // 4, offset % 4
def _slot_to_print_ams_index(self, global_index: int) -> int:
"""Convert UI/global slot index to printer print/start ams_index.
In ace_hub mode, print/start uses global channel numbering where
toolhead channels occupy 1..3 and ACE0 starts at index 4.
"""
idx = int(global_index)
if self._filament_mode == "ace_hub":
box_id, local_slot = self._global_to_box_slot(idx)
if box_id >= 0:
return 4 + box_id * 4 + int(local_slot)
return idx
return idx
def _slot_usable_for_print(self, global_index: int) -> bool:
"""Whether a global slot can be used for current filament mode."""
slot = next((s for s in self._ams_slots if int(s.get("global_index", -1)) == int(global_index)), None)
if not slot:
return False
if int(slot.get("status", 0)) != 5:
return False
box_id = int(slot.get("box_id", -1))
if self._filament_mode == "ace_hub":
# In hub mode, toolhead channels (0..2) and ACE channels are both printable.
return box_id == -1 or box_id >= 0
if self._filament_mode == "ace_direct":
return box_id >= 0
return box_id == -1
def _loaded_slots_for_print(self) -> list[tuple[int, dict]]:
"""Loaded slots filtered for current filament mode."""
loaded = [
(int(s.get("global_index", i)), s)
for i, s in enumerate(self._ams_slots)
if s.get("status") == 5 and self._slot_usable_for_print(int(s.get("global_index", i)))
]
return loaded
def _select_loaded_slots_for_print(self, warn_on_empty_default: bool = False) -> list[tuple[int, dict]]:
"""Return loaded slots, honoring default_ams_slot when configured."""
default_slot = getattr(self._args, "default_ams_slot", "auto")
all_loaded = self._loaded_slots_for_print()
if default_slot == "auto":
return all_loaded
try:
slot_idx = int(default_slot)
except ValueError:
return all_loaded
selected = [(i, s) for i, s in all_loaded if i == slot_idx]
if selected:
return selected
if warn_on_empty_default:
log.warning(f"Default slot {slot_idx} is empty - falling back to auto")
return all_loaded
@staticmethod
def _slot_color_rgba(slot: dict) -> list[int]:
color = slot.get("color", [255, 255, 255])
if isinstance(color, list) and len(color) >= 3:
return [int(color[0]), int(color[1]), int(color[2]), 255]
return [255, 255, 255, 255]
def _build_auto_ams_box_mapping(
self,
warn_on_empty_default: bool = False,
loaded_slots: list[tuple[int, dict]] | None = None,
) -> list[dict]:
"""Build print mapping from currently loaded slots (no explicit dialog assignments)."""
loaded = loaded_slots
if loaded is None:
loaded = self._select_loaded_slots_for_print(warn_on_empty_default=warn_on_empty_default)
if not loaded:
return []
loaded_map = {gidx: s for gidx, s in loaded}
max_idx = max(loaded_map.keys())
# The printer interprets ams_box_mapping as an ordered list (entry N = TN).
# Missing slots must be inserted as placeholders, otherwise everything shifts.
# A placeholder must NOT reference a physically empty tray: the printer
# rejects such an entry even for a tool the GCode never calls (printing
# Filament 4 with the slot below it empty fails; all-full works). Point
# gap placeholders at a definitely-loaded tray instead of the gap's own
# (empty) index.
fallback_gidx = max_idx # highest loaded slot -> loaded + printable
fallback_slot = loaded_map[fallback_gidx]
fallback_ams = self._slot_to_print_ams_index(fallback_gidx)
result = []
for i in range(max_idx + 1):
if i in loaded_map:
s = loaded_map[i]
result.append({
"paint_index": i,
"ams_index": self._slot_to_print_ams_index(i),
"paint_color": [255, 255, 255, 255],
"ams_color": self._slot_color_rgba(s),
"material_type": s.get("type", "PLA"),
})
else:
result.append({
"paint_index": i,
"ams_index": fallback_ams,
"paint_color": [255, 255, 255, 255],
"ams_color": self._slot_color_rgba(fallback_slot),
"material_type": fallback_slot.get("type", "PLA"),
})
return result
def _build_assigned_ams_box_mapping(self, assignments: list) -> tuple[list[dict], int, int]:
"""Build print mapping from UI filament assignments.
Returns (mapping, unused_count, invalid_count).
"""
slot_by_global_index = {
int(s.get("global_index", i)): s
for i, s in enumerate(self._ams_slots)
}
ams_box_mapping: list[dict] = []
unused_count = 0
invalid_count = 0
for i, a in enumerate(assignments):
try:
if a.get("is_used") is False:
unused_count += 1
continue
global_slot = int(a["slot_index"])
except (ValueError, TypeError, KeyError):
invalid_count += 1
continue
if global_slot < 0:
unused_count += 1
continue
if not self._slot_usable_for_print(global_slot):
invalid_count += 1
continue
slot = slot_by_global_index.get(global_slot, {})
ams_box_mapping.append({
# Preserve slicer paint indices (can be sparse when paint 0 is unused).
"paint_index": a.get("paint_index", i),
"ams_index": self._slot_to_print_ams_index(global_slot),
"paint_color": a.get("paint_color", [255, 255, 255, 255]),
"ams_color": self._slot_color_rgba(slot),
"material_type": slot.get("type", a.get("material", "PLA")),
})
return ams_box_mapping, unused_count, invalid_count
def _box_local_to_global(self, box_id: int, local_slot: int, boxes: list) -> int:
"""Convert (box_id, local slot) to global slot index for current topology."""
if box_id == -1:
return local_slot
if self._filament_mode == "ace_direct":
# Multi-ACE (Issue #95): each unit occupies its own block of 4.
# Identical to the old `return local_slot` for a single unit (id 0).
return box_id * 4 + local_slot
return 3 + box_id * 4 + local_slot
def _slot_activity_map(self, boxes: list, global_loaded: int = -1) -> dict:
"""Build {global_slot_index: loading|unloading} from feed_status data."""
# Note: all boxes are considered — the old primary_ace_id filter (skip
# every ACE box except the first in ace_direct mode) is gone since the
# slot aggregation now handles multiple ACE units (Issue #95).
activity: dict = {}
for box in boxes:
fs = box.get("feed_status") or {}
current_status = int(fs.get("current_status", -1))
local_slot = int(fs.get("slot_index", -1))
feed_type = int(fs.get("type", -1))
if current_status in (-1, 10, 11) or local_slot < 0:
continue
box_slots = box.get("slots") or []
if local_slot >= len(box_slots) or (box_slots[local_slot] or {}).get("status") != 5:
continue
if feed_type == 1:
act = "loading"
elif feed_type == 2:
act = "unloading"
else:
continue
global_slot = self._box_local_to_global(int(box.get("id", -1)), local_slot, boxes)
if feed_type == 1 and self._pending_load_slot >= 0 and global_slot != self._pending_load_slot:
# Ignore transient firmware-reported loading slots that differ from the requested target.
if global_loaded >= 0 and global_loaded != self._pending_load_slot:
activity[global_loaded] = "unloading"
continue
if feed_type == 1 and global_loaded >= 0 and global_slot != global_loaded:
# During a slot swap the firmware reports the target slot immediately,
# while the previously loaded slot is still being unloaded first.
activity[global_loaded] = "unloading"
activity[global_slot] = act
return activity
def _on_multicolor_box(self, payload: dict):
if payload.get("state") == "failed":
req = getattr(self, "_last_ams_set_request", None)
log.warning(
f"multiColorBox setInfo rejected by printer: request={req} raw_response={payload.get('data')}"
)
self._state["last_ams_set_error"] = True
return
data = payload.get("data") or {}
if not isinstance(data, dict):
log.warning(f"multiColorBox/report: unexpected data shape: {data!r}")
return
boxes = data.get("multi_color_box") or []
if not boxes:
return
self._state["last_ams_set_error"] = False
self._head_tools_model = int(data.get("head_tools_model", self._head_tools_model))
self._filament_mode = self._detect_filament_mode(boxes, self._head_tools_model)
self._state["filament_mode"] = self._filament_mode
global_slots, global_loaded = self._aggregate_slots(boxes, self._filament_mode)
self._ams_loaded_slot = global_loaded
self._update_ace_drying_state(data, boxes)
for box in boxes:
bid = int(box.get("id", -1))
if 0 <= bid <= 3 and "auto_feed" in box:
self._ace_auto_feed[bid] = int(box["auto_feed"])
if self._pending_load_slot >= 0 and global_loaded == self._pending_load_slot:
self._pending_load_slot = -1
activity_map = self._slot_activity_map(boxes, global_loaded)
for s in global_slots:
s["activity"] = activity_map.get(s.get("global_index"), "")
# Tip forming: after feed-in (status=10) or feed-out (status=11)
# the original slicer automatically sends type=3 (extruder retract).
# Check ALL boxes so ACE-triggered events are handled correctly.
for box in boxes:
fs = box.get("feed_status") or {}
current_status = fs.get("current_status")
slot_index = fs.get("slot_index", 0)
box_id = box.get("id", -1)
if current_status in (10, 11):
def _tip_form(bi=box_id, si=slot_index, cs=current_status):
import time; time.sleep(2)
self.client.publish(
"multiColorBox", "feedFilament",
{"multi_color_box": [{"id": bi, "feed_status": {"slot_index": si, "type": 3}}]},
timeout=0
)
log.info(f"Tip forming (type=3) after status={cs} box={bi} slot={si}")
threading.Thread(target=_tip_form, daemon=True).start()
if global_slots:
self._ams_slots = global_slots
log.info(f"AMS slots received: {len(global_slots)}, loaded_slot={self._ams_loaded_slot}")
self._push_status_update()
def _update_ace_drying_state(self, data: dict, boxes: list):
"""Extract ACE drying state from multiColorBox report/getInfo payloads."""
ace_ids = sorted({int(b.get("id", -1)) for b in boxes if int(b.get("id", -1)) >= 0})
self._ace_box_ids = [i for i in ace_ids if 0 <= i <= 3]
def _num_from(src: dict, keys: tuple[str, ...], default=None):
for k in keys:
v = src.get(k)
if v is not None:
try:
return float(v)
except Exception:
return default
return default
def _humidity_from(src: dict, default=None):
return _num_from(src, ("humidity", "current_humidity", "cur_humidity", "relative_humidity", "humidity_value"), default)
def _current_temp_from(src: dict, default=None):
return _num_from(src, ("current_temp", "cur_temp", "temperature", "temp", "drying_temp", "chamber_temp"), default)
def _minutes_from(src: dict, key: str, default=0):
raw = src.get(key, default)
try:
value = int(float(raw))
except Exception:
return int(default)
# Some firmware payloads report dryer times in seconds while the UI uses minutes.
if value > (24 * 60):
return max(0, int(round(value / 60.0)))
return max(0, value)
per_unit: list[dict] = []
for box in boxes:
bid = int(box.get("id", -1))
if bid < 0:
continue
bs = box.get("drying_status") or box.get("drying_settings")
bs = bs if isinstance(bs, dict) else {}
hu = _humidity_from(bs, _humidity_from(box))
ct = _current_temp_from(bs, _current_temp_from(box))
if bs or hu is not None or ct is not None:
per_unit.append({
"id": bid,
"status": int(bs.get("status", 0)),
"target_temp": int(bs.get("target_temp", 0)),
"duration": _minutes_from(bs, "duration", 0),
"remain_time": _minutes_from(bs, "remain_time", 0),
"humidity": hu,
"current_temp": ct,
})
src = data.get("drying_status") or data.get("drying_settings")
if not isinstance(src, dict):
for box in boxes:
if int(box.get("id", -1)) < 0:
continue
cand = box.get("drying_status") or box.get("drying_settings")
if isinstance(cand, dict):
src = cand
break
if isinstance(src, dict):
cur = self._state.get("ace_drying") or {}
active = [u for u in per_unit if u.get("status", 0)]
primary = active[0] if active else (per_unit[0] if per_unit else {})
self._state["ace_drying"] = {
"status": int(src.get("status", cur.get("status", 0))),
"target_temp": int(src.get("target_temp", cur.get("target_temp", 0))),
"duration": _minutes_from(src, "duration", cur.get("duration", 0)),
"remain_time": _minutes_from(src, "remain_time", cur.get("remain_time", 0)),
"humidity": _humidity_from(src, primary.get("humidity", cur.get("humidity"))),
"current_temp": _current_temp_from(src, primary.get("current_temp", cur.get("current_temp"))),
"units": per_unit,
}
elif per_unit:
active = [u for u in per_unit if u.get("status", 0)]
primary = active[0] if active else per_unit[0]
self._state["ace_drying"] = {
"status": int(primary.get("status", 0)),
"target_temp": int(primary.get("target_temp", 0)),
"duration": int(primary.get("duration", 0)),
"remain_time": int(primary.get("remain_time", 0)),
"humidity": primary.get("humidity"),
"current_temp": primary.get("current_temp"),
"units": per_unit,
}
def _on_light(self, payload: dict):
d = payload.get("data") or {}
self._state["light_on"] = bool(d.get("status", 0))
self._state["light_brightness"] = int(d.get("brightness", 80))
self._push_status_update()
# OrcaSlicer filament preset IDs (MoonrakerPrinterAgent.cpp mapping)
# Default mapping per material type when the user has not set a slot
# profile override. For the Kobra X we prefer Anycubic's own
# filament IDs from the `@Anycubic Kobra X 0.4 nozzle` profiles - those
# are printer-specific is_compatible and are picked up by OrcaSlicer directly
# matched. Library fallbacks (OGF*) only for material types without
# Kobra X-specific Anycubic profile - their @system profiles have
# `compatible_printers: []` (= compatible with all printers).
_TRAY_INFO_IDX = {
# Anycubic-eigene Kobra-X-Profile
"PLA": "GFPLA",
"PLA+": "GFPLA+",
"PLA SILK": "GFPLA Silk",
"PLA-SILK": "GFPLA Silk",
"PLASILK": "GFPLA Silk",
"SILK PLA": "GFPLA Silk",
"PLA MATTE": "GFPLA",
"PLA-MATTE": "GFPLA",
"PLA MARBLE": "GFPLA",
"PLA WOOD": "GFPLA",
"PETG": "GFPETG",
"PETG+": "GFPETG",
"ABS": "GFABS",
"ASA": "GFASA",
"TPU": "GFTPU 95A",
"TPE": "GFTPU 95A",
"PVA": "GFPVA",
# Kein Anycubic-Kobra-X-Profil → Library-Fallback
"PLA-CF": "OGFL98",
"PLA CF": "OGFL98",
"PETG-CF": "OGFG98",
"PETG CF": "OGFG98",
"PA": "OGFN99",
"PA-CF": "OGFN98",
"PA CF": "OGFN98",
"PC": "OGFC99",
"HIPS": "OGFS98",
}
# Normalizes material type strings to the canonical key for _TRAY_INFO_IDX
# and _default_filament_name. PLA variants without an exact match fall
# back to their base family (PLA+ -> PLA+, PLA Matte -> PLA, etc.).
@staticmethod
def _normalize_material(mat: str) -> str:
m = mat.upper().strip().replace("-", " ").replace("_", " ")
# Bekannte Varianten normalisieren
_ALIASES = {
"PLAPLUS": "PLA+", "PLA PLUS": "PLA+",
"SILK PLA": "PLA SILK", "PLASILK": "PLA SILK",
"PLA MATTE": "PLA MATTE", "PLA MARBLE": "PLA MARBLE",
"PLA WOOD": "PLA WOOD",
"TPE": "TPU",
"PETG PLUS": "PETG+",
"PA6": "PA", "PA12": "PA", "PA66": "PA",
}
if m in _ALIASES:
return _ALIASES[m]
return m
@staticmethod
def _material_family(mat: str) -> str:
"""Reduce a material to its base polymer family.
PLA / PLA+ / PLA SILK / PLA MATTE -> "PLA"; PETG / PETG+ -> "PETG"; etc.
Used by the stale-profile guard: only a change of *family* (e.g. PETG ->
PLA) invalidates a saved slot profile — a change within the family
(PLA -> PLA SILK) must not discard an otherwise valid profile.
"""
if not mat:
return ""
m = AmsFilamentMixin._normalize_material(mat)
# Longer prefixes first so "PETG" is not swallowed by "PET".
for fam in ("PETG", "PLA", "ABS", "ASA", "TPU", "PVA", "HIPS", "PA", "PC", "PET"):
if m.startswith(fam):
return fam
return m
def _parse_combined_rfid_type(self, raw_type: str) -> tuple[str, str]:
"""Split a combined ACE-RFID "VENDOR TYPE SERIAL" string (e.g.
"GEEETECH PLA Bas", written via third-party RFID tools) into
(vendor, material_family).
Anycubic's ACE RFID system concatenates vendor + material + a
truncated serial/variant into one `type` string for custom tags -
unlike a normal spool report where `type` is just "PLA"/"PETG"/etc.
Returns ("", "") when the first token isn't a known vendor (from the
merged system+user filament library), which leaves plain type
strings like "PLA" completely unaffected (Issue #101).
"""
tokens = raw_type.split()
if len(tokens) < 2:
return "", ""
first = tokens[0].strip().lower()
vendors = {p.get("vendor", "").lower(): p.get("vendor", "") for p in self._load_orca_filaments()}
vendor = vendors.get(first)
if not vendor:
return "", ""
family = self._material_family(" ".join(tokens[1:]))
if not family:
return "", ""
return vendor, family
@staticmethod
def _rfid_variant_tokens(raw_type: str) -> list[str]:
"""Tokens after "VENDOR TYPE" in a combined ACE-RFID string (e.g.
["bas"] for "GEEETECH PLA Bas") - the truncated variant/serial that
distinguishes multiple profiles of the same (vendor, material family),
e.g. "Basic" vs. "Matte". Kept separate from _parse_combined_rfid_type()
so that function's 2-tuple signature (and its existing callers/tests)
stay unchanged (Issue #101)."""
tokens = raw_type.split()
return [t.lower() for t in tokens[2:]]
def _match_profile_by_vendor_family(self, vendor: str, family: str,
variant_tokens: list[str] | None = None) -> dict:
"""Find an imported/system filament profile by (vendor, material
family) - used to auto-resolve a combined ACE-RFID type string to
the user's already-imported OrcaSlicer profile (Issue #101), since
the exact profile `name` never appears verbatim in the truncated
RFID string.
When multiple profiles share the same (vendor, family) - e.g. "Geeetech
PLA Basic" and "Geeetech PLA Matte" both matching (Geeetech, PLA) -
variant_tokens (the RFID string's remaining tokens, e.g. ["bas"] for
"Basic") are scored against each candidate's name: a word-prefix match
scores higher than a plain substring match, so "bas" prefers "Basic"
over "Matte" or an unrelated profile name containing "bas" as noise.
Falls back to the first match when nothing disambiguates."""
matches = [
p for p in self._load_orca_filaments()
if p.get("vendor", "").lower() == vendor.lower()
and self._material_family(p.get("type", "")) == family
]
if not matches:
return {}
if len(matches) == 1 or not variant_tokens:
return matches[0]
best = matches[0]
best_score = -1
for p in matches:
name_words = p.get("name", "").lower().split()
score = 0
for tok in variant_tokens:
if any(w.startswith(tok) for w in name_words):
score += 2
elif tok in p.get("name", "").lower():
score += 1
if score > best_score:
best_score = score
best = p
log.debug(
f"_match_profile_by_vendor_family: {len(matches)} profiles match "
f"vendor={vendor!r} family={family!r}, variant_tokens={variant_tokens!r} "
f"-> {best.get('name')!r} (score={best_score})"
)
return best
def _profile_material(self, profile: dict) -> str:
"""Material type (e.g. "PETG") of a saved slot profile, resolved by
(vendor, name) from the Orca filament library. Returns "" when the
profile is not in the library — we do NOT guess in that case."""
name = (profile or {}).get("name", "")
if not name:
return ""
vendor = profile.get("vendor", "")
for p in self._load_orca_filaments():
if p.get("vendor") == vendor and p.get("name") == name:
return p.get("type", "") or ""
return ""
def _effective_slot_profile(self, global_idx: int, ams_material: str) -> dict:
"""Saved slot-profile override — but only while its material *family*
still matches the material currently loaded in the AMS. Falls back to
auto-resolving a combined ACE-RFID type string (Issue #101) when there
is no (usable) manual override.
Non-destructive suppression (Option A): when the family no longer matches
(e.g. a PETG profile but PLA loaded) the override is skipped → falls
through to the RFID auto-match / generic default. The override stays in
config.ini and reactivates as soon as the matching material is loaded
again. When the profile's family is unknown we do NOT suppress (fail-safe).
Centralized here (rather than duplicated per caller) so every consumer -
the dashboard's /kx/filament/slots, Happy-Hare gate data, and the
OrcaSlicer lane-data sync - benefits from RFID auto-matching identically,
instead of only the one call site that happened to also call
_parse_combined_rfid_type() directly."""
# A combined ACE-RFID string ("GEEETECH PLA Bas") carries a vendor
# prefix that _material_family() alone can't see past (it only
# strips known polymer prefixes, so "GEEETECH PLA BAS" resolves to
# itself, not "PLA") - resolve the plain material family through the
# RFID parser first so the stale-profile guard below compares against
# the actual polymer family, not the raw combined string.
vendor, family = self._parse_combined_rfid_type(ams_material)
plain_material = family or ams_material
profile = self._filament_profiles.get(global_idx) or {}
if profile.get("name"):
prof_fam = self._material_family(self._profile_material(profile))
ams_fam = self._material_family(plain_material)
if not (prof_fam and ams_fam and prof_fam != ams_fam):
return profile
if vendor:
variant_tokens = self._rfid_variant_tokens(ams_material)
auto = self._match_profile_by_vendor_family(vendor, family, variant_tokens)
if auto.get("name"):
return auto
return {}
def _build_lane_data(self) -> dict:
"""Builds BBL AMS JSON for OrcaSlicer DevFilaSystemParser::ParseV1_0.
POSITION-FAITHFUL: every physical slot keeps its position (tray id =
slot position). Empty slots are reported as placeholder trays, NOT
filtered out/compacted - otherwise colors shift to wrong positions
(e.g. slot 1=yellow, 2=empty, 3=red -> red must not land on position 2).
"""
slots = self._ams_slots
total = len(slots)
if total == 0:
return {"ams": [], "ams_exist_bits": "0", "tray_exist_bits": "0"}
ams_count = (total + 3) // 4
ams_exist_bits = 0
tray_exist_bits = 0
ams_array = []
for ams_id in range(ams_count):
ams_exist_bits |= (1 << ams_id)
tray_array = []
max_slot = min(3, total - ams_id * 4 - 1)
for slot_id in range(max_slot + 1):
slot_index = ams_id * 4 + slot_id
slot = slots[slot_index] if slot_index < total else {}
occupied = slot.get("status") == 5
if occupied:
tray_exist_bits |= (1 << slot_index)
color_raw = slot.get("color", [255, 255, 255])
if isinstance(color_raw, list) and len(color_raw) >= 3:
color_hex = "{:02X}{:02X}{:02X}FF".format(
int(color_raw[0]), int(color_raw[1]), int(color_raw[2])
)
elif isinstance(color_raw, str) and len(color_raw) >= 6:
color_hex = color_raw[:6].upper() + "FF"
else:
color_hex = "FFFFFFFF"
material = self._normalize_material(slot.get("type", "PLA"))
# User override from config.ini [filament_profiles].slot_N_id
# takes precedence over the default mapping by material type.
# The vendor is sent along (tray_sub_brands + filament_vendor),
# so a patched OrcaSlicer can match by brand + type +
# color (analogous to SnapmakerPrinterAgent).
# Three-layer resolution for the filament hint sent to OrcaSlicer,
# all handled inside _effective_slot_profile() (Issue #101):
# 1. User-Wahl (config.ini [filament_profiles]) — exakte Kontrolle
# 2. Combined ACE-RFID "VENDOR TYPE SERIAL" string (e.g.
# "GEEETECH PLA Bas") auto-matched against the user's
# already-imported profile library. Not persisted to
# config.ini - re-derives on every call, so a differently
# tagged spool loaded later isn't stuck with a stale match.
# 3. Generic fallback (_TRAY_INFO_IDX) per material type - no
# vendor hint; OrcaSlicer then picks its own generic preset
user_profile = self._effective_slot_profile(slot_index, material)
if user_profile.get("name"):
material = self._material_family(user_profile.get("type", material)) or material
vendor = user_profile.get("vendor", "")
fila_name = user_profile.get("name", "")
tray_info_idx = user_profile.get("id") or self._TRAY_INFO_IDX.get(material, "OGFL99")
else:
# Default: Library-Generic-Profil (siehe _default_filament_name) —
# is compatible with all printers and guaranteed to be visible.
# The user deliberately picks a concrete brand per slot if they
# want one; the default stays neutral.
fila_name = self._default_filament_name(material)
vendor = "Generic" if fila_name.startswith("Generic ") else ""
tray_info_idx = self._lookup_filament_id(vendor, fila_name) or self._TRAY_INFO_IDX.get(material, "OGFL99")
tray_array.append({
"id": str(slot_id),
"tag_uid": "0000000000000000",
"tray_info_idx": tray_info_idx,
"tray_type": material,
"tray_color": color_hex,
"tray_sub_brands": vendor,
# OrcaSlicer-Empfangs-Patch PR #13719 erwartet `name` +
# `vendor_name` pro Lane (Stufen-Matching: Vendor+Name → Name →
# filament_id_by_type). We send both spellings so that
# older patch variants + future upstream PRs are both
# covered.
"name": fila_name,
"vendor_name": vendor,
# Aliases for older patch variants (variant 2,
# MoonrakerPrinterAgent.cpp): filament_id direkt (exakt),
# otherwise resolve the preset name via find_preset().
"filament_id": tray_info_idx,
"filament_vendor": vendor,
"filament_name": fila_name,
"preset": fila_name,
})
else:
tray_array.append({
"id": str(slot_id),
"tag_uid": "0000000000000000",
"tray_info_idx": "",
"tray_type": "",
"tray_color": "00000000",
"tray_slot_placeholder": "1",
})
ams_array.append({"id": str(ams_id), "info": "0002", "tray": tray_array})
return {
"ams": ams_array,
"ams_exist_bits": format(ams_exist_bits, "X"),
"tray_exist_bits": format(tray_exist_bits, "X"),
}

56
bridge_constants.py Normal file
View File

@@ -0,0 +1,56 @@
"""
bridge_constants.py - shared constants for the bridge modules.
Extracted so the mixin modules can import these without a circular dependency
on the kobrax_moonraker_bridge facade. Re-exported from there for callers.
────────────────────────────────────────────────────────────────────────────
Copyright (C) 2026 viewit (KX-Bridge contributors)
Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md.
"""
# Maps the printer's own MQTT state strings to Klipper/Moonraker print states.
KOBRA_TO_KLIPPER_STATE = {
"free": "standby",
"busy": "printing",
"printing": "printing",
"preheating": "printing",
"auto_leveling": "printing",
"checking": "printing",
"updated": "printing",
"init": "printing",
"pausing": "paused",
"paused": "paused",
"resuming": "printing",
"resumed": "printing",
"stopping": "printing",
"stoped": "standby",
"finished": "complete",
"failed": "error",
"canceled": "standby",
}
MOONRAKER_VERSION = "v0.9.3-1"
KLIPPER_VERSION = "v0.12.0-1"
# Default kobra_state -> KXGauge emotion mapping (used when [kxgauge_mapping]
# is absent or incomplete in config.ini).
DEFAULT_KXGAUGE_MAPPING = {
"free": "neutral",
"printing": "happy",
"paused": "worried",
"pause": "worried",
"finished": "glee",
"error": "scared",
"offline": "tired",
}
# Valid KXGauge emotion names (see kxgauge/API.md) - used to reject garbage
# values coming in through the settings save.
KXGAUGE_VALID_EMOTIONS = {
"neutral", "happy", "angry", "tired", "surprised", "curious", "confused",
"sleepy", "love", "dizzy", "crying", "focused", "glee", "sad", "worried",
"annoyed", "skeptic", "frustrated", "unimpressed", "suspicious", "squint",
"furious", "scared", "awe",
}

2688
bridge_endpoints.py Normal file

File diff suppressed because it is too large Load Diff

60
bridge_logging.py Normal file
View File

@@ -0,0 +1,60 @@
"""
bridge_logging.py - browser log stream plumbing for the bridge.
Holds the ring buffer + SSE queues that feed the Web UI's live log view, the
logging.Handler that populates them, and the verbose-HTTP-log toggle.
Extracted from kobrax_moonraker_bridge.py; the buffer/queues are re-imported
there so the log-stream/download endpoints keep working against the same
shared objects.
────────────────────────────────────────────────────────────────────────────
Copyright (C) 2026 viewit (KX-Bridge contributors)
Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md.
"""
import collections as _collections
import logging
def _set_verbose_http_log(enabled: bool):
logging.getLogger("aiohttp.access").setLevel(logging.INFO if enabled else logging.WARNING)
# Ring buffer for the browser log stream + the open SSE consumer queues.
# These are shared mutable objects: kobrax_moonraker_bridge re-imports the
# same buffer/list so its log-stream/download handlers append/read the very
# same instances this handler writes to.
_log_buffer: "_collections.deque[dict]" = _collections.deque(maxlen=500)
_log_sse_queues: list = []
class _BrowserLogHandler(logging.Handler):
"""Sends log records to the ring buffer and all open SSE queues."""
_fmt = logging.Formatter(datefmt="%H:%M:%S")
def emit(self, record: logging.LogRecord):
msg = record.getMessage()
# Pass exceptions with traceback through to the browser (otherwise the
# user only sees "Error: X" without context).
if record.exc_info:
try:
msg += "\n" + self._fmt.formatException(record.exc_info)
except Exception:
pass
entry = {
"ts": self._fmt.formatTime(record, "%H:%M:%S"),
"lvl": record.levelname,
"name": record.name,
"msg": msg,
}
_log_buffer.append(entry)
for q in list(_log_sse_queues):
try:
q.put_nowait(entry)
except Exception:
pass
_browser_handler = _BrowserLogHandler()
logging.getLogger().addHandler(_browser_handler)

289
bridge_moonraker.py Normal file
View File

@@ -0,0 +1,289 @@
"""
bridge_moonraker.py - MoonrakerCompatMixin for KobraXBridge.
The Moonraker/Klipper-compatible HTTP surface (/server/*, /printer/*,
/machine/*) that Mainsail/Fluidd/OrcaSlicer/moonraker-obico talk to:
server/printer info, printer.objects query/list/subscribe, files list +
metadata, history, webcams, access api-key + update-manager stubs.
Mixed into KobraXBridge.
────────────────────────────────────────────────────────────────────────────
Copyright (C) 2026 viewit (KX-Bridge contributors)
Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md.
"""
import sys
import time
import logging
from aiohttp import web
from bridge_constants import MOONRAKER_VERSION, KLIPPER_VERSION
log = logging.getLogger("bridge")
class MoonrakerCompatMixin:
async def handle_server_info(self, request):
return web.json_response({
"result": {
"klippy_connected": True,
"klippy_state": "ready",
"components": ["file_manager", "job_state", "virtual_sdcard"],
"failed_components":[],
"registered_directories": ["gcodes"],
"warnings": [],
"websocket_count": len(self.ws_clients),
"moonraker_version": MOONRAKER_VERSION,
"api_version": [1, 3, 0],
"api_version_string": "1.3.0",
}
})
async def handle_printer_info(self, request):
s = self._state
return web.json_response({
"result": {
"state": "ready",
"state_message": "Printer is ready",
"hostname": "kobrax-bridge",
"klipper_path": "/home/pi/klipper",
"python_path": "/home/pi/klippy-env/bin/python",
"log_file": "/tmp/klippy.log",
"config_file": "/home/pi/printer.cfg",
"software_version": KLIPPER_VERSION,
"cpu_info": s["printer_name"],
}
})
async def handle_machine_system_info(self, request):
return web.json_response({
"result": {
"system_info": {
"cpu_info": {"cpu_count": 4, "bits": "64bit", "processor": "armv7l",
"cpu_desc": "Anycubic Kobra X Bridge", "serial_number": "",
"hardware_desc": "", "model": "Kobra X Bridge",
"total_memory": 524288, "memory_units": "kB"},
"sd_info": {},
"distribution": {"name": "Linux", "id": "linux", "version": "1.0",
"version_parts": {}, "like": "", "codename": ""},
"available_services": [],
"service_state": {},
"python": {"version": list(sys.version_info[:3]), "version_string": sys.version},
"network": {},
"canbus": {},
}
}
})
async def handle_objects_query(self, request):
objects = self._build_printer_objects()
requested = []
query = request.rel_url.query
if "objects" in query:
requested = [x.strip() for x in str(query.get("objects", "")).split(",") if x.strip()]
elif query:
requested = [k for k in query.keys() if k]
filtered = {k: objects[k] for k in requested if k in objects} if requested else objects
return web.json_response({"result": {"status": filtered, "eventtime": time.time()}})
async def handle_objects_list(self, request):
return web.json_response({
"result": {
"objects": list(self._build_printer_objects().keys())
}
})
async def handle_objects_subscribe(self, request):
return web.json_response({
"result": {
"status": self._build_printer_objects(),
"eventtime": time.time(),
}
})
async def handle_files_list(self, request):
filename = self._state.get("filename", "")
files = []
if filename:
files.append({
"path": filename,
"modified": time.time(),
"size": 0,
"permissions": "rw",
})
return web.json_response({"result": files})
def _build_file_metadata(self, filename: str) -> dict:
"""Builds the Moonraker file metadata for a file. Shared source
for HTTP /server/files/metadata AND the WS RPC server.files.metadata
(previously the WS path had its own broken logic with a non-existent
existierenden Store-Methode → leere Antwort → Mobileraker fragte in
endless loop, app hung on refresh, Issue #48).
Liefert Mobileraker-kompatible Pflichtfelder: `filename`, `size`,
`modified` are non-nullable in GCodeFile; `print_start_time` and the
Slicer-Felder optional."""
s = self._state
# Live _state values are only relevant for the currently/last tracked
# job's own file - using them as a starting point for a DIFFERENT
# filename leaked the tracked job's layer count/time into unrelated
# metadata queries (Issue #102). For any other filename, rely solely
# on that file's own GCodeStore row.
is_tracked_file = bool(filename) and filename == s.get("filename")
layer_h = float(s.get("layer_height") or 0.0) if is_tracked_file else 0.0
first_h = float(s.get("first_layer_height") or 0.0) if is_tracked_file else 0.0
total_layers = int(s.get("total_layers") or 0) if is_tracked_file else 0
est_time = int(s.get("slicer_time") or 0) if is_tracked_file else 0
size_bytes = 0
try:
gf = self._store.get_file_by_name(filename) or {}
if not layer_h:
layer_h = float(gf.get("layer_height") or 0.0)
first_h = float(gf.get("first_layer_height") or layer_h)
if not total_layers:
total_layers = int(gf.get("layer_count") or 0)
if not est_time:
est_time = int(gf.get("est_print_time_sec") or 0)
size_bytes = int(gf.get("size_bytes") or 0)
except Exception:
pass
# Third fallback: the printer's own buried/report analytics event
# (fires once per print start regardless of slicer), for files that
# are neither the currently-tracked job nor in our own GCodeStore -
# e.g. printed directly via Anycubic Slicer Next (Issue #102).
buried = self._buried_cache
if buried and buried.get("task_name") == filename:
if not total_layers:
total_layers = buried.get("total_layers") or total_layers
if not est_time:
est_time = buried.get("estimate_duration") or est_time
if not size_bytes:
size_bytes = buried.get("gcode_size") or size_bytes
if not layer_h:
layer_h = self._layer_height_from_filename(filename)
if layer_h and not first_h:
first_h = layer_h
object_height = round(first_h + max(0, total_layers - 1) * layer_h, 3) if (layer_h and total_layers) else 0.0
return {
"filename": filename,
# GCodeFile (Mobileraker) requires size as a non-nullable int.
"size": size_bytes or 1,
"modified": time.time(),
"estimated_time": est_time or None,
"layer_height": layer_h or None,
"first_layer_height": first_h or None,
"layer_count": total_layers or None,
"object_height": object_height or None,
"thumbnails": [],
}
async def handle_files_metadata(self, request):
"""Moonraker /server/files/metadata — moonraker-obico + Mobileraker
holen Datei-Metadaten (Slicer-Zeit, Layer, object_height).
Logic in _build_file_metadata (shared with WS RPC)."""
filename = request.rel_url.query.get("filename", "") or self._state.get("filename", "")
if not filename:
return web.json_response({"result": {}})
return web.json_response({"result": self._build_file_metadata(filename)})
# -- Moonraker stubs for moonraker-obico ----------------------------------
async def handle_access_api_key(self, request):
"""Moonraker /access/api_key - we have no auth, return a dummy.
moonraker-obico logs a WARNING otherwise."""
return web.json_response({"result": "kx-bridge-no-auth-required"})
async def handle_machine_update_status(self, request):
"""Moonraker /machine/update/status - Obico uses this to show installed plugins."""
return web.json_response({
"result": {
"busy": False,
"github_rate_limit": 60,
"github_requests_remaining": 60,
"github_limit_reset_time": time.time() + 3600,
"version_info": {},
}
})
async def handle_history_list(self, request):
"""Moonraker /server/history/list - job history from the GCodeStore.
moonraker-obico only uses the last element (limit=1, order=desc)."""
try:
limit = int(request.rel_url.query.get("limit", "50"))
except ValueError:
limit = 50
try:
jobs = self._store.list_jobs(limit=limit) or []
except Exception:
jobs = []
# Mapping to the Moonraker schema. Moonraker returns start_time as a Unix
# timestamp (float), not an ISO string - moonraker-obico parses it with
# int(start_time) and crashes otherwise.
def _to_unix_ts(iso: str | None) -> float:
if not iso:
return 0.0
try:
from datetime import datetime
# Format from GCodeStore: "2026-05-27T21:22:25Z"
dt = datetime.strptime(iso, "%Y-%m-%dT%H:%M:%SZ")
return dt.replace(tzinfo=__import__("datetime").timezone.utc).timestamp()
except Exception:
return 0.0
result_jobs = []
for j in jobs:
start_ts = _to_unix_ts(j.get("started_at"))
dur = j.get("duration_sec") or 0
result_jobs.append({
"job_id": j.get("id"),
"exists": True,
"end_time": (start_ts + dur) if start_ts and dur else None,
"filament_used": 0.0,
"filename": j.get("filename", ""),
"metadata": {},
"print_duration": dur,
"status": j.get("status") or "completed",
"start_time": start_ts,
"total_duration": dur,
})
return web.json_response({"result": {"count": len(result_jobs), "jobs": result_jobs}})
async def handle_webcams_list(self, request):
"""Moonraker /server/webcams/list - Obico fetches the webcam URLs here.
When the client comes from another host (e.g. moonraker-obico on a
separate server), it needs absolute URLs to reach the stream.
A Host header with localhost/127.0.0.1 is replaced by the real LAN IP."""
host_hdr = request.headers.get("Host", "") if request else ""
host_name = (host_hdr or "").split(":")[0]
port_part = f":{host_hdr.split(':')[1]}" if ":" in (host_hdr or "") else f":{self._args.port}"
local_ip = getattr(self, "_local_ip", None) or host_name
if host_name in ("localhost", "127.0.0.1", ""):
host_name = local_ip
base = f"http://{host_name}{port_part}"
stream_url = f"{base}/api/camera/stream"
snapshot_url = f"{base}/api/camera/snapshot"
return web.json_response({
"result": {
"webcams": [
{
"name": "KX-Bridge",
"location": "printer",
"service": "mjpegstreamer",
"enabled": True,
"icon": "mdiWebcam",
"target_fps": 5,
"target_fps_idle": 2,
"stream_url": stream_url,
"snapshot_url": snapshot_url,
"flip_horizontal": False,
"flip_vertical": False,
"rotation": 0,
"aspect_ratio": "16:9",
"extra_data": {},
}
]
}
})

418
bridge_mqtt.py Normal file
View File

@@ -0,0 +1,418 @@
"""
bridge_mqtt.py - MqttCallbacksMixin for KobraXBridge.
The MQTT reader-thread callbacks (_on_temp/_on_print/_on_info/_on_skip/
_on_buried/_on_file) and their helpers (_wait_for_file_action,
_delete_printer_file_fire_and_forget, _apply_preprint_skip_after_start).
Mixed into KobraXBridge; relies on shared bridge state (self._state,
self.client, self._store, self._current_job_*, self._spoolman*, etc.).
────────────────────────────────────────────────────────────────────────────
Copyright (C) 2026 viewit (KX-Bridge contributors)
Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md.
"""
import copy
import time
import asyncio
import logging
import threading
try:
import config_loader as env_loader
except ImportError:
import env_loader
from bridge_constants import KOBRA_TO_KLIPPER_STATE
log = logging.getLogger("bridge")
class MqttCallbacksMixin:
# -------------------------------------------------------------------------
# MQTT callbacks (called from reader thread)
# -------------------------------------------------------------------------
def _on_temp(self, payload: dict):
d = payload.get("data") or {}
self._state["nozzle_temp"] = float(d.get("curr_nozzle_temp", 0))
self._state["nozzle_target"] = float(d.get("target_nozzle_temp", 0))
self._state["bed_temp"] = float(d.get("curr_hotbed_temp", 0))
self._state["bed_target"] = float(d.get("target_hotbed_temp", 0))
if self._kxgauge:
self._kxgauge.set_heat_celsius(self._state["nozzle_temp"])
self._push_status_update()
def _on_print(self, payload: dict):
d = payload.get("data") or {}
kobra_state = payload.get("state", "")
self._state["print_state"] = KOBRA_TO_KLIPPER_STATE.get(kobra_state, "printing")
if kobra_state:
self._state["kobra_state"] = kobra_state
if self._kxgauge:
emotion = self._kxgauge_mapping.get(kobra_state)
if emotion:
self._kxgauge.set_emotion(emotion)
# Automatically switch on the camera at print start (settings option).
# Centralized here so it covers all print start paths (OrcaSlicer + UI).
# _camera_autostarted verhindert Mehrfach-Trigger pro Druck.
if kobra_state == "printing":
if (getattr(self._args, "camera_on_print", 0)
and not self._camera_autostarted
and not self._camera_user_stopped):
self._camera_autostarted = True
try:
self.client.start_camera()
log.info("Camera switched on automatically at print start")
except Exception as e:
log.warning(f"Camera auto-start failed: {e}")
elif kobra_state in ("free", "finished", "stoped", "canceled"):
self._camera_autostarted = False
self._camera_user_stopped = False # release for the next print
if kobra_state in ("pause", "paused"):
pause_msg = payload.get("msg", "")
if pause_msg:
error_code = payload.get("code", 0)
self._state["error_code"] = error_code
self._state["pause_msg"] = pause_msg
log.warning(f"Printer paused: [{error_code}] {pause_msg}")
elif kobra_state in ("resuming", "resumed", "printing", "finished", "stoped", "canceled"):
self._state["error_code"] = 0
self._state["pause_msg"] = ""
# Job-History: Druckstart erkennen
if kobra_state == "printing" and not self._current_job_id:
filename = d.get("filename", self._state.get("filename", ""))
if filename:
gf = self._store.get_file_by_name(filename)
if gf:
self._current_job_id = self._store.start_job(
gcode_file_id=gf["id"],
printer_id=self._printer_id,
)
self._current_job_filename = filename
log.info(f"Job started: {self._current_job_id} for {filename}")
self._spoolman_slot_usage = {}
self._spoolman_slot_reported = {}
self._spoolman_last_usage = 0.0
self._spoolman_last_sync = 0.0
# Job-History: Druckende erkennen
if kobra_state in ("finished",) and self._current_job_id:
self._store.finish_job(self._current_job_id, status="completed")
log.info(f"Job abgeschlossen: {self._current_job_id}")
self._spoolman_notify_end()
self._current_job_id = ""
# Optional cleanup (Settings -> Print): only for files that are
# also backed by the bridge's own GCode store - never for prints
# started directly from the printer/Anycubic Slicer, which would
# otherwise be deleted with no copy left anywhere (Issue: delete
# printer file after successful print). Deliberately only on a
# clean "finished" - stoped/canceled prints keep their file.
if getattr(self._args, "delete_printer_file_after_print", 0) and self._current_job_filename:
self._delete_printer_file_fire_and_forget(self._current_job_filename)
self._current_job_filename = ""
elif kobra_state in ("stoped", "canceled") and self._current_job_id:
self._store.finish_job(self._current_job_id, status="cancelled")
log.info(f"Job abgebrochen: {self._current_job_id}")
self._spoolman_notify_end()
self._current_job_id = ""
self._current_job_filename = ""
# Terminal states (successful finish AND stop/cancel) must leave the
# same clean end state - a "finished" print used to only clear
# file_ready (Issue #29), leaving progress/filename/duration/layer
# fields stuck at the last job's values until the *next* print
# happened to overwrite them (Issue #102).
if kobra_state in ("finished", "stoped", "canceled"):
self._state["progress"] = 0.0
self._state["filename"] = ""
self._state["file_ready"] = ""
self._state["print_duration"] = 0
self._state["remain_time"] = 0
self._state["slicer_time"] = 0
self._state["layer_height"] = 0.0
self._state["first_layer_height"] = 0.0
self._state["supplies_usage"] = 0
self._state["curr_layer"] = 0
self._state["total_layers"] = 0
self._thumbnail_b64 = ""
else:
# Only adopt the payload's filename outside terminal states - the
# printer often still reports the just-finished job's filename in
# the same "finished"/"stoped"/"canceled" message that triggered
# the reset above, which would otherwise immediately undo it.
self._state["filename"] = d.get("filename", self._state["filename"])
# Pre-print phases (leveling/preheating/checking) report their own
# "progress" - passing it through would make display_status.progress/
# virtual_sdcard.progress jump non-monotonically once real printing
# starts and the value resets (Issue #102).
if "progress" in d and kobra_state not in ("preheating", "auto_leveling", "checking", "updated", "init"):
self._state["progress"] = float(d["progress"]) / 100.0
if "print_time" in d:
self._state["print_duration"] = int(d["print_time"]) * 60
if "remain_time" in d:
self._state["remain_time"] = int(d["remain_time"]) * 60
if "curr_layer" in d:
self._state["curr_layer"] = d["curr_layer"]
if "total_layers" in d:
self._state["total_layers"] = d["total_layers"]
if "taskid" in d:
self._state["taskid"] = str(d["taskid"])
if "supplies_usage" in d:
self._state["supplies_usage"] = int(d["supplies_usage"])
settings = d.get("settings") or {}
if "print_speed_mode" in settings:
self._state["print_speed_mode"] = int(settings["print_speed_mode"])
self._push_status_update()
def _on_info(self, payload: dict):
d = payload.get("data") or {}
# Only adopt the MQTT name if no custom name is set (env or per-printer config)
if not env_loader.get("BRIDGE_PRINTER_NAME") and not getattr(self, "_name_locked", False):
self._state["printer_name"] = d.get("printerName", self._state["printer_name"])
self._state["firmware_version"] = d.get("version", self._state["firmware_version"])
# The real print state lives in info/report inside the nested
# project.state ("printing"/"paused"/...). The top-level data.state is only
# the device state ("busy"/"free") and would swallow "paused".
project = d.get("project") or {}
proj_state = project.get("state", "")
kobra_state = proj_state or d.get("state", "")
if kobra_state:
self._state["print_state"] = KOBRA_TO_KLIPPER_STATE.get(kobra_state, "standby")
self._state["kobra_state"] = kobra_state
# Hide the upload banner after the print ends (Issue #29) - the state also
# arrives via info/report (project.state) depending on the printer, not only print/report.
# Layer fields must reset here too (Issue #102) - info/report is the
# only source for curr_layer/total_layers on some printers, and they
# otherwise stay stuck at the last job's values indefinitely.
if kobra_state in ("finished", "stoped", "canceled"):
self._state["file_ready"] = ""
self._state["curr_layer"] = 0
self._state["total_layers"] = 0
# Camera auto-start here as well (OrcaSlicer often reports the start via info/report).
# The _camera_autostarted guard prevents a double start with _on_print.
if kobra_state == "printing":
if (getattr(self._args, "camera_on_print", 0)
and not self._camera_autostarted
and not self._camera_user_stopped):
self._camera_autostarted = True
try:
self.client.start_camera()
log.info("Camera switched on automatically at print start")
except Exception as e:
log.warning(f"Camera auto-start failed: {e}")
elif kobra_state in ("free", "finished", "stoped", "canceled"):
self._camera_autostarted = False
self._camera_user_stopped = False # release for the next print
if project:
if "filename" in project:
self._state["filename"] = project["filename"]
# Same non-monotonic-progress guard as _on_print (Issue #102).
if "progress" in project and kobra_state not in ("preheating", "auto_leveling", "checking", "updated", "init"):
self._state["progress"] = float(project["progress"]) / 100.0
if "print_time" in project:
self._state["print_duration"] = int(project["print_time"]) * 60
if "remain_time" in project:
self._state["remain_time"] = int(project["remain_time"]) * 60
if "curr_layer" in project:
self._state["curr_layer"] = project["curr_layer"]
if "total_layers" in project:
self._state["total_layers"] = project["total_layers"]
t = d.get("temp") or {}
if t:
self._state["nozzle_temp"] = float(t.get("curr_nozzle_temp", 0))
self._state["nozzle_target"] = float(t.get("target_nozzle_temp", 0))
self._state["bed_temp"] = float(t.get("curr_hotbed_temp", 0))
self._state["bed_target"] = float(t.get("target_hotbed_temp", 0))
urls = d.get("urls") or {}
if urls.get("fileUploadurl"):
self._state["upload_url"] = urls["fileUploadurl"]
if urls.get("rtspUrl"):
self._state["camera_url"] = urls["rtspUrl"]
self.camera_cache.set_url(urls["rtspUrl"])
fan = d.get("fan_speed_pct")
if fan is not None:
self._state["fan_speed"] = int(fan)
speed_mode = d.get("print_speed_mode")
if speed_mode is not None:
self._state["print_speed_mode"] = int(speed_mode)
self._push_status_update()
def _on_skip(self, payload: dict):
"""skip/report-Callback (Part-Skip-Feature, v0.9.10).
The printer ALWAYS reports the list of already-skipped objects here
(objects_skip_parts), whether on query_obj or after skip/start.
The full object list comes from file/report.
"""
d = payload.get("data") or {}
skipped = d.get("objects_skip_parts") or d.get("skipped") or d.get("skipped_parts") or []
# While a pre-print skip is still pending, ignore empty early reports
# so the UI doesn't snap back before the printer confirms the skip.
now = time.time()
if (not skipped and self._pending_preprint_skip
and now <= self._pending_preprint_skip_deadline):
return
# During an active print, skip states are effectively monotonic.
# Some firmware reports come back empty/partial in between;
# those must not remove already-confirmed skip objects from the UI.
existing_skipped = [str(n) for n in (self._skip_state.get("skipped") or []) if n]
existing_set = set(existing_skipped)
incoming_skipped = [str(n) for n in (skipped or []) if n]
incoming_set = set(incoming_skipped)
active_print = self._state.get("print_state") in ("printing", "paused")
if active_print and existing_set:
if not incoming_set:
skipped = list(existing_skipped)
elif not incoming_set.issuperset(existing_set):
merged = list(existing_skipped)
for n in incoming_skipped:
if n not in existing_set:
merged.append(n)
skipped = merged
# Release the pending lock once the printer confirms the requested objects
if self._pending_preprint_skip and set(skipped) >= set(self._pending_preprint_skip):
self._pending_preprint_skip = []
self._pending_preprint_skip_deadline = 0.0
self._skip_state = {
"skipped": list(skipped),
"ts": int(time.time()),
}
if payload.get("state") == "done" or payload.get("code") == 200:
log.info(f"Skip response: state={payload.get('state')} code={payload.get('code')} skipped={skipped}")
def _delete_printer_file_fire_and_forget(self, filename: str) -> None:
"""Deletes a file from the printer's own storage without waiting for
the response - called from _on_print(), which runs on the MQTT
reader thread itself, so blocking here (like _wait_for_file_action
does) would deadlock: the file/report reply that would unblock it is
dispatched from that same thread. Fire-and-forget is safe because the
bridge's own copy in the GCode store is what matters for correctness
here; a failed delete just leaves the printer's storage as it is
(Settings -> Print -> "Delete file from printer after successful print")."""
try:
self.client.publish(
"file", "deleteBatch",
{"root": "local", "files": [{"path": "/", "filename": filename}]},
timeout=0,
)
log.info(f"Requested printer-storage delete for {filename} after successful print")
except Exception as e:
log.warning(f"Delete-after-print request failed for {filename}: {e}")
def _wait_for_file_action(self, action: str, send_fn, timeout: float = 8.0) -> dict | None:
"""Sends a file/* MQTT request (via send_fn, which must call
self.client.publish(..., timeout=0) fire-and-forget) and blocks the
calling thread until a matching file/report with this `action`
arrives via _on_file, or the timeout elapses.
Needed because the printer's publish() return value for actions like
listLocal/deleteBatch is just a generic immediate ACK skeleton
(code=0, empty fields) - the real response is a separate, later
file/report message, same as the existing fileDetails pattern.
Must be called from a worker thread (e.g. via run_in_executor), not
the asyncio event loop, since it blocks on a threading.Event.
"""
event = threading.Event()
waiter = {"event": event, "result": None}
self._file_action_waiters[action] = waiter
try:
send_fn()
event.wait(timeout)
return waiter["result"]
finally:
if self._file_action_waiters.get(action) is waiter:
del self._file_action_waiters[action]
def _on_buried(self, payload: dict):
"""buried/report - the printer's own analytics event, fired once per
print start (verified live against a real Kobra X: fires identically
for prints started via Anycubic Slicer Next and via OrcaSlicer/the
bridge). Carries gcode_size/estimate_duration/total_layers, which
_build_file_metadata() falls back to for files not in our own
GCodeStore (Issue #102), plus printer storage usage."""
d = payload.get("data") or {}
task_name = d.get("task_name") or ""
if not task_name:
return
self._buried_cache = {
"task_name": task_name,
"gcode_size": int(d.get("gcode_size") or 0),
"estimate_duration": int(d.get("estimate_duration") or 0),
"total_layers": int(d.get("total_layers") or 0),
}
self._state["storage_total_mb"] = int(d.get("storage_total") or 0)
self._state["storage_used_mb"] = int(d.get("storage_used") or 0)
log.info(
f"buried/report: {task_name} size={d.get('gcode_size')} "
f"est={d.get('estimate_duration')}s layers={d.get('total_layers')}"
)
def _on_file(self, payload: dict):
# Deliver to any pending listLocal/deleteBatch waiter first (see
# _wait_for_file_action) - these actions carry no file_details/
# thumbnail payload of their own, so this doesn't interfere with the
# handling below.
action = payload.get("action") or ""
waiter = self._file_action_waiters.get(action)
if waiter is not None:
waiter["result"] = payload
waiter["event"].set()
d = payload.get("data") or {}
details = d.get("file_details") or {}
thumb = details.get("thumbnail") or details.get("png_image") or ""
file_name = d.get("filename") or details.get("filename") or self._last_uploaded_file
active_print = self._state.get("print_state") in ("printing", "paused")
current_print_file = self._state.get("filename") or ""
# Uploads during a running print must not overwrite the active
# progress preview.
if thumb and (not active_print or (file_name and file_name == current_print_file)):
self._thumbnail_b64 = thumb
log.info(f"Thumbnail received: {len(thumb)} base64 chars")
# Part-Skip: Objekt-Liste + optionales SVG (v0.9.10)
objs = details.get("objects_skip_parts") or []
svg = details.get("svg_image") or ""
if objs:
filename = file_name
if filename:
try:
self._store.update_file_objects(filename, objs, svg)
log.info(f"Skip objects for {filename}: {len(objs)} ({'with SVG' if svg else 'no SVG'})")
except Exception as e:
log.warning(f"update_file_objects failed: {e}")
self._push_status_update()
def _apply_preprint_skip_after_start(self, names: list[str], retries: int = 20, delay_s: float = 0.75):
"""Sends the skip command only after the printer switched to the printing state.
Before that, the command goes nowhere (no active print).
"""
wanted = [str(n) for n in (names or []) if isinstance(n, str) and n]
if not wanted:
return False
for i in range(max(1, int(retries))):
try:
if self._state.get("print_state") not in ("printing", "paused"):
time.sleep(max(0.1, float(delay_s)))
continue
resp = self.client.skip_objects(wanted)
if resp is not None:
log.info(f"Pre-Print skip applied ({len(wanted)} objects) on attempt {i+1}/{retries}")
self._pending_preprint_skip = []
self._pending_preprint_skip_deadline = 0.0
return True
except Exception as e:
log.debug(f"Pre-Print skip attempt {i+1}/{retries} failed: {e}")
time.sleep(max(0.1, float(delay_s)))
log.warning(f"Pre-Print skip could not be confirmed after {retries} attempts")
self._pending_preprint_skip = []
self._pending_preprint_skip_deadline = 0.0
return False

154
bridge_spoolman.py Normal file
View File

@@ -0,0 +1,154 @@
"""
bridge_spoolman.py - SpoolmanMixin for KobraXBridge.
Filament-usage attribution + reporting to a Spoolman server, and the
/kx/spoolman/* endpoints. Mixed into KobraXBridge; relies on the shared bridge
state (self._state, self._store, self._spoolman*, self._json_cors) provided by
the core class.
────────────────────────────────────────────────────────────────────────────
Copyright (C) 2026 viewit (KX-Bridge contributors)
Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md.
"""
import asyncio
import logging
import threading
log = logging.getLogger("bridge")
class SpoolmanMixin:
# ── Spoolman helpers ──────────────────────────────────────────────────────
def _spoolman_filament_mm(self) -> float:
"""Total filament_used_mm for the current print file from the GCode DB."""
filename = self._state.get("filename", "")
if not filename:
return 0.0
try:
gf = self._store.get_file_by_name(filename)
return float(gf.get("filament_used_mm") or 0.0) if gf else 0.0
except Exception:
return 0.0
def _spoolman_attribute_tick(self, activity_map: dict) -> None:
"""Attribute the supplies_usage delta since last tick to the active slot.
Skips attribution during loading/unloading transitions (tool changes +
purges) to avoid charging the wrong spool for purge material."""
if not self._spoolman or not self._spoolman_slot_spools:
return
if self._state.get("print_state") != "printing":
return
current = self._state.get("supplies_usage", 0)
delta = current - self._spoolman_last_usage
self._spoolman_last_usage = current
if delta <= 0:
return
loaded = self._ams_loaded_slot
if loaded < 0:
return
if activity_map.get(loaded):
return
self._spoolman_slot_usage[loaded] = self._spoolman_slot_usage.get(loaded, 0.0) + delta
def _spoolman_unreported(self) -> dict[int, float]:
"""Return {slot_idx: mm} of usage not yet reported to Spoolman.
Falls back to equal split of total supplies_usage when per-slot
attribution data is absent (e.g. single-extruder with no AMS)."""
total_used = self._state.get("supplies_usage", 0)
if self._spoolman_slot_usage:
return {
slot: self._spoolman_slot_usage.get(slot, 0.0)
- self._spoolman_slot_reported.get(slot, 0.0)
for slot in self._spoolman_slot_spools
}
n = len(self._spoolman_slot_spools)
already = sum(self._spoolman_slot_reported.values())
per = (total_used - already) / n if n else 0.0
return {slot: per for slot in self._spoolman_slot_spools}
def _spoolman_report(self, unreported: dict[int, float], min_mm: float = 0.1) -> None:
"""Fire-and-forget report of unreported mm to each mapped spool."""
sm = self._spoolman
for slot_idx, mm in unreported.items():
if mm < min_mm:
continue
spool_id = self._spoolman_slot_spools.get(slot_idx)
if not spool_id:
continue
self._spoolman_slot_reported[slot_idx] = (
self._spoolman_slot_reported.get(slot_idx, 0.0) + mm
)
def _send(sid=spool_id, length=mm):
try:
sm.use_filament(sid, length)
log.info(f"Spoolman: {length:.1f} mm → spool {sid}")
except Exception as e:
log.warning(f"Spoolman: report failed (spool {sid}): {e}")
threading.Thread(target=_send, daemon=True, name="spoolman-report").start()
def _spoolman_notify_end(self):
"""Report remaining filament on print end."""
if not self._spoolman or not self._spoolman_slot_spools:
return
self._spoolman_report(self._spoolman_unreported())
def _spoolman_sync_midprint(self):
"""Report incremental filament usage during a print (sync_rate interval)."""
if not self._spoolman or not self._spoolman_slot_spools:
return
self._spoolman_report(self._spoolman_unreported(), min_mm=10.0)
# ── Spoolman API handlers ─────────────────────────────────────────────────
async def handle_kx_spoolman_status(self, request):
"""GET /kx/spoolman/status"""
return self._json_cors({
"configured": bool(self._spoolman),
"reachable": self._spoolman_reachable if self._spoolman else False,
"server": self._spoolman.server_url if self._spoolman else "",
"sync_rate": self._spoolman.sync_rate if self._spoolman else 0,
"slot_spools": {str(k): v for k, v in self._spoolman_slot_spools.items()},
})
async def handle_kx_spoolman_spools(self, request):
"""GET /kx/spoolman/spools — proxied from Spoolman."""
if not self._spoolman:
return self._json_cors({"error": "Spoolman not configured"}, status=503)
try:
spools = await asyncio.get_event_loop().run_in_executor(
None, self._spoolman.list_spools
)
return self._json_cors({"spools": spools})
except Exception as e:
log.warning(f"Spoolman: list_spools failed: {e}")
return self._json_cors({"error": str(e)}, status=502)
async def handle_kx_spoolman_set_active(self, request):
"""POST /kx/spoolman/active-spool
Body: {"slot_map": {"0": 42, "2": 17}} — AMS slot index → Spoolman spool ID."""
try:
data = await request.json()
except Exception:
return self._json_cors({"error": "invalid JSON"}, status=400)
slot_map = data.get("slot_map") or data.get("slot_spools") or {}
self._spoolman_slot_spools = {
int(k): int(v) for k, v in slot_map.items()
if str(v).isdigit() and int(v) > 0
}
# Persist per printer (own [spoolman_<id>] section) so the
# assignment survives bridge restarts and two AMS units don't overwrite each other.
# (Previously: NameError on `config_loader` -> nothing was ever saved.)
try:
import config_loader as _cl
_cl.save_spool_map(self._spoolman_slot_spools, self._printer_id)
except Exception as _e:
log.warning("Spoolman: failed to save slot map: %s", _e)
self._spoolman_slot_usage = {}
self._spoolman_slot_reported = {}
self._spoolman_last_usage = 0.0
return self._json_cors({"slot_spools": {str(k): v for k, v in self._spoolman_slot_spools.items()}})

409
camera.py Normal file
View File

@@ -0,0 +1,409 @@
"""
camera.py - central camera demuxer (CameraCache) plus the ffmpeg locator.
Keeps one ffmpeg process per output type (jpeg/h264/mjpeg) open, reading the
printer's FLV stream and fanning it out to dashboard/OrcaSlicer/Obico
consumers. Extracted from kobrax_moonraker_bridge.py; re-exported from there
so existing imports keep working.
────────────────────────────────────────────────────────────────────────────
Copyright (C) 2026 viewit (KX-Bridge contributors)
Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md.
"""
import os
import sys
import time
import asyncio
import logging
log = logging.getLogger("bridge")
# Same base-path logic as the main module: next to sys.executable in a
# PyInstaller binary, otherwise next to this file.
_BASE = os.path.dirname(sys.executable) if getattr(sys, "frozen", False) else os.path.dirname(os.path.abspath(__file__))
try:
import imageio_ffmpeg
def _find_ffmpeg() -> str:
return imageio_ffmpeg.get_ffmpeg_exe()
except ImportError:
def _find_ffmpeg() -> str:
exe_name = "ffmpeg.exe" if sys.platform == "win32" else "ffmpeg"
local = os.path.join(_BASE, exe_name)
if os.path.isfile(local):
return local
return "ffmpeg"
class CameraCache:
"""Zentraler Kamera-Demuxer.
Keeps ONE ffmpeg process per output type open that reads the FLV stream
from the printer and produces:
- MJPEG @ 2fps -> last frame in RAM for /api/camera/snapshot
- MPEG-TS (-c:v copy) -> fanout to all /api/camera/h264 subscribers
- MJPEG @ 15fps/640px -> fanout to all /api/camera/stream subscribers
(the live-view used by the dashboard AND by every Moonraker-compatible
client, since server.webcams.list advertises this same stream_url)
Damit:
* Only ONE FLV connection to the printer per output type (solves the
single-client limit / 429) - previously /api/camera/stream opened a
brand-new, uncached ffmpeg + printer connection per HTTP client, which
competed with the cached jpeg/h264 connections for the printer's very
limited number of concurrent camera clients and caused intermittent
"stream unavailable" failures.
* Snapshots are instant (memory read, no ffmpeg spawn per request)
* Multiple parallel H.264/MJPEG consumers possible (plugin + web UI + ...)
Lazy start on the first consumer, auto-restart on ffmpeg crash.
"""
JPEG_SOI = b"\xff\xd8"
JPEG_EOI = b"\xff\xd9"
TS_CHUNK = 65536
def __init__(self):
self._url: str = ""
self.latest_jpeg: bytes = b""
self.latest_jpeg_ts: float = 0.0
self.h264_subscribers: "set[asyncio.Queue[bytes]]" = set()
self.mjpeg_subscribers: "set[asyncio.Queue[bytes]]" = set()
self._proc_jpeg: "asyncio.subprocess.Process | None" = None
self._proc_h264: "asyncio.subprocess.Process | None" = None
self._proc_mjpeg: "asyncio.subprocess.Process | None" = None
self._task_jpeg: "asyncio.Task | None" = None
self._task_h264: "asyncio.Task | None" = None
self._task_mjpeg: "asyncio.Task | None" = None
self._lock = asyncio.Lock()
self._fail_count_jpeg: int = 0
self._fail_count_h264: int = 0
self._fail_count_mjpeg: int = 0
def set_url(self, url: str):
# A changed URL means the printer rotated its stream token (typically
# after a reboot). Running ffmpeg processes still hold the stale URL
# and will never pick it up on their own - they only re-read self._url
# at the top of their outer loop, which they never reach while blocked
# in a stdout read on the old, now-silent connection. Tear them down;
# the next ensure_running() respawns them against the new URL.
changed = bool(url and self._url and url != self._url)
self._url = url
if changed:
self.reset()
def reset(self):
"""Reset backoff counters and forcefully tear down any running
ffmpeg loops - including cancelling their background tasks.
Only killing the ffmpeg subprocess is not enough: the owning task
might currently be sitting in `await asyncio.sleep(delay)` from a
previous exponential backoff (up to 300s) after an earlier failure.
Resetting the fail-count doesn't wake it up early, so a user
clicking "reset" could see nothing happen for minutes. Cancelling
the task guarantees an immediate, clean restart on the next
ensure_running() call.
"""
self._fail_count_jpeg = 0
self._fail_count_h264 = 0
self._fail_count_mjpeg = 0
for task in (self._task_jpeg, self._task_h264, self._task_mjpeg):
if task is not None and not task.done():
task.cancel()
for proc in (self._proc_jpeg, self._proc_h264, self._proc_mjpeg):
if proc is not None:
try:
proc.kill()
except Exception:
pass
self._task_jpeg = self._task_h264 = self._task_mjpeg = None
self._proc_jpeg = self._proc_h264 = self._proc_mjpeg = None
async def ensure_running(self):
# NOTE: we check the *task* state, not self._proc_* - the process
# handle is only assigned later, inside the task body, once ffmpeg
# has actually been spawned. Checking self._proc_* here left a race
# window: two callers arriving before the newly-created task got a
# chance to run would both see "no process yet" and each spawn a
# duplicate ffmpeg + duplicate printer connection, silently
# orphaning the older one (whichever task's coroutine runs last
# overwrites the shared self._proc_* reference, so nobody keeps a
# handle to kill the earlier orphaned process). Task creation is
# synchronous, so checking self._task_* here is race-free.
if self._task_jpeg is None or self._task_jpeg.done():
self._task_jpeg = asyncio.create_task(self._run_jpeg_loop())
if self._task_h264 is None or self._task_h264.done():
self._task_h264 = asyncio.create_task(self._run_h264_loop())
if self._task_mjpeg is None or self._task_mjpeg.done():
self._task_mjpeg = asyncio.create_task(self._run_mjpeg_loop())
def _input_args(self, url: str) -> list[str]:
args = ["-fflags", "nobuffer", "-flags", "low_delay",
# Bail out if the source goes silent. A printer reboot or
# network loss leaves the TCP connection ESTABLISHED with no
# data and no FIN, so a passive stdout read blocks forever
# without this (Issue #99). Value is microseconds.
"-timeout", "10000000"]
if url.lower().startswith("rtsp://"):
args += ["-probesize", "32", "-analyzeduration", "0", "-rtsp_transport", "tcp"]
else:
# The printer's FLV source occasionally emits non-monotonic container
# timestamps (PTS jumps of days) while the video data itself stays
# valid. Without this flag ffmpeg's realtime pacing breaks on such a
# jump and the stream stalls after ~15-30 min (Issue #90).
args += ["-use_wallclock_as_timestamps", "1",
"-probesize", "500000", "-analyzeduration", "500000"]
return args
async def _run_jpeg_loop(self):
"""Keeps an ffmpeg process alive that writes MJPEG@2fps into the cache."""
while True:
url = self._url
if not url:
await asyncio.sleep(2.0)
continue
try:
proc = await asyncio.create_subprocess_exec(
_find_ffmpeg(), "-loglevel", "warning",
*self._input_args(url), "-i", url,
"-vf", "fps=2",
"-f", "image2pipe", "-vcodec", "mjpeg", "-q:v", "3",
"-flush_packets", "1", "pipe:1",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
self._proc_jpeg = proc
except Exception as e:
log.warning(f"CameraCache: ffmpeg-jpeg start failed: {e}")
await asyncio.sleep(3.0)
continue
buf = b""
rc = None
try:
while True:
chunk = await proc.stdout.read(self.TS_CHUNK)
if not chunk:
break
buf += chunk
# extract complete JPEG frames
while True:
start = buf.find(self.JPEG_SOI)
if start == -1:
buf = b""
break
end = buf.find(self.JPEG_EOI, start + 2)
if end == -1:
buf = buf[start:]
break
self.latest_jpeg = buf[start:end + 2]
self.latest_jpeg_ts = time.time()
buf = buf[end + 2:]
except Exception as e:
log.debug(f"CameraCache: jpeg-loop unterbrochen: {e}")
finally:
# NOTE: cleanup operates on the local `proc` reference, not on
# self._proc_jpeg - see _run_mjpeg_loop's identical comment.
# If this task got cancelled (e.g. by reset()), a new task may
# already have started and assigned its own process to
# self._proc_jpeg by the time we reach here; killing that
# shared attribute instead of our own local proc would kill
# the WRONG (newer) process and leak this one as an orphan.
try:
proc.kill()
except Exception:
pass
try:
await proc.wait()
except Exception:
pass
rc = proc.returncode
if rc:
try:
err = await proc.stderr.read(500)
if err:
log.warning(f"CameraCache: ffmpeg-jpeg stderr: {err.decode(errors='replace').strip()}")
except Exception:
pass
if self._proc_jpeg is proc:
self._proc_jpeg = None
if rc:
self._fail_count_jpeg += 1
delay = min(2.0 * (2 ** self._fail_count_jpeg), 300.0)
log.warning(f"CameraCache: ffmpeg-jpeg exit {rc}, retry in {delay:.0f}s (Versuch {self._fail_count_jpeg})")
await asyncio.sleep(delay)
else:
self._fail_count_jpeg = 0
await asyncio.sleep(2.0)
async def _run_h264_loop(self):
"""Keeps an ffmpeg process alive that fans out MPEG-TS to all subscribers."""
while True:
url = self._url
if not url:
await asyncio.sleep(2.0)
continue
try:
proc = await asyncio.create_subprocess_exec(
_find_ffmpeg(), "-loglevel", "warning",
*self._input_args(url), "-i", url,
"-c:v", "copy", "-an",
"-f", "mpegts", "pipe:1",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
self._proc_h264 = proc
except Exception as e:
log.warning(f"CameraCache: ffmpeg-h264 start failed: {e}")
await asyncio.sleep(3.0)
continue
rc = None
try:
while True:
chunk = await proc.stdout.read(self.TS_CHUNK)
if not chunk:
break
# Fanout: non-blocking per subscriber; slow clients
# get their oldest chunk dropped (queue full -> drop).
for q in list(self.h264_subscribers):
if q.full():
try:
q.get_nowait()
except Exception:
pass
try:
q.put_nowait(chunk)
except Exception:
pass
except Exception as e:
log.debug(f"CameraCache: h264-loop unterbrochen: {e}")
finally:
# NOTE: cleanup operates on the local `proc` reference, not on
# self._proc_h264 - see _run_mjpeg_loop's identical comment.
# If this task got cancelled (e.g. by reset()), a new task may
# already have started and assigned its own process to
# self._proc_h264 by the time we reach here; killing that
# shared attribute instead of our own local proc would kill
# the WRONG (newer) process and leak this one as an orphan.
try:
proc.kill()
except Exception:
pass
try:
await proc.wait()
except Exception:
pass
rc = proc.returncode
if rc:
try:
err = await proc.stderr.read(500)
if err:
log.warning(f"CameraCache: ffmpeg-h264 stderr: {err.decode(errors='replace').strip()}")
except Exception:
pass
if self._proc_h264 is proc:
self._proc_h264 = None
if rc:
self._fail_count_h264 += 1
delay = min(2.0 * (2 ** self._fail_count_h264), 300.0)
log.warning(f"CameraCache: ffmpeg-h264 exit {rc}, retry in {delay:.0f}s (Versuch {self._fail_count_h264})")
await asyncio.sleep(delay)
else:
self._fail_count_h264 = 0
await asyncio.sleep(2.0)
async def _run_mjpeg_loop(self):
"""Keeps an ffmpeg process alive that fans out MJPEG@15fps/640px
(complete JPEG frames) to all /api/camera/stream subscribers."""
while True:
url = self._url
if not url:
await asyncio.sleep(2.0)
continue
try:
proc = await asyncio.create_subprocess_exec(
_find_ffmpeg(), "-loglevel", "warning",
*self._input_args(url), "-i", url,
"-vf", "fps=15,scale=640:-1",
"-f", "image2pipe", "-vcodec", "mjpeg", "-q:v", "3",
"-flush_packets", "1", "pipe:1",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
self._proc_mjpeg = proc
except Exception as e:
log.warning(f"CameraCache: ffmpeg-mjpeg start failed: {e}")
await asyncio.sleep(3.0)
continue
buf = b""
rc = None
try:
while True:
chunk = await proc.stdout.read(self.TS_CHUNK)
if not chunk:
break
buf += chunk
# extract complete JPEG frames and fan them out whole
# (so every subscriber gets clean multipart boundaries,
# not arbitrary byte chunks like the h264/mpegts fanout)
while True:
start = buf.find(self.JPEG_SOI)
if start == -1:
buf = b""
break
end = buf.find(self.JPEG_EOI, start + 2)
if end == -1:
buf = buf[start:]
break
frame = buf[start:end + 2]
buf = buf[end + 2:]
for q in list(self.mjpeg_subscribers):
if q.full():
try:
q.get_nowait()
except Exception:
pass
try:
q.put_nowait(frame)
except Exception:
pass
except Exception as e:
log.debug(f"CameraCache: mjpeg-loop unterbrochen: {e}")
finally:
# NOTE: cleanup operates on the local `proc` reference, not
# on self._proc_mjpeg. If this task got cancelled (e.g. by
# reset()) a new task may already have started and assigned
# its own process to self._proc_mjpeg by the time we reach
# here - killing that shared attribute instead of our own
# local proc would kill the WRONG (newer) process.
try:
proc.kill()
except Exception:
pass
try:
await proc.wait()
except Exception:
pass
rc = proc.returncode
if rc:
try:
err = await proc.stderr.read(500)
if err:
log.warning(f"CameraCache: ffmpeg-mjpeg stderr: {err.decode(errors='replace').strip()}")
except Exception:
pass
if self._proc_mjpeg is proc:
self._proc_mjpeg = None
if rc:
self._fail_count_mjpeg += 1
delay = min(2.0 * (2 ** self._fail_count_mjpeg), 300.0)
log.warning(f"CameraCache: ffmpeg-mjpeg exit {rc}, retry in {delay:.0f}s (Versuch {self._fail_count_mjpeg})")
await asyncio.sleep(delay)
else:
self._fail_count_mjpeg = 0
await asyncio.sleep(2.0)

View File

@@ -116,3 +116,23 @@ custom_3_duration_sec = 14400
# Wie oft (Sekunden) der Filamentverbrauch während des Drucks gemeldet wird
# (0 = nur beim Druckende)
# sync_rate = 0
[kxgauge]
# URL des KXGauge-Displays (leer = deaktiviert), siehe
# https://gitea.it-drui.de/viewit/kxgauge
# url = http://192.168.x.x
enabled = 0
# Ziel-Temperatur für die Heat-Ring-Skala (Grad Celsius, meist Hotend-Solltemperatur)
heat_peak = 250
[kxgauge_mapping]
# kobra_state -> KXGauge-Emotion. Nur belegte States werden gesendet, gültige
# Emotionsnamen siehe KXGauge API.md.
free = neutral
printing = happy
paused = worried
pause = worried
finished = glee
error = scared
offline = tired

View File

@@ -18,6 +18,7 @@ CONFIG_SECTION_CONNECTION = "connection"
CONFIG_SECTION_PRINT = "print"
CONFIG_SECTION_BRIDGE = "bridge"
CONFIG_SECTION_SPOOLMAN = "spoolman"
CONFIG_SECTION_KXGAUGE = "kxgauge"
def _find_config_file() -> pathlib.Path | None:
@@ -78,6 +79,9 @@ CONFIG_ENV_MAPPING = {
"VERBOSE_HTTP_LOG": (CONFIG_SECTION_BRIDGE, "verbose_http_log"),
"SPOOLMAN_SERVER": (CONFIG_SECTION_SPOOLMAN, "server"),
"SPOOLMAN_SYNC_RATE": (CONFIG_SECTION_SPOOLMAN, "sync_rate"),
"KXGAUGE_URL": (CONFIG_SECTION_KXGAUGE, "url"),
"KXGAUGE_ENABLED": (CONFIG_SECTION_KXGAUGE, "enabled"),
"KXGAUGE_HEAT_PEAK": (CONFIG_SECTION_KXGAUGE, "heat_peak"),
}
@@ -492,6 +496,9 @@ DELETE_PRINTER_FILE_AFTER_PRINT = _safe_int(get("DELETE_PRINTER_FILE_AFTER_PRINT
PRINT_START_DIALOG = _safe_int(get("PRINT_START_DIALOG", get("FILE_READY_DIALOG", "1")), 1)
SPOOLMAN_SERVER = get("SPOOLMAN_SERVER", "")
SPOOLMAN_SYNC_RATE = _safe_int(get("SPOOLMAN_SYNC_RATE", "0"), 0)
KXGAUGE_URL = get("KXGAUGE_URL", "")
KXGAUGE_ENABLED = _safe_int(get("KXGAUGE_ENABLED", "0"), 0)
KXGAUGE_HEAT_PEAK = _safe_int(get("KXGAUGE_HEAT_PEAK", "250"), 250)
BRIDGE_HOST_IP = get("BRIDGE_HOST_IP", "")
POLL_INTERVAL = _safe_int(get("POLL_INTERVAL", "3"), 3)
VERBOSE_HTTP_LOG = _safe_int(get("VERBOSE_HTTP_LOG", "0"), 0)

75
credentials.py Normal file
View File

@@ -0,0 +1,75 @@
"""
credentials.py - fetch + decrypt Anycubic Kobra printer credentials.
Talks to the printer's HTTP /info + /ctrl endpoints (port 18910) and decrypts
the AES-256-CBC response to recover username/password/device_id/mode_id.
Algorithm reverse-engineered in tools/fetch_credentials.py. Extracted from
kobrax_moonraker_bridge.py; _kx_fetch_credentials is re-imported there for the
"add printer" flow.
────────────────────────────────────────────────────────────────────────────
Copyright (C) 2026 viewit (KX-Bridge contributors)
Licensed under GPLv3 — see LICENSE in the project root.
Protocol reverse-engineered for interoperability (§69e UrhG). See NOTICE.md.
"""
import json
import time
import hashlib
import aiohttp
try:
import base64 as _base64
from Crypto.Cipher import AES as _AES
from Crypto.Util.Padding import unpad as _unpad
_HAS_CRYPTO = True
except ImportError:
_HAS_CRYPTO = False
def _kx_generate_signature(token: str, ts: int, nonce: str) -> str:
first = hashlib.md5(token[:16].encode()).hexdigest()
return hashlib.md5((first + str(ts) + nonce).encode()).hexdigest()
def _kx_decrypt_info(encrypted_b64: str, key: str, iv: str) -> dict:
cipher = _AES.new(key.encode(), _AES.MODE_CBC, iv.encode())
raw = _base64.b64decode(encrypted_b64)
return json.loads(_unpad(cipher.decrypt(raw), _AES.block_size).decode())
async def _kx_fetch_credentials(ip: str, port: int = 18910) -> dict:
"""Fetches + decrypts printer credentials via HTTP /info + /ctrl.
Raises an exception on network/decrypt errors. Algorithm from
tools/fetch_credentials.py (AES-256-CBC, Key=token[16:32], IV=ctrl-token).
"""
if not _HAS_CRYPTO:
raise RuntimeError("pycryptodome is not installed")
import random, string
nonce = "".join(random.choice(string.ascii_letters + string.digits) for _ in range(6))
timeout = aiohttp.ClientTimeout(total=10)
async with aiohttp.ClientSession() as s:
async with s.get(f"http://{ip}:{port}/info", timeout=timeout) as r:
r.raise_for_status()
info = await r.json()
token = info["token"]
ts = int(time.time() * 1000)
sign = _kx_generate_signature(token, ts, nonce)
params = {"ts": ts, "nonce": nonce, "sign": sign, "did": "random"}
async with s.post(f"http://{ip}:{port}/ctrl", params=params, timeout=timeout) as r:
r.raise_for_status()
data = await r.json()
result = _kx_decrypt_info(data["data"]["info"], token[16:32], data["data"]["token"])
if "error" in result:
raise RuntimeError(result.get("error", "decrypt failed"))
return {
"printer_ip": result.get("ip", ip),
"username": result.get("username", ""),
"password": result.get("password", ""),
"device_id": result.get("deviceId", ""),
"mode_id": str(result.get("modeId", "20030")),
"model": result.get("modelName", "Anycubic Kobra"),
}

182
gcode_meta.py Normal file
View File

@@ -0,0 +1,182 @@
"""
gcode_meta.py - GCode file metadata extraction helpers (estimated print time,
layer heights, embedded thumbnail, per-slot filament info).
Extracted from kobrax_moonraker_bridge.py; re-exported from there so existing
call sites keep working. Used by the file-upload and print-start paths.
────────────────────────────────────────────────────────────────────────────
Copyright (C) 2026 viewit (KX-Bridge contributors)
Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md.
"""
import re
import base64
import logging
log = logging.getLogger("bridge")
def _parse_gcode_estimated_time(data: bytes) -> int:
"""Reads the estimated print time from GCode (OrcaSlicer + PrusaSlicer).
Returns seconds, 0 when not found.
PrusaSlicer writes the time into the header (first 16KB),
OrcaSlicer writes it at the end of the file (last 16KB)."""
import re
# Search the beginning + end of the file (OrcaSlicer writes the time at the end)
search_text = (data[:16384] + data[-65536:]).decode("utf-8", errors="ignore")
# OrcaSlicer: ; total estimated time: 9m 20s
# PrusaSlicer: ; estimated printing time (normal mode) = 1h 9m 20s
m = (re.search(r";\s*total estimated time:\s*(.*)", search_text) or
re.search(r";\s*estimated printing time \(normal mode\)\s*=\s*(.*)", search_text))
if not m:
return 0
parts = re.findall(r"(\d+)\s*([hms])", m.group(1))
secs = 0
for val, unit in parts:
if unit == "h": secs += int(val) * 3600
elif unit == "m": secs += int(val) * 60
elif unit == "s": secs += int(val)
if secs:
log.info(f"Slicer estimate: {secs}s ({m.group(1).strip()})")
return secs
def _parse_gcode_layer_heights(data: bytes) -> tuple[float, float]:
"""Reads (layer_height, initial_layer_height) from the OrcaSlicer/PrusaSlicer
GCode header. Both are stored as a config block at the end of the GCode.
Beispiel-Zeilen:
; layer_height = 0.2
; initial_layer_print_height = 0.2
Returns (0.0, 0.0) when not found - the caller decides what to do
(typisch: keinen Z-Wert anzeigen)."""
import re
head = data[:16384].decode("utf-8", errors="ignore")
tail = data[-65536:].decode("utf-8", errors="ignore")
search = head + "\n" + tail
def _grab(pat):
m = re.search(pat, search)
if not m:
return 0.0
try:
return float(m.group(1))
except Exception:
return 0.0
layer_h = _grab(r";\s*layer_height\s*=\s*([0-9.]+)")
first_h = (_grab(r";\s*initial_layer_print_height\s*=\s*([0-9.]+)") or
_grab(r";\s*first_layer_height\s*=\s*([0-9.]+)") or
layer_h)
return layer_h, first_h
def _extract_thumbnail(data: bytes) -> str:
"""Extracts the base64 PNG thumbnail from GCode (OrcaSlicer format)."""
try:
marker = b"; thumbnail begin"
end_marker = b"; thumbnail end"
start = data.find(marker)
if start == -1:
return ""
start = data.find(b"\n", start) + 1
end = data.find(end_marker, start)
if end == -1:
return ""
lines = data[start:end].split(b"\n")
b64 = b"".join(
line[2:].strip() if line.startswith(b"; ") else line.strip()
for line in lines
)
return b64.decode("ascii")
except Exception:
return ""
def _extract_filament_info(data: bytes) -> list[dict]:
"""Reads filament colors/materials incl. tool order from Orca/Prusa GCode.
Gibt Liste von {slot_index, color_hex, material} in Tool-/Paint-Reihenfolge
(T0, T1, ...).
Searches both the start and the end of the file since Orca can insert
large thumbnail blocks, pushing the metadata into the tail.
"""
try:
head = data[:131072]
tail = data[-131072:] if len(data) > 131072 else b""
header = (head + b"\n" + tail).decode("utf-8", errors="ignore")
colors, materials = [], []
paint_count_hint = 0
tool_filament_order = []
for line in header.splitlines():
if re.match(r"^\s*;\s*filament_colour\s*=", line):
val = line.split("=", 1)[-1].strip()
colors = [c.strip().lstrip("#") for c in val.split(";") if c.strip()]
elif re.match(r"^\s*;\s*filament_multi_colour\s*=", line) and not colors:
val = line.split("=", 1)[-1].strip()
colors = [c.strip().lstrip("#") for c in val.split(";") if c.strip()]
elif re.match(r"^\s*;\s*filament_type\s*=", line):
val = line.split("=", 1)[-1].strip()
parts = [m.strip() for m in re.split(r"[;,]", val) if m.strip()]
materials = parts
paint_count_hint = max(paint_count_hint, len(parts))
elif re.match(r"^\s*;\s*filament_density\s*:", line):
val = line.split(":", 1)[-1].strip()
parts = [x.strip() for x in re.split(r"[;,]", val) if x.strip()]
paint_count_hint = max(paint_count_hint, len(parts))
elif re.match(r"^\s*;\s*filament_diameter\s*:", line):
val = line.split(":", 1)[-1].strip()
parts = [x.strip() for x in re.split(r"[;,]", val) if x.strip()]
paint_count_hint = max(paint_count_hint, len(parts))
elif re.match(r"^\s*;\s*filament\s*:", line):
raw = line.split(":", 1)[-1]
parsed = []
for p in [x.strip() for x in raw.split(",") if x.strip()]:
try:
parsed.append(int(p))
except Exception:
pass
if parsed:
tool_filament_order = parsed
total_paints = max(len(colors), len(materials), paint_count_hint)
if tool_filament_order:
total_paints = max(total_paints, max(tool_filament_order))
if total_paints <= 0:
return []
# Keep full paint list visible; mark paints referenced by Orca tool order as used.
if len(colors) < total_paints:
colors.extend(["FFFFFF"] * (total_paints - len(colors)))
if len(materials) < total_paints:
materials.extend(["PLA"] * (total_paints - len(materials)))
# Prefer actual tool-change commands from the GCode body.
# This avoids forwarding paints that are present in metadata but never used.
used_paints_zero_based = set()
try:
for m in re.finditer(br"(?m)^[ \t]*T([0-9]+)\b", data):
used_paints_zero_based.add(int(m.group(1)))
except Exception:
used_paints_zero_based = set()
# Fallback for slicers that only provide paint usage in header metadata.
used_paints_from_header = set()
for n in tool_filament_order:
try:
# Orca/Prusa filament: list is typically 1-based.
used_paints_from_header.add(max(0, int(n) - 1))
except Exception:
pass
result = []
for i in range(total_paints):
hex_color = colors[i] if i < len(colors) else "FFFFFF"
result.append({
"slot_index": i,
"color_hex": "#" + hex_color.upper() if hex_color else "#FFFFFF",
"material": materials[i] if i < len(materials) else "PLA",
"is_used": (i in used_paints_zero_based) if used_paints_zero_based else ((i in used_paints_from_header) if used_paints_from_header else True),
})
return result
except Exception:
return []

226
gcode_store.py Normal file
View File

@@ -0,0 +1,226 @@
"""
gcode_store.py - persistent per-bridge SQLite store for uploaded GCode files
and print-job history.
Extracted from kobrax_moonraker_bridge.py; re-exported from there so existing
imports keep working.
────────────────────────────────────────────────────────────────────────────
Copyright (C) 2026 viewit (KX-Bridge contributors)
Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md.
"""
import os
import json
import time
import uuid
import sqlite3
import threading
class GCodeStore:
"""Persistenter GCode-Store pro Bridge-Instanz (SQLite)."""
def __init__(self, data_dir: str):
os.makedirs(data_dir, exist_ok=True)
self._gcode_dir = os.path.join(data_dir, "gcodes")
os.makedirs(self._gcode_dir, exist_ok=True)
db_path = os.path.join(data_dir, "kx-bridge.db")
self._conn = sqlite3.connect(db_path, check_same_thread=False)
self._conn.row_factory = sqlite3.Row
self._lock = threading.Lock()
self._init_schema()
def _init_schema(self):
with self._lock:
self._conn.executescript("""
CREATE TABLE IF NOT EXISTS gcode_files (
id TEXT PRIMARY KEY,
filename TEXT NOT NULL,
path TEXT NOT NULL,
size_bytes INTEGER NOT NULL,
uploaded_at TEXT NOT NULL,
thumbnail_b64 TEXT,
est_print_time_sec INTEGER,
filament_used_mm REAL,
layer_count INTEGER,
gcode_filaments TEXT,
objects_skip_parts TEXT,
svg_image TEXT,
web_unverified INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS print_jobs (
id TEXT PRIMARY KEY,
gcode_file_id TEXT NOT NULL,
printer_id TEXT NOT NULL,
started_at TEXT NOT NULL,
ended_at TEXT,
status TEXT NOT NULL,
duration_sec INTEGER,
filament_assignments TEXT,
abort_reason TEXT
);
""")
# Migration: add gcode_filaments column for older databases
try:
self._conn.execute("ALTER TABLE gcode_files ADD COLUMN gcode_filaments TEXT")
self._conn.commit()
except Exception:
pass
# Migration: columns objects_skip_parts + svg_image (part-skip feature, v0.9.10)
# Plus layer_height / first_layer_height (Obico Z height, v0.9.18)
for col, typ in (
("objects_skip_parts", "TEXT"),
("svg_image", "TEXT"),
("layer_height", "REAL"),
("first_layer_height", "REAL"),
):
try:
self._conn.execute(f"ALTER TABLE gcode_files ADD COLUMN {col} {typ}")
self._conn.commit()
except Exception:
pass
# Migration: flag for web uploads (warning before print)
try:
self._conn.execute("ALTER TABLE gcode_files ADD COLUMN web_unverified INTEGER NOT NULL DEFAULT 0")
self._conn.commit()
except Exception:
pass
def save_file(self, file_id: str, filename: str, data: bytes,
est_time_sec: int = 0, thumbnail_b64: str = "",
gcode_filaments: list | None = None,
web_unverified: bool = False,
layer_height: float = 0.0,
first_layer_height: float = 0.0) -> str:
"""Saves a GCode file to disk and DB. Returns the path."""
safe_name = os.path.basename(filename)
path = os.path.join(self._gcode_dir, safe_name)
with open(path, "wb") as f:
f.write(data)
now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
with self._lock:
filaments_json = json.dumps(gcode_filaments) if gcode_filaments else None
self._conn.execute(
"""INSERT OR REPLACE INTO gcode_files
(id, filename, path, size_bytes, uploaded_at, thumbnail_b64, est_print_time_sec, gcode_filaments, web_unverified, layer_height, first_layer_height)
VALUES (?,?,?,?,?,?,?,?,?,?,?)""",
(file_id, filename, path, len(data), now, thumbnail_b64 or None, est_time_sec or None, filaments_json, 1 if web_unverified else 0, layer_height or None, first_layer_height or None)
)
self._conn.commit()
return path
def list_files(self) -> list:
with self._lock:
rows = self._conn.execute(
"SELECT * FROM gcode_files ORDER BY uploaded_at DESC"
).fetchall()
return [dict(r) for r in rows]
def get_file(self, file_id: str) -> dict | None:
with self._lock:
row = self._conn.execute(
"SELECT * FROM gcode_files WHERE id=?", (file_id,)
).fetchone()
return dict(row) if row else None
def get_file_by_name(self, filename: str) -> dict | None:
with self._lock:
row = self._conn.execute(
"SELECT * FROM gcode_files WHERE filename=? ORDER BY uploaded_at DESC LIMIT 1",
(filename,)
).fetchone()
return dict(row) if row else None
def update_file_objects(self, filename: str, objects: list, svg: str = "") -> None:
"""Saves the object list + optional SVG for a file (matched via filename)."""
if not filename:
return
with self._lock:
self._conn.execute(
"UPDATE gcode_files SET objects_skip_parts=?, svg_image=? "
"WHERE filename=?",
(json.dumps(objects), svg or "", filename),
)
self._conn.commit()
def update_file_filaments(self, file_id: str, gcode_filaments: list | None) -> None:
"""Updates parsed GCode filaments for an existing DB entry."""
with self._lock:
self._conn.execute(
"UPDATE gcode_files SET gcode_filaments=? WHERE id=?",
(json.dumps(gcode_filaments) if gcode_filaments else None, file_id),
)
self._conn.commit()
def clear_web_unverified(self, file_id: str) -> bool:
with self._lock:
cur = self._conn.execute(
"UPDATE gcode_files SET web_unverified=0 WHERE id=?",
(file_id,),
)
self._conn.commit()
return cur.rowcount > 0
def delete_file(self, file_id: str) -> bool:
row = self.get_file(file_id)
if not row:
return False
try:
os.remove(row["path"])
except OSError:
pass
with self._lock:
self._conn.execute("DELETE FROM gcode_files WHERE id=?", (file_id,))
self._conn.commit()
return True
def start_job(self, gcode_file_id: str, printer_id: str,
filament_assignments: list | None = None) -> str:
job_id = str(uuid.uuid4())
now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
assignments_json = json.dumps(filament_assignments) if filament_assignments else None
with self._lock:
self._conn.execute(
"""INSERT INTO print_jobs
(id, gcode_file_id, printer_id, started_at, status, filament_assignments)
VALUES (?,?,?,?,'printing',?)""",
(job_id, gcode_file_id, printer_id, now, assignments_json)
)
self._conn.commit()
return job_id
def finish_job(self, job_id: str, status: str = "completed",
abort_reason: str = "") -> None:
now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
with self._lock:
row = self._conn.execute(
"SELECT started_at FROM print_jobs WHERE id=?", (job_id,)
).fetchone()
duration = None
if row:
try:
import calendar
start = time.strptime(row["started_at"], "%Y-%m-%dT%H:%M:%SZ")
duration = int(time.time() - calendar.timegm(start))
except Exception:
pass
self._conn.execute(
"""UPDATE print_jobs SET ended_at=?, status=?, duration_sec=?, abort_reason=?
WHERE id=?""",
(now, status, duration, abort_reason or None, job_id)
)
self._conn.commit()
def list_jobs(self, limit: int = 50, offset: int = 0) -> list:
with self._lock:
rows = self._conn.execute(
"""SELECT j.*, f.filename, f.thumbnail_b64
FROM print_jobs j
LEFT JOIN gcode_files f ON j.gcode_file_id = f.id
ORDER BY j.started_at DESC LIMIT ? OFFSET ?""",
(limit, offset)
).fetchall()
return [dict(r) for r in rows]

File diff suppressed because it is too large Load Diff

74
kxgauge_client.py Normal file
View File

@@ -0,0 +1,74 @@
"""
kxgauge_client.py - thin synchronous HTTP client for KXGauge (ESP32 round-face
display, https://gitea.it-drui.de/viewit/kxgauge).
KXGauge exposes a simple GET-only HTTP API (no push, no websocket) - the
bridge has to actively poke it whenever printer state/temperature changes.
Designed to be called from daemon threads (MQTT reader thread callbacks),
mirrors spoolman_client.py's shape: plain `requests`, no event-loop dependency.
────────────────────────────────────────────────────────────────────────────
Copyright (C) 2026 viewit (KX-Bridge contributors)
Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md.
"""
import logging
log = logging.getLogger("kobrax.kxgauge")
# Tolerance in °C before a new heat value is worth another GET - avoids
# spamming the device on every MQTT tick when the temperature is basically
# holding steady.
_HEAT_TOLERANCE = 2.0
class KXGaugeClient:
"""Thin synchronous HTTP client for a KXGauge display.
All calls swallow their own exceptions and just log a warning - a
disconnected/misconfigured display must never interrupt MQTT callback
processing or the poll loop.
"""
def __init__(self, base_url: str, heat_peak: float = 250.0):
self.base_url = base_url.rstrip("/")
self.heat_peak = heat_peak
self._last_emotion: str | None = None
self._last_heat_celsius: float | None = None
self._peak_sent = False
def _get(self, path: str) -> bool:
try:
import requests
r = requests.get(f"{self.base_url}{path}", timeout=3)
r.raise_for_status()
return True
except Exception as e:
log.warning(f"KXGauge request failed ({path}): {e}")
return False
def health_check(self) -> bool:
try:
import requests
r = requests.get(f"{self.base_url}/status", timeout=3)
r.raise_for_status()
return True
except Exception:
return False
def set_emotion(self, name: str) -> None:
name = (name or "").lower().strip()
if not name or name == self._last_emotion:
return
if self._get(f"/emotion/{name}"):
self._last_emotion = name
def set_heat_celsius(self, value: float) -> None:
if not self._peak_sent:
if self._get(f"/heat/peak/celsius/{self.heat_peak}"):
self._peak_sent = True
if (self._last_heat_celsius is not None
and abs(value - self._last_heat_celsius) < _HEAT_TOLERANCE):
return
if self._get(f"/heat/celsius/{value:.1f}"):
self._last_heat_celsius = value

45
spoolman_client.py Normal file
View File

@@ -0,0 +1,45 @@
"""
spoolman_client.py - thin synchronous HTTP client for Spoolman filament tracking.
Extracted from kobrax_moonraker_bridge.py as part of splitting that module up;
re-exported from there so existing imports keep working.
────────────────────────────────────────────────────────────────────────────
Copyright (C) 2026 viewit (KX-Bridge contributors)
Licensed under GPLv3 — see LICENSE in the project root. See NOTICE.md.
"""
class SpoolmanClient:
"""Thin synchronous HTTP client for Spoolman filament tracking.
Designed to be called from daemon threads (poll loop, _on_print callbacks).
Uses requests (already in requirements) so no event-loop dependency.
"""
def __init__(self, server_url: str, sync_rate: int = 0):
self.server_url = server_url.rstrip("/")
self.sync_rate = sync_rate
def _req(self, method: str, path: str, **kwargs):
import requests
r = requests.request(method, f"{self.server_url}{path}", timeout=5, **kwargs)
r.raise_for_status()
return r.json()
def health_check(self) -> bool:
try:
self._req("GET", "/api/v1/health")
return True
except Exception:
return False
def list_spools(self) -> list:
return self._req("GET", "/api/v1/spool")
def use_filament(self, spool_id: int, use_length_mm: float) -> None:
"""Report consumed filament length in mm. Spoolman converts to weight
using the spool's filament profile density."""
self._req("PUT", f"/api/v1/spool/{spool_id}/use",
json={"use_length": round(use_length_mm, 2)})

271
tests/test_kxgauge.py Normal file
View File

@@ -0,0 +1,271 @@
"""KXGauge display integration (https://gitea.it-drui.de/viewit/kxgauge).
KXGauge is a small ESP32 round-face display with a GET-only HTTP API and no
push/websocket - the bridge has to actively poke it on printer state/temp
changes. Covers: the KXGaugeClient dedupe logic, settings GET/POST roundtrip
for the new fields, the mapping config loader, and the /api/kxgauge/test
connectivity check endpoint.
"""
import argparse
import tempfile
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
import pytest_asyncio
from aiohttp.test_utils import TestClient, TestServer
from kobrax_moonraker_bridge import KobraXBridge, build_app
from kxgauge_client import KXGaugeClient
# ── KXGaugeClient (sync HTTP client, dedupe logic) ──────────────────────────
def test_set_emotion_sends_get_and_dedupes():
kg = KXGaugeClient("http://192.168.1.99")
with patch("requests.get") as mock_get:
mock_get.return_value = MagicMock(raise_for_status=lambda: None)
kg.set_emotion("happy")
kg.set_emotion("happy") # same emotion again - must not re-send
assert mock_get.call_count == 1
assert mock_get.call_args[0][0] == "http://192.168.1.99/emotion/happy"
def test_set_emotion_sends_again_after_change():
kg = KXGaugeClient("http://192.168.1.99")
with patch("requests.get") as mock_get:
mock_get.return_value = MagicMock(raise_for_status=lambda: None)
kg.set_emotion("happy")
kg.set_emotion("scared")
assert mock_get.call_count == 2
assert mock_get.call_args[0][0] == "http://192.168.1.99/emotion/scared"
def test_set_heat_celsius_sends_peak_once_then_dedupes_within_tolerance():
kg = KXGaugeClient("http://192.168.1.99", heat_peak=230)
with patch("requests.get") as mock_get:
mock_get.return_value = MagicMock(raise_for_status=lambda: None)
kg.set_heat_celsius(200.0)
kg.set_heat_celsius(200.5) # within tolerance - no new /heat/celsius call
urls = [c.args[0] for c in mock_get.call_args_list]
assert "http://192.168.1.99/heat/peak/celsius/230" in urls
assert urls.count("http://192.168.1.99/heat/celsius/200.0") == 1
assert not any("/heat/celsius/200.5" in u for u in urls)
def test_set_heat_celsius_sends_again_beyond_tolerance():
kg = KXGaugeClient("http://192.168.1.99")
with patch("requests.get") as mock_get:
mock_get.return_value = MagicMock(raise_for_status=lambda: None)
kg.set_heat_celsius(200.0)
kg.set_heat_celsius(210.0)
heat_calls = [c.args[0] for c in mock_get.call_args_list if "/heat/celsius/" in c.args[0]]
assert len(heat_calls) == 2
def test_client_swallows_connection_errors():
kg = KXGaugeClient("http://192.168.1.99")
with patch("requests.get", side_effect=OSError("unreachable")):
kg.set_emotion("happy") # must not raise
assert kg._last_emotion is None # failed send - state not updated
def test_health_check_true_on_200():
kg = KXGaugeClient("http://192.168.1.99")
with patch("requests.get") as mock_get:
mock_get.return_value = MagicMock(raise_for_status=lambda: None)
assert kg.health_check() is True
def test_health_check_false_on_error():
kg = KXGaugeClient("http://192.168.1.99")
with patch("requests.get", side_effect=OSError("unreachable")):
assert kg.health_check() is False
# ── Bridge integration: settings GET/POST, test endpoint ───────────────────
def _make_bridge(pid="1", **arg_overrides):
c = MagicMock()
c.callbacks = {}
c.connected = False
args = argparse.Namespace(
printer_ip="192.168.1.50", mqtt_port=9883, username="", password="",
mode_id="20030", device_id="", host="127.0.0.1", port=7125,
data_dir=tempfile.mkdtemp(prefix="kxgauge-"),
kxgauge_url="", kxgauge_enabled=0, kxgauge_heat_peak=250,
)
for k, v in arg_overrides.items():
setattr(args, k, v)
all_bridges = {}
bridge = KobraXBridge(c, args=args, printer_id=pid, all_bridges=all_bridges)
all_bridges[pid] = bridge
return bridge
@pytest_asyncio.fixture
async def kxgauge_client():
bridge = _make_bridge(kxgauge_url="http://192.168.1.77", kxgauge_enabled=1, kxgauge_heat_peak=230)
# _find_config_path() otherwise resolves to the real repo config/config.ini
# (env_loader.find_config_path() walks from the script location, ignoring
# args.data_dir) - a settings POST in a test must never touch that file.
import pathlib
sandbox_cfg = pathlib.Path(tempfile.mkdtemp(prefix="kxgauge-cfg-")) / "config.ini"
bridge._find_config_path = lambda: sandbox_cfg
bridge._restart_bridge = lambda: None # a settings POST schedules a restart - don't kill the test process
app = build_app(bridge)
async with TestClient(TestServer(app)) as c:
yield c, bridge
def _fake_get_response(status=200, json_body=None):
resp = MagicMock()
resp.status = status
resp.json = AsyncMock(return_value=json_body or {})
ctx = MagicMock()
ctx.__aenter__ = AsyncMock(return_value=resp)
ctx.__aexit__ = AsyncMock(return_value=False)
return ctx
@pytest.mark.asyncio
async def test_bridge_creates_kxgauge_client_when_enabled():
bridge = _make_bridge(kxgauge_url="http://192.168.1.77", kxgauge_enabled=1)
assert bridge._kxgauge is not None
assert bridge._kxgauge.base_url == "http://192.168.1.77"
@pytest.mark.asyncio
async def test_bridge_no_kxgauge_client_when_disabled():
bridge = _make_bridge(kxgauge_url="http://192.168.1.77", kxgauge_enabled=0)
assert bridge._kxgauge is None
@pytest.mark.asyncio
async def test_bridge_no_kxgauge_client_when_no_url():
bridge = _make_bridge(kxgauge_enabled=1, kxgauge_url="")
assert bridge._kxgauge is None
@pytest.mark.asyncio
async def test_settings_roundtrip_persists_kxgauge_fields(kxgauge_client):
c, bridge = kxgauge_client
resp = await c.get("/api/settings")
data = await resp.json()
assert data["kxgauge_url"] == "http://192.168.1.77"
assert data["kxgauge_enabled"] == 1
assert data["kxgauge_heat_peak"] == 230
assert isinstance(data["kxgauge_mapping"], dict)
assert data["kxgauge_mapping"]["printing"] == "happy"
@pytest.mark.asyncio
async def test_kxgauge_mapping_default_when_no_config_section():
bridge = _make_bridge()
assert bridge._kxgauge_mapping["free"] == "neutral"
assert bridge._kxgauge_mapping["error"] == "scared"
@pytest.mark.asyncio
async def test_kxgauge_test_endpoint_success(kxgauge_client):
c, bridge = kxgauge_client
with patch("aiohttp.ClientSession.get", return_value=_fake_get_response(200, {"emotion": "HAPPY"})) as mock_get:
resp = await c.post("/api/kxgauge/test", json={})
data = await resp.json()
assert resp.status == 200
assert data["result"] == "ok"
assert data["emotion"] == "HAPPY"
assert mock_get.call_args[0][0] == "http://192.168.1.77/status"
@pytest.mark.asyncio
async def test_kxgauge_test_endpoint_uses_url_from_body_over_configured():
bridge = _make_bridge() # no configured kxgauge_url
app = build_app(bridge)
async with TestClient(TestServer(app)) as c:
with patch("aiohttp.ClientSession.get", return_value=_fake_get_response(200, {"emotion": "NEUTRAL"})) as mock_get:
resp = await c.post("/api/kxgauge/test", json={"url": "http://192.168.1.88"})
data = await resp.json()
assert resp.status == 200
assert mock_get.call_args[0][0] == "http://192.168.1.88/status"
@pytest.mark.asyncio
async def test_kxgauge_test_endpoint_no_url_configured_error():
bridge = _make_bridge()
app = build_app(bridge)
async with TestClient(TestServer(app)) as c:
resp = await c.post("/api/kxgauge/test", json={})
assert resp.status == 400
@pytest.mark.asyncio
async def test_kxgauge_test_endpoint_unreachable_returns_502(kxgauge_client):
c, bridge = kxgauge_client
with patch("aiohttp.ClientSession.get", side_effect=OSError("connection refused")):
resp = await c.post("/api/kxgauge/test", json={})
assert resp.status == 502
@pytest.mark.asyncio
async def test_settings_post_saves_custom_mapping(kxgauge_client):
c, bridge = kxgauge_client
resp = await c.post("/api/settings", json={
"printer_ip": "192.168.1.50", "mqtt_port": 9883,
"kxgauge_url": "http://192.168.1.77", "kxgauge_enabled": 1,
"kxgauge_heat_peak": 240,
"kxgauge_mapping": {"printing": "love", "error": "furious"},
})
assert resp.status == 200
assert bridge._kxgauge_mapping["printing"] == "love"
assert bridge._kxgauge_mapping["error"] == "furious"
@pytest.mark.asyncio
async def test_settings_post_rejects_invalid_emotion(kxgauge_client):
c, bridge = kxgauge_client
before = dict(bridge._kxgauge_mapping)
resp = await c.post("/api/settings", json={
"printer_ip": "192.168.1.50", "mqtt_port": 9883,
"kxgauge_mapping": {"printing": "not_a_real_emotion"},
})
assert resp.status == 200
# invalid value must be rejected, mapping keeps its previous value
assert bridge._kxgauge_mapping["printing"] == before["printing"]
# ── MQTT callback hooks fire the mapped emotion / heat value ───────────────
@pytest.mark.asyncio
async def test_on_temp_pushes_heat_to_kxgauge(kxgauge_client):
c, bridge = kxgauge_client
bridge._kxgauge.set_heat_celsius = MagicMock()
bridge._on_temp({"data": {"curr_nozzle_temp": 205, "target_nozzle_temp": 210,
"curr_hotbed_temp": 60, "target_hotbed_temp": 60}})
bridge._kxgauge.set_heat_celsius.assert_called_once_with(205.0)
@pytest.mark.asyncio
async def test_on_print_pushes_mapped_emotion_to_kxgauge(kxgauge_client):
c, bridge = kxgauge_client
bridge._kxgauge.set_emotion = MagicMock()
bridge._on_print({"state": "printing", "data": {}})
bridge._kxgauge.set_emotion.assert_called_once_with("happy")
@pytest.mark.asyncio
async def test_on_print_unmapped_state_does_not_call_kxgauge(kxgauge_client):
c, bridge = kxgauge_client
bridge._kxgauge_mapping = {} # no entries at all
bridge._kxgauge.set_emotion = MagicMock()
bridge._on_print({"state": "printing", "data": {}})
bridge._kxgauge.set_emotion.assert_not_called()
@pytest.mark.asyncio
async def test_no_kxgauge_configured_callbacks_are_noop():
bridge = _make_bridge() # kxgauge disabled
assert bridge._kxgauge is None
# must not raise even though _kxgauge is None
bridge._on_temp({"data": {"curr_nozzle_temp": 200, "target_nozzle_temp": 200,
"curr_hotbed_temp": 60, "target_hotbed_temp": 60}})
bridge._on_print({"state": "printing", "data": {}})

View File

@@ -30,6 +30,9 @@ async def test_update_check_testing_channel_is_docker_only(client):
c, bridge = client
bridge._read_version = lambda: "testing-2e4dbf0"
# Patch the API so that if the handler wrongly tried to fetch releases,
# the test would notice (mock returns something, but the handler must not
# reach it).
with patch("aiohttp.ClientSession.get") as mock_get:
resp = await c.get("/api/update/check")
data = await resp.json()

View File

@@ -76,6 +76,58 @@ function _updateSpoolmanStatusDot(){
}
}
// ── KXGauge ──
var _kxgaugeMapping={};
var KXGAUGE_EMOTIONS=['neutral','happy','angry','tired','surprised','curious','confused',
'sleepy','love','dizzy','crying','focused','glee','sad','worried','annoyed','skeptic',
'frustrated','unimpressed','suspicious','squint','furious','scared','awe'];
var KXGAUGE_STATES=['free','printing','paused','pause','finished','error','offline'];
function renderKxgaugeMapping(){
var wrap=document.getElementById('kxgauge-mapping-list');
if(!wrap)return;
wrap.innerHTML='';
KXGAUGE_STATES.forEach(function(state){
var row=document.createElement('div');
row.className='set-row';
var lbl=document.createElement('label');
lbl.textContent=state;
var sel=document.createElement('select');
sel.dataset.state=state;
KXGAUGE_EMOTIONS.forEach(function(em){
var opt=document.createElement('option');
opt.value=em;opt.textContent=em;
if((_kxgaugeMapping[state]||'')===em)opt.selected=true;
sel.appendChild(opt);
});
sel.onchange=function(){_kxgaugeMapping[state]=sel.value;};
row.appendChild(lbl);row.appendChild(sel);
wrap.appendChild(row);
});
}
function _resetKxgaugeStatusDot(){
var dot=document.getElementById('kxgauge-status-dot');
var lbl=document.getElementById('kxgauge-status-lbl');
if(!dot||!lbl)return;
dot.style.color='var(--txt2)';lbl.textContent='';
}
function testKxgaugeConnection(){
var dot=document.getElementById('kxgauge-status-dot');
var lbl=document.getElementById('kxgauge-status-lbl');
var url=(document.getElementById('s-kxgauge-url')||{}).value||'';
if(lbl)lbl.textContent='…';
post('/api/kxgauge/test',{url:url}).then(function(r){return r.json()}).then(function(d){
if(d.error){
if(dot)dot.style.color='var(--err)';
if(lbl)lbl.textContent=d.error;
return;
}
if(dot)dot.style.color='var(--ok)';
if(lbl)lbl.textContent='verbunden'+(d.emotion?' ('+d.emotion+')':'');
}).catch(function(){
if(dot)dot.style.color='var(--err)';
if(lbl)lbl.textContent='nicht erreichbar';
});
}
function _buildSpoolmanSection(){
var sec=document.getElementById('fd-spoolman-section');
var rows=document.getElementById('fd-spoolman-rows');
@@ -411,6 +463,11 @@ function applyLang(){
setText('modal-sec-spoolman',T.modal_sec_spoolman||'Spoolman');
setText('lbl-spoolman-url',T.lbl_spoolman_url||'Server-URL');
setText('lbl-spoolman-sync-rate',T.lbl_spoolman_sync_rate||'Sync-Rate (s, 0=aus)');
setText('modal-sec-kxgauge',T.modal_sec_kxgauge||'KXGauge');
setText('lbl-kxgauge-enabled',T.settings_kxgauge_enabled||'Aktiviert');
setText('lbl-kxgauge-url',T.settings_kxgauge_url||'Geräte-URL');
setText('lbl-kxgauge-heat-peak',T.settings_kxgauge_heat_peak||'Ziel-Temperatur (°C)');
setText('lbl-kxgauge-test',T.settings_kxgauge_test||'Verbindung testen');
setText('modal-sec-obico',T.modal_sec_obico||'Obico');
setText('setcat-lbl-system',T.settings_version||'System');
setText('lbl-set-lang',T.settings_cat_language||'Sprache');
@@ -1157,6 +1214,13 @@ function openSettings(){
var su=document.getElementById('s-spoolman-url');if(su)su.value=d.spoolman_server||'';
var sr=document.getElementById('s-spoolman-sync-rate');if(sr)sr.value=(d.spoolman_sync_rate!==undefined?d.spoolman_sync_rate:30);
_updateSpoolmanStatusDot();
// KXGauge
var kge=document.getElementById('s-kxgauge-enabled');if(kge)kge.checked=!!d.kxgauge_enabled;
var kgu=document.getElementById('s-kxgauge-url');if(kgu)kgu.value=d.kxgauge_url||'';
var kgp=document.getElementById('s-kxgauge-heat-peak');if(kgp)kgp.value=(d.kxgauge_heat_peak!==undefined?d.kxgauge_heat_peak:250);
_kxgaugeMapping=d.kxgauge_mapping||{};
renderKxgaugeMapping();
_resetKxgaugeStatusDot();
});
// Sprach-Select im Settings-Panel mit aktueller Sprache spiegeln
var ls=document.getElementById('s-lang-select');
@@ -1920,6 +1984,10 @@ function saveSettings(){
verbose_http_log: (document.getElementById('s-verbose-http-log')||{}).checked?1:0,
spoolman_server: (document.getElementById('s-spoolman-url')||{}).value||'',
spoolman_sync_rate: Math.max(0,parseInt((document.getElementById('s-spoolman-sync-rate')||{}).value||'30',10)),
kxgauge_enabled: (document.getElementById('s-kxgauge-enabled')||{}).checked?1:0,
kxgauge_url: (document.getElementById('s-kxgauge-url')||{}).value||'',
kxgauge_heat_peak: Math.max(0,Math.min(400,parseFloat((document.getElementById('s-kxgauge-heat-peak')||{}).value)||250)),
kxgauge_mapping: _kxgaugeMapping,
}).then(function(){
btn.textContent=T.update_restarting;
setTimeout(function(){

View File

@@ -697,6 +697,29 @@
<span id="spoolman-status-dot"></span> <span id="spoolman-status-lbl"></span>
</div>
</div>
<!-- KXGauge -->
<div class="card" style="margin-top:10px">
<div class="card-title"><span>🙂</span> <span id="modal-sec-kxgauge">KXGauge</span></div>
<div class="set-row">
<label id="lbl-kxgauge-enabled">Aktiviert</label>
<input type="checkbox" id="s-kxgauge-enabled">
</div>
<div class="set-row">
<label id="lbl-kxgauge-url">Geräte-URL</label>
<input type="text" id="s-kxgauge-url" placeholder="http://192.168.x.x" style="width:200px">
</div>
<div class="set-row">
<label id="lbl-kxgauge-heat-peak">Ziel-Temperatur (°C)</label>
<input type="number" id="s-kxgauge-heat-peak" min="0" max="400" value="250" style="width:80px">
</div>
<div class="set-row">
<button type="button" id="btn-kxgauge-test" onclick="testKxgaugeConnection()"><span id="lbl-kxgauge-test">Verbindung testen</span></button>
</div>
<div id="kxgauge-status-row" style="margin-top:6px;font-size:12px;color:var(--txt2)">
<span id="kxgauge-status-dot"></span> <span id="kxgauge-status-lbl"></span>
</div>
<div id="kxgauge-mapping-list" style="margin-top:10px"></div>
</div>
<!-- Obico -->
<div class="card" style="margin-top:10px">
<div class="card-title"><span>🕵</span> <span id="modal-sec-obico">Obico</span></div>

View File

@@ -173,6 +173,7 @@
"log_topic_label": "Thema:",
"log_topic_print": "Druck",
"log_topic_status": "Status",
"modal_sec_kxgauge": "KXGauge",
"modal_sec_obico": "Obico",
"modal_sec_spoolman": "Spoolman",
"nav_ams": "AMS",
@@ -258,6 +259,10 @@
"settings_file_ready_dialog": "Druckdialog",
"settings_file_ready_mode": "Nach Upload: Druckstart-Verhalten",
"settings_integrations": "Integrationen",
"settings_kxgauge_enabled": "Aktiviert",
"settings_kxgauge_heat_peak": "Ziel-Temperatur (°C)",
"settings_kxgauge_test": "Verbindung testen",
"settings_kxgauge_url": "Geräte-URL",
"settings_language": "Sprache",
"settings_mode_id": "Mode-ID",
"settings_mode_id_placeholder": "20030",

View File

@@ -173,6 +173,7 @@
"log_topic_label": "Topic:",
"log_topic_print": "Print",
"log_topic_status": "Status",
"modal_sec_kxgauge": "KXGauge",
"modal_sec_obico": "Obico",
"modal_sec_spoolman": "Spoolman",
"nav_ams": "AMS",
@@ -258,6 +259,10 @@
"settings_file_ready_dialog": "Print dialog",
"settings_file_ready_mode": "After upload: Start print behavior",
"settings_integrations": "Integrations",
"settings_kxgauge_enabled": "Enabled",
"settings_kxgauge_heat_peak": "Target temperature (°C)",
"settings_kxgauge_test": "Test connection",
"settings_kxgauge_url": "Device URL",
"settings_language": "Language",
"settings_mode_id": "Mode ID",
"settings_mode_id_placeholder": "20030",

View File

@@ -173,6 +173,7 @@
"log_topic_label": "Tema:",
"log_topic_print": "Impresión",
"log_topic_status": "Estado",
"modal_sec_kxgauge": "KXGauge",
"modal_sec_obico": "Obico",
"modal_sec_spoolman": "Spoolman",
"nav_ams": "AMS",
@@ -258,6 +259,10 @@
"settings_file_ready_dialog": "Diálogo de impresión",
"settings_file_ready_mode": "Después de carga: Comportamiento de inicio de impresión",
"settings_integrations": "Integraciones",
"settings_kxgauge_enabled": "Activado",
"settings_kxgauge_heat_peak": "Temperatura objetivo (°C)",
"settings_kxgauge_test": "Probar conexión",
"settings_kxgauge_url": "URL del dispositivo",
"settings_language": "Idioma",
"settings_mode_id": "ID de modo",
"settings_mode_id_placeholder": "20030",

View File

@@ -161,6 +161,7 @@
"log_topic_label": "Sujet :",
"log_topic_print": "Impression",
"log_topic_status": "Statut",
"modal_sec_kxgauge": "KXGauge",
"modal_sec_obico": "Obico",
"modal_sec_spoolman": "Spoolman",
"nav_ams": "AMS",
@@ -246,6 +247,10 @@
"settings_file_ready_dialog": "Dialogue d'impression",
"settings_file_ready_mode": "Après téléchargement : Comportement de démarrage d'impression",
"settings_integrations": "Intégrations",
"settings_kxgauge_enabled": "Activé",
"settings_kxgauge_heat_peak": "Température cible (°C)",
"settings_kxgauge_test": "Tester la connexion",
"settings_kxgauge_url": "URL de l'appareil",
"settings_language": "Langue",
"settings_mode_id": "ID du mode",
"settings_mode_id_placeholder": "20030",

View File

@@ -161,6 +161,7 @@
"log_topic_label": "Argomento:",
"log_topic_print": "Stampa",
"log_topic_status": "Stato",
"modal_sec_kxgauge": "KXGauge",
"modal_sec_obico": "Obico",
"modal_sec_spoolman": "Spoolman",
"nav_ams": "AMS",
@@ -246,6 +247,10 @@
"settings_file_ready_dialog": "Finestra di dialogo stampa",
"settings_file_ready_mode": "Dopo il caricamento: Comportamento di avvio stampa",
"settings_integrations": "Integrazioni",
"settings_kxgauge_enabled": "Abilitato",
"settings_kxgauge_heat_peak": "Temperatura obiettivo (°C)",
"settings_kxgauge_test": "Verifica connessione",
"settings_kxgauge_url": "URL dispositivo",
"settings_language": "Lingua",
"settings_mode_id": "ID modalità",
"settings_mode_id_placeholder": "20030",

View File

@@ -173,6 +173,7 @@
"log_topic_label": "主题:",
"log_topic_print": "打印",
"log_topic_status": "状态",
"modal_sec_kxgauge": "KXGauge",
"modal_sec_obico": "Obico",
"modal_sec_spoolman": "Spoolman",
"nav_ams": "AMS",
@@ -258,6 +259,10 @@
"settings_file_ready_dialog": "打印对话框",
"settings_file_ready_mode": "上传后:开始打印行为",
"settings_integrations": "集成",
"settings_kxgauge_enabled": "启用",
"settings_kxgauge_heat_peak": "目标温度 (°C)",
"settings_kxgauge_test": "测试连接",
"settings_kxgauge_url": "设备地址",
"settings_language": "语言",
"settings_mode_id": "模式 ID",
"settings_mode_id_placeholder": "20030",