diff --git a/docs/architecture/lib-map.md b/docs/architecture/lib-map.md index 7e1ec4bc..0045e8b7 100644 --- a/docs/architecture/lib-map.md +++ b/docs/architecture/lib-map.md @@ -112,6 +112,8 @@ Target package: `lib/geometry/`, with subpackages only when a real boundary appe `dossier.py` / `dossier_render.py` / `webui.py` / `render.py` (`3d printers show NAME`: name resolution, the concurrent read-only live reads, the web UI probe, the text; `render.py` is the status text shared with `3d printer status`), `preflight.py` (pre-print checks), + `colour_fix.py` (the CIEDE2000 slot colour tolerance and the `--fix` / `--fix-colour` rewrite of a + G-code + project 3MF + sidecars to the loaded colour), `move_bounds.py` (G-code moves and arc extremes vs axis travel / printable area), `force_hatch.py` (the owner's Telegram hatch approval that `3d printer start --force` needs), `watch.py` / `watch_run.py` / `watch_sinks.py` (`3d printer watch`: detectors, streams, diff --git a/docs/commands/print-preflight.md b/docs/commands/print-preflight.md index e98856cf..3f6d5526 100644 --- a/docs/commands/print-preflight.md +++ b/docs/commands/print-preflight.md @@ -10,17 +10,18 @@ input is missing is SKIP and its line says what to pass — it never passes sile ``` 3d print-preflight FILE.gcode [--printer NAME] [--axis-limits X0:X1,Y0:Y1|NAME] [--source-3mf FILE.3mf] [--visible [OBJ:]SPEC]... [--bridge-max MM] [--printops-api URL] [--offline] - [--printops-snapshot FILE] [--accept preset_match|mix_calibration|spool_in_sync --reason TEXT] [--json] [--secs S] [--no-bed-move] [--fix] + [--printops-snapshot FILE] [--accept preset_match|mix_calibration|spool_in_sync --reason TEXT] [--json] [--secs S] [--no-bed-move] + [--colour-delta-e N] [--fix [--fix-max-delta-e N]] ``` -`--fix` adds the thumbnails the G-code lacks, in place (header only; mtime kept, a current slice-warnings record follows the new SHA-256), then runs the checks. It is the only option that writes. +`--fix` first adds the thumbnails the G-code lacks, in place (header only; mtime kept, a current slice-warnings record follows the new SHA-256), and, with a live `--printer`, rewrites the file's filament colours to the ones the printer reports; then it runs the checks. It is the only option that writes. | Check | When | FAIL means | |---|---|---| | `thumbnails` | always | The G-code lacks preview images its `; thumbnails =` spec asks for (48x48, 96x96 and 300x300 PNG when it names none). The line names the exact fix, `3d gcode-thumbnails add 'FILE'`; `--fix` runs it. [Layers](gcode-thumbnails.md#layers-a-thumbnail-cannot-be-missing) add thumbnails earlier, so this FAIL means the G-code came from outside them. | | `printer` | `--printer` | The printer is printing, paused, busy, or its state cannot be read. | | `bed` | `--printer` with `watch.bed.enabled` (SKIP otherwise) | The camera sees objects on the bed or the build plate missing against the empty-bed reference; WARN when it cannot tell (no reference at this Z, plate out of view, camera down). See [Bed state](printer.md#bed-state). On an idle printer with `watch.bed.auto_view` (the U1: its camera sees 15 % of the plate at the Z 200 rest) the bed is moved in Z only to where the whole plate is in view, judged there and put back, and the line names the Z used; never while printing/paused, never homing; `--no-bed-move` judges at the current Z. | -| `slot Tn` | `--printer` on a printer with per-head data (Snapmaker U1) | Tool *n* is used by the G-code but head T*n+1* is empty, or its colour/material differs from `; filament_colour` / `; filament_type`. | +| `slot Tn` | `--printer` on a printer with per-head data (Snapmaker U1) | Tool *n* is used by the G-code but head T*n+1* is empty, its material differs from `; filament_type`, or its colour is further than `--colour-delta-e` (CIEDE2000, default 10) from `; filament_colour`. The line shows the distance; `--fix` aligns the file with the printer, [`3d printer slot set`](printer.md#slot-colour) the printer with the file. | | `axis travel` | always (SKIP when the limits are unknown) | A move, or the bulge of an arc, leaves the printer's X/Y travel anywhere in the file. See [`3d gcode-bounds`](gcode-bounds.md). | | `print area` | always | A plate move leaves the G-code's `; printable_area`, or an extruding path crosses `; bed_exclude_area` (vendor blocks exempt: `;TYPE:Custom`; on Bambu Studio G-code everything outside its object blocks, wipe tower and features — see [`3d gcode-bounds`](gcode-bounds.md)). | | `macros` | when the G-code calls macros | A macro (PAUSE, a vendor toolchange, …) is not in the printer's `trusted_macros`: its moves cannot be checked. | @@ -136,6 +137,9 @@ Options: - `--bridge-max MM` — longest span the overhangs gate counts as a bridge; overrides the printer's `bridge_max_mm`: `3d print-preflight out/porch.gcode --axis-limits u1 --bridge-max 25` - `--json` — JSON report (`checks[]` with `name`, `status`, `detail`, `lines`): `3d print-preflight out/part.gcode --printer u1 --json | jq '.checks[] | select(.status=="FAIL")'` - `--no-bed-move` — the `bed` line judges at the current Z: the printer's bed is not moved to the best inspection Z (`watch.bed.auto_view`): `3d print-preflight out/part.gcode --printer u1 --no-bed-move` +- `--colour-delta-e N` — the `slot Tn` colour tolerance, CIEDE2000 (default 10, the `colour.warn_delta_e` of `fleet/policy.yaml`). The slot line shows the distance either way. The beige PETG of 2026-11-04 (sliced `#E8D8B0`, shown by the U1 as `#FEE5A5`) is ΔE 6.4 and passes; white against that beige is 20.0 and fails; red against blue is 52.9. A **material** mismatch is always a FAIL, whatever the colour: `3d print-preflight out/part.gcode --printer u1 --colour-delta-e 5` +- `--fix` — before the checks, first add the thumbnails the G-code lacks (in place, header only, mtime kept; no printer needed), then, with a live `--printer`, rewrite `; filament_colour` / `; extruder_colour` for every tool the G-code uses to the colour the printer reports, in the G-code **and** in its project 3MF (`--source-3mf`, else `NAME.3mf` next to the G-code; `Metadata/project_settings.config`, `Metadata/plate_N.json`). The toolpath bytes, mode and mtime stay; `NAME.slice-warnings.json` follows the new G-code SHA-256 and `NAME.manifest.json` gets the new slot colour; the old and new SHA-256 of every rewritten file go to stderr (the file's hash changed: upload it again). A head further than `--fix-max-delta-e` away is refused with nothing written (another filament, not the same one reported differently). Without a live `--printer` (none given, `--offline`, an offline printer) only the thumbnails are added and stderr says the colours were not aligned. It is the one option that writes: `3d print-preflight out/plate.gcode --printer u1 --fix` +- `--fix-max-delta-e N` — the largest ΔE `--fix` rewrites across (default 30): `3d print-preflight out/plate.gcode --printer u1 --fix --fix-max-delta-e 40` - `--secs S` — Creality: seconds to listen for pushed state: `3d print-preflight out/roof.gcode --printer k1se --secs 5` - `--printops-api URL` — 3d-api for the PrintOps gates (default `$PRINTOPS_API`, else `printops.yaml`): `3d print-preflight out/part.gcode --printer u1 --printops-api http://192.168.0.39:8765` - `--offline` — no network: no printer contact, PrintOps state from the snapshot (the slicer hook mode); together with `--printops-api` it is a usage error: `3d print-preflight out/part.gcode --offline` diff --git a/docs/commands/printer.md b/docs/commands/printer.md index fca03efd..93273252 100644 --- a/docs/commands/printer.md +++ b/docs/commands/printer.md @@ -18,6 +18,9 @@ Epic: [#90](https://github.com/alex-mextner/3d-cli/issues/90). 3d printer status NAME [--json] [--secs S] 3d printer upload NAME FILE.gcode [--as REMOTE] 3d printer start NAME FILE.gcode --yes [--as REMOTE] [--source-3mf X.3mf] [--bridge-max MM] [--force] [--secs S] [--no-bed-move] + [--colour-delta-e N] [--fix-colour [--fix-max-delta-e N]] +3d printer slot set NAME T --material M --color #RRGGBB [--brand B] [--sub-type S] [--dry-run] [--force-official] [--secs S] [--json] +3d printer slot show NAME [--secs S] [--json] 3d printer cancel NAME --yes 3d printer evidence NAME [--job ID | --last] [--out DIR] [--gcode FILE.gcode] [--note TEXT] [--klippy-log FILE] [--job-file FILE] 3d printer guard NAME --job-id ID --yes [--interval S] [--evidence DIR] @@ -86,6 +89,56 @@ Options: head-sequence check on it and `3d halves-check` (without it, the head-sequence check uses `NAME.3mf` next to `NAME.gcode` when present). `3d printer start u1 out/base.gcode --yes --source-3mf print/base.3mf` +- `--colour-delta-e N` — `start`: the preflight `slot Tn` colour tolerance, CIEDE2000, default 10 + (see [Slot colour](#slot-colour)). `3d printer start u1 out/plate.gcode --yes --colour-delta-e 8` +- `--fix-colour` — `start`: first rewrite the file (and its project 3MF) to the colours the printer + reports, print the old/new SHA-256, upload the fixed file, then run the preflight and start. + `3d printer start u1 out/plate.gcode --yes --fix-colour` +- `--fix-max-delta-e N` — the largest ΔE `--fix-colour` rewrites across (default 30). + `3d printer start u1 out/plate.gcode --yes --fix-colour --fix-max-delta-e 40` + +## Slot colour + +The preflight `slot Tn` line compares the colour the slicer wrote (`; filament_colour`) with the +colour the printer reports for that head by **CIEDE2000 ΔE**, default tolerance 10 +(`--colour-delta-e`), and prints the distance. The same beige PETG is `#E8D8B0` in the slice and +`#FEE5A5` on the U1 screen: ΔE 6.4, PASS. A material mismatch is always a FAIL. Three ways out of a +real mismatch, none of them `--force`: + +```bash +# 1. the file is wrong about the colour: rewrite it at the source (G-code + project 3MF + sidecars) +3d print-preflight out/plate.gcode --printer u1 --fix +3d printer start u1 out/plate.gcode --yes --fix-colour # rewrite, upload the fixed file, preflight, start +# 2. the printer is wrong about what is loaded: write the head's record (idle printers only) +3d printer slot set u1 T4 --material PETG --color '#E8D8B0' --brand Elegoo --dry-run +3d printer slot set u1 T4 --material PETG --color '#E8D8B0' --brand Elegoo +# 3. load the filament the file was sliced for +``` + +`--fix` / `--fix-colour` set entry *n* of `filament_colour` and `extruder_colour` to the printer's +colour for every tool the G-code uses, in the G-code and in its project 3MF +(`Metadata/project_settings.config`, `plate_N.json`), keep mode and mtime, move +`NAME.slice-warnings.json` to the new G-code SHA-256 and update `NAME.manifest.json`, and print the +old and new SHA-256 of each file (the hash is part of the stored file name, hence the upload in +`start --fix-colour`). A head whose colour is further than `--fix-max-delta-e` (default 30) from the +slice is refused with nothing written: that is another filament, and relabelling it would hide a +wrong-colour print. + +`slot set NAME T --material M --color #RRGGBB [--brand B] [--sub-type S] [--dry-run] +[--force-official]` is `3d printers filament set NAME/T` (`--brand` is its `--vendor`); `T` +is the label the printer shows (U1: T1..T4 = G-code tools T0..T3). `slot show NAME` reads the heads. + +**What the Snapmaker U1 accepts** (Snapmaker/u1-klipper `print_task_config.py`, issue #400): the +touchscreen's own Klipper command `SET_PRINT_FILAMENT_CONFIG CONFIG_EXTRUDER= VENDOR=… FILAMENT_TYPE=… +FILAMENT_SUBTYPE=… FILAMENT_COLOR_RGBA=RRGGBBFF`, sent through `POST /printer/gcode/script` (no +webhook endpoint exists for it) and read back from `print_task_config`. It needs vendor, type and +sub-type together (defaults Generic / Basic), persists to `print_task.json`, resets the head's +pressure advance (the answer prints the K it has now), and refuses a head with official RFID filament +unless `--force-official`. 3d refuses before sending when the fleet entry is not `access: full` +(read-only and offline printers), while printing or paused, or when Klipper is not ready. Not used: +the RFID path `FILAMENT_DT_UPDATE` / the Extended Firmware's `POST /printer/filament_detect/set` +(it marks the slot official, which locks later writes). Other adapters (Creality) have no +per-head write: `slot set` exits 2 and `slot show` is the read-only comparison. ## Examples @@ -827,6 +880,16 @@ RIG_HATCH_REQUEST_3D_PRINTER_START_FORCE="loaded_filament_match: slice has no co `RIG_HATCH_REQUEST_3D_PRINTER_START_FORCE=approved:hgr-…`, redeems. - Approved: the start continues, prints `FORCED: starting despite preflight FAIL (...)`, and the helper appends the use to `~/.config/agent-tools/overrides.log`. +- If the daemon could not even post the card (`tg-ctl ask-forward not-posted … daemon declined: + reason=send-failed`: its Telegram `sendMessage` failed, 2026-11-04 it was `getaddrinfo EAI_AGAIN + api.telegram.org`, a resolver blip while plain `tg` worked) nothing was posted, so the ask is + repeated after 3, 10 and 30 s (four requests in all, each reported on stderr). If it still + fails the start is refused with the way around Telegram: the file is already stored on the + printer, so start it **from the printer screen** (Files, then Print) — the screen does not run + these checks, do it only when the listed FAILs are understood — and with the log to read + (`~/.config/tg-cli/tg-ctl.*.log`, lines `sendMessage failed` / `ask-forward dropped`). + The daemon itself sends once and never retries (tg-cli `forwardAndPostQuestion`), which is why + the retry lives here. - Without `--force`, without the variable, denied, timed out, or without a trusted `tg-ctl` / helper (`~/dev/agent-tools`, then `~/xp/agent-tools`, owned by you or root and not group/world-writable): exit 1, nothing is sent to the printer. diff --git a/lib/commands/print_preflight.py b/lib/commands/print_preflight.py index cc6cdf9e..bf8f11c1 100644 --- a/lib/commands/print_preflight.py +++ b/lib/commands/print_preflight.py @@ -37,12 +37,11 @@ from printer_ops.devices import Device from printer_ops.move_bounds import Limits from printer_ops.preflight import Check - from printer_ops.printops_checks import PrintopsContext USAGE = """3d print-preflight FILE.gcode [--printer NAME] [--axis-limits X0:X1,Y0:Y1|NAME] [--source-3mf FILE.3mf] [--visible [OBJ:]SPEC]... [--bridge-max MM] [--printops-api URL] [--offline] [--printops-snapshot FILE] [--accept preset_match|mix_calibration|spool_in_sync --reason TEXT] [--json] [--secs S] - [--no-bed-move] [--fix] + [--no-bed-move] [--colour-delta-e N] [--fix [--fix-max-delta-e N]] Read-only checks before a print (the one write is opt-in: --fix). `3d printer start` runs the same checks and refuses on FAIL. Every G-code gate runs; one whose input is missing is SKIP and says what to pass. @@ -57,8 +56,10 @@ is in view, judged there and put back; the line names the Z used. Never while printing/paused, never homing; --no-bed-move judges at the current Z slot Tn (--printer, per-head printers such as the Snapmaker U1) each head the G-code - uses is loaded, and its colour and material equal `; filament_colour` / - `; filament_type` for that tool + uses is loaded, its material equals `; filament_type` for that tool (a + mismatch is always a FAIL) and its colour is within --colour-delta-e of + `; filament_colour` (CIEDE2000, default 10; the line shows the distance: the + beige #E8D8B0 sliced for a head reporting #FEE5A5 is ΔE 6.4 and passes) axis travel every G0-G3 move (arcs by their real extremes, start/end G-code included) stays inside the printer's X/Y limits; SKIP when the limits are unknown print area plate moves stay inside the G-code's `; printable_area`; no extruding path @@ -138,9 +139,21 @@ into a WARN that carries --reason; only those three; needs --reason TEXT --reason TEXT why the accepted mismatch is printed --no-bed-move the bed line judges at the current Z: the printer's bed is not moved - --fix add the thumbnails the G-code lacks, in place (header only: the toolpath is - unchanged, mtime kept, a current slice-warnings record follows the new - SHA-256), then run the checks. The only option that writes. + --colour-delta-e N slot Tn colour tolerance, CIEDE2000 (default 10) + --fix first add the thumbnails the G-code lacks, in place (header only: the toolpath + is unchanged, mtime kept, a current slice-warnings record follows the new + SHA-256); with a live --printer then also rewrite `; filament_colour` / + `; extruder_colour` of + every tool the G-code uses to the colour the printer reports, in the G-code and + its project 3MF (--source-3mf, else NAME.3mf beside it; Metadata/ + project_settings.config), keeping mode/mtime; NAME.slice-warnings.json and + NAME.manifest.json follow; old and new sha256 of each file are printed to + stderr. A head whose colour is further than --fix-max-delta-e away is refused + (another filament, nothing written). Then the checks run. The only option + that writes (re-upload the file after it: its hash changed). Without a live + --printer (none given, --offline, an offline printer) only the thumbnails are + added and stderr says the colours were not aligned + --fix-max-delta-e N the largest ΔE --fix rewrites across (default 30) --json JSON report on stdout --secs S Creality: seconds to listen for pushed state (default 3) @@ -154,6 +167,7 @@ 3d print-preflight out/roof.gcode --axis-limits -5:229,-0.5:227 3d print-preflight print/k1se/k1_base_4_v3.gcode --axis-limits k1se 3d print-preflight out/walls.gcode --visible 'WALL*:y<104.8' + 3d print-preflight out/plate.gcode --printer u1 --fix # same beige PETG, other hex: rewrite + sha256s 3d print-preflight out/part.gcode --printer u1 --json | jq '.checks[] | select(.status=="FAIL")' 3d print-preflight out/part.gcode --printer u1 --printops-api http://192.168.0.39:8765 3d print-preflight out/part.gcode --offline # slicer hook: PrintOps snapshot, no network @@ -161,7 +175,8 @@ _VALUE_OPTIONS = ("--printer", "--source-3mf", "--secs", "--axis-limits", "--visible", "--bridge-max", - "--printops-api", "--printops-snapshot", "--accept", "--reason") + "--printops-api", "--printops-snapshot", "--accept", "--reason", "--colour-delta-e", + "--fix-max-delta-e") _FLAG_OPTIONS = ("--json", "--offline", "--no-bed-move", "--fix") @@ -222,20 +237,11 @@ def _resolve_inputs(pos: list[str], opts: dict[str, str]) -> tuple[Path, Path | return gcode, source -def _apply_fix(gcode: Path) -> None: - """--fix: add the missing thumbnails in place before the checks run.""" - from slicing.thumbnails import fix_in_place - - fixed = fix_in_place(gcode) - if fixed is not None: - print(f"--fix: added {', '.join(map(str, fixed.added))} to {gcode}", file=sys.stderr) - - def _offline_profile(opts: dict[str, str], printer: str | None, offline: bool) -> Device | None: """The static profile standing in for --printer when it is offline or --offline is given. - Such a printer is never contacted: its profile replaces --printer, and an explicit - --axis-limits replaces only its limits. Mutates `opts` accordingly.""" + Such a printer is never contacted: its profile replaces --printer, and an explicit --axis-limits + replaces only its limits. Mutates `opts` accordingly.""" from printer_ops.devices import get_device profile = get_device(printer, offline_ok=True) if printer is not None else None @@ -258,8 +264,29 @@ def _static_device(opts: dict[str, str], profile: Device | None) -> tuple[Device return device_for(spec), trusted_macros_for(spec) -def _connect(opts: dict[str, str], profile: Device | None) -> _Target: - """Device, live status, axis limits and trusted macros for the run; reads the printer if --printer.""" +def _fix_thumbnails(gcode: Path) -> None: + """--fix: add the missing thumbnails in place before the checks run.""" + from slicing.thumbnails import fix_in_place + + fixed = fix_in_place(gcode) + if fixed is not None: + print(f"--fix: added {', '.join(map(str, fixed.added))} to {gcode}", file=sys.stderr) + + +def _fix_colours(gcode: Path, source: Path | None, status: PrinterStatus, opts: dict[str, str]) -> None: + """--fix: rewrite the file to the loaded colours before the checks run (report on stderr).""" + from printer_ops.colour_fix import DEFAULT_FIX_MAX_DELTA_E, fix_to_loaded, parse_delta_e + + limit = parse_delta_e(opts.get("--fix-max-delta-e"), "--fix-max-delta-e", DEFAULT_FIX_MAX_DELTA_E, + command="print-preflight") + fixed = fix_to_loaded(gcode, status, source, limit, command="print-preflight") + print("\n".join(fixed.lines()) if fixed else "--fix: the loaded colours already equal the G-code's", + file=sys.stderr) + + +def _connect(opts: dict[str, str], profile: Device | None, gcode: Path, source: Path | None) -> _Target: + """Device, live status, axis limits and trusted macros for the run; reads the printer if --printer + (and, with --fix, rewrites the file to the loaded colours).""" from printer_ops import adapter_by_name, limits_from_option device, trusted = _static_device(opts, profile) @@ -268,29 +295,19 @@ def _connect(opts: dict[str, str], profile: Device | None) -> _Target: if "--axis-limits" in opts: limits, limits_source = limits_from_option(opts["--axis-limits"], command="print-preflight") if "--printer" not in opts: + if "--fix" in opts: + print("--fix: no live --printer, the colours were not aligned (thumbnails only)", file=sys.stderr) return _Target(device, None, limits, limits_source, trusted) adapter = adapter_by_name(opts["--printer"], secs=opts.get("--secs"), command="print-preflight") status = adapter.status() + if "--fix" in opts: + _fix_colours(gcode, source, status, opts) if "--axis-limits" not in opts: limits, limits_source = adapter.axis_limits() # this printer's macros only, whatever --axis-limits names return _Target(adapter.device, status, limits, limits_source, adapter.trusted_macros) -def _setup_printer(opts: dict[str, str]) -> tuple[PrintopsContext, _Target]: - """PrintOps gate context plus the printer target, from --printer / --axis-limits / --offline.""" - from printer_ops.printops_checks import context_from_options - - printer = opts.get("--printer") - offline = "--offline" in opts - profile = _offline_profile(opts, printer, offline) - printops = context_from_options(printer, opts.get("--axis-limits"), api=opts.get("--printops-api"), - offline=offline, snapshot=opts.get("--printops-snapshot"), - command="print-preflight", accept=opts.get("--accept"), - reason=opts.get("--reason")) - return printops, _connect(opts, profile) - - def _bed_line(opts: dict[str, str], device: Device | None) -> Check | None: """The camera bed line, only for a real --printer whose device is known.""" if "--printer" not in opts or device is None: @@ -311,16 +328,28 @@ def run(argv: list[str]) -> int: gcode, source = _resolve_inputs(pos, opts) from printer_ops import bridge_max_for + from printer_ops.colour_fix import DEFAULT_SLOT_DELTA_E, parse_delta_e from printer_ops.preflight import FAIL, run_preflight + from printer_ops.printops_checks import context_from_options + colour_de = parse_delta_e(opts.get("--colour-delta-e"), "--colour-delta-e", DEFAULT_SLOT_DELTA_E, + command="print-preflight") + printer = opts.get("--printer") + offline = "--offline" in opts + profile = _offline_profile(opts, printer, offline) if "--fix" in opts: - _apply_fix(gcode) - printops, target = _setup_printer(opts) + _fix_thumbnails(gcode) + printops = context_from_options(printer, opts.get("--axis-limits"), api=opts.get("--printops-api"), + offline=offline, snapshot=opts.get("--printops-snapshot"), + command="print-preflight", accept=opts.get("--accept"), + reason=opts.get("--reason")) + target = _connect(opts, profile, gcode, source) bed = _bed_line(opts, target.device) bridge, bridge_source = bridge_max_for(opts.get("--bridge-max"), target.device, command="print-preflight") report = run_preflight(gcode, status=target.status, source_3mf=source, axis_limits=target.limits, limits_source=target.limits_source, trusted_macros=target.trusted, visible=visible, - bridge_max=bridge, bridge_source=bridge_source, bed=bed, printops=printops) + bridge_max=bridge, bridge_source=bridge_source, bed=bed, printops=printops, + colour_delta_e=colour_de) print(json.dumps(report.to_json(), indent=1) if "--json" in opts else report.render()) if report.overall == FAIL: raise GateFailure("preflight FAIL", command="print-preflight", silent=True) diff --git a/lib/commands/printer.py b/lib/commands/printer.py index 570aa93a..7397fbdc 100644 --- a/lib/commands/printer.py +++ b/lib/commands/printer.py @@ -53,7 +53,7 @@ A FILE without its preview thumbnails gets them first (in place, sha256 printed: `3d gcode-thumbnails check FILE`) start NAME FILE.gcode --yes [--as REMOTE] [--source-3mf X.3mf] [--bridge-max MM] [--force] [--secs S] - [--no-bed-move] + [--no-bed-move] [--colour-delta-e N] [--fix-colour [--fix-max-delta-e N]] start a file that was uploaded before (also adds missing thumbnails to FILE in place: then upload FILE again); runs `3d print-preflight` first, plus the camera bed check @@ -61,6 +61,14 @@ printer has watch.bed.enabled (an idle printer with watch.bed.auto_view has its bed moved in Z to where the camera sees the whole plate, and back: --no-bed-move) + slot set NAME T --material M --color #RRGGBB [--brand B] [--sub-type S] [--dry-run] [--force-official] + [--secs S] [--json] + write the filament one head shows it holds (U1: the touchscreen's + SET_PRINT_FILAMENT_CONFIG, read back) so the printer's record equals + what is loaded; T is the label on the printer (T1..T4 = G-code + T0..T3). Idle, `access: full` printers only; --dry-run prints the + before/after and every rule that would refuse, writing nothing. + `slot show NAME` reads the heads (see `3d printers filament --help`) cancel NAME --yes cancel the running/paused job evidence NAME [--job ID | --last] [--out DIR] [--gcode F] [--note T] after EVERY print (done/cancelled/failed): save the Moonraker @@ -124,6 +132,14 @@ --bridge-max MM preflight overhangs gate: longest bridge; overrides the printer's bridge_max_mm in printer-devices.yaml (default 15) --no-bed-move start: the camera bed check judges at the current Z (the bed is not moved) + --colour-delta-e N start: slot Tn colour tolerance, CIEDE2000 (default 10; the preflight line shows the + distance; a material mismatch is always a FAIL) + --fix-colour start: first rewrite `; filament_colour` / `; extruder_colour` of every tool the file + uses to the colour the printer reports, in FILE and its project 3MF (--source-3mf, + else NAME.3mf beside it), keeping mtime/mode; slice-warnings and manifest follow; + old/new sha256 are printed; the fixed FILE is uploaded (its hash, so its stored + name, changed), then the preflight runs. A head whose colour is further than + --fix-max-delta-e (default 30) away is refused: another filament Safety: start refuses while the printer is printing, paused, busy or unreadable, and when the content-addressed remote file is missing, differs in size, or (Moonraker, read back) @@ -141,6 +157,8 @@ 3d printer upload k1se "out/roof left.gcode" --as roof_left.gcode 3d printer start u1 out/part.gcode --yes 3d printer start u1 out/base.gcode --yes --source-3mf print/base.3mf + 3d printer start u1 out/plate.gcode --yes --fix-colour # slicer beige != the screen's beige: rewrite, upload, start + 3d printer slot set u1 T3 --material PETG --color '#FEE5A5' --brand Elegoo --dry-run RIG_HATCH_REQUEST_3D_PRINTER_START_FORCE="roof overhang FAIL is a stale map, issue #N" \\ 3d printer start k1se out/roof.gcode --yes --force # waits for the owner's Telegram tap 3d printer cancel k1se --yes @@ -157,8 +175,8 @@ 3d printer todo list u1 3d printer -n k1se issues add --title "X belt squeaks" --type inspect""" -_VALUE = {"--secs", "--as", "--source-3mf", "--bridge-max"} -_FLAGS = {"--json", "--yes", "--force", "--no-bed-move"} +_VALUE = {"--secs", "--as", "--source-3mf", "--bridge-max", "--colour-delta-e", "--fix-max-delta-e"} +_FLAGS = {"--json", "--yes", "--force", "--no-bed-move", "--fix-colour"} def _parse(argv: list[str], allowed: set[str], n_pos: int, sub: str) -> tuple[list[str], dict[str, str]]: @@ -391,7 +409,8 @@ def _cmd_start(argv: list[str]) -> int: import tempfile from pathlib import Path - (name, file), opts = _parse(argv, {"--yes", "--force", "--as", "--source-3mf", "--secs", "--bridge-max", "--no-bed-move"}, 2, + (name, file), opts = _parse(argv, {"--yes", "--force", "--as", "--source-3mf", "--secs", "--bridge-max", "--no-bed-move", + "--colour-delta-e", "--fix-colour", "--fix-max-delta-e"}, 2, "start") if "--yes" not in opts: raise UsageError( @@ -450,6 +469,7 @@ def _preflight_report(name: str, adapter: Adapter, st: PrinterStatus, original: source: Path | None, opts: dict[str, str]) -> Any: """Run and print the preflight on the snapshot; the report is relabelled with the original path.""" from printer_ops import bridge_max_for + from printer_ops.colour_fix import DEFAULT_SLOT_DELTA_E, parse_delta_e from printer_ops.preflight import check_bed, run_preflight from printer_ops.printops_checks import context_from_options @@ -464,6 +484,8 @@ def _preflight_report(name: str, adapter: Adapter, st: PrinterStatus, original: report = run_preflight(local, status=st, source_3mf=source, axis_limits=limits, limits_source=limits_source, trusted_macros=adapter.trusted_macros, origin=original, bridge_max=bridge, bridge_source=bridge_source, printops=printops, + colour_delta_e=parse_delta_e(opts.get("--colour-delta-e"), "--colour-delta-e", + DEFAULT_SLOT_DELTA_E, command="printer"), bed=check_bed(adapter.device, move="--no-bed-move" not in opts) if adapter.device is not None else None) report.file = f"{original} (snapshot)" @@ -471,7 +493,7 @@ def _preflight_report(name: str, adapter: Adapter, st: PrinterStatus, original: return report -def _require_preflight_pass(report: Any, name: str, original: Path, local_hash: str, +def _require_preflight_pass(report: Any, name: str, original: Path, local_hash: str, remote: str, opts: dict[str, str]) -> None: """Refuse on a preflight FAIL unless --force is given and the owner approves it.""" from printer_ops.base import PrinterRefused @@ -488,7 +510,8 @@ def _require_preflight_pass(report: Any, name: str, original: Path, local_hash: from printer_ops.force_hatch import require_force_approval forced = ", ".join(c.name for c in report.checks if c.status == FAIL) - approval = require_force_approval(printer=name, file=str(original), sha256=local_hash, checks=report.checks) + approval = require_force_approval(printer=name, file=str(original), sha256=local_hash, checks=report.checks, + remote_name=remote) print(f"FORCED: starting despite preflight FAIL ({forced}); owner approved on Telegram ({approval})") @@ -524,24 +547,77 @@ def _start_and_report(adapter: Adapter, name: str, remote: str) -> None: print(render_status(name, adapter, after)) +def _fix_colour_and_upload(name: str, original: Path, local: Path, source: Path | None, st: PrinterStatus, + adapter: Adapter, opts: dict[str, str]) -> None: + """start --fix-colour: rewrite ORIGINAL (and its project 3MF) to the colours the printer reports, + refresh the snapshot LOCAL from it and upload it: its hash, so its stored name, changed.""" + import shutil + + from printer_ops.base import safe_remote_name + from printer_ops.colour_fix import DEFAULT_FIX_MAX_DELTA_E, fix_to_loaded, parse_delta_e + + limit = parse_delta_e(opts.get("--fix-max-delta-e"), "--fix-max-delta-e", DEFAULT_FIX_MAX_DELTA_E, + command="printer") + fixed = fix_to_loaded(original, st, source, limit, command="printer") + if fixed is None: + print("--fix-colour: the loaded colours already equal the G-code's; nothing rewritten") + return + print("\n".join(fixed.lines())) + shutil.copyfile(original, local) + res = adapter.upload(local, safe_remote_name(local, opts.get("--as"))) + print(f"uploaded the colour-fixed file to {name} as {res.name}: {res.size} B, size verified on the printer") + + def _start_snapshot(name: str, file: str, original: Path, local: Path, source: Path | None, opts: dict[str, str]) -> int: from printer_ops.base import safe_remote_name - remote = safe_remote_name(local, opts.get("--as")) # content-addressed: -.gcode - size = local.stat().st_size + safe_remote_name(local, opts.get("--as")) # a name the printer would refuse fails before anything is written adapter = _adapter(name, opts, write=True) st = adapter.status() _refuse_if_busy(name, st, "before preflight") + if "--fix-colour" in opts: + _fix_colour_and_upload(name, original, local, source, st, adapter, opts) + remote = safe_remote_name(local, opts.get("--as")) # content-addressed: -.gcode + size = local.stat().st_size _require_stored_file(adapter, name, file, remote, size, opts) local_hash = _verify_remote_identity(adapter, name, file, remote, local) report = _preflight_report(name, adapter, st, original, local, source, opts) - _require_preflight_pass(report, name, original, local_hash, opts) + _require_preflight_pass(report, name, original, local_hash, remote, opts) _start_and_report(adapter, name, remote) return 0 +_SLOT_PASS_THROUGH = {"--brand": "--vendor", "--colour": "--color"} # `slot set` spellings -> `printers filament set` + + +def _cmd_slot(argv: list[str]) -> int: + """`slot set NAME T --material M --color #RRGGBB [--brand B]` and `slot show NAME`: the head + filament record of the printer, through `3d printers filament` (ONE implementation of the write: + Adapter.set_head_filament, read back; refusals for non-`full` access, a running job, official RFID).""" + import re + + from printers_cli.filament import run as filament + + if not argv or argv[0] in ("-h", "--help", "help"): + print(USAGE) + return 0 if argv else 1 + action, rest = argv[0], argv[1:] + if action == "show": + return filament(["show", *rest]) + if action != "set": + raise UsageError(f"unknown slot action {action!r}; accepted: set, show", command="printer", + remediation=["3d printer slot set NAME T --material PETG --color '#RRGGBB'"]) + if len(rest) < 2 or rest[0].startswith("-") or rest[1].startswith("-"): + raise UsageError("slot set needs the printer and the head: slot set NAME T --material M --color #RRGGBB", + command="printer", remediation=["3d printer slot set u1 T3 --material PETG --color '#FEE5A5'"]) + name, head, options = rest[0], rest[1], rest[2:] + if not re.fullmatch(r"[Tt][0-9]+", head): + raise UsageError(f"{head!r} is not a head label: T as the printer shows it (U1: T1..T4)", command="printer") + return filament(["set", f"{name}/{head.upper()}", *(_SLOT_PASS_THROUGH.get(o, o) for o in options)]) + + def _cmd_cancel(argv: list[str]) -> int: from printer_ops.base import ACTIVE_JOB_STATES, PrinterRefused from printer_ops.render import render_status @@ -643,6 +719,7 @@ def handler(rest: list[str], name: str | None) -> int: "status": lambda rest, name: _cmd_status(rest), "upload": lambda rest, name: _cmd_upload(rest), "start": lambda rest, name: _cmd_start(rest), + "slot": lambda rest, name: _cmd_slot(rest), "cancel": lambda rest, name: _cmd_cancel(rest), "evidence": _run_evidence, "watch": _run_watch, @@ -678,7 +755,7 @@ def run(argv: list[str]) -> int: handler = _SUBCOMMANDS.get(sub) if handler is None: raise UsageError(f"unknown subcommand '{sub}'", command="printer", - remediation=["Use list, status, upload, start, cancel, evidence, watch, supervise, vision, guard, bed, subscribe, unsubscribe, " + remediation=["Use list, status, upload, start, slot, cancel, evidence, watch, supervise, vision, guard, bed, subscribe, unsubscribe, " "subscriptions, agent, doctor or todo (issues) — see `3d printer --help`."]) return handler(rest, name) @@ -702,7 +779,7 @@ def _printer_name(argv: list[str]) -> tuple[str | None, list[str]]: COMMAND = Command( name="printer", group="SLICING", - summary="drive configured printers: list / status / upload / start --yes / cancel --yes / evidence / watch / vision / subscribe / agent / todo", + summary="drive configured printers: list / status / upload / start --yes / slot set / cancel --yes / evidence / watch / vision / subscribe / agent / todo", usage=USAGE, run=run, ) diff --git a/lib/printer_ops/colour_fix.py b/lib/printer_ops/colour_fix.py new file mode 100644 index 00000000..7bd15a42 --- /dev/null +++ b/lib/printer_ops/colour_fix.py @@ -0,0 +1,438 @@ +"""colour_fix.py — slot colour tolerance and the rewrite that aligns a G-code with the loaded filament. + +WHAT: two things behind the preflight `slot Tn` line. + - `colour_delta(a, b)`: CIEDE2000 between two `#RRGGBB` colours (CIELAB, D65; printops/color.py). + `DEFAULT_SLOT_DELTA_E` (10) is the tolerance of the slot check: two beiges of the same PETG + (#E8D8B0 sliced, #FEE5A5 loaded: ΔE 6.4) are the same filament; white vs beige (16.7) or red vs + blue (52.9) are not. A material mismatch is never tolerated (preflight.py). + - `plan_colour_fix` + `apply_colour_fix` (`3d print-preflight --fix`, `3d printer start + --fix-colour`): for every tool the G-code uses, write the colour the PRINTER reports into the + `; filament_colour` / `; extruder_colour` entries of the G-code and into the 3MF project it was + sliced from (`Metadata/project_settings.config`, `Metadata/plate_N.json filament_colors`), so the + files record what is really loaded. Sidecars follow: `NAME.slice-warnings.json` gets the new + G-code SHA-256 (`slice_warnings.follow_edit`), `NAME.manifest.json` the new slot colour. Old and + new SHA-256 of every rewritten file are returned for the operator to print. + +WHY: on 2026-11-04 a SIMBOL plate sliced with #E8D8B0 met a printer that showed the same beige PETG + as #FEE5A5: the preflight FAILed on string inequality and the only way out was `--force`, whose + Telegram approval then failed. The cause is metadata, and the right fix is at the source, in + seconds, with the old and new hashes on record. + +INVARIANTS: stdlib only (print-preflight imports it in-process). The toolpath bytes are never + touched: only the colour entries of `; filament_colour =` / `; extruder_colour =` lines change; + mode and mtime are kept (a staleness check G-code vs project must not be fooled), and both files + are staged to temp files first, so a 3MF that cannot be rewritten leaves the G-code untouched. A + tool whose loaded colour is further than `max_delta_e` from the sliced one is REFUSED, not + rewritten: that is a different filament, and relabelling it would hide a wrong-colour print. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import tempfile +import zipfile +from dataclasses import dataclass, field +from pathlib import Path +from typing import TYPE_CHECKING + +from printops.color import hex_delta_e, parse_hex + +from printer_ops.base import PrinterRefused + +if TYPE_CHECKING: + from printer_ops.base import PrinterStatus + from printer_ops.preflight import GcodeMeta + +DEFAULT_SLOT_DELTA_E = 10.0 # CIEDE2000; the `colour.warn_delta_e` of fleet/policy.yaml +DEFAULT_FIX_MAX_DELTA_E = ( + 30.0 # above this the loaded filament is another colour: no rewrite +) + +_COLOUR_LINE = re.compile( + rb"^(;[ \t]*(?:filament_colour|extruder_colour)[ \t]*=[ \t]*)(.*?)(\r?\n?)$" +) +_COLOUR_ENTRY = re.compile(r"#?[0-9A-Fa-f]{6}(?:[0-9A-Fa-f]{2})?") +_PROJECT_SETTINGS = "Metadata/project_settings.config" +_PLATE_JSON = re.compile(r"Metadata/plate_\d+\.json") +_CHUNK = 1 << 22 + + +def colour_delta(a: str | None, b: str | None) -> float | None: + """CIEDE2000 between two colours (`#RRGGBB`, RGBA and bare forms accepted); None when either is not one.""" + pa, pb = parse_hex(a), parse_hex(b) + if pa is None or pb is None: + return None + return hex_delta_e(pa, pb) + + +def parse_delta_e( + value: str | None, option: str, default: float, *, command: str +) -> float: + """`--colour-delta-e` / `--fix-max-delta-e` value: a positive number (CIEDE2000), else `default`.""" + from errors import InvalidArgument + + if value is None: + return default + try: + number = float(value) + except ValueError: + number = -1.0 + if not number > 0: + raise InvalidArgument( + option, value, ["a positive number (CIEDE2000), e.g. 10"], command=command + ) + return number + + +@dataclass(frozen=True) +class ColourEdit: + """One tool's sliced colour replaced by the loaded one.""" + + tool: int # G-code tool number (0-based) + label: str # the printer's name of that head (U1: T1..T4) + old: str + new: str + delta_e: float + + +@dataclass(frozen=True) +class FileChange: + path: Path + old_sha256: str + new_sha256: str + + +@dataclass +class FixResult: + edits: list[ColourEdit] + gcode: FileChange + project: FileChange | None = None + manifest: Path | None = None # NAME.manifest.json whose slot colours were updated + sidecar: Path | None = ( + None # NAME.slice-warnings.json re-bound to the new G-code hash + ) + notes: list[str] = field(default_factory=list) + + def lines(self) -> list[str]: + out = [ + f"colour fix: tool T{e.tool} ({e.label}) {e.old} -> {e.new} (ΔE {e.delta_e:.1f})" + for e in self.edits + ] + out.extend( + f" {c.path}\n sha256 old {c.old_sha256}\n sha256 new {c.new_sha256}" + for c in (self.gcode, self.project) + if c is not None + ) + if self.manifest is not None: + out.append(f" {self.manifest}: slot colours updated") + if self.sidecar is not None: + out.append(f" {self.sidecar}: re-bound to the new G-code sha256") + out.extend(f" note: {n}" for n in self.notes) + return out + + +def plan_colour_fix( + meta: GcodeMeta, status: PrinterStatus, max_delta_e: float = DEFAULT_FIX_MAX_DELTA_E +) -> tuple[list[ColourEdit], list[str]]: + """(edits, refusals): the used tools whose sliced colour differs from the loaded one, and the + reasons a tool is not rewritten (empty head, a different colour beyond `max_delta_e`).""" + edits: list[ColourEdit] = [] + refusals: list[str] = [] + for tool in meta.used_tools: + slot = status.slots[tool] if tool < len(status.slots) else None + old = meta.colours[tool] if tool < len(meta.colours) else None + if ( + slot is None + or not slot.present + or not slot.color + or not old + or old == slot.color + ): + continue + de = colour_delta(old, slot.color) + if de is None: + continue + if de > max_delta_e: + refusals.append( + f"tool T{tool} ({slot.label}): G-code {old}, loaded {slot.color}: ΔE {de:.1f} > {max_delta_e:g}, " + "another colour, not the same filament reported differently" + ) + else: + edits.append(ColourEdit(tool, slot.label, old, slot.color, de)) + return edits, refusals + + +def _ident(st: os.stat_result) -> tuple[int, int, int]: + return st.st_ino, st.st_size, st.st_mtime_ns + + +def _entries(value: str, wanted: dict[int, str]) -> str: + """`a;b;c` with entry `t` of `wanted` replaced by its colour (quotes and spacing kept).""" + parts = value.split(";") + for tool, new in wanted.items(): + if tool < len(parts) and _COLOUR_ENTRY.search(parts[tool]): + parts[tool] = _COLOUR_ENTRY.sub(new, parts[tool], count=1) + return ";".join(parts) + + +def _patch_line(line: bytes, wanted: dict[int, str]) -> bytes: + m = _COLOUR_LINE.match(line) + if m is None: + return line + value = m.group(2).decode("utf-8", errors="surrogateescape") + return ( + m.group(1) + + _entries(value, wanted).encode("utf-8", errors="surrogateescape") + + m.group(3) + ) + + +def _stage_gcode( + path: Path, wanted: dict[int, str] +) -> tuple[Path, FileChange, os.stat_result]: + """Copy `path` with the colour lines patched to a temp file beside it; (temp, hashes, source stat).""" + old, new = hashlib.sha256(), hashlib.sha256() + fd, tmp = tempfile.mkstemp(prefix=".colour-fix-", dir=path.resolve().parent) + changed = 0 + try: + with open(path, "rb") as src, os.fdopen(fd, "wb") as dst: + st = os.fstat(src.fileno()) + for line in src: + old.update(line) + out = ( + _patch_line(line, wanted) + if line[:1] == b";" and b"colour" in line + else line + ) + changed += out != line + new.update(out) + dst.write(out) + if not changed: + raise PrinterRefused( + f"{path} has no `; filament_colour` / `; extruder_colour` entry to rewrite for tools " + f"{', '.join(f'T{t}' for t in sorted(wanted))}", + command="print-preflight", + remediation=[ + "The G-code carries no slicer config block: re-slice it, or load the colour that was sliced." + ], + ) + except BaseException: + Path(tmp).unlink(missing_ok=True) + raise + return Path(tmp), FileChange(path, old.hexdigest(), new.hexdigest()), st + + +def _indent_of(text: str) -> int: + for line in text.splitlines()[1:3]: + stripped = line.lstrip(" ") + if stripped and stripped != line: + return len(line) - len(stripped) + return 4 + + +def _patch_json( + raw: bytes, keys: tuple[str, ...], wanted: dict[int, str] +) -> bytes | None: + """JSON with entry `t` of each list in `keys` set to its colour; None when nothing changes.""" + text = raw.decode("utf-8") + data = json.loads(text) + if not isinstance(data, dict): + return None + changed = False + for key in keys: + arr = data.get(key) + if not isinstance(arr, list): + continue + for tool, new in wanted.items(): + if ( + tool < len(arr) + and isinstance(arr[tool], str) + and arr[tool].upper() != new.upper() + ): + arr[tool] = new + changed = True + if not changed: + return None + out = json.dumps(data, indent=_indent_of(text), ensure_ascii=False) + return (out + "\n" if text.endswith("\n") else out).encode("utf-8") + + +class _NotAProject(ValueError): + """The 3MF is not a slicer project: there is no colour list in it to rewrite.""" + + +def _sha256_and_stat(path: Path) -> tuple[str, os.stat_result]: + digest = hashlib.sha256() + with open(path, "rb") as fh: + st = os.fstat(fh.fileno()) + while chunk := fh.read(_CHUNK): + digest.update(chunk) + return digest.hexdigest(), st + + +def _patched_members(zin: zipfile.ZipFile, wanted: dict[int, str]) -> dict[str, bytes]: + """The members of the project whose colour lists change, as their new bytes.""" + patched: dict[str, bytes] = {} + for info in zin.infolist(): + if info.filename == _PROJECT_SETTINGS: + keys: tuple[str, ...] = ("filament_colour", "extruder_colour") + elif _PLATE_JSON.fullmatch(info.filename): + keys = ("filament_colors",) + else: + continue + if (new := _patch_json(zin.read(info), keys, wanted)) is not None: + patched[info.filename] = new + return patched + + +def _write_patched(zin: zipfile.ZipFile, patched: dict[str, bytes], tmp: Path) -> None: + """Copy every member of `zin` to the zip `tmp`, the `patched` ones with their new bytes.""" + with tmp.open("wb") as raw, zipfile.ZipFile(raw, "w") as zout: + for info in zin.infolist(): + zout.writestr( + info, + patched.get(info.filename) or zin.read(info), + compress_type=info.compress_type, + ) + + +def _stage_project( + path: Path, wanted: dict[int, str] +) -> tuple[Path, FileChange, os.stat_result] | None: + """The 3MF with its colour lists patched, as a temp file beside it; None when it already agrees.""" + old_sha, st = _sha256_and_stat(path) + try: + zin = zipfile.ZipFile(path) + except zipfile.BadZipFile as exc: + raise _NotAProject(f"{path} is not a 3MF (zip): {exc}") from exc + with zin: + if _PROJECT_SETTINGS not in zin.namelist(): + raise _NotAProject( + f"{path} has no {_PROJECT_SETTINGS}: not a slicer project 3MF" + ) + patched = _patched_members(zin, wanted) + if _PROJECT_SETTINGS not in patched: + return None + fd, name = tempfile.mkstemp(prefix=".colour-fix-", dir=path.resolve().parent) + os.close(fd) + tmp = Path(name) + try: + _write_patched(zin, patched, tmp) + except BaseException: + tmp.unlink(missing_ok=True) + raise + new_sha = hashlib.sha256(tmp.read_bytes()).hexdigest() + return tmp, FileChange(path, old_sha, new_sha), st + + +def _install(tmp: Path, dst: Path, st: os.stat_result) -> None: + """Replace `dst` by the staged `tmp`, refusing when `dst` changed meanwhile; mode and mtime kept.""" + os.chmod(tmp, st.st_mode & 0o7777) + os.utime(tmp, ns=(st.st_atime_ns, st.st_mtime_ns)) + if _ident(os.stat(dst)) != _ident(st): + raise PrinterRefused( + f"{dst} changed while its colours were being rewritten; run again", + command="print-preflight", + ) + os.replace(tmp, dst) + + +def _update_manifest(project: Path, wanted: dict[int, str]) -> Path | None: + """NAME.manifest.json beside NAME.3mf: the slot colours (`filaments[].hex`, tool = slot - 1).""" + path = project.with_name(project.stem + ".manifest.json") + if not path.is_file(): + return None + text = path.read_text(encoding="utf-8") + data = json.loads(text) + changed = False + for fil in data.get("filaments", []) if isinstance(data, dict) else []: + tool = fil.get("tool") if isinstance(fil, dict) else None + new = wanted.get(tool) if isinstance(tool, int) else None + if new is not None and str(fil.get("hex", "")).upper() != new.upper(): + fil["hex"] = new + changed = True + if not changed: + return None + path.write_text( + json.dumps(data, ensure_ascii=False, indent=_indent_of(text)), encoding="utf-8" + ) + return path + + +def apply_colour_fix( + gcode: Path, project: Path | None, edits: list[ColourEdit] +) -> FixResult: + """Rewrite the colours of `edits` in `gcode` (and `project`, a 3MF) in place; see the module doc.""" + wanted = {e.tool: e.new for e in edits} + staged_g, change_g, st_g = _stage_gcode(gcode, wanted) + staged_p = None + result = FixResult(edits, change_g) + try: + staged_p = _stage_project(project, wanted) if project is not None else None + except _NotAProject as exc: + project = None + result.notes.append(f"{exc}; left untouched") + except BaseException: + staged_g.unlink(missing_ok=True) + raise + try: + _install(staged_g, gcode, st_g) + if staged_p is not None and project is not None: + _install(staged_p[0], project, staged_p[2]) + result.project = staged_p[1] + finally: + staged_g.unlink(missing_ok=True) + if staged_p is not None: + staged_p[0].unlink(missing_ok=True) + from slicing.slice_warnings import follow_edit + + result.sidecar = follow_edit( + gcode, gcode, change_g.old_sha256, change_g.new_sha256, "colour fix" + ) + if project is not None: + result.manifest = _update_manifest(project, wanted) + if staged_p is None: + result.notes.append(f"{project} already carries the loaded colours") + elif not result.notes: + result.notes.append( + "no project 3MF found (--source-3mf / NAME.3mf): only the G-code was rewritten" + ) + return result + + +def fix_to_loaded( + gcode: Path, + status: PrinterStatus, + source_3mf: Path | None, + max_delta_e: float = DEFAULT_FIX_MAX_DELTA_E, + *, + command: str, +) -> FixResult | None: + """Rewrite `gcode` (+ its project 3MF: `source_3mf`, else NAME.3mf beside it) to the colours the + printer reports for the tools it uses. None when nothing differs; PrinterRefused (nothing written) + when the printer reports no per-head data or a tool's loaded colour is another colour.""" + from printer_ops.preflight import find_project_3mf, read_gcode_meta + + if not status.slots: + raise PrinterRefused( + "the printer reports no per-head filament data: there is no loaded colour to rewrite the file to", + command=command, + ) + edits, refusals = plan_colour_fix(read_gcode_meta(gcode), status, max_delta_e) + if refusals: + raise PrinterRefused( + "colour fix refused, nothing written: " + refusals[0], + command=command, + remediation=[ + *refusals[1:], + "Load the filament that was sliced (`3d printer slot set` fixes the printer's record), or, if " + "this really is the same filament, raise the limit: --fix-max-delta-e N.", + ], + ) + if not edits: + return None + project, _ = find_project_3mf(gcode, source_3mf) + return apply_colour_fix(gcode, project, edits) diff --git a/lib/printer_ops/force_hatch.py b/lib/printer_ops/force_hatch.py index a1217fba..d18980cd 100644 --- a/lib/printer_ops/force_hatch.py +++ b/lib/printer_ops/force_hatch.py @@ -17,6 +17,14 @@ (fix the gate). Each approved or refused force is therefore an issue to analyse and a root cause to fix, not a habit — the refusal message says so. +RETRY (issue https://github.com/alex-mextner/3d-cli/issues/728): the tg-ctl daemon answers a card it + could not send with `ask-forward not-posted: … daemon declined: reason=send-failed` (tg-cli + `forwardAndPostQuestion`: ONE `sendMessage`, no retry; on 2026-11-04 22:02 and 22:03 it was + `getaddrinfo EAI_AGAIN api.telegram.org`, a resolver blip, while a plain `tg` a moment later + worked). Nothing was posted, so asking again is safe: the ask is repeated after each pause of + RETRY_DELAYS_S (3, 10, 30 s). When it still fails the refusal says how to start from the + printer's own screen and where the daemon logs the cause. + INVARIANTS: the helper is loaded from a FIXED path under the account's real home (never an env var, never the repo or $PATH), and only when its file is owned by you or root and not group/world writable; a helper that is missing or untrusted refuses the start (fail closed). @@ -31,6 +39,7 @@ import shlex import stat import sys +import time from collections.abc import Sequence from pathlib import Path from typing import Protocol, cast @@ -45,7 +54,15 @@ # agent-tools checkouts searched, relative to the real home; the first one that has the helper wins. _CHECKOUTS = (Path("dev") / "agent-tools", Path("xp") / "agent-tools") _DETAIL_CHARS = 240 -_CARD_VALUE_CHARS = 1000 # the shared hatch helper cuts each context value at this length +_CARD_VALUE_CHARS = ( + 1000 # the shared hatch helper cuts each context value at this length +) +# The daemon's own wording for a card it could not post (agent-tools `_ask_forward_verdict`, tg-cli +# question-cards.ts `dropAsk(socket, "send-failed")`). +_SEND_FAILED = "reason=send-failed" +# Pauses before each further ask while the card is not posted: a transient resolver/network failure +# of the daemon clears in seconds, so the owner still gets the card and the start does not stall. +RETRY_DELAYS_S = (3.0, 10.0, 30.0) class _Verdict(Protocol): @@ -129,7 +146,9 @@ def failing_summary(checks: Sequence[Check], budget: int | None = None) -> str: return "\n".join(lines) -def _refusal(failing: str, why: str, command: str) -> PrinterRefused: +def _refusal( + failing: str, why: str, command: str, extra: Sequence[str] = () +) -> PrinterRefused: return PrinterRefused( f"--force does not override preflight FAIL by itself: {why}", command="printer", @@ -139,13 +158,59 @@ def _refusal(failing: str, why: str, command: str) -> PrinterRefused: f"Only if forcing is truly needed, ask the owner for a one-time Telegram approval: " f'{ENV_VAR}="" {command}', "Every forced case is analysed afterwards and its root cause fixed.", + *extra, "Failing checks:\n" + failing, ], ) +def _send_failed(verdict: _Verdict) -> bool: + """The daemon declined to post the card (`ask-forward not-posted … reason=send-failed`): its + Telegram send failed (a DNS or network blip, 2026-11-04 `getaddrinfo EAI_AGAIN api.telegram.org`). + No card exists then, so asking again cannot double-approve.""" + return not verdict.approved and _SEND_FAILED in verdict.reason + + +def _ask_with_retry( + hatch: _HatchLib, context: dict[str, object], command: str +) -> _Verdict: + """`request_hatch_approval`, asked again after each `RETRY_DELAYS_S` pause while the card could not + be posted. Any other verdict (approved, denied, timed out, not requested) is final at once.""" + verdict = hatch.request_hatch_approval( + HOOK_ID, context, cwd=os.getcwd(), command=command, critical=True + ) + for delay in RETRY_DELAYS_S: + if not _send_failed(verdict): + break + print( + f"hatch: the approval card was not posted to Telegram ({verdict.reason}); asking again in {delay:g} s", + file=sys.stderr, + ) + time.sleep(delay) + verdict = hatch.request_hatch_approval( + HOOK_ID, context, cwd=os.getcwd(), command=command, critical=True + ) + return verdict + + +def _screen_hint(remote_name: str | None) -> list[str]: + """How to start without Telegram: from the printer's own screen.""" + what = f"`{remote_name}`" if remote_name else "the file" + return [ + f"Start from the printer screen instead: {what} is already stored on the printer (Files, then Print). " + "The screen does NOT run these checks: only do it when the FAIL lines below are understood.", + "Why Telegram failed: tg-ctl's daemon could not send the card (plain `tg` can still work); see " + "`~/.config/tg-cli/tg-ctl.*.log`, lines `sendMessage failed` / `ask-forward dropped: reason=send-failed`.", + ] + + def require_force_approval( - *, printer: str, file: str, sha256: str, checks: Sequence[Check] + *, + printer: str, + file: str, + sha256: str, + checks: Sequence[Check], + remote_name: str | None = None, ) -> str: """Return the approval reason when the owner approved the forced start, else raise PrinterRefused. @@ -169,11 +234,17 @@ def require_force_approval( "sha256": sha256, "failing_checks": failing_summary(checks, _CARD_VALUE_CHARS - 50), } - verdict = hatch.request_hatch_approval( - HOOK_ID, context, cwd=os.getcwd(), command=command, critical=True - ) + verdict = _ask_with_retry(hatch, context, command) if verdict.approved: return verdict.reason + if _send_failed(verdict): + raise _refusal( + failing, + f"the approval request could not be posted to Telegram after {1 + len(RETRY_DELAYS_S)} attempts " + f"({verdict.reason})", + command, + extra=_screen_hint(remote_name), + ) if verdict.should_stop: raise _refusal( failing, diff --git a/lib/printer_ops/preflight.py b/lib/printer_ops/preflight.py index 94556737..18363f15 100644 --- a/lib/printer_ops/preflight.py +++ b/lib/printer_ops/preflight.py @@ -6,8 +6,11 @@ - printer the target printer is free (not printing/paused/busy/unknown); - bed (printers with `watch.bed.enabled`) the camera saw the bed clear with the plate on (printer_ops/bed.py, issue #303): objects or no plate FAIL, unknown WARN; - - slot Tn for every tool the G-code uses: the printer's head is loaded, and its colour - and material equal the G-code's `; filament_colour` / `; filament_type` + - slot Tn for every tool the G-code uses: the printer's head is loaded, its material + equals the G-code's `; filament_type` and its colour is within a CIEDE2000 + tolerance of `; filament_colour` (default 10, the distance is shown; a beige + reported as #FEE5A5 for a #E8D8B0 slice passes) — printer_ops/colour_fix.py, + which `print-preflight --fix` uses to rewrite the file to the loaded colour (needs per-head data: Moonraker printers exposing print_task_config, i.e. U1); - halves `3d halves-check <3mf> --json` when the source 3MF is given; - axis travel every move (arcs by their real extremes) inside the printer's X/Y limits, @@ -58,6 +61,7 @@ from typing import TYPE_CHECKING, Any from printer_ops.base import PrinterStatus, Slot +from printer_ops.colour_fix import DEFAULT_SLOT_DELTA_E, colour_delta from printer_ops.move_bounds import ( BoundsReport, Limits, @@ -151,20 +155,25 @@ def _hex(value: str | None) -> str | None: ) -def _scan_gcode_line(line: str, config: dict[str, str], tool_lines: set[int]) -> None: - """Fold one G-code line into the slicer config comments / the `Tn` tool selections seen.""" - if line.startswith(";"): - m = _CONFIG_LINE.match(line.rstrip("\n")) - if m: - config.setdefault(m.group(1).strip(), m.group(2).strip()) - elif line[:1] == "T": - t = _TOOL_LINE.match(line.strip()) - if t: - tool_lines.add(int(t.group(1))) - - -def _filament_amount(value: str) -> float: - """One `filament used [mm]` entry; an unparsable amount counts as unused, keep reading.""" +def _scan_gcode_config(path: Path) -> tuple[dict[str, str], set[int]]: + """(first value of every `; key = value` slicer comment, tool numbers of the `Tn` lines).""" + config: dict[str, str] = {} + tool_lines: set[int] = set() + with open(path, encoding="utf-8", errors="replace") as fh: + for line in fh: + if line.startswith(";"): + m = _CONFIG_LINE.match(line.rstrip("\n")) + if m: + config.setdefault(m.group(1).strip(), m.group(2).strip()) + elif line[:1] == "T": + t = _TOOL_LINE.match(line.strip()) + if t: + tool_lines.add(int(t.group(1))) + return config, tool_lines + + +def _amount(value: str) -> float: + """One entry of `; filament used [mm]`; an unparsable amount counts as unused.""" try: return float(value) except ValueError: @@ -172,23 +181,18 @@ def _filament_amount(value: str) -> float: def _used_tools(config: dict[str, str], tool_lines: set[int]) -> list[int]: - """Tools that print: those with filament used, else those selected by `Tn`, else tool 0.""" + """Non-zero entries of `; filament used [mm]`, else the `Tn` lines, else tool 0.""" used_mm = config.get("filament used [mm]") if used_mm: - amounts = [_filament_amount(v) for v in used_mm.split(",")] - return [i for i, a in enumerate(amounts) if a > 0] - if tool_lines: - return sorted(tool_lines) - return [0] + return [ + i for i, a in enumerate(_amount(v) for v in used_mm.split(",")) if a > 0 + ] + return sorted(tool_lines) or [0] def read_gcode_meta(path: Path) -> GcodeMeta: """Filament colours/types and the tools actually used, from slicer comments + Tn lines.""" - config: dict[str, str] = {} - tool_lines: set[int] = set() - with open(path, encoding="utf-8", errors="replace") as fh: - for line in fh: - _scan_gcode_line(line, config, tool_lines) + config, tool_lines = _scan_gcode_config(path) return GcodeMeta( colours=[_hex(c) for c in _split(config.get("filament_colour", ""))], types=[(t or None) for t in _split(config.get("filament_type", ""))], @@ -224,55 +228,74 @@ def check_printer_free(status: PrinterStatus) -> Check: return Check("printer", PASS, f"{status.state} [{status.raw_state}]") -def _slot_problems(slot: Slot, want_c: str | None, want_m: str | None) -> list[str]: - """How the loaded head differs from the colour/material the G-code expects.""" - problems = [] +def _slot_problems( + want_c: str | None, want_m: str | None, slot: Slot, colour_delta_e: float +) -> tuple[list[str], str]: + """(problems, colour note): the mismatches of one loaded slot against the G-code's colour and + material, and the ΔE note of a colour that differs but is within tolerance.""" + problems: list[str] = [] + note = "" if want_c and slot.color and want_c != slot.color: - problems.append(f"colour: G-code {want_c}, loaded {slot.color}") + de = colour_delta(want_c, slot.color) + if de is None or de > colour_delta_e: + shown = "" if de is None else f" (ΔE {de:.1f} > {colour_delta_e:g})" + problems.append( + f"colour: G-code {want_c}, loaded {slot.color}{shown}; " + "fix: print-preflight --fix / printer start --fix-colour" + ) + else: + note = ( + f" (colour ΔE {de:.1f} from G-code {want_c}, within {colour_delta_e:g})" + ) if want_m and slot.material and want_m.upper() != slot.material.upper(): problems.append(f"material: G-code {want_m}, loaded {slot.material}") - return problems + return problems, note -def _check_slot(meta: GcodeMeta, status: PrinterStatus, t: int) -> Check: - """The `slot Tn` line: is the head the G-code uses present and loaded as expected.""" - name = f"slot T{t}" - if t >= len(status.slots): +def _slot_check( + tool: int, meta: GcodeMeta, status: PrinterStatus, colour_delta_e: float +) -> Check: + name = f"slot T{tool}" + if tool >= len(status.slots): return Check( name, FAIL, - f"G-code uses tool {t} but the printer has {len(status.slots)} heads", + f"G-code uses tool {tool} but the printer has {len(status.slots)} heads", ) - slot = status.slots[t] + slot = status.slots[tool] if not slot.present: return Check( name, FAIL, f"head {slot.label} is used by the G-code but has no filament / is disabled", ) - want_c = meta.colours[t] if t < len(meta.colours) else None - want_m = meta.types[t] if t < len(meta.types) else None - problems = _slot_problems(slot, want_c, want_m) + want_c = meta.colours[tool] if tool < len(meta.colours) else None + want_m = meta.types[tool] if tool < len(meta.types) else None + problems, note = _slot_problems(want_c, want_m, slot, colour_delta_e) if problems: return Check(name, FAIL, f"head {slot.label}: " + "; ".join(problems)) if not want_c: return Check( name, WARN, - f"head {slot.label}: G-code has no filament_colour for tool {t}; " + f"head {slot.label}: G-code has no filament_colour for tool {tool}; " f"loaded {slot.color} {slot.material}", ) return Check( name, PASS, - f"head {slot.label}: {slot.color} {slot.material or ''}".rstrip(), + f"head {slot.label}: {slot.color} {slot.material or ''}".rstrip() + note, ) -def check_slots(meta: GcodeMeta, status: PrinterStatus) -> list[Check]: +def check_slots( + meta: GcodeMeta, status: PrinterStatus, colour_delta_e: float = DEFAULT_SLOT_DELTA_E +) -> list[Check]: + """One line per used tool. The colour matches within `colour_delta_e` (CIEDE2000, the distance is + in the line); a material mismatch is always a FAIL.""" if not status.slots: return [Check("slots", SKIP, "the printer reports no per-head filament data")] - return [_check_slot(meta, status, t) for t in meta.used_tools] + return [_slot_check(t, meta, status, colour_delta_e) for t in meta.used_tools] def run_halves_check(source: Path) -> tuple[int, str, str]: @@ -371,8 +394,27 @@ def find_project_3mf(gcode: Path, source: Path | None) -> tuple[Path | None, str return None, "" +def check_head_sequence( + gcode: Path, project: Path | None, project_source: str +) -> Check: + """Issue #103: every object prints each layer on the heads its filaments call for.""" + if project is None: + return Check( + "head seq", + SKIP, + "no project 3MF: pass --source-3mf FILE.3mf or keep NAME.3mf next to NAME.gcode", + ) + from slicing.head_sequence import ProjectError, preflight_check + + try: + result = preflight_check(str(gcode), str(project)) + except ProjectError as exc: + return Check("head seq", SKIP, f"{project_source}: {exc}") + return _head_sequence_verdict(result, project_source) + + def _head_sequence_verdict(result: dict[str, Any], project_source: str) -> Check: - """The `head seq` line for a successful `preflight_check` result.""" + """The `head seq` line of one `slicing.head_sequence.preflight_check` result.""" objects = result.get("objects") objs = ( [o for o in objects if isinstance(o, dict)] if isinstance(objects, list) else [] @@ -405,25 +447,6 @@ def _head_sequence_verdict(result: dict[str, Any], project_source: str) -> Check ) -def check_head_sequence( - gcode: Path, project: Path | None, project_source: str -) -> Check: - """Issue #103: every object prints each layer on the heads its filaments call for.""" - if project is None: - return Check( - "head seq", - SKIP, - "no project 3MF: pass --source-3mf FILE.3mf or keep NAME.3mf next to NAME.gcode", - ) - from slicing.head_sequence import ProjectError, preflight_check - - try: - result = preflight_check(str(gcode), str(project)) - except ProjectError as exc: - return Check("head seq", SKIP, f"{project_source}: {exc}") - return _head_sequence_verdict(result, project_source) - - def check_mix_faces(project: Path | None, project_source: str, gcode: Path) -> Check: """Issues #411 / #419: a part in a mixed filament shows its colour on walls, not on a flat top. By the project 3MF when there is one (exact: it knows which parts are mixed, FAIL); else by the @@ -549,7 +572,6 @@ def fmt(names: list[str]) -> str: def _axis_travel_check( rep: BoundsReport, axis_limits: Limits | None, limits_source: str ) -> Check: - """The `axis travel` line: every move inside the printer's axis limits.""" if axis_limits is None: why = f"{limits_source}; " if limits_source else "" return Check( @@ -574,9 +596,8 @@ def _axis_travel_check( def _print_area_violations(rep: BoundsReport) -> list[str]: - """Plate moves outside printable_area or inside bed_exclude_area, one phrase each. - - bed_exclude_area is evaluated on its own: a missing printable_area must not hide it.""" + """printable_area / bed_exclude_area breaches; each is evaluated on its own, so a missing + printable_area must not hide an exclusion.""" parts = [] if rep.area_out_count and rep.printable_area: parts.append( @@ -592,7 +613,6 @@ def _print_area_violations(rep: BoundsReport) -> list[str]: def _print_area_check(rep: BoundsReport) -> Check: - """The `print area` line: the plate section stays inside the printable area.""" if not rep.plate_found: return Check( "print area", @@ -609,9 +629,7 @@ def _print_area_check(rep: BoundsReport) -> Check: ) if rep.printable_area is None: return Check( - "print area", - WARN, - "UNVERIFIED: the G-code has no `; printable_area`", + "print area", WARN, "UNVERIFIED: the G-code has no `; printable_area`" ) return Check( "print area", @@ -621,20 +639,6 @@ def _print_area_check(rep: BoundsReport) -> Check: ) -def _unverified_check(rep: BoundsReport) -> Check | None: - """The `unverified` WARN for moves the bounds check could only partly judge, if any.""" - if not rep.unverified_count: - return None - more = ", ..." if rep.unverified_count > len(rep.unverified) else "" - return Check( - "unverified", - WARN, - f"{rep.unverified_count} move(s) only partly checked: " - + ", ".join(rep.unverified) - + more, - ) - - def check_moves( rep: BoundsReport, axis_limits: Limits | None, @@ -645,9 +649,17 @@ def check_moves( _axis_travel_check(rep, axis_limits, limits_source), _print_area_check(rep), ] - unverified = _unverified_check(rep) - if unverified is not None: - out.append(unverified) + if rep.unverified_count: + more = ", ..." if rep.unverified_count > len(rep.unverified) else "" + out.append( + Check( + "unverified", + WARN, + f"{rep.unverified_count} move(s) only partly checked: " + + ", ".join(rep.unverified) + + more, + ) + ) out.extend(check_macros(rep, trusted_macros)) return out @@ -668,7 +680,7 @@ def bounds_report( def _bed_when(v: dict[str, Any], view: dict[str, Any]) -> str: - """Where the verdict came from, how old it is, and the Z it was judged at if the bed was moved.""" + """Where the verdict came from, and how the bed got to the Z it was judged at.""" age = v.get("age_s") when = ( f"{v.get('source', '?')}, {age:.0f} s ago" @@ -683,10 +695,9 @@ def _bed_when(v: dict[str, Any], view: dict[str, Any]) -> str: return when -def _bed_judgement( - bed: Any, plate: Any, when: str, why: str, frame: list[str] +def _bed_verdict_line( + bed: object, plate: object, when: str, why: str, frame: list[str] ) -> Check: - """The bed/plate state pair -> the `bed` line.""" if plate == "missing": return Check("bed", FAIL, f"build plate missing ({when}): {why}", frame) if bed == "occupied": @@ -713,7 +724,7 @@ def check_bed_verdict(v: dict[str, Any]) -> Check: ) if plate != "missing" and bed != "occupied": return Check("bed", WARN, f"bed state {bed} ({when}): {why}", frame) - return _bed_judgement(bed, plate, when, why, frame) + return _bed_verdict_line(bed, plate, when, why, frame) def check_bed( @@ -779,8 +790,10 @@ def run_preflight( bridge_source: str = "default", bed: Check | None = None, printops: PrintopsContext | None = None, + colour_delta_e: float = DEFAULT_SLOT_DELTA_E, ) -> Report: - """`limits_source` names where `axis_limits` came from, or why they are unknown. + """`colour_delta_e` is the CIEDE2000 tolerance of the `slot Tn` colour (default 10; the line + shows the distance). `limits_source` names where `axis_limits` came from, or why they are unknown. `origin` is where the G-code lives when `gcode` is a snapshot copy (`printer start`): the sibling NAME.3mf and NAME.slice-warnings.json are looked up there. `bed` is the camera bed line (`check_bed` above), computed by the caller that knows the device.""" @@ -788,7 +801,9 @@ def run_preflight( report.checks.append(check_thumbnails(gcode, origin)) if status is not None: report.checks.append(check_printer_free(status)) - report.checks.extend(check_slots(read_gcode_meta(gcode), status)) + report.checks.extend( + check_slots(read_gcode_meta(gcode), status, colour_delta_e) + ) if bed is not None: report.checks.append(bed) report.checks.extend( diff --git a/tests/colour_fix_fixtures.py b/tests/colour_fix_fixtures.py new file mode 100644 index 00000000..895c9e39 --- /dev/null +++ b/tests/colour_fix_fixtures.py @@ -0,0 +1,93 @@ +"""Fixtures for the slot colour tolerance and `--fix` tests: a G-code, its project 3MF, manifest and +slice-warnings sidecar for the SIMBOL case (beige PETG sliced as #E8D8B0, the U1 reporting #FEE5A5), +and a fake U1 that holds that PETG in head T4 (G-code tool 3). Never a real printer.""" + +from __future__ import annotations + +import json +import zipfile +from pathlib import Path + +import numpy as np +from printer_fakes import FakeMoonraker + +import gcode_thumbnails as gt + +SLICED = "#E8D8B0" # what the slicer wrote +LOADED = "#FEE5A5" # what the U1 screen shows for the same beige PETG +COLOURS = ["#000000", "#808080", "#1B5E20", SLICED] +# the three preview sizes a G-code must carry (print-preflight's `thumbnails` gate); drawn once, flat grey +THUMBNAILS = "".join( + gt.block( + gt.Thumb(side, side, "PNG"), + gt.encode_png(np.full((side, side, 4), 128, np.uint8)), + ) + for side in (48, 96, 300) +) +TOOLPATH = "G28\nT3\nG1 X10 Y10 E1\nG1 X20 Y20 E1\n" + + +def gcode_text(colour: str = SLICED, material: str = "PETG") -> str: + colours = ";".join([*COLOURS[:3], colour]) + return ( + "; HEADER_BLOCK_START\n; HEADER_BLOCK_END\n" + + THUMBNAILS + + TOOLPATH + + "; filament used [mm] = 0.00, 0.00, 0.00, 30.50\n" + + f"; extruder_colour = {colours}\n" + + f"; filament_colour = {colours}\n" + + f"; filament_type = PLA;PLA;PLA;{material}\n" + ) + + +def write_project(path: Path, colour: str = SLICED) -> None: + """A slicer project 3MF: project_settings.config (4-space JSON), plate json, one model part.""" + colours = [*COLOURS[:3], colour] + settings = { + "filament_colour": colours, + "extruder_colour": list(colours), + "filament_type": ["PLA", "PLA", "PLA", "PETG"], + "layer_height": "0.2", + } + plate = {"filament_colors": list(colours), "filament_ids": [0, 1, 2, 3]} + with zipfile.ZipFile(path, "w", zipfile.ZIP_DEFLATED) as z: + z.writestr("[Content_Types].xml", "") + z.writestr("3D/3dmodel.model", "kept byte for byte") + z.writestr( + "Metadata/project_settings.config", json.dumps(settings, indent=4) + "\n" + ) + z.writestr("Metadata/plate_1.json", json.dumps(plate, indent=4)) + + +def write_manifest(path: Path, colour: str = SLICED) -> None: + filaments = [ + {"slot": i + 1, "key": f"f{i}", "name": f"F{i}", "hex": c, "tool": i} + for i, c in enumerate([*COLOURS[:3], colour]) + ] + path.write_text( + json.dumps({"file": "plate.3mf", "filaments": filaments}, indent=1), + encoding="utf-8", + ) + + +def write_plate(directory: Path, name: str = "plate", material: str = "PETG") -> Path: + """NAME.gcode + NAME.3mf + NAME.manifest.json + NAME.slice-warnings.json; the G-code's path. + A `material` other than the loaded PETG makes the `slot` gate FAIL (the way a start needs --force).""" + from slicing import slice_warnings as sw + + gcode = directory / f"{name}.gcode" + gcode.write_text(gcode_text(material=material), encoding="utf-8") + write_project(directory / f"{name}.3mf") + write_manifest(directory / f"{name}.manifest.json") + sw.record(gcode, logs=[("slice.log", "")]) + return gcode + + +def load_beige_petg( + moon: FakeMoonraker, head: int = 3, colour: str = LOADED, material: str = "PETG" +) -> None: + """The fake U1 holds `material` of `colour` in head index `head` (label T).""" + cfg = moon.status["print_task_config"] + cfg["filament_exist"][head] = True + cfg["filament_color_rgba"][head] = colour.lstrip("#") + "FF" + cfg["filament_type"][head] = material diff --git a/tests/e2e/hatch_runner.py b/tests/e2e/hatch_runner.py index 3ef6b8fe..a02f51c6 100644 --- a/tests/e2e/hatch_runner.py +++ b/tests/e2e/hatch_runner.py @@ -3,8 +3,10 @@ An e2e test cannot tap Approve on a real phone, and the shared hatch helper refuses every tg-ctl outside the test sandbox. This runner is `python hatch_runner.py <3d args>`: it swaps `printer_ops.force_hatch.load_hatch` for a fake whose verdict comes from THREED_TEST_HATCH_VERDICT -("approve" | "deny") and appends each request (hook id, context, command) as one JSON line to the +("approve" | "deny" | "send-failed") and appends each request (hook id, context, command) as one JSON line to the file in THREED_TEST_HATCH_LOG, then runs the real dispatcher. Everything else is the real CLI. +Verdict "send-failed" answers every request as the tg-ctl daemon does when its Telegram send failed +(`ask-forward not-posted … reason=send-failed`); the retry pauses are zeroed so the test does not wait. """ from __future__ import annotations @@ -52,8 +54,11 @@ def request_hatch_approval( ) + "\n" ) - if os.environ["THREED_TEST_HATCH_VERDICT"] == "approve": + verdict = os.environ["THREED_TEST_HATCH_VERDICT"] + if verdict == "approve": return _Verdict(True, True, "test double approved") + if verdict == "send-failed": + return _Verdict(False, True, "tg-ctl ask: ask-forward not-posted: daemon declined: reason=send-failed") return _Verdict(False, True, "test double denied") @@ -62,5 +67,5 @@ def _fake_hatch() -> _FakeHatch: if __name__ == "__main__": - with patch.object(force_hatch, "load_hatch", _fake_hatch): + with patch.object(force_hatch, "load_hatch", _fake_hatch), patch.object(force_hatch, "RETRY_DELAYS_S", (0.0,) * 3): sys.exit(main(sys.argv[1:])) diff --git a/tests/e2e/test_print_preflight_colour_workflows.py b/tests/e2e/test_print_preflight_colour_workflows.py new file mode 100644 index 00000000..b2c9285b --- /dev/null +++ b/tests/e2e/test_print_preflight_colour_workflows.py @@ -0,0 +1,162 @@ +"""E2E stories through `bin/3d print-preflight` for the 2026-11-04 SIMBOL print: beige PETG sliced as +#E8D8B0 on a U1 that shows it as #FEE5A5. The slot colour is judged by CIEDE2000 distance (no `--force` +for a same-family beige), and `--fix` rewrites the files to the printer's colour. Loopback fakes only.""" + +from __future__ import annotations + +import json +import zipfile +from pathlib import Path + +from colour_fix_fixtures import ( + LOADED, + SLICED, + THUMBNAILS, + gcode_text, + load_beige_petg, + write_plate, +) +from printer_fakes import FakeMoonraker, write_devices +from slicing import slice_warnings as sw + +from .workflow_helper import run_cli + + +def slot_line(stdout: str) -> str: + return next(line for line in stdout.splitlines() if "slot T3" in line) + + +def test_preflight_accepts_the_same_beige_and_prints_the_distance( + tmp_path: Path, +) -> None: + """The user preflights the SIMBOL plate: the slot line passes and says how far the colours are.""" + moon = FakeMoonraker() + load_beige_petg(moon) + g = write_plate(tmp_path) + with moon.serve(): + write_devices(tmp_path / "xdg-config", moon=moon) + default = run_cli(tmp_path, "print-preflight", str(g), "--printer", "u1") + strict = run_cli( + tmp_path, + "print-preflight", + str(g), + "--printer", + "u1", + "--colour-delta-e", + "5", + ) + + assert slot_line(default.stdout).split()[0] == "PASS" and "ΔE 6.4" in slot_line( + default.stdout + ) + assert slot_line(strict.stdout).split()[ + 0 + ] == "FAIL" and "(ΔE 6.4 > 5)" in slot_line(strict.stdout) + assert moon.writes() == [] + + +def test_preflight_still_fails_a_wrong_material_and_a_different_colour( + tmp_path: Path, +) -> None: + """A PLA file on a PETG head and a blue file on a beige head still FAIL; `--fix` refuses to rewrite across + another colour and leaves the file as it was.""" + moon = FakeMoonraker() + load_beige_petg(moon) + pla = tmp_path / "pla.gcode" + pla.write_text(gcode_text(SLICED, "PLA"), encoding="utf-8") + blue = tmp_path / "blue.gcode" + blue.write_text(gcode_text("#0000FF"), encoding="utf-8") + with moon.serve(): + write_devices(tmp_path / "xdg-config", moon=moon) + wrong_material = run_cli( + tmp_path, "print-preflight", str(pla), "--printer", "u1" + ) + wrong_colour = run_cli( + tmp_path, "print-preflight", str(blue), "--printer", "u1" + ) + refused_fix = run_cli( + tmp_path, "print-preflight", str(blue), "--printer", "u1", "--fix" + ) + + assert ( + slot_line(wrong_material.stdout).split()[0] == "FAIL" + and "material: G-code PLA, loaded PETG" in wrong_material.stdout + ) + assert slot_line(wrong_colour.stdout).split()[0] == "FAIL" and "ΔE" in slot_line( + wrong_colour.stdout + ) + assert refused_fix.returncode == 1 and "another colour" in refused_fix.stderr + assert blue.read_text(encoding="utf-8") == gcode_text( + "#0000FF" + ) # nothing was rewritten + + +def test_preflight_fix_aligns_gcode_3mf_and_sidecars_and_prints_the_hashes( + tmp_path: Path, +) -> None: + """`--fix`: one command rewrites the colours at the source; old and new sha256 are on record.""" + moon = FakeMoonraker() + load_beige_petg(moon) + g = write_plate(tmp_path) + old = sw.sha256_file(g) + with moon.serve(): + write_devices(tmp_path / "xdg-config", moon=moon) + fixed = run_cli(tmp_path, "print-preflight", str(g), "--printer", "u1", "--fix") + + assert f"colour fix: tool T3 (T4) {SLICED} -> {LOADED} (ΔE 6.4)" in fixed.stderr + assert ( + old in fixed.stderr + and sw.sha256_file(g) in fixed.stderr + and "sha256 new" in fixed.stderr + ) + assert ( + slot_line(fixed.stdout).split()[0] == "PASS" and "slicer warn" in fixed.stdout + ) + assert f"; filament_colour = #000000;#808080;#1B5E20;{LOADED}" in g.read_text( + encoding="utf-8" + ) + with zipfile.ZipFile(g.with_suffix(".3mf")) as z: + assert ( + json.loads(z.read("Metadata/project_settings.config"))["filament_colour"][3] + == LOADED + ) + assert ( + json.loads((tmp_path / "plate.manifest.json").read_text())["filaments"][3][ + "hex" + ] + == LOADED + ) + assert sw.load(sw.sidecar_path(g))["gcode_sha256"] == sw.sha256_file(g) + assert [line for line in fixed.stdout.splitlines() if "slicer warn" in line][ + 0 + ].split()[0] == "PASS" + assert moon.writes() == [] + + +def test_one_fix_adds_the_missing_thumbnails_and_aligns_the_colours( + tmp_path: Path, +) -> None: + """A G-code without previews, sliced in another beige than the printer's: one `--fix` repairs both and the + preflight passes; its thumbnails gate would have FAILed before.""" + moon = FakeMoonraker() + load_beige_petg(moon) + g = write_plate(tmp_path) + g.write_text(gcode_text().replace(THUMBNAILS, ""), encoding="utf-8") + sw.record(g, logs=[("slice.log", "")]) + with moon.serve(): + write_devices(tmp_path / "xdg-config", moon=moon) + plain = run_cli(tmp_path, "print-preflight", str(g), "--printer", "u1") + fixed = run_cli(tmp_path, "print-preflight", str(g), "--printer", "u1", "--fix") + offline = run_cli( + tmp_path, "print-preflight", str(g), "--axis-limits", "u1", "--fix" + ) + + assert plain.returncode == 1 and "FAIL thumbnails" in plain.stdout + assert fixed.returncode == 0, fixed.stdout + fixed.stderr + assert "--fix: added 48x48/PNG, 96x96/PNG, 300x300/PNG" in fixed.stderr + assert f"colour fix: tool T3 (T4) {SLICED} -> {LOADED}" in fixed.stderr + text = g.read_text(encoding="utf-8") + assert "; thumbnail begin 300x300" in text and LOADED in text + assert sw.load(sw.sidecar_path(g))["gcode_sha256"] == sw.sha256_file(g) + # nothing to align to without a live printer: the colours stay, stderr says so, the thumbnails are done + assert offline.returncode == 0 and "colours were not aligned" in offline.stderr diff --git a/tests/e2e/test_printer_colour_workflows.py b/tests/e2e/test_printer_colour_workflows.py new file mode 100644 index 00000000..c7fe05b9 --- /dev/null +++ b/tests/e2e/test_printer_colour_workflows.py @@ -0,0 +1,158 @@ +"""E2E stories through `bin/3d printer` for the slot colour: `start --fix-colour` (rewrite, upload the +fixed file, start) and `slot set` / `slot show` (align the printer's head record with the file) on a +loopback fake U1; never a real printer.""" + +from __future__ import annotations + +from pathlib import Path + +from colour_fix_fixtures import LOADED, SLICED, load_beige_petg, write_plate +from fleet_fixtures import u1_entry, write_fleet +from printer_fakes import FakeMoonraker, u1_status, write_devices + +from .workflow_helper import run_cli + + +def slot_line(stdout: str) -> str: + return next(line for line in stdout.splitlines() if "slot T3" in line) + + +def test_printer_start_fix_colour_needs_no_force_for_the_slot_colour( + tmp_path: Path, +) -> None: + """The slot colour no longer blocks: `start --fix-colour` rewrites, uploads the fixed file and starts it + with no force and no approval request.""" + moon = FakeMoonraker() + load_beige_petg(moon) + g = write_plate(tmp_path) + with moon.serve(): + write_devices(tmp_path / "xdg-config", moon=moon) + started = run_cli( + tmp_path, "printer", "start", "u1", str(g), "--yes", "--fix-colour" + ) + + assert started.returncode == 0, started.stdout + started.stderr + assert ( + f"{SLICED} -> {LOADED}" in started.stdout + and "uploaded the colour-fixed file" in started.stdout + ) + assert slot_line(started.stdout).split()[0] == "PASS" + assert "FORCED" not in started.stdout + assert [r.path for r in moon.writes()] == [ + "/server/files/upload", + "/printer/print/start", + ] + assert moon.contents[moon.writes()[1].query["filename"][0]] == g.read_bytes() + + +def test_printer_slot_set_dry_run_then_write_aligns_the_printer_with_the_file( + tmp_path: Path, +) -> None: + """The other direction: the printer's record is set to the file's colour (U1 SET_PRINT_FILAMENT_CONFIG).""" + moon = FakeMoonraker() + with moon.serve(): + env = { + "THREED_FLEET": str(write_fleet(tmp_path / "fleet.yaml", u1_entry(moon))) + } + dry = run_cli( + tmp_path, + "printer", + "slot", + "set", + "u1", + "T4", + "--material", + "PETG", + "--color", + SLICED, + "--brand", + "Elegoo", + "--dry-run", + env_extra=env, + ) + assert moon.scripts == [] + wrote = run_cli( + tmp_path, + "printer", + "slot", + "set", + "u1", + "T4", + "--material", + "PETG", + "--color", + SLICED, + "--brand", + "Elegoo", + env_extra=env, + ) + shown = run_cli(tmp_path, "printer", "slot", "show", "u1", env_extra=env) + moon.status = u1_status("printing", "big.gcode") + busy = run_cli( + tmp_path, + "printer", + "slot", + "set", + "u1", + "T4", + "--material", + "PLA", + "--color", + "#FFFFFF", + env_extra=env, + ) + + assert dry.returncode == 0, dry.stderr + assert "DRY RUN" in dry.stdout and SLICED in dry.stdout + assert wrote.returncode == 0, wrote.stderr + assert f"u1/T4 now shows PETG {SLICED} Elegoo Basic (read back)" in wrote.stdout + assert any( + f"PETG {SLICED}" in line and line.lstrip().startswith("T4") + for line in shown.stdout.splitlines() + ) + assert busy.returncode == 1 and "never done during a job" in busy.stderr + assert len(moon.scripts) == 1 and moon.scripts[0].startswith( + "SET_PRINT_FILAMENT_CONFIG CONFIG_EXTRUDER=3 " + ) + + +def test_slot_set_refuses_an_offline_printer_and_a_missing_head(tmp_path: Path) -> None: + """The owner tries `slot set` on a read-only printer and on a head that is not there: both are refused.""" + moon = FakeMoonraker() + with moon.serve(): + env = { + "THREED_FLEET": str( + write_fleet(tmp_path / "fleet.yaml", u1_entry(moon, access="read-only")) + ) + } + readonly = run_cli( + tmp_path, + "printer", + "slot", + "set", + "u1", + "T4", + "--material", + "PETG", + "--color", + SLICED, + env_extra=env, + ) + no_head = run_cli( + tmp_path, + "printer", + "slot", + "set", + "u1", + "--material", + "PETG", + "--color", + SLICED, + env_extra=env, + ) + + assert readonly.returncode == 1 and "read-only" in readonly.stderr + assert ( + no_head.returncode == 2 and "needs the printer and the head" in no_head.stderr + ) + assert moon.scripts == [] diff --git a/tests/e2e/test_printer_force_retry_workflows.py b/tests/e2e/test_printer_force_retry_workflows.py new file mode 100644 index 00000000..e9bb9bfb --- /dev/null +++ b/tests/e2e/test_printer_force_retry_workflows.py @@ -0,0 +1,44 @@ +"""E2E: a `3d printer start --force` approval that tg-ctl could not post to Telegram +(`ask-forward not-posted … reason=send-failed`, issue #728) is asked again, then refused with the way to +start from the printer's own screen. The hatch is a test double; the printer a loopback fake.""" + +from __future__ import annotations + +from pathlib import Path + +from colour_fix_fixtures import load_beige_petg, write_plate +from printer_fakes import FakeMoonraker, write_devices + +from .workflow_helper import hatch_requests, run_cli, run_cli_with_hatch + + +def test_force_approval_that_telegram_could_not_post_is_retried_then_points_to_the_printer_screen( + tmp_path: Path, +) -> None: + """tg-ctl said `ask-forward not-posted … reason=send-failed`: the CLI asks again (4 requests in all), + then refuses with how to start from the printer's own screen; nothing is started.""" + moon = FakeMoonraker() + load_beige_petg(moon) + g = write_plate(tmp_path, material="PLA") # PLA sliced, PETG loaded: a real slot FAIL + with moon.serve(): + write_devices(tmp_path / "xdg-config", moon=moon) + assert run_cli(tmp_path, "printer", "upload", "u1", str(g)).returncode == 0 + refused = run_cli_with_hatch( + tmp_path, + "printer", + "start", + "u1", + str(g), + "--yes", + "--force", + verdict="send-failed", + ) + + assert refused.returncode == 1, refused.stdout + refused.stderr + assert "could not be posted to Telegram after 4 attempts" in refused.stderr + assert ( + "Start from the printer screen" in refused.stderr and "plate-" in refused.stderr + ) + assert refused.stderr.count("asking again in 0 s") == 3 + assert len(hatch_requests(tmp_path)) == 4 + assert "/printer/print/start" not in [r.path for r in moon.writes()] diff --git a/tests/test_colour_fix.py b/tests/test_colour_fix.py new file mode 100644 index 00000000..934bcf24 --- /dev/null +++ b/tests/test_colour_fix.py @@ -0,0 +1,331 @@ +"""Unit tests for lib/printer_ops/colour_fix.py: the CIEDE2000 slot colour tolerance and the rewrite +that aligns a G-code, its project 3MF, manifest and slice-warnings sidecar with the loaded colour +(issue https://github.com/alex-mextner/3d-cli/issues/728).""" + +from __future__ import annotations + +import hashlib +import json +import os +import zipfile +from pathlib import Path + +import pytest +from colour_fix_fixtures import ( + LOADED, + SLICED, + THUMBNAILS, + TOOLPATH, + gcode_text, + load_beige_petg, + write_plate, + write_project, +) +from printer_fakes import FakeMoonraker + +from errors import InvalidArgument +from printer_ops import colour_fix, moonraker, preflight +from printer_ops.base import PrinterRefused, PrinterStatus +from slicing import slice_warnings as sw + + +def status(colour: str = LOADED) -> PrinterStatus: + moon = FakeMoonraker() + load_beige_petg(moon, colour=colour) + return moonraker.parse_status(moon.status) + + +def sha(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +# ---- the distance ------------------------------------------------------------------------ + + +def test_the_real_beige_pair_is_close_and_other_colours_are_far() -> None: + assert colour_fix.colour_delta(SLICED, LOADED) == pytest.approx(6.4, abs=0.1) + assert colour_fix.colour_delta("#FFFFFF", SLICED) == pytest.approx(16.7, abs=0.1) + assert colour_fix.colour_delta("#FF0000", "#0000FF") == pytest.approx(52.9, abs=0.1) + assert colour_fix.colour_delta(SLICED, SLICED) == 0.0 + assert colour_fix.colour_delta("nonsense", SLICED) is None + + +@pytest.mark.parametrize("bad", ["0", "-3", "abc", "nan"]) +def test_delta_e_option_must_be_a_positive_number(bad: str) -> None: + with pytest.raises(InvalidArgument): + colour_fix.parse_delta_e( + bad, "--colour-delta-e", 10.0, command="print-preflight" + ) + assert colour_fix.parse_delta_e(None, "--colour-delta-e", 10.0, command="x") == 10.0 + assert colour_fix.parse_delta_e("7.5", "--colour-delta-e", 10.0, command="x") == 7.5 + + +# ---- the slot check ---------------------------------------------------------------------- + + +def slots_of( + tmp_path: Path, text: str, st: PrinterStatus, **kw: float +) -> preflight.Check: + g = tmp_path / "p.gcode" + g.write_text(text, encoding="utf-8") + (check,) = preflight.check_slots(preflight.read_gcode_meta(g), st, **kw) + return check + + +def test_same_family_beige_passes_and_reports_the_distance(tmp_path: Path) -> None: + check = slots_of(tmp_path, gcode_text(), status()) + assert check.status == "PASS" + assert ( + "ΔE 6.4" in check.detail and SLICED in check.detail and "PETG" in check.detail + ) + + +def test_a_different_colour_fails_with_the_distance_and_the_fix(tmp_path: Path) -> None: + check = slots_of(tmp_path, gcode_text("#FFFFFF"), status()) + assert check.status == "FAIL" + assert "G-code #FFFFFF, loaded #FEE5A5 (ΔE 20.0 > 10)" in check.detail + assert "--fix" in check.detail + + +def test_the_threshold_is_configurable(tmp_path: Path) -> None: + assert ( + slots_of(tmp_path, gcode_text(), status(), colour_delta_e=5.0).status == "FAIL" + ) + assert ( + slots_of(tmp_path, gcode_text("#FFFFFF"), status(), colour_delta_e=25.0).status + == "PASS" + ) + + +def test_a_material_mismatch_stays_a_fail_even_for_the_same_colour( + tmp_path: Path, +) -> None: + check = slots_of(tmp_path, gcode_text(SLICED, "PLA"), status()) + assert ( + check.status == "FAIL" and "material: G-code PLA, loaded PETG" in check.detail + ) + assert ( + slots_of(tmp_path, gcode_text(LOADED, "PLA"), status(colour=LOADED)).status + == "FAIL" + ) + + +# ---- the plan ---------------------------------------------------------------------------- + + +def test_plan_edits_the_used_tool_and_refuses_another_colour(tmp_path: Path) -> None: + g = write_plate(tmp_path) + edits, refusals = colour_fix.plan_colour_fix(preflight.read_gcode_meta(g), status()) + assert refusals == [] and [(e.tool, e.label, e.old, e.new) for e in edits] == [ + (3, "T4", SLICED, LOADED) + ] + assert edits[0].delta_e == pytest.approx(6.4, abs=0.1) + + edits, refusals = colour_fix.plan_colour_fix( + preflight.read_gcode_meta(g), status(colour="#0000FF") + ) + assert edits == [] and len(refusals) == 1 + assert "tool T3 (T4)" in refusals[0] and "another colour" in refusals[0] + # the operator may widen the limit on purpose + edits, _ = colour_fix.plan_colour_fix( + preflight.read_gcode_meta(g), status(colour="#0000FF"), max_delta_e=100.0 + ) + assert [e.new for e in edits] == ["#0000FF"] + + +def test_plan_leaves_matching_and_empty_heads_alone(tmp_path: Path) -> None: + g = write_plate(tmp_path) + assert colour_fix.plan_colour_fix( + preflight.read_gcode_meta(g), status(colour=SLICED) + ) == ([], []) + moon = FakeMoonraker() + moon.status["print_task_config"]["filament_exist"][3] = False + empty = moonraker.parse_status(moon.status) + assert colour_fix.plan_colour_fix(preflight.read_gcode_meta(g), empty) == ([], []) + + +# ---- the rewrite ------------------------------------------------------------------------- + + +def fix( + g: Path, st: PrinterStatus | None = None, source: Path | None = None +) -> colour_fix.FixResult: + result = colour_fix.fix_to_loaded( + g, st or status(), source, command="print-preflight" + ) + assert result is not None + return result + + +def test_fix_rewrites_gcode_3mf_manifest_and_sidecar_consistently( + tmp_path: Path, +) -> None: + g = write_plate(tmp_path) + project, manifest, side = ( + g.with_suffix(".3mf"), + tmp_path / "plate.manifest.json", + sw.sidecar_path(g), + ) + old_g, old_p = sha(g), sha(project) + g.chmod(0o640) + os.utime(g, (1_700_000_000, 1_700_000_000)) + os.utime(project, (1_700_000_100, 1_700_000_100)) + model_before = zipfile.ZipFile(project).read("3D/3dmodel.model") + + result = fix(g) + + # G-code: only the colour entries of tool 3 changed, the toolpath bytes did not + text = g.read_text(encoding="utf-8") + assert text.startswith( + "; HEADER_BLOCK_START\n; HEADER_BLOCK_END\n" + THUMBNAILS + TOOLPATH + ) + assert f"; filament_colour = #000000;#808080;#1B5E20;{LOADED}\n" in text + assert f"; extruder_colour = #000000;#808080;#1B5E20;{LOADED}\n" in text + assert "; filament_type = PLA;PLA;PLA;PETG\n" in text + assert (g.stat().st_mode & 0o777, g.stat().st_mtime_ns) == ( + 0o640, + 1_700_000_000 * 10**9, + ) + assert project.stat().st_mtime_ns == 1_700_000_100 * 10**9 + # the reported hashes are the real ones + assert (result.gcode.old_sha256, result.gcode.new_sha256) == ( + old_g, + sha(g), + ) and old_g != sha(g) + assert result.project is not None and ( + result.project.old_sha256, + result.project.new_sha256, + ) == (old_p, sha(project)) + # 3MF: colour lists + plate json updated, the other members untouched + with zipfile.ZipFile(project) as z: + cfg = json.loads(z.read("Metadata/project_settings.config")) + assert cfg["filament_colour"][3] == cfg["extruder_colour"][3] == LOADED + assert ( + cfg["filament_colour"][:3] == ["#000000", "#808080", "#1B5E20"] + and cfg["layer_height"] == "0.2" + ) + assert ( + z.read("Metadata/project_settings.config") + .decode() + .startswith('{\n "filament_colour"') + ) + assert ( + json.loads(z.read("Metadata/plate_1.json"))["filament_colors"][3] == LOADED + ) + assert z.read("3D/3dmodel.model") == model_before and z.testzip() is None + # manifest + slice-warnings sidecar follow + assert [f["hex"] for f in json.loads(manifest.read_text())["filaments"]][ + 3 + ] == LOADED + data = json.loads(side.read_text(encoding="utf-8")) + assert ( + data["gcode_sha256"] == sha(g) + and data["post_processed"][-1]["step"] == "colour fix" + ) + assert sw.evaluate(g)[0] == "PASS" + assert result.manifest == manifest and result.sidecar == side + lines = "\n".join(result.lines()) + assert ( + f"{SLICED} -> {LOADED} (ΔE 6.4)" in lines and old_g in lines and sha(g) in lines + ) + # nothing is left behind and a second run finds nothing to do + assert [ + p.name for p in tmp_path.iterdir() if p.name.startswith(".colour-fix") + ] == [] + assert ( + colour_fix.fix_to_loaded(g, status(), None, command="print-preflight") is None + ) + + +def test_fix_uses_an_explicit_project_and_tolerates_a_non_project_3mf( + tmp_path: Path, +) -> None: + g = write_plate(tmp_path) + other = tmp_path / "other.3mf" + write_project(other) + assert fix(g, source=other).project is not None + assert ( + json.loads(zipfile.ZipFile(other).read("Metadata/project_settings.config"))[ + "filament_colour" + ][3] + == LOADED + ) + assert ( + json.loads( + zipfile.ZipFile(g.with_suffix(".3mf")).read( + "Metadata/project_settings.config" + ) + )["filament_colour"][3] + == SLICED + ) # the sibling was not the project asked for + + h = write_plate(tmp_path, "model") + with zipfile.ZipFile(h.with_suffix(".3mf"), "w") as z: + z.writestr( + "3D/3dmodel.model", "" + ) # a plain model, not a slicer project + result = fix(h) + assert result.project is None and "not a slicer project 3MF" in "\n".join( + result.lines() + ) + assert LOADED in h.read_text(encoding="utf-8") + + +def test_fix_without_a_project_rewrites_only_the_gcode(tmp_path: Path) -> None: + g = tmp_path / "solo.gcode" + g.write_text(gcode_text(), encoding="utf-8") + result = fix(g) + assert result.project is None and result.manifest is None and result.sidecar is None + assert "no project 3MF found" in "\n".join(result.lines()) + + +def test_fix_refuses_another_colour_and_writes_nothing(tmp_path: Path) -> None: + g = write_plate(tmp_path) + before = {p.name: p.read_bytes() for p in tmp_path.iterdir()} + with pytest.raises(PrinterRefused, match="another colour"): + colour_fix.fix_to_loaded( + g, status(colour="#0000FF"), None, command="print-preflight" + ) + assert {p.name: p.read_bytes() for p in tmp_path.iterdir()} == before + + +def test_fix_without_a_colour_line_refuses_before_touching_the_3mf( + tmp_path: Path, +) -> None: + g = write_plate(tmp_path) + g.write_text( + TOOLPATH + + "; filament used [mm] = 0.00, 0.00, 0.00, 30.50\n; filament_type = PLA;PLA;PLA;PETG\n" + ) + project = g.with_suffix(".3mf") + before = project.read_bytes() + edits = [colour_fix.ColourEdit(3, "T4", SLICED, LOADED, 6.4)] + with pytest.raises(PrinterRefused, match="no `; filament_colour`"): + colour_fix.apply_colour_fix(g, project, edits) + assert project.read_bytes() == before + + +def test_fix_refuses_a_printer_without_per_head_data(tmp_path: Path) -> None: + g = write_plate(tmp_path) + with pytest.raises(PrinterRefused, match="no per-head filament data"): + colour_fix.fix_to_loaded( + g, PrinterStatus("idle", "standby"), None, command="print-preflight" + ) + + +def test_fix_keeps_quotes_crlf_and_the_other_entries(tmp_path: Path) -> None: + g = tmp_path / "crlf.gcode" + g.write_bytes( + b"G28\r\nT3\r\n" + b'; filament_colour = "#000000";"#808080";"#1B5E20";"#E8D8B0FF"\r\n' + b"; filament used [mm] = 0.00, 0.00, 0.00, 30.50\r\n" + b"; filament_type = PLA;PLA;PLA;PETG\r\n" + ) + fix(g) + assert ( + b'; filament_colour = "#000000";"#808080";"#1B5E20";"#FEE5A5"\r\n' + in g.read_bytes() + ) + assert g.read_bytes().count(b"\r\n") == 5 and g.read_bytes().startswith( + b"G28\r\nT3\r\n" + ) diff --git a/tests/test_force_hatch.py b/tests/test_force_hatch.py index 81a746c5..4180b571 100644 --- a/tests/test_force_hatch.py +++ b/tests/test_force_hatch.py @@ -179,3 +179,79 @@ def test_missing_helper_fails_closed( _home(monkeypatch, tmp_path) with pytest.raises(force_hatch.HatchUnavailable, match="not found"): force_hatch.load_hatch() + + +# ---- tg-ctl `ask-forward not-posted … reason=send-failed` (issue #728) -------------------- + +NOT_POSTED = ( + "tg-ctl ask: ask-forward not-posted: agent=claude kind=permission req=hatch-3d-printer-start-force-d6543f0fc5b0 " + "daemon declined: reason=send-failed" +) + + +class ScriptedHatch: + """The shared hatch helper answering a fixed sequence of verdicts, one per request.""" + + def __init__(self, *verdicts: tuple[bool, bool, str]) -> None: + self.verdicts = list(verdicts) + self.requests = 0 + + def request_hatch_approval(self, hook_id: str, context: dict[str, object], **_: object) -> SimpleNamespace: + self.requests += 1 + approved, should_stop, reason = self.verdicts.pop(0) + return SimpleNamespace(approved=approved, should_stop=should_stop, reason=reason) + + +@pytest.fixture() +def sleeps(monkeypatch: pytest.MonkeyPatch) -> list[float]: + slept: list[float] = [] + monkeypatch.setattr(force_hatch.time, "sleep", slept.append) + return slept + + +def ask(hatch: ScriptedHatch, monkeypatch: pytest.MonkeyPatch) -> str: + monkeypatch.setattr(force_hatch, "load_hatch", lambda: hatch) + return force_hatch.require_force_approval( + printer="u1", file="plate.gcode", sha256="0" * 64, checks=CHECKS, remote_name="plate-0123456789abcdef.gcode" + ) + + +def test_a_card_that_was_not_posted_is_asked_again_with_backoff_until_the_owner_answers( + monkeypatch: pytest.MonkeyPatch, sleeps: list[float], capsys: pytest.CaptureFixture[str] +) -> None: + hatch = ScriptedHatch((False, True, NOT_POSTED), (False, True, NOT_POSTED), (True, True, "approved by tg-ctl ask")) + assert ask(hatch, monkeypatch) == "approved by tg-ctl ask" + assert hatch.requests == 3 and sleeps == [3.0, 10.0] + assert "asking again in 3 s" in capsys.readouterr().err + + +def test_a_card_that_never_posts_ends_with_the_printer_screen_way( + monkeypatch: pytest.MonkeyPatch, sleeps: list[float] +) -> None: + hatch = ScriptedHatch(*[(False, True, NOT_POSTED)] * 4) + with pytest.raises(PrinterRefused) as exc: + ask(hatch, monkeypatch) + assert hatch.requests == 4 and sleeps == [3.0, 10.0, 30.0] + assert "could not be posted to Telegram after 4 attempts" in exc.value.message + text = "\n".join(exc.value.remediation) + assert "Start from the printer screen" in text and "plate-0123456789abcdef.gcode" in text + assert "does NOT run these checks" in text and "tg-ctl.*.log" in text and "- macros:" in text + + +@pytest.mark.parametrize( + "verdict", + [ + (False, True, "tg-ctl ask decision was 'deny' (expected 'allow')"), + (False, True, "tg-ctl ask: ask-forward expired: messageId=7 — posted to Telegram; no decision"), + (False, True, "tg-ctl ask: ask-forward not-posted: daemon declined: reason=not-registered"), + (False, False, "RIG_HATCH_REQUEST_3D_PRINTER_START_FORCE is not set"), + ], +) +def test_other_outcomes_are_final_and_never_retried( + monkeypatch: pytest.MonkeyPatch, sleeps: list[float], verdict: tuple[bool, bool, str] +) -> None: + hatch = ScriptedHatch(verdict) + with pytest.raises(PrinterRefused) as exc: + ask(hatch, monkeypatch) + assert hatch.requests == 1 and sleeps == [] + assert "printer screen" not in "\n".join(exc.value.remediation) diff --git a/tests/test_printer_colour_commands.py b/tests/test_printer_colour_commands.py new file mode 100644 index 00000000..2ba0b95f --- /dev/null +++ b/tests/test_printer_colour_commands.py @@ -0,0 +1,372 @@ +"""`3d print-preflight --colour-delta-e/--fix`, `3d printer start --fix-colour` and `3d printer slot set` +against a loopback fake U1 (tests/printer_fakes.py); the printer is never real. Issue +https://github.com/alex-mextner/3d-cli/issues/728.""" + +from __future__ import annotations + +import json +from collections.abc import Iterator +from pathlib import Path + +import pytest +from colour_fix_fixtures import LOADED, SLICED, gcode_text, load_beige_petg, write_plate +from fleet_fixtures import u1_entry, write_fleet +from printer_fakes import FakeMoonraker, u1_status, write_devices + +from errors import GateFailure, InvalidArgument, UsageError +from printer_ops import preflight +from printer_ops.base import PrinterRefused, safe_remote_name +from slicing import slice_warnings as sw + + +@pytest.fixture() +def moon(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[FakeMoonraker]: + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "xdg")) + monkeypatch.chdir(tmp_path) + fake = FakeMoonraker() + load_beige_petg(fake) + with fake.serve(): + write_devices(tmp_path / "xdg", moon=fake) + yield fake + + +@pytest.fixture() +def gates_ok(monkeypatch: pytest.MonkeyPatch) -> None: + """The thumbnail / overhang / slicer-warning gates have their own tests: pin them to PASS.""" + monkeypatch.setattr( + preflight, + "check_thumbnails", + lambda p, shown=None: preflight.Check("thumbnails", "PASS", "stub"), + ) + monkeypatch.setattr( + preflight, + "check_overhangs", + lambda p, *_: preflight.Check("overhangs", "PASS", "stub"), + ) + monkeypatch.setattr( + preflight, + "check_slice_warnings", + lambda o, g: preflight.Check("slicer warn", "PASS", "stub"), + ) + + +def preflight_cli(*argv: str) -> int: + from commands.print_preflight import run + + return run(list(argv)) + + +def printer(*argv: str) -> int: + from commands.printer import run + + return run(list(argv)) + + +# ---- print-preflight --------------------------------------------------------------------- + + +def test_preflight_passes_the_beige_pair_and_shows_the_distance( + moon: FakeMoonraker, + tmp_path: Path, + gates_ok: None, + capsys: pytest.CaptureFixture[str], +) -> None: + g = write_plate(tmp_path) + assert preflight_cli(str(g), "--printer", "u1") == 0 + out = capsys.readouterr().out + assert "slot T3" in out and "ΔE 6.4" in out and f"G-code {SLICED}" in out + assert moon.writes() == [] + + +def test_preflight_threshold_option_and_validation( + moon: FakeMoonraker, + tmp_path: Path, + gates_ok: None, + capsys: pytest.CaptureFixture[str], +) -> None: + g = write_plate(tmp_path) + with pytest.raises(GateFailure): + preflight_cli(str(g), "--printer", "u1", "--colour-delta-e", "5") + assert ( + f"colour: G-code {SLICED}, loaded {LOADED} (ΔE 6.4 > 5)" + in capsys.readouterr().out + ) + with pytest.raises(InvalidArgument): + preflight_cli(str(g), "--printer", "u1", "--colour-delta-e", "0") + + +def test_preflight_fails_a_different_colour_and_a_wrong_material( + moon: FakeMoonraker, + tmp_path: Path, + gates_ok: None, + capsys: pytest.CaptureFixture[str], +) -> None: + other = tmp_path / "other.gcode" + other.write_text(gcode_text("#0000FF"), encoding="utf-8") + with pytest.raises(GateFailure): + preflight_cli(str(other), "--printer", "u1") + assert "G-code #0000FF, loaded #FEE5A5 (ΔE " in capsys.readouterr().out + petg_as_pla = tmp_path / "pla.gcode" + petg_as_pla.write_text(gcode_text(SLICED, "PLA"), encoding="utf-8") + with pytest.raises(GateFailure): + preflight_cli(str(petg_as_pla), "--printer", "u1") + assert "material: G-code PLA, loaded PETG" in capsys.readouterr().out + + +def test_preflight_fix_rewrites_then_passes_and_prints_old_and_new_sha( + moon: FakeMoonraker, + tmp_path: Path, + gates_ok: None, + capsys: pytest.CaptureFixture[str], +) -> None: + g = write_plate(tmp_path) + old = sw.sha256_file(g) + assert preflight_cli(str(g), "--printer", "u1", "--fix") == 0 + cap = capsys.readouterr() + assert ( + f"{SLICED} -> {LOADED}" in cap.err + and old in cap.err + and sw.sha256_file(g) in cap.err + ) + assert f"G-code {LOADED}" not in cap.out # exact match now: no note + assert "slot T3" in cap.out and "PASS" in cap.out + assert f"; filament_colour = #000000;#808080;#1B5E20;{LOADED}" in g.read_text( + encoding="utf-8" + ) + assert moon.writes() == [] # --fix writes files, never the printer + + assert preflight_cli(str(g), "--printer", "u1", "--fix") == 0 + assert "already equal" in capsys.readouterr().err + + +def test_preflight_fix_json_stays_machine_readable( + moon: FakeMoonraker, + tmp_path: Path, + gates_ok: None, + capsys: pytest.CaptureFixture[str], +) -> None: + g = write_plate(tmp_path) + assert preflight_cli(str(g), "--printer", "u1", "--fix", "--json") == 0 + cap = capsys.readouterr() + assert any(c["name"] == "slot T3" for c in json.loads(cap.out)["checks"]) + assert "sha256 old" in cap.err + + +def test_preflight_fix_refuses_another_colour_and_aligns_colours_only_with_a_live_printer( + moon: FakeMoonraker, + tmp_path: Path, + gates_ok: None, + capsys: pytest.CaptureFixture[str], +) -> None: + g = tmp_path / "blue.gcode" + g.write_text(gcode_text("#0000FF"), encoding="utf-8") + before = g.read_bytes() + with pytest.raises(PrinterRefused, match="another colour"): + preflight_cli(str(g), "--printer", "u1", "--fix") + assert g.read_bytes() == before + # widening the limit is an explicit, per-run decision + assert ( + preflight_cli(str(g), "--printer", "u1", "--fix", "--fix-max-delta-e", "100") + == 0 + ) + assert LOADED in g.read_text(encoding="utf-8") + g.write_bytes(before) + capsys.readouterr() + # no live printer (none, or --offline): the colours have nothing to align to, the rest of --fix still runs + for argv in (("--axis-limits", "u1"), ("--printer", "u1", "--offline")): + assert preflight_cli(str(g), *argv, "--fix") == 0 + assert "colours were not aligned" in capsys.readouterr().err + assert g.read_bytes() == before + + +# ---- printer start --fix-colour ---------------------------------------------------------- + + +def test_start_fix_colour_rewrites_uploads_the_fixed_file_and_starts( + moon: FakeMoonraker, + tmp_path: Path, + gates_ok: None, + capsys: pytest.CaptureFixture[str], +) -> None: + g = write_plate(tmp_path) + old_name = safe_remote_name(g) + assert printer("start", "u1", str(g), "--yes", "--fix-colour") == 0 + out = capsys.readouterr().out + assert f"{SLICED} -> {LOADED}" in out and "uploaded the colour-fixed file" in out + new_name = safe_remote_name(g) + assert ( + new_name != old_name and new_name in moon.files and old_name not in moon.files + ) + assert ( + moon.contents[new_name] == g.read_bytes() + and LOADED.encode() in moon.contents[new_name] + ) + assert [(r.path, r.query.get("filename")) for r in moon.writes()] == [ + ("/server/files/upload", None), + ("/printer/print/start", [new_name]), + ] + assert "slot T3" in out and "start acknowledged by u1" in out + + +def test_start_fix_colour_refuses_another_colour_before_writing_anything( + moon: FakeMoonraker, tmp_path: Path, gates_ok: None +) -> None: + g = tmp_path / "blue.gcode" + g.write_text(gcode_text("#0000FF"), encoding="utf-8") + before = g.read_bytes() + with pytest.raises(PrinterRefused, match="another colour"): + printer("start", "u1", str(g), "--yes", "--fix-colour") + assert g.read_bytes() == before and moon.writes() == [] + + +def test_start_without_fix_colour_passes_the_beige_without_rewriting( + moon: FakeMoonraker, + tmp_path: Path, + gates_ok: None, + capsys: pytest.CaptureFixture[str], +) -> None: + g = write_plate(tmp_path) + moon.put(safe_remote_name(g), g.read_bytes()) + before = g.read_bytes() + assert printer("start", "u1", str(g), "--yes") == 0 + assert "ΔE 6.4" in capsys.readouterr().out and g.read_bytes() == before + + +def test_start_fix_colour_is_refused_while_printing( + moon: FakeMoonraker, tmp_path: Path, gates_ok: None +) -> None: + g = write_plate(tmp_path) + before = g.read_bytes() + moon.status = u1_status("printing", "x.gcode") + load_beige_petg(moon) + with pytest.raises(PrinterRefused, match="printing"): + printer("start", "u1", str(g), "--yes", "--fix-colour") + assert g.read_bytes() == before and moon.writes() == [] + + +# ---- printer slot set -------------------------------------------------------------------- + + +@pytest.fixture() +def fleet_moon( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> Iterator[FakeMoonraker]: + fake = FakeMoonraker() + with fake.serve(): + monkeypatch.setenv( + "THREED_FLEET", str(write_fleet(tmp_path / "fleet.yaml", u1_entry(fake))) + ) + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "xdg")) + monkeypatch.chdir(tmp_path) + yield fake + + +def test_slot_set_writes_the_head_and_reads_it_back( + fleet_moon: FakeMoonraker, capsys: pytest.CaptureFixture[str] +) -> None: + assert ( + printer( + "slot", + "set", + "u1", + "t4", + "--material", + "PETG", + "--color", + "fee5a5", + "--brand", + "Elegoo", + ) + == 0 + ) + assert ( + "u1/T4 now shows PETG #FEE5A5 Elegoo Basic (read back)" + in capsys.readouterr().out + ) + (script,) = fleet_moon.scripts + assert ( + script.startswith("SET_PRINT_FILAMENT_CONFIG CONFIG_EXTRUDER=3 ") + and 'VENDOR="Elegoo"' in script + ) + assert "FILAMENT_COLOR_RGBA=FEE5A5FF" in script and 'FILAMENT_TYPE="PETG"' in script + + +def test_slot_set_dry_run_sends_nothing_and_predicts_refusals( + fleet_moon: FakeMoonraker, capsys: pytest.CaptureFixture[str] +) -> None: + assert ( + printer( + "slot", + "set", + "u1", + "T4", + "--material", + "PETG", + "--color", + "#FEE5A5", + "--dry-run", + ) + == 0 + ) + out = capsys.readouterr().out + assert "DRY RUN" in out and "FEE5A5" in out + assert fleet_moon.writes() == [] + + fleet_moon.status = u1_status("printing", "x.gcode") + with pytest.raises(GateFailure): + printer( + "slot", + "set", + "u1", + "T4", + "--material", + "PETG", + "--color", + "#FEE5A5", + "--dry-run", + ) + assert "WOULD BE REFUSED" in capsys.readouterr().out and fleet_moon.writes() == [] + + +def test_slot_set_refuses_a_running_job_a_read_only_printer_and_bad_arguments( + tmp_path: Path, fleet_moon: FakeMoonraker, monkeypatch: pytest.MonkeyPatch +) -> None: + fleet_moon.status = u1_status("printing", "x.gcode") + with pytest.raises(PrinterRefused, match="never done during a job"): + printer("slot", "set", "u1", "T4", "--material", "PETG", "--color", "#FEE5A5") + assert fleet_moon.scripts == [] + + monkeypatch.setenv( + "THREED_FLEET", + str( + write_fleet(tmp_path / "ro.yaml", u1_entry(fleet_moon, access="read-only")) + ), + ) + with pytest.raises(PrinterRefused, match="read-only"): + printer( + "slot", + "set", + "u1", + "T4", + "--material", + "PETG", + "--color", + "#FEE5A5", + "--dry-run", + ) + assert fleet_moon.scripts == [] + + with pytest.raises(UsageError, match="not a head label"): + printer("slot", "set", "u1", "4", "--material", "PETG", "--color", "#FEE5A5") + with pytest.raises(UsageError, match="needs the printer and the head"): + printer("slot", "set", "u1") + with pytest.raises(UsageError, match="unknown slot action"): + printer("slot", "wipe", "u1") + + +def test_slot_show_lists_the_heads( + fleet_moon: FakeMoonraker, capsys: pytest.CaptureFixture[str] +) -> None: + assert printer("slot", "show", "u1") == 0 + out = capsys.readouterr().out + assert "T4" in out and "PLA" in out and fleet_moon.writes() == []