diff --git a/docs/commands/printer.md b/docs/commands/printer.md index 773930ec..6cdb9637 100644 --- a/docs/commands/printer.md +++ b/docs/commands/printer.md @@ -42,6 +42,10 @@ Epic: [#90](https://github.com/alex-mextner/3d-cli/issues/90). 3d printer todo dismiss ID --reason TEXT | snooze ID [--until DATE] [--reason TEXT] | reopen ID 3d printer todo import [FILE...] [--dry-run] [--json] 3d printer -n PRINTER issues|todo list|add|done|dismiss|snooze|reopen|import … +3d printer -n PRINTER macro list [--offline | --dry-run] [--all] [--json] +3d printer -n PRINTER macro run MACRO [KEY=VALUE ...] [--yes] [--dry-run] [--offline] [--timeout S] +3d printer -n PRINTER aliases [--json] +3d printer -n PRINTER ALIAS [args] [--yes] [--dry-run] [--offline] [--timeout S] # home, level, load, preheat, ... ``` | Subcommand | What | @@ -60,6 +64,8 @@ Epic: [#90](https://github.com/alex-mextner/3d-cli/issues/90). | `subscribe` / `unsubscribe` / `subscriptions` | Route printer alerts to an AI agent first: it looks at the frame and tells the user only if the failure is real — see [printer-subscribe.md](printer-subscribe.md). | | `agent list` / `stop` / `handoff` | The temporary per-printer omp agent (tmux) that gets the warning+ alerts no live subscriber took — see [printer-subscribe.md#the-temporary-printer-agent](printer-subscribe.md#the-temporary-printer-agent). | | `doctor [NAME\|--all]` | Diagnose the printer's setup (Klippy, Moonraker, Spoolman link and CORS, active spool, camera, watch services, ...) and with `--fix` repair what is safe — the same command as `3d printers doctor`, see [printers-doctor.md](printers-doctor.md). | +| `macro list\|run` | The macros a Klipper printer defines, and ONE macro run with typed `KEY=VALUE` parameters — see [Macros and aliases](#macros-and-aliases-3d-printer-n-name-macro-home-load-). | +| `aliases`, `home`, `level`, `load`, `preheat`, ... | Short names for the usual operations, mapped per adapter, model and printer to the real macros — see [Macros and aliases](#macros-and-aliases-3d-printer-n-name-macro-home-load-). | | `todo …` (alias `issues`) | The to-do list of each printer: needed calibrations, due maintenance, repairs of open faults, manual tasks — see [To-do list](#to-do-list-3d-printer-todo). A printer's problems are tasks of this list, not GitHub tickets. | Options: @@ -393,6 +399,95 @@ Telegram approval). With `--printer A --axis-limits B` only A's list counts. print). The preflight overhangs gate counts uncarried runs up to it as bridges; `--bridge-max` overrides it, and without either the limit is 15 mm. The gate's line names where the limit came from. +## Macros and aliases (`3d printer -n NAME macro`, `home`, `load`, ...) + +Issue [#718](https://github.com/alex-mextner/3d-cli/issues/718). Two ways to run what the printer's own +firmware knows, both through the adapter's G-code channel (`moonraker`: `POST /printer/gcode/script`). +The `creality-stock` adapter has no G-code channel (ws:9999 only starts and stops a file): every command +below refuses it with "does not support macros and aliases". The printer is named with `-n NAME`, like `todo`. + +### `macro list` and `macro run` + +``` +3d printer -n u1 macro list # what the printer defines; * = in its trusted_macros +3d printer -n u1 macro list --offline --json # registry data only (trusted_macros + alias targets); nothing contacted +3d printer -n u1 macro list --dry-run # print the two GET requests, send none +3d printer -n u1 macro run SM_PRINT_AUTO_FEED EXTRUDER=1 --yes --dry-run +3d printer -n k1se macro run PAUSE +``` + +`macro list` reads `GET /printer/objects/list` (the `gcode_macro NAME` objects) and `GET /printer/gcode/help` +(every extended command with its description); `_private` macros need `--all`. `macro run` prints the exact +line that is sent (`gcode:` block) and the request on `--dry-run`. Gates, in this order, nothing is sent before +all passed: + +1. **Typed tokens.** The name is an identifier (`[A-Za-z_][A-Za-z0-9_]*`, upper-cased); each value is one token of + `A-Za-z0-9_.,:+/-` (at most 64 characters). A newline, `;`, space, quote, `{`, `%`, `#` or `*` is refused as an + invalid argument, so a value can never append a second command. +2. **No emergency stop.** `M112` and `EMERGENCY_STOP`, as a name or inside a value, are refused whatever the flags: + the printer's own stop control is the way to stop it. +3. **Access.** Only `access: full` printers; `read-only` and `offline` (never contacted) refuse, also with `--dry-run`. +4. **Defined on the printer.** The name must appear in the printer's `gcode/help`. A standard code (`G28`, `M104`, `T0`) + is run only if the printer defines a macro of that name (the K1's `G29`) or through an alias. +5. **`--yes`** for a macro outside the printer's `trusted_macros` and for dangerous commands (`FIRMWARE_RESTART`, `RESTART`, + `SAVE_CONFIG`, `FORCE_MOVE`, `M84`, calibrations that write config, ...), trusted or not. +6. **An idle printer** (state re-read now; printing, paused, busy and unreadable all refuse) for motion, heating and + unknown macros. `PAUSE`, `RESUME`, `CANCEL_PRINT`, `M600`, `M220`, `M221` and the `SET_PAUSE_*` pair are allowed during a job. + +A macro body is the printer's own config and is not checked, so any macro that is not known to be harmless is treated +as motion. `--offline` prints the script from registry data without contacting the printer (it cannot check 4 and 6). + +**There is no raw `gcode` passthrough.** `3d printer gcode SCRIPT` does not exist: `/printer/gcode/script` runs any text, newlines +and `;` included, the model of this tool is that an agent never sends a free-text line to hardware, and a stock G-code +(`G1`, `M104`) cannot be tied to a trusted list. A command the table lacks is added as a named, typed alias in the +printer's own alias file (below) or run as a macro the printer defines. + +### Aliases + +`3d printer -n NAME ALIAS [args] [--yes] [--dry-run] [--offline]`. An alias is a table entry (`kind`, a `summary`, typed +`args`, `gcode` lines); its lines are vetted, so only `dangerous` ones and entries marked `yes: true` ask for `--yes`; the +idle-printer rule (6) and the definition check (4) still apply. The table is +[`lib/data/printer-aliases.yaml`](../../lib/data/printer-aliases.yaml) (schema in its header), layered per adapter, then +every model whose name contains the entry's `contains` text, then the printer's own entries in +`~/.config/3d-cli/printer-aliases.yaml` and `./printer-aliases.yaml` (the project file wins per alias): + +```yaml +# ~/.config/3d-cli/printer-aliases.yaml or ./printer-aliases.yaml: printer name -> its aliases +u1: + home: {summary: home X and Y only, kind: motion, gcode: ["G28 X Y"]} # replaces the table's home + nozzle-wipe: {summary: wipe the nozzle, kind: motion, gcode: [ROUGHLY_CLEAN_NOZZLE]} # a new alias + park: null # removed for this printer + cooldown: {unsupported: "close the chamber door first"} # removed, with the reason shown +``` + +An alias that a model does not have says so (`alias park is not defined for k1se ...`, with the aliases it does have); +it is never guessed. `3d printer -n NAME aliases` lists what each alias sends on that printer and which are unsupported; +`3d printer ALIAS --help` shows one. + +| Alias | Sends | Where | +|---|---|---| +| `home [XYZ]` | `G28` / `G28 X Y` | every `moonraker` printer | +| `level` | `BED_MESH_CALIBRATE`; Creality K1: the stock `G29` (heats, probes) | `moonraker`; K1 family | +| `qgl`, `z-tilt` | `QUAD_GANTRY_LEVEL`, `Z_TILT_ADJUST` | `moonraker`; refused with a reason on the U1 and the K1 SE (their stock `printer.cfg` define neither) and, on any printer, when the firmware does not define the command | +| `pause`, `resume` | `PAUSE`, `RESUME` (allowed during a job) | `moonraker` | +| `preheat MATERIAL` / `--nozzle C` / `--bed C` / `--head T2` | `SET_HEATER_TEMPERATURE` for the nozzle of the head (fleet `heads:`, default T1) and `heater_bed`; PLA 210/60, PETG 235/80, ABS 250/100, ASA 255/100, TPU 225/50 (nozzle/bed, typical values), nozzle 0-300, bed 0-120 | `moonraker` | +| `cooldown` | `TURN_OFF_HEATERS` | `moonraker` | +| `load`, `unload` | generic: `LOAD_FILAMENT`, `UNLOAD_FILAMENT` (Fluidd/Mainsail convention, refused when the printer has no such macro); K1: `LOAD_MATERIAL`, `QUIT_MATERIAL`; U1: `load T1..T4` = `AUTO_FEEDING EXTRUDER=n LOAD=1`, `unload T1..T4` = `AUTO_FEEDING EXTRUDER=n UNLOAD=1 STAGE=prepare` then `STAGE=doing` | per adapter/model | +| `level-heated [--bed C]` | `AUTO_BED_MESH_CALIBRATE TEMP=C` (heats the bed to 55 by default, soaks 2 min, meshes) | Snapmaker U1 | +| `clean-nozzle` | `ROUGHLY_CLEAN_NOZZLE` | Snapmaker U1 | +| `park [T1..T4]`, `idle-position [T1..T4]` | `PARK_EXTRUDER[n]`, `MOVE_TO_XY_IDLE_POSITION_EXTRUDER[n]` | Snapmaker U1 | +| `purifier idle\|cool\|preheat\|hot` | `SET_PURIFIER_MODE MODE=0..3` | Snapmaker U1 | + +U1 facts come from the firmware config of [Snapmaker/u1-klipper](https://github.com/Snapmaker/u1-klipper) +(`lava/fluidd.cfg`, `lava/printer.cfg`, `klippy/extras/filament_feed.py`, `purifier.py`, `kinematics/extruder.py`); the +K1 SE ones from [CrealityOfficial/K1_Series_Klipper](https://github.com/CrealityOfficial/K1_Series_Klipper) +`config/K1_SE_CR4CU220812S12/gcode_macro.cfg`. None of them was run on a real printer by the author: the tests use +the loopback fake Moonraker only. Head arguments are the fleet labels (`T1`..), never a bare number, and an unusable +head (fleet `usable: false`) is refused with its reason. + +Models without an adapter here (Bambu Lab, Prusa) have no aliases: they are `offline` in the fleet and every command above +refuses them; a Bambu/Prusa adapter would bring its own `adapters:` layer in the table. + ## Adapters | Adapter | Printers | Transport | diff --git a/lib/commands/printer.py b/lib/commands/printer.py index 758076a4..47eba167 100644 --- a/lib/commands/printer.py +++ b/lib/commands/printer.py @@ -1,4 +1,4 @@ -"""3d printer — drive configured physical printers: list, status, upload, start, cancel, watch, vision, bed, subscribe, agent, todo. +"""3d printer — drive configured physical printers: list, status, upload, start, slot, cancel, macro, aliases, watch, vision, bed, subscribe, agent, todo. WHAT: talks to the printers named in printer-devices.yaml through a protocol adapter — `moonraker` (Klipper/Moonraker, Snapmaker U1) or `creality-stock` (Creality K1 family on @@ -20,6 +20,9 @@ never reaches a printer screen without its preview; - a failed/timed-out start is reported as UNKNOWN and never resent automatically. +MACROS AND ALIASES (`-n NAME macro list|run`, `aliases`, `home`, `load`, ...): printer_ops/macro_cli.py and alias_cli.py; + a macro is a checked passthrough, an alias a per-adapter/model/printer table entry (issue #718). + INVARIANTS: stdlib-only at import time (registry contract); printer_ops is imported lazily. """ from __future__ import annotations @@ -137,8 +140,35 @@ maintenance, repairs, manual tasks (3d-api; offline a JSONL buffer — see `3d printer todo --help`). A printer's problems are tasks of this list, not GitHub tickets. - -n, --name NAME the printer for `todo`/`issues`: - `3d printer -n u1 issues list` + macro list|run MACRO [KEY=VALUE ...] [--yes] [--dry-run] [--offline] + the macros the printer defines / run ONE macro (Klipper: any + gcode_macro or extended command) with typed KEY=VALUE params; + the exact line sent is printed. Names are identifiers, values + one safe token; M112/EMERGENCY_STOP never; read-only/offline + printers refuse; a macro outside the printer's trusted_macros + and dangerous ones need --yes; motion/heating ones need an + idle printer (PAUSE/RESUME/CANCEL_PRINT style do not). + There is no raw `gcode` passthrough (see docs/commands/printer.md) + (see `3d printer macro --help`) + aliases [--json] what each alias below sends on this printer (-n NAME) or per + adapter/model; `3d printer ALIAS --help` for one + -n, --name NAME the printer for `todo`/`issues`, `macro`, `aliases` and the aliases: + `3d printer -n u1 issues list`, `3d printer -n u1 home` + (own aliases per printer: ~/.config/3d-cli/printer-aliases.yaml, ./printer-aliases.yaml) + +Aliases (thin subcommands, need -n NAME; each maps per adapter/model to real macros and says so +when a model has none: `3d printer -n NAME ALIAS [args] [--yes] [--dry-run] [--offline]`): + home [XYZ] home all axes, or only X/Y/Z (G28) + level bed mesh (BED_MESH_CALIBRATE; Creality K1: stock G29) + qgl | z-tilt quad gantry level / Z tilt adjust (printers that define them) + pause | resume PAUSE / RESUME (allowed while a job runs) + preheat MATERIAL | --nozzle C --bed C [--head T2] + PLA|PETG|ABS|ASA|TPU typical temperatures (SET_HEATER_TEMPERATURE) + cooldown TURN_OFF_HEATERS + load T1..T4 | unload T1..T4 filament in/out: U1 AUTO_FEEDING per head; K1 LOAD_MATERIAL / + QUIT_MATERIAL (no head); other Klipper LOAD_FILAMENT / UNLOAD_FILAMENT + Snapmaker U1 only: level-heated [--bed C], clean-nozzle, park [T1..T4], idle-position [T1..T4], + purifier idle|cool|preheat|hot Options: --json list/status as JSON (status: normalized state + raw firmware payload) @@ -203,6 +233,11 @@ 3d printer todo list u1 3d printer -n k1se issues add --title "X belt squeaks" --type inspect""" +# Subcommands that take the printer as a positional NAME: `-n` is refused for them. +_NAME_AS_POSITIONAL = frozenset({ + "list", "status", "upload", "start", "slot", "cancel", "evidence", "watch", "supervise", "vision", "guard", "bed", + "subscribe", "unsubscribe", "subscriptions", "agent", "doctor", +}) _VALUE = {"--secs", "--as", "--source-3mf", "--bridge-max", "--colour-delta-e", "--fix-max-delta-e"} _FLAGS = {"--json", "--yes", "--force", "--no-bed-move", "--fix-colour", "--watch"} @@ -744,6 +779,30 @@ def _run_todo(rest: list[str], name: str | None) -> int: return todo(rest, printer=name) +def _run_macro(rest: list[str], name: str | None) -> int: + from printer_ops.macro_cli import run_macro + + return run_macro(name, rest) + + +def _run_aliases(rest: list[str], name: str | None) -> int: + from printer_ops.alias_cli import run_aliases + + return run_aliases(name, rest) + + +def _run_alias(sub: str, rest: list[str], name: str | None) -> int: + """A subcommand that is none of the above: an alias of printer NAME (home, load, ...), else unknown.""" + from printer_ops.alias_cli import run_alias + + aliased = run_alias(name, sub, rest) + if aliased is not None: + return aliased + raise UsageError(f"unknown subcommand '{sub}'", command="printer", + remediation=["Use list, status, upload, start, slot, cancel, evidence, watch, supervise, vision, guard, bed, subscribe, unsubscribe, " + "subscriptions, agent, doctor, todo (issues), macro, aliases or an alias (home, level, load, ...) — see `3d printer --help`."]) + + def _run_subscriptions(sub: str) -> Callable[[list[str], str | None], int]: """Handler for subscribe/unsubscribe/subscriptions: the module takes the subcommand word too.""" def handler(rest: list[str], name: str | None) -> int: @@ -774,6 +833,8 @@ def handler(rest: list[str], name: str | None) -> int: "doctor": _run_doctor, "todo": _run_todo, "issues": _run_todo, + "macro": _run_macro, + "aliases": _run_aliases, } @@ -789,14 +850,12 @@ def run(argv: list[str]) -> int: if sub in ("-h", "--help", "help"): print(USAGE) return 0 - if name is not None and sub not in ("todo", "issues"): - raise UsageError(f"-n/--name selects the printer for `todo`/`issues`; `{sub}` takes the printer as NAME", + if name is not None and sub in _NAME_AS_POSITIONAL: + raise UsageError(f"-n/--name selects the printer for `todo`/`issues`, `macro`, `aliases` and the aliases; `{sub}` takes the printer as NAME", command="printer", remediation=[f"3d printer {sub} {name} …"]) handler = _SUBCOMMANDS.get(sub) if handler is None: - raise UsageError(f"unknown subcommand '{sub}'", command="printer", - 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 _run_alias(sub, rest, name) return handler(rest, name) diff --git a/lib/data/printer-aliases.yaml b/lib/data/printer-aliases.yaml new file mode 100644 index 00000000..3b44aacb --- /dev/null +++ b/lib/data/printer-aliases.yaml @@ -0,0 +1,183 @@ +# printer-aliases.yaml — the `3d printer -n NAME ` shortcuts, per adapter and per printer model. +# +# Read by lib/printer_ops/aliases.py (schema checked at load). Issue #718. An alias is a name for +# G-code lines that THIS kind of printer understands, so what exists differs per firmware: +# a missing alias is reported as "not defined for ", never guessed. +# +# LAYERS (later wins per alias name, whole entry replaced): +# 1. adapters. the protocol's own commands (stock Klipper for `moonraker`) +# 2. models[].aliases every `contains` (case-insensitive, in the printer's `model:`) that matches +# 3. the printer's own entries in ~/.config/3d-cli/printer-aliases.yaml, then ./printer-aliases.yaml: +# `NAME: {alias: entry, ...}` (same entry schema; the project file wins per alias) +# An entry `unsupported: ` (or `null` in a printer's own entries) removes an alias from that +# layer on, with the reason shown to the user. +# +# ENTRY: +# summary one line for `--help` +# kind inert | job | heating | motion | dangerous (lib/printer_ops/macros.py: job = allowed while +# a job runs; heating/motion/dangerous need an idle printer; dangerous also needs --yes) +# yes true: the alias needs --yes (extra confirmation) +# timeout seconds Moonraker may take to answer (default 120) +# args ordered mapping, filled from positional words, `--name VALUE` or `NAME=VALUE`: +# type axes X, Y, Z letters, any order -> "X Y" +# type head T1..Tn (the fleet heads); also {name}_extruder, {name}_suffix ("" for T1, "1" for T2) +# type int | float min / max +# type choice choices: {name: {placeholder: value}}; the extra values fill unset placeholders +# optional: true | default: VALUE; a line whose placeholder has no value is left out +# gcode lines sent as ONE script; {arg} placeholders are replaced by validated values only +# +# Sources of the printer-specific entries (nothing was run on a real printer): +# Snapmaker U1: https://github.com/Snapmaker/u1-klipper lava/fluidd.cfg, lava/printer.cfg, +# klippy/extras/filament_feed.py, klippy/extras/purifier.py, klippy/kinematics/extruder.py +# Creality K1 SE: https://github.com/CrealityOfficial/K1_Series_Klipper +# config/K1_SE_CR4CU220812S12/gcode_macro.cfg +# Stock Klipper: https://www.klipper3d.org/G-Codes.html + +adapters: + moonraker: + home: + summary: "home all axes (G28), or only the given ones - `home X Y`" + kind: motion + timeout: 300 + args: + axes: {type: axes, optional: true, default: ""} + gcode: ["G28 {axes}"] + level: + summary: "bed mesh (BED_MESH_CALIBRATE)" + kind: motion + timeout: 600 + gcode: [BED_MESH_CALIBRATE] + qgl: + summary: "quad gantry level (QUAD_GANTRY_LEVEL), only on printers that define it" + kind: motion + timeout: 300 + gcode: [QUAD_GANTRY_LEVEL] + z-tilt: + summary: "Z tilt adjust (Z_TILT_ADJUST), only on printers that define it" + kind: motion + timeout: 300 + gcode: [Z_TILT_ADJUST] + pause: + summary: "pause the running print (PAUSE)" + kind: job + gcode: [PAUSE] + resume: + summary: "resume a paused print (RESUME)" + kind: job + gcode: [RESUME] + preheat: + summary: "heat to a material's typical temperatures, or `--nozzle C` / `--bed C` (`--head T2`)" + kind: heating + args: + material: + type: choice + optional: true + choices: + PLA: {nozzle: 210, bed: 60} + PETG: {nozzle: 235, bed: 80} + ABS: {nozzle: 250, bed: 100} + ASA: {nozzle: 255, bed: 100} + TPU: {nozzle: 225, bed: 50} + nozzle: {type: float, min: 0, max: 300, optional: true} + bed: {type: float, min: 0, max: 120, optional: true} + head: {type: head, default: T1} + gcode: + - SET_HEATER_TEMPERATURE HEATER={head_extruder} TARGET={nozzle} + - SET_HEATER_TEMPERATURE HEATER=heater_bed TARGET={bed} + cooldown: + summary: "switch every heater off (TURN_OFF_HEATERS)" + kind: heating + gcode: [TURN_OFF_HEATERS] + load: + summary: "load filament with the printer's LOAD_FILAMENT macro (a Fluidd/Mainsail convention, not core Klipper)" + kind: motion + timeout: 300 + gcode: [LOAD_FILAMENT] + unload: + summary: "unload filament with the printer's UNLOAD_FILAMENT macro (a Fluidd/Mainsail convention, not core Klipper)" + kind: motion + timeout: 300 + gcode: [UNLOAD_FILAMENT] + +models: + - contains: snapmaker u1 + aliases: + level-heated: + summary: "heat the bed, soak, then bed mesh (AUTO_BED_MESH_CALIBRATE; `--bed C`, default 55; waits 2 min)" + kind: motion + timeout: 900 + args: + bed: {type: float, min: 30, max: 100, default: 55} + gcode: ["AUTO_BED_MESH_CALIBRATE TEMP={bed}"] + clean-nozzle: + summary: "rough nozzle clean at the discard position (ROUGHLY_CLEAN_NOZZLE)" + kind: motion + timeout: 300 + gcode: [ROUGHLY_CLEAN_NOZZLE] + load: + summary: "feed the filament of head T1..T4 (AUTO_FEEDING EXTRUDER=n LOAD=1)" + kind: motion + timeout: 600 + args: + slot: {type: head} + gcode: ["AUTO_FEEDING EXTRUDER={slot} LOAD=1"] + unload: + summary: "unload the filament of head T1..T4 (AUTO_FEEDING UNLOAD=1, stages prepare then doing)" + kind: motion + timeout: 600 + args: + slot: {type: head} + gcode: + - AUTO_FEEDING EXTRUDER={slot} UNLOAD=1 STAGE=prepare + - AUTO_FEEDING EXTRUDER={slot} UNLOAD=1 STAGE=doing + park: + summary: "park a head in its dock (PARK_EXTRUDER, PARK_EXTRUDER1..; default T1)" + kind: motion + timeout: 120 + args: + head: {type: head, default: T1} + gcode: ["PARK_EXTRUDER{head_suffix}"] + idle-position: + summary: "move a head to the idle XY position (MOVE_TO_XY_IDLE_POSITION_EXTRUDER; default T1)" + kind: motion + timeout: 120 + args: + head: {type: head, default: T1} + gcode: ["MOVE_TO_XY_IDLE_POSITION_EXTRUDER{head_suffix}"] + purifier: + summary: "chamber purifier mode idle|cool|preheat|hot (SET_PURIFIER_MODE MODE=0..3)" + kind: heating + args: + mode: + type: choice + choices: + idle: {code: 0} + cool: {code: 1} + preheat: {code: 2} + hot: {code: 3} + gcode: ["SET_PURIFIER_MODE MODE={code}"] + qgl: + unsupported: "the U1 printer.cfg (Snapmaker/u1-klipper lava/printer.cfg) defines no quad_gantry_level" + z-tilt: + unsupported: "the U1 printer.cfg (Snapmaker/u1-klipper lava/printer.cfg) defines no z_tilt" + - contains: creality k1 + aliases: + level: + summary: "heat and bed mesh with the stock G29 macro (it heats the bed and nozzle itself)" + kind: motion + timeout: 900 + gcode: [G29] + load: + summary: "load filament (stock LOAD_MATERIAL: heats to the default temperature, feeds 80 mm)" + kind: motion + timeout: 300 + gcode: [LOAD_MATERIAL] + unload: + summary: "unload filament (stock QUIT_MATERIAL: heats, purges and retracts)" + kind: motion + timeout: 300 + gcode: [QUIT_MATERIAL] + qgl: + unsupported: "the K1 SE stock printer.cfg defines no quad_gantry_level (bed mesh only: `level`)" + z-tilt: + unsupported: "the K1 SE stock printer.cfg defines no z_tilt (bed mesh only: `level`)" diff --git a/lib/printer_ops/alias_cli.py b/lib/printer_ops/alias_cli.py new file mode 100644 index 00000000..a3aebb01 --- /dev/null +++ b/lib/printer_ops/alias_cli.py @@ -0,0 +1,230 @@ +"""alias_cli.py — `3d printer -n NAME aliases` and the alias subcommands (home, level, load, preheat, park, ...). + +WHAT: the CLI side of printer_ops/aliases.py: lists what each alias sends on a printer (or per adapter/model), + and runs one alias through the same gates as `macro run` (printer_ops/macro_cli.py `execute`: typed values, + `access: full`, a G-code channel, defined on the printer, `--yes`, an idle printer; `--dry-run`, `--offline`). + +WHY: an alias a printer model lacks must say so and list what it has — never guess a macro name — and every alias + must show the exact G-code it sends. Issue https://github.com/alex-mextner/3d-cli/issues/718. + +INVARIANTS: stdlib-only at import time; imported lazily by `commands/printer.py`. +""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING, Any + +from errors import UsageError +from printer_ops import aliases, macros +from printer_ops.macro_cli import ( + FLAGS, + VALUE_OPTS, + adapter_of, + device_of, + execute, + heads_of, + require_full_access, + split_opts, + timeout_of, +) + +if TYPE_CHECKING: + from printer_ops.devices import Device + +ALIASES_USAGE = """3d printer [-n NAME] aliases [--json] + With -n NAME: the aliases that printer has (adapter + model + its own printer-aliases.yaml entries), what each + sends, and which are not supported on it and why. Without -n: the whole built-in table + (lib/data/printer-aliases.yaml).""" + + +def _wants_help(argv: list[str]) -> bool: + return any(a in ("-h", "--help") for a in argv) + + +def _alias_line(spec: aliases.AliasSpec) -> str: + args = " ".join(f"[{a.name}]" if a.optional else f"<{a.name}>" for a in spec.args) + return f"{spec.name} {args}".rstrip() + + +def _arg_line(a: aliases.ArgSpec) -> str: + choices = f" {', '.join(a.choices)}" if a.choices else "" + limits = ( + f" {a.minimum:g}..{a.maximum:g}" + if a.minimum is not None and a.maximum is not None + else "" + ) + default = f" (default {a.default})" if a.default else "" + return f" {a.name}: {a.type}{choices}{limits}{default}" + + +def _detail(spec: aliases.AliasSpec) -> str: + if spec.unsupported is not None: + return f" {spec.name}: NOT supported - {spec.unsupported}" + confirm = ", needs --yes" if spec.yes or spec.kind == macros.DANGEROUS else "" + return "\n".join( + [ + f" {_alias_line(spec)}", + f" {spec.summary}", + f" sends: {'; '.join(spec.gcode)} [{spec.kind}{confirm}] (from {spec.source})", + *(_arg_line(a) for a in spec.args), + ] + ) + + +def _as_json(spec: aliases.AliasSpec) -> dict[str, Any]: + return { + "summary": spec.summary, + "kind": spec.kind, + "gcode": list(spec.gcode), + "source": spec.source, + "unsupported": spec.unsupported, + } + + +def _list_builtin(as_json: bool) -> int: + layers = aliases.builtin_layers() + if as_json: + print( + json.dumps( + {label: {n: _as_json(s) for n, s in e.items()} for label, e in layers}, + indent=2, + ) + ) + return 0 + for label, entries in layers: + print(f"{label}:") + for spec in entries.values(): + print(_detail(spec)) + return 0 + + +def _list_printer(name: str, as_json: bool) -> int: + device = device_of(name, "aliases") + table = aliases.resolve( + device.adapter, device.model, aliases.overrides_for(device.name) + ) + if as_json: + print(json.dumps({n: _as_json(s) for n, s in sorted(table.items())}, indent=2)) + return 0 + print( + f"{device.name} ({device.model or device.adapter}, adapter {device.adapter}):" + ) + ordered = sorted( + table.values(), key=lambda s: s.unsupported is not None + ) # the usable ones first + for spec in ordered: + print(_detail(spec)) + return 0 + + +def run_aliases(name: str | None, argv: list[str]) -> int: + """`aliases [--json]`: the aliases of printer NAME (resolved), or the whole built-in table.""" + if _wants_help(argv): + print(ALIASES_USAGE) + return 0 + words, opts = split_opts(argv, frozenset({"--json"})) + if words: + raise UsageError( + f"`aliases` takes no arguments, got {words}", command="printer" + ) + return ( + _list_builtin("--json" in opts) + if name is None + else _list_printer(name, "--json" in opts) + ) + + +def _print_builtin_help(sub: str) -> int: + for label, entries in aliases.builtin_layers(): + if sub in entries: + print(f"{label}:") + print(_detail(entries[sub])) + return 0 + + +def _no_printer(sub: str, argv: list[str]) -> int | None: + """An alias word without -n NAME: its built-in help, or a usage error; None when `sub` is no alias at all.""" + if sub not in aliases.builtin_names(): + return None + if _wants_help(argv): + return _print_builtin_help(sub) + raise UsageError( + f"`{sub}` is an alias and needs the printer: 3d printer -n NAME {sub} ...", + command="printer", + remediation=["3d printer aliases # what each alias sends, per model"], + ) + + +def _unknown_alias( + name: str, + sub: str, + model: str | None, + adapter: str, + table: dict[str, aliases.AliasSpec], +) -> UsageError: + return UsageError( + f"alias `{sub}` is not defined for {name} ({model or 'no model'}, adapter {adapter})", + command="printer", + remediation=[ + f"Defined for it: {', '.join(aliases.supported(table)) or 'none'}", + f"Run a macro it defines: `3d printer -n {name} macro list`", + f"Or add `{name}: {{{sub}: {{summary: ..., kind: motion, gcode: [...]}}}}` to " + f"./{aliases.OVERRIDES_FILENAME} (or the user config dir's)", + ], + ) + + +def _lookup(device: Device, sub: str, argv: list[str]) -> aliases.AliasSpec | None: + """The alias to run; None when `--help` was printed instead. Raises for what the printer cannot do.""" + helping = _wants_help(argv) + if not helping: + require_full_access(device, f"alias {sub}") + adapter_of( + device + ) # an adapter without a G-code channel says so before any "alias not defined" + table = aliases.resolve( + device.adapter, device.model, aliases.overrides_for(device.name) + ) + spec = table.get(sub) + if spec is None: + raise _unknown_alias(device.name, sub, device.model, device.adapter, table) + if spec.unsupported is not None: + raise UsageError( + f"alias `{sub}` is not supported on {device.name} ({device.model or device.adapter}): " + f"{spec.unsupported}", + command="printer", + remediation=[ + f"Defined for it: {', '.join(aliases.supported(table)) or 'none'}" + ], + ) + if helping: + print( + f"3d printer -n {device.name} {_alias_line(spec)} [--yes] [--dry-run] [--offline] [--timeout S]" + ) + print(_detail(spec)) + return None + return spec + + +def run_alias(name: str | None, sub: str, argv: list[str]) -> int | None: + """Subcommand `sub` as an alias of printer `name`; None when `sub` is no alias at all (the caller reports it).""" + if name is None: + return _no_printer(sub, argv) + device = device_of(name, sub) + spec = _lookup(device, sub, argv) + if spec is None: + return 0 + words, opts = split_opts(argv, FLAGS, VALUE_OPTS) + bound = aliases.bind(spec, words, heads_of(device)) + return execute( + device, + list(bound.lines), + what=f"alias {sub}", + again=" ".join([sub, *words]), + kind=spec.kind, + curated=True, + entry_yes=spec.yes, + timeout=timeout_of(opts, spec.timeout), + opts=opts, + ) diff --git a/lib/printer_ops/aliases.py b/lib/printer_ops/aliases.py new file mode 100644 index 00000000..d206537c --- /dev/null +++ b/lib/printer_ops/aliases.py @@ -0,0 +1,529 @@ +"""aliases.py — the `3d printer -n NAME ` registry: per adapter, per model, per printer. + +WHAT: loads lib/data/printer-aliases.yaml (the built-in table), merges the layers adapter -> + matching models -> the printer's own aliases (`printer-aliases.yaml`, user then project), and + binds the words a user typed to the alias's `args`, producing the G-code lines to send. + +WHY: "home", "load filament", "park" are the same intent on every printer but a different + command on each firmware (Klipper `G28`, Snapmaker U1 `AUTO_FEEDING EXTRUDER=n LOAD=1`, + Creality K1 `LOAD_MATERIAL`). The mapping is data, overridable per printer, and an alias a + model does not have is reported as such — never guessed. The table schema is in the data + file's header. Issue https://github.com/alex-mextner/3d-cli/issues/718. + +PER-PRINTER FILES (later wins per alias name, whole entry replaced): + ~/.config/3d-cli/printer-aliases.yaml, then ./printer-aliases.yaml: + u1: # the printer's fleet / printer-devices.yaml name + home: {summary: ..., kind: motion, gcode: ["G28 X Y"]} + park: null # removed for this printer + cooldown: {unsupported: "why"} # removed, reason shown + +INVARIANTS: stdlib + lazy pyyaml. Every value reaching a G-code line went through a typed + check here (axes letters, a fleet head label, a bounded number, a listed choice) or the + table itself; there is no free-text argument. +""" + +from __future__ import annotations + +import functools +import math +import re +from collections.abc import Mapping +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Final + +from errors import InvalidArgument, MissingDependency, ThreeDError, UsageError +from printer_ops import macros + +DATA_FILE: Final = ( + Path(__file__).resolve().parent.parent / "data" / "printer-aliases.yaml" +) +OVERRIDES_FILENAME: Final = "printer-aliases.yaml" +TYPES: Final = ("axes", "head", "int", "float", "choice") +_ENTRY_KEYS: Final = frozenset( + {"summary", "kind", "yes", "timeout", "args", "gcode", "unsupported"} +) +_ARG_KEYS: Final = frozenset({"type", "optional", "default", "min", "max", "choices"}) +_ALIAS_NAME: Final = re.compile(r"[a-z][a-z0-9-]*") +_PLACEHOLDER: Final = re.compile(r"\{([A-Za-z_][A-Za-z0-9_]*)\}") +_HEAD_LABEL: Final = re.compile(r"T[0-9]+", re.IGNORECASE) + + +class AliasConfigError(ThreeDError): + """The alias table (built-in or a printer's own file) breaks its schema. Exit 2.""" + + exit_code = 2 + + +@dataclass(frozen=True) +class HeadInfo: + index: int + extruder: str # Klipper object name: extruder, extruder1, ... + usable: bool + reason: str | None + + +@dataclass(frozen=True) +class ArgSpec: + name: str + type: str + optional: bool = False + default: str | None = None + minimum: float | None = None + maximum: float | None = None + choices: dict[str, dict[str, str]] = field(default_factory=dict, hash=False) + + +@dataclass(frozen=True) +class AliasSpec: + name: str + summary: str + kind: str + gcode: tuple[str, ...] + source: str # which layer defined it, for messages + args: tuple[ArgSpec, ...] = () + yes: bool = False + timeout: float = 120.0 + unsupported: str | None = None # set: the alias does NOT exist here; this is why + + +def _yaml() -> Any: + try: + import yaml + except ImportError as exc: # pragma: no cover - pyyaml is a core dependency + raise MissingDependency( + "pyyaml", + install="uv sync", + degrades="printer aliases cannot be read", + command="printer", + ) from exc + return yaml + + +def _bad(where: str, why: str) -> AliasConfigError: + return AliasConfigError( + f"{where}: {why}", + command="printer", + remediation=[f"Schema: the header of {DATA_FILE}"], + ) + + +def _number(value: Any, where: str, key: str) -> float: + if ( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not math.isfinite(value) + ): + raise _bad(where, f"`{key}` must be a number, got {value!r}") + return float(value) + + +def _text(value: Any) -> str: + """A scalar of the YAML as the text a G-code line gets (210.0 -> 210).""" + return f"{value:g}" if isinstance(value, float) else str(value) + + +# ---- parsing one entry ---------------------------------------------------------------------- + + +def _choices(spec: dict[str, Any], where: str) -> dict[str, dict[str, str]]: + if spec["type"] != "choice": + if "choices" in spec: + raise _bad(where, "`choices` is for type choice only") + return {} + raw = spec.get("choices") + if not isinstance(raw, dict) or not raw: + raise _bad(where, "a choice needs a non-empty `choices` mapping") + out: dict[str, dict[str, str]] = {} + for label, fills in raw.items(): + if not isinstance(fills, dict): + raise _bad(where, f"choice {label!r} must map placeholder -> value") + out[str(label)] = {str(k): _text(v) for k, v in fills.items()} + return out + + +def _arg(name: str, spec: Any, where: str) -> ArgSpec: + where = f"{where} arg {name}" + if ( + not isinstance(spec, dict) + or set(spec) - _ARG_KEYS + or spec.get("type") not in TYPES + ): + raise _bad( + where, + f"must be a mapping with type in {', '.join(TYPES)} and keys {', '.join(sorted(_ARG_KEYS))}", + ) + lo = _number(spec["min"], where, "min") if "min" in spec else None + hi = _number(spec["max"], where, "max") if "max" in spec else None + default = _text(spec["default"]) if "default" in spec else None + optional = bool(spec.get("optional", False)) or default is not None + return ArgSpec(name, spec["type"], optional, default, lo, hi, _choices(spec, where)) + + +def _placeholders(arg: ArgSpec) -> set[str]: + if arg.type == "head": + return {arg.name, f"{arg.name}_extruder", f"{arg.name}_suffix"} + if arg.type == "choice": + return {arg.name, *(k for fills in arg.choices.values() for k in fills)} + return {arg.name} + + +def _removed(name: str, spec: Any, source: str, where: str) -> AliasSpec | None: + """`None` / `{unsupported: why}`: the alias is removed from this layer on; None when `spec` is a real entry.""" + if spec is None: + return AliasSpec( + name, "", macros.INERT, (), source, unsupported="disabled for this printer" + ) + if not isinstance(spec, dict) or set(spec) - _ENTRY_KEYS: + raise _bad( + where, f"must be a mapping with keys {', '.join(sorted(_ENTRY_KEYS))}" + ) + if "unsupported" not in spec: + return None + reason = spec["unsupported"] + if len(spec) != 1 or not isinstance(reason, str) or not reason.strip(): + raise _bad(where, "`unsupported` takes a reason and no other key") + return AliasSpec(name, "", macros.INERT, (), source, unsupported=reason.strip()) + + +def _gcode_lines( + spec: dict[str, Any], args: tuple[ArgSpec, ...], where: str +) -> tuple[str, ...]: + gcode = spec.get("gcode") + if ( + not isinstance(gcode, list) + or not gcode + or not all(isinstance(g, str) and g.strip() for g in gcode) + ): + raise _bad(where, "`gcode` must be a non-empty list of lines") + known = {p for a in args for p in _placeholders(a)} + for line in gcode: + macros.refuse_denied(line, f"{where} gcode") + unknown = set(_PLACEHOLDER.findall(line)) - known + if unknown: + raise _bad( + where, + f"gcode line {line!r} uses {', '.join(sorted(unknown))}, which no arg provides", + ) + if re.search(r"[;\n\r]", line): + raise _bad(where, f"gcode line {line!r} must be one command without `;`") + return tuple(g.strip() for g in gcode) + + +def parse_entry(name: str, spec: Any, source: str) -> AliasSpec: + """One alias entry (`None` or `{unsupported: why}` = removed) -> AliasSpec; every schema problem is reported.""" + where = f"{source} alias {name}" + if not _ALIAS_NAME.fullmatch(name): + raise _bad(where, "an alias name is lower-case letters, digits and -") + removed = _removed(name, spec, source, where) + if removed is not None: + return removed + if spec.get("kind") not in macros.KINDS: + raise _bad(where, f"`kind` must be one of {', '.join(macros.KINDS)}") + summary = spec.get("summary") + if not isinstance(summary, str) or not summary.strip(): + raise _bad(where, "`summary` is required (it is the --help text)") + raw_args = spec.get("args") or {} + if not isinstance(raw_args, dict): + raise _bad(where, "`args` must be a mapping") + args = tuple(_arg(str(k), v, where) for k, v in raw_args.items()) + timeout = _number(spec["timeout"], where, "timeout") if "timeout" in spec else 120.0 + if not 0 < timeout <= 3600: + raise _bad(where, "`timeout` is seconds, 1-3600") + return AliasSpec( + name, + summary.strip(), + str(spec["kind"]), + _gcode_lines(spec, args, where), + source, + args, + bool(spec.get("yes", False)), + timeout, + ) + + +def parse_entries(value: Any, source: str) -> dict[str, AliasSpec]: + if value is None: + return {} + if not isinstance(value, dict): + raise _bad( + source, + "must be a mapping of alias name -> entry (schema: lib/data/printer-aliases.yaml)", + ) + return {str(k): parse_entry(str(k), v, source) for k, v in value.items()} + + +# ---- the layers ----------------------------------------------------------------------------- + + +def _model_layer(item: Any) -> tuple[str, dict[str, AliasSpec]]: + if ( + not isinstance(item, dict) + or set(item) != {"contains", "aliases"} + or not isinstance(item["contains"], str) + ): + raise _bad(str(DATA_FILE), "each model is {contains: TEXT, aliases: {...}}") + return item["contains"].lower(), parse_entries( + item["aliases"], f"model {item['contains']}" + ) + + +@functools.cache +def _table() -> tuple[ + dict[str, dict[str, AliasSpec]], tuple[tuple[str, dict[str, AliasSpec]], ...] +]: + data = _yaml().safe_load(DATA_FILE.read_text(encoding="utf-8")) + if not isinstance(data, dict) or set(data) - {"adapters", "models"}: + raise _bad(str(DATA_FILE), "top level must have `adapters` and `models` only") + adapters = { + str(a): parse_entries(entries, f"adapter {a}") + for a, entries in (data.get("adapters") or {}).items() + } + return adapters, tuple(_model_layer(item) for item in data.get("models") or []) + + +def builtin_names() -> set[str]: + """Every alias name any built-in layer defines (they become `3d printer` subcommands).""" + return {n for _, entries in builtin_layers() for n in entries} + + +def builtin_layers() -> list[tuple[str, dict[str, AliasSpec]]]: + """(layer label, entries) of the whole built-in table, adapters first.""" + adapters, models = _table() + return [ + *((f"adapter {a}", e) for a, e in adapters.items()), + *((f"model {c}", e) for c, e in models), + ] + + +def overrides_for(name: str, cwd: Path | None = None) -> dict[str, AliasSpec]: + """The aliases printer `name` has of its own: user file, then project file (later wins per alias).""" + from cli.paths import config_dir + + merged: dict[str, AliasSpec] = {} + for path in ( + config_dir() / OVERRIDES_FILENAME, + (cwd or Path.cwd()) / OVERRIDES_FILENAME, + ): + if not path.is_file(): + continue + try: + data = _yaml().safe_load(path.read_text(encoding="utf-8")) + except _yaml().YAMLError as exc: + raise AliasConfigError( + f"{path}: invalid YAML: {exc}", command="printer" + ) from exc + if data is not None and not isinstance(data, dict): + raise _bad(str(path), "top level must map a printer name -> its aliases") + merged.update(parse_entries((data or {}).get(name), f"{path} {name}")) + return merged + + +def resolve( + adapter: str, model: str | None, overrides: Mapping[str, AliasSpec] | None = None +) -> dict[str, AliasSpec]: + """Aliases of one printer, unsupported ones included (their `unsupported` says why).""" + adapters, models = _table() + merged: dict[str, AliasSpec] = dict(adapters.get(adapter, {})) + text = (model or "").lower() + for contains, entries in models: + if contains in text: + merged.update(entries) + merged.update(overrides or {}) + return merged + + +def supported(table: Mapping[str, AliasSpec]) -> dict[str, AliasSpec]: + return {n: s for n, s in sorted(table.items()) if s.unsupported is None} + + +# ---- binding the words of a call ------------------------------------------------------------ + + +@dataclass(frozen=True) +class Bound: + lines: tuple[str, ...] + + +def _scan(words: list[str], names: list[str]) -> tuple[dict[str, str], list[str]]: + """Words -> (named values, positional words): `--name V`, `--name=V`, `NAME=V`; the rest is positional.""" + named: dict[str, str] = {} + positional: list[str] = [] + it = iter(words) + for w in it: + key, value = _named(w, names, it) + if key is None: + positional.append(w) + continue + if key not in names: + raise UsageError( + f"unknown argument {key!r}", + command="printer", + remediation=[f"Accepted: {', '.join(names) or 'none'}"], + ) + if key in named: + raise UsageError(f"argument {key!r} given twice", command="printer") + named[key] = value + return named, positional + + +def _named(word: str, names: list[str], rest: Any) -> tuple[str | None, str]: + """(argument name, value) when `word` names an argument (consuming the next word for `--name V`).""" + if word.startswith("--"): + key, sep, value = word[2:].partition("=") + if not sep: + value = next(rest, "") + if not value: + raise UsageError(f"{word} needs a value", command="printer") + return key.lower(), value + key, sep, value = word.partition("=") + return (key.lower(), value) if sep and key.lower() in names else (None, "") + + +def _split(words: list[str], spec: AliasSpec) -> dict[str, str]: + """Words -> {arg name: raw text}: named ones first, positional ones fill the rest in `args` order.""" + names = [a.name for a in spec.args] + out, positional = _scan(words, names) + free = [n for n in names if n not in out] + if len(positional) > len(free): + raise UsageError( + f"`{spec.name}` takes {len(names)} argument(s) ({', '.join(names) or 'none'}), " + f"got extra {positional[len(free) :]}", + command="printer", + ) + out.update(zip(free, positional, strict=False)) + return out + + +def _number_arg(arg: ArgSpec, text: str) -> str: + try: + v = float(text) if arg.type == "float" else float(int(text)) + except ValueError: + raise InvalidArgument( + arg.name, + text, + [f"a {'number' if arg.type == 'float' else 'whole number'}"], + command="printer", + ) from None + below = arg.minimum is not None and v < arg.minimum + above = arg.maximum is not None and v > arg.maximum + if not math.isfinite(v) or below or above: + lo = "-inf" if arg.minimum is None else f"{arg.minimum:g}" + hi = "inf" if arg.maximum is None else f"{arg.maximum:g}" + raise InvalidArgument(arg.name, text, [f"{lo} .. {hi}"], command="printer") + return f"{v:g}" + + +def _head_values( + arg: ArgSpec, text: str, heads: Mapping[str, HeadInfo], alias: str +) -> dict[str, str]: + label = text.upper() + if not _HEAD_LABEL.fullmatch(text): + raise InvalidArgument( + arg.name, + text, + ["a head label: " + ", ".join(sorted(heads))], + command="printer", + extra="Heads are named by their label (T1..), not by a 0- or 1-based number.", + ) + info = heads.get(label) + if info is None: + raise InvalidArgument( + arg.name, + text, + sorted(heads), + command="printer", + extra=f"`{alias}` knows only the heads listed for this printer in fleet/printers.yaml.", + ) + if not info.usable: + from printer_ops.base import PrinterRefused + + raise PrinterRefused( + f"head {label} is marked unusable: {info.reason or 'no reason recorded'}", + command="printer", + ) + suffix = str(info.index) if info.index else "" + return { + arg.name: str(info.index), + f"{arg.name}_extruder": info.extruder, + f"{arg.name}_suffix": suffix, + } + + +def _axes(text: str) -> str: + letters = list(text.upper()) + if ( + not letters + or any(c not in "XYZ" for c in letters) + or len(set(letters)) != len(letters) + ): + raise InvalidArgument( + "axes", + text, + ["X, Y, Z in any order, each once, e.g. XY"], + command="printer", + ) + return " ".join(letters) + + +def _typed( + arg: ArgSpec, text: str, heads: Mapping[str, HeadInfo], alias: str +) -> dict[str, str]: + """The placeholder values one argument contributes, every one checked against its type.""" + if arg.type == "axes": + return {arg.name: _axes(text) if text else ""} + if arg.type == "head": + return _head_values(arg, text, heads, alias) + if arg.type == "choice": + match = {k.upper(): k for k in arg.choices}.get(text.upper()) + if match is None: + raise InvalidArgument( + arg.name, text, sorted(arg.choices), command="printer" + ) + return {arg.name: match} + return {arg.name: _number_arg(arg, text)} + + +def _collect( + spec: AliasSpec, words: list[str], heads: Mapping[str, HeadInfo] +) -> dict[str, str]: + """Placeholder -> checked value for everything the call provided; a choice's fills only where unset.""" + raw = _split(words, spec) + values: dict[str, str] = {} + fills: dict[str, str] = {} + for arg in spec.args: + text = raw.get(arg.name, arg.default) + if text is None: + if not arg.optional: + raise UsageError( + f"`{spec.name}` needs {arg.name}", + command="printer", + remediation=[f"3d printer {spec.name} --help"], + ) + continue + values.update(_typed(arg, text, heads, spec.name)) + if arg.type == "choice": + fills.update(arg.choices[values[arg.name]]) + by_name = {a.name: a for a in spec.args} + for key, fill in fills.items(): + if key not in values: + values[key] = _number_arg(by_name[key], fill) if key in by_name else fill + return values + + +def bind(spec: AliasSpec, words: list[str], heads: Mapping[str, HeadInfo]) -> Bound: + """Typed values from `words` -> the alias's G-code lines. A line whose placeholder has no value is left out; + no line left is a usage error (e.g. `preheat` without a material, --nozzle or --bed).""" + values = _collect(spec, words, heads) + lines = [ + _PLACEHOLDER.sub(lambda m: values[m.group(1)], line).strip() + for line in spec.gcode + if set(_PLACEHOLDER.findall(line)) <= set(values) + ] + if not lines: + raise UsageError( + f"`{spec.name}` needs at least one of: {', '.join(a.name for a in spec.args)}", + command="printer", + remediation=[f"3d printer {spec.name} --help"], + ) + return Bound(tuple(lines)) diff --git a/lib/printer_ops/base.py b/lib/printer_ops/base.py index 9fae875d..120098e1 100644 --- a/lib/printer_ops/base.py +++ b/lib/printer_ops/base.py @@ -26,6 +26,7 @@ if TYPE_CHECKING: from printer_ops.devices import Device + from printer_ops.macros import MacroCatalog from urllib.parse import urlsplit from errors import InvalidArgument, ThreeDError @@ -252,6 +253,18 @@ def run_gcode(self, script: str, *, timeout: float = 120.0) -> None: once the printer ran it, waiting at most `timeout` seconds.""" raise self._unsupported("run_gcode") + def gcode_request(self, script: str) -> str: + """The request `run_gcode(script)` sends, as one line (`POST `), for `--dry-run`. Sends nothing.""" + raise self._unsupported("macros and aliases (no G-code channel)") + + def macro_catalog(self) -> MacroCatalog: + """The macros and extended commands the printer defines. Read-only.""" + raise self._unsupported("macros and aliases (no G-code channel)") + + def catalog_requests(self) -> list[str]: + """The requests `macro_catalog()` sends, one line each (`GET `), for `macro list --dry-run`.""" + raise self._unsupported("macros and aliases (no G-code channel)") + def set_head_filament(self, index: int, filament: HeadFilament, *, force_official: bool = False) -> HeadWrite: """Make head `index` show `filament` on the printer, verified by reading it back. diff --git a/lib/printer_ops/macro_cli.py b/lib/printer_ops/macro_cli.py new file mode 100644 index 00000000..f9f1e37a --- /dev/null +++ b/lib/printer_ops/macro_cli.py @@ -0,0 +1,368 @@ +"""macro_cli.py — `3d printer -n NAME macro list|run`, (the shared gates `execute`, `device_of`, `adapter_of`, `heads_of` also serve printer_ops/alias_cli.py). + +WHAT: the CLI side of printer_ops/macros.py (what may be sent) and printer_ops/aliases.py (the per-adapter, + per-model, per-printer shortcut table): discovery of the macros a Klipper printer defines, a checked + passthrough of one macro with `KEY=VALUE` parameters, and the aliases that expand to G-code lines. + +SAFETY, in the order it is enforced (nothing is sent before every step passed): + 1. names/values are typed tokens (macros.build_line, aliases.bind): no newline, `;`, quote, M112; + 2. the printer must be `access: full` (read-only and offline fleet printers refuse; offline are never contacted); + 3. the adapter must have a G-code channel (Moonraker; creality-stock has none); + 4. the command must be defined on the printer (`/printer/gcode/help`); a standard G/M code only through an alias; + 5. `--yes` for a macro outside `trusted_macros`, for dangerous ones and for aliases that ask for it; + 6. an idle printer for motion/heating/dangerous (state re-read here; pause/resume/cancel style is exempt). + `--dry-run` runs 1-6 (reads only) and prints the exact request instead of sending it; `--offline` contacts + nothing and prints the script from registry data only. + +INVARIANTS: stdlib-only at import time; printer_ops pieces are imported lazily by the caller. +Issue https://github.com/alex-mextner/3d-cli/issues/718. +""" + +from __future__ import annotations + +import difflib +import json +from typing import TYPE_CHECKING, Any + +from errors import InvalidArgument, UsageError +from printer_ops import aliases, macros +from printer_ops.base import BUSY_STATES, PrinterRefused + +if TYPE_CHECKING: + from printer_ops.base import Adapter + from printer_ops.devices import Device + +FLAGS = frozenset({"--yes", "--dry-run", "--offline", "--json", "--all"}) +VALUE_OPTS = frozenset({"--timeout"}) + +MACRO_USAGE = """3d printer -n NAME macro ... + list [--offline] [--dry-run] [--all] [--json] + the macros NAME defines (Moonraker: /printer/objects/list + /printer/gcode/help), + with trust (* = in its trusted_macros), risk class and description; `_private` + ones with --all. --offline: registry data only (trusted_macros + alias targets), + nothing is contacted. --dry-run: print the requests, send none. + run MACRO [KEY=VALUE ...] [--yes] [--dry-run] [--offline] [--timeout S] + run ONE macro: the line `MACRO KEY=VALUE ...` is printed exactly as sent. + Refused: names that are not identifiers, values with anything but A-Za-z0-9_.,:+/-, + M112/EMERGENCY_STOP anywhere, read-only/offline printers, G/M codes that the + printer does not define as macros, names the printer does not define. + --yes: needed for a macro outside the printer's trusted_macros and for dangerous + ones (FIRMWARE_RESTART, RESTART, SAVE_CONFIG, ...). Motion/heating/unknown macros + need an idle printer; PAUSE/RESUME/CANCEL_PRINT-style ones do not. + --dry-run: checks everything (reads only), prints the request, sends nothing. + +Examples: + 3d printer -n u1 macro list + 3d printer -n u1 macro list --offline --json + 3d printer -n u1 macro run SM_PRINT_AUTO_FEED EXTRUDER=1 --yes --dry-run + 3d printer -n k1se macro run PAUSE + 3d printer -n u1 macro run MOVE_TO_XY_IDLE_POSITION_EXTRUDER X_OFFSET=3.5 --yes""" + + +def split_opts( + argv: list[str], flags: frozenset[str], values: frozenset[str] = frozenset() +) -> tuple[list[str], dict[str, str]]: + """Pull known option words out of argv; everything else (macro/alias words, `--nozzle 200`) stays in order.""" + rest: list[str] = [] + opts: dict[str, str] = {} + i = 0 + while i < len(argv): + a = argv[i] + if a in flags: + opts[a] = "1" + elif a in values: + if i + 1 >= len(argv): + raise UsageError(f"{a} needs a value", command="printer") + opts[a] = argv[i + 1] + i += 1 + else: + rest.append(a) + i += 1 + return rest, opts + + +def timeout_of(opts: dict[str, str], default: float) -> float: + if "--timeout" not in opts: + return default + try: + value = float(opts["--timeout"]) + except ValueError: + value = -1.0 + if not 0 < value <= 3600: + raise InvalidArgument( + "--timeout", opts["--timeout"], ["seconds, 1-3600"], command="printer" + ) + return value + + +def device_of(name: str | None, sub: str) -> Device: + from printer_ops.devices import get_device + + if name is None: + raise UsageError( + f"`{sub}` needs the printer: 3d printer -n NAME {sub} ...", + command="printer", + remediation=["`3d printer list` shows the configured printers."], + ) + return get_device(name, offline_ok=True) + + +def heads_of(device: Device) -> dict[str, aliases.HeadInfo]: + """Heads of the printer by label (fleet/printers.yaml); a printer the fleet does not describe has T1 only.""" + from printer_ops.moonraker import extruder_name + from printops.fleet import load_fleet + + fleet = load_fleet() + entry = fleet.printers.get(device.name) if fleet is not None else None + if entry is None or not entry.heads: + return {"T1": aliases.HeadInfo(0, "extruder", True, None)} + return { + h.label.upper(): aliases.HeadInfo( + h.index, h.extruder or extruder_name(h.index), h.usable, h.reason + ) + for h in entry.heads + } + + +def adapter_of(device: Device) -> Adapter: + from printer_ops import make_adapter + + adapter = make_adapter(device) + adapter.gcode_request( + "" + ) # UnsupportedOperation when the adapter has no G-code channel (creality-stock) + return adapter + + +def _check_defined( + device: Device, lines: list[str], catalog: macros.MacroCatalog, *, curated: bool +) -> None: + for line in lines: + cmd = macros.command_of(line) + if cmd in device.trusted_macros: + continue + if macros.is_numeric_code(cmd): + if curated or cmd in catalog.macros: + continue + raise PrinterRefused( + f"{cmd} is a standard G-code, not a macro {device.name} defines: `macro run` sends macros only", + command="printer", + remediation=[ + "Use an alias for the usual ones (`3d printer aliases`), or define the macro in the " + "printer's config." + ], + ) + if not catalog.knows(cmd): + near = difflib.get_close_matches( + cmd, sorted(catalog.commands.keys() | catalog.macros), n=3 + ) + raise PrinterRefused( + f"{device.name} does not define {cmd}", + command="printer", + remediation=[ + f"Did you mean: {', '.join(near)}" + if near + else f"`3d printer -n {device.name} macro list` shows what it defines." + ], + ) + + +def require_full_access(device: Device, what: str) -> None: + """Only `access: full` printers take a write; read-only ones may still be read, offline ones are never contacted.""" + if device.access == "full": + return + hint = (f"Only read it: 3d printer status {device.name}" if device.access == "read-only" + else "3d never contacts an offline fleet printer; run its jobs by hand.") + raise PrinterRefused(f"{device.name} is {device.access} in {device.source}: {what} refused", + command="printer", remediation=[hint]) + + +def execute( + device: Device, + lines: list[str], + *, + what: str, + again: str, + kind: str | None, + curated: bool, + entry_yes: bool, + timeout: float, + opts: dict[str, str], +) -> int: + require_full_access(device, what) + verdict = macros.assess( + lines, device.trusted_macros, curated=curated, kind=kind, entry_yes=entry_yes + ) + script = "\n".join(lines) + adapter = adapter_of(device) + dry, offline = "--dry-run" in opts, "--offline" in opts + header = f"{device.name} ({device.adapter} {device.host}:{device.port}): {what}" + if not offline: + _check_defined(device, lines, adapter.macro_catalog(), curated=curated) + need_yes = bool(verdict.yes_reasons) and "--yes" not in opts + if need_yes and not (dry or offline): + raise UsageError( + f"{what} needs --yes: {'; '.join(verdict.yes_reasons)}", + command="printer", + remediation=[ + f"Confirm explicitly: 3d printer -n {device.name} {again} --yes" + ], + ) + if verdict.needs_idle and not offline: + st = adapter.status() + if st.state in BUSY_STATES: + raise PrinterRefused( + f"{device.name} is {st.state} [{st.raw_state}]: {what} is {verdict.kind} and needs an idle printer", + command="printer", + remediation=[ + "Wait for the job to end; pause/resume/cancel style macros (PAUSE, RESUME, CANCEL_PRINT) " + "are allowed while it runs." + ], + ) + print(header) + print("gcode:") + for line in lines: + print(f" {line}") + if need_yes: + print(f"would need --yes: {'; '.join(verdict.yes_reasons)}") + if offline: + print( + f"{adapter.gcode_request(script)}\noffline: nothing was contacted; the printer's idle state and its " + "macro list were not checked" + ) + return 0 + if dry: + print(f"{adapter.gcode_request(script)}\ndry-run: nothing was sent") + return 0 + adapter.run_gcode(script, timeout=timeout) + print(f"sent to {device.name}; the printer answered ok") + return 0 + + +def _registry_rows(device: Device) -> list[dict[str, Any]]: + """Macros known without contacting the printer: its trusted_macros and the commands of its aliases.""" + rows: dict[str, dict[str, Any]] = {} + for n in sorted(device.trusted_macros): + rows[n] = { + "name": n, + "trusted": True, + "kind": macros.classify(n), + "description": "", + "source": "trusted_macros", + } + table = aliases.supported( + aliases.resolve(device.adapter, device.model, aliases.overrides_for(device.name)) + ) + for alias, spec in table.items(): + for line in spec.gcode: + cmd = macros.command_of(line) + if cmd and "{" not in cmd and cmd not in rows: + rows[cmd] = { + "name": cmd, + "trusted": False, + "kind": macros.classify(cmd), + "description": "", + "source": f"alias {alias}", + } + return sorted(rows.values(), key=lambda r: str(r["name"])) + + +def _print_rows(rows: list[dict[str, Any]], as_json: bool, title: str) -> None: + if as_json: + print(json.dumps(rows, indent=2)) + return + print(title) + width = max((len(str(r["name"])) for r in rows), default=4) + print(f" {'NAME':<{width}} T {'KIND':<9} DESCRIPTION") + for r in rows: + print( + f" {r['name']:<{width}} {'*' if r['trusted'] else ' '} {r['kind']:<9} {r['description'] or r.get('source', '')}" + ) + + +def _macro_list(device: Device, opts: dict[str, str]) -> int: + as_json = "--json" in opts + if device.access == "offline" or "--offline" in opts: + rows = _registry_rows(device) + _print_rows( + rows, + as_json, + f"{device.name}: {len(rows)} macros known from the registry (nothing contacted; " + "T * = in trusted_macros)", + ) + return 0 + adapter = adapter_of(device) + if "--dry-run" in opts: + print("\n".join(adapter.catalog_requests())) + print("dry-run: nothing was sent") + return 0 + catalog = adapter.macro_catalog() + names = sorted( + catalog.macros + if "--all" in opts + else {m for m in catalog.macros if not m.startswith("_")} + ) + rows = [ + { + "name": n, + "trusted": n in device.trusted_macros, + "kind": macros.classify(n), + "description": catalog.describe(n), + "source": "printer", + } + for n in names + ] + hidden = len(catalog.macros) - len(names) + note = f", {hidden} private (_NAME) hidden: --all" if hidden else "" + _print_rows( + rows, + as_json, + f"{device.name}: {len(rows)} macros defined by the printer{note} (T * = in trusted_macros)", + ) + return 0 + + +def _macro_run(device: Device, words: list[str], opts: dict[str, str]) -> int: + if not words: + raise UsageError( + "`macro run` needs the macro name", + command="printer", + remediation=["3d printer macro --help"], + ) + line = macros.build_line(words[0], words[1:]) + return execute( + device, + [line], + what=f"macro run {line}", + again=f"macro run {line}", + kind=None, + curated=False, + entry_yes=False, + timeout=timeout_of(opts, 120.0), + opts=opts, + ) + + +def run_macro(name: str | None, argv: list[str]) -> int: + if not argv or argv[0] in ("-h", "--help", "help"): + print(MACRO_USAGE) + return 0 if argv else 1 + sub, rest = argv[0], argv[1:] + if sub not in ("list", "run"): + raise UsageError( + f"unknown macro subcommand '{sub}'", + command="printer", + remediation=["Use list or run."], + ) + if any(a in ("-h", "--help") for a in rest): + print(MACRO_USAGE) + return 0 + words, opts = split_opts(rest, FLAGS, VALUE_OPTS) + device = device_of(name, f"macro {sub}") + if sub == "list": + if words: + raise UsageError( + f"`macro list` takes no arguments, got {words}", command="printer" + ) + return _macro_list(device, opts) + return _macro_run(device, words, opts) diff --git a/lib/printer_ops/macros.py b/lib/printer_ops/macros.py new file mode 100644 index 00000000..f9403d1a --- /dev/null +++ b/lib/printer_ops/macros.py @@ -0,0 +1,269 @@ +"""macros.py — what `3d printer macro run` and the alias subcommands may send to a Klipper printer. + +WHAT: the pure rules between a user's words and one G-code script: a macro name and its + `KEY=VALUE` parameters are checked character by character (`build_line`), a command is put + into a risk class (`classify`), and `assess` turns a script into the gates it must pass + (`--yes`, an idle printer). `MacroCatalog` is what the printer says it defines + (Moonraker `/printer/objects/list` + `/printer/gcode/help`). + +WHY: Moonraker's `/printer/gcode/script` runs ANY text, newlines and `;` included, so one + unchecked value could append `M112` or a whole second command. Nothing here is free text: + a name is an identifier, a value is a short token from a fixed alphabet, and the emergency + stop is never sent (the printer's own button is the way to stop, not a remote script). + Issue https://github.com/alex-mextner/3d-cli/issues/718. + +RISK CLASSES (`classify`): `inert` (reads, messages: any time), `job` (pause/resume/cancel + style: allowed while a job runs), `heating`, `motion` (everything not known, because a + macro body is not checked), `dangerous` (restarts, config writes, kinematics). heating, + motion and dangerous need an idle printer; dangerous and any macro outside the printer's + `trusted_macros` also need `--yes`. + +INVARIANTS: stdlib only; nothing here touches the network. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from typing import Final + +from errors import InvalidArgument +from printer_ops.base import PrinterRefused + +INERT, JOB, HEATING, MOTION, DANGEROUS = ( + "inert", + "job", + "heating", + "motion", + "dangerous", +) +KINDS: Final = (INERT, JOB, HEATING, MOTION, DANGEROUS) +IDLE_KINDS: Final = frozenset({HEATING, MOTION, DANGEROUS}) + +# Never sent, whatever the flags: the emergency stop is the printer's own button. +DENIED: Final = frozenset({"M112", "EMERGENCY_STOP"}) +_JOB: Final = frozenset( + { + "PAUSE", + "RESUME", + "CANCEL_PRINT", + "CLEAR_PAUSE", + "M600", + "M220", + "M221", + "SET_PAUSE_NEXT_LAYER", + "SET_PAUSE_AT_LAYER", + } +) +_INERT: Final = frozenset( + { + "M105", + "M114", + "M115", + "M117", + "GET_POSITION", + "STATUS", + "HELP", + "RESPOND", + "SET_GCODE_VARIABLE_QUERY", + } +) +_INERT_PREFIXES: Final = ("QUERY_", "GET_") +_HEATING: Final = frozenset( + { + "M104", + "M109", + "M140", + "M141", + "M190", + "M191", + "SET_HEATER_TEMPERATURE", + "TURN_OFF_HEATERS", + "TEMPERATURE_WAIT", + "SET_TEMPERATURE_FAN_TARGET", + "SET_PURIFIER_MODE", + } +) +_DANGEROUS: Final = frozenset( + { + "FIRMWARE_RESTART", + "RESTART", + "SAVE_CONFIG", + "SDCARD_RESET_FILE", + "SET_KINEMATIC_POSITION", + "FORCE_MOVE", + "M999", + "M84", + "M18", + "SHUTDOWN", + "SHAPER_CALIBRATE", + "PID_CALIBRATE", + "PROBE_CALIBRATE", + "Z_OFFSET_APPLY_PROBE", + "Z_OFFSET_APPLY_ENDSTOP", + } +) + +_NAME: Final = re.compile(r"[A-Za-z_][A-Za-z0-9_]{0,63}") +_NUMERIC_CODE: Final = re.compile( + r"[GMT][0-9]+" +) # G28, M104, T0: standard codes, not macros +# A value is one short token: no space, quote, `;`, `{`, `%`, newline. `,` and `:` carry lists/pairs. +_VALUE: Final = re.compile(r"[A-Za-z0-9_.,:+/-]{1,64}") +_WORD: Final = re.compile(r"[A-Za-z0-9_]+") + + +def is_numeric_code(name: str) -> bool: + """`G28`, `M104`, `T2`: a standard G-code, which only a printer-defined macro of that name may override.""" + return _NUMERIC_CODE.fullmatch(name.upper()) is not None + + +def check_name(name: str) -> str: + """The upper-cased command name; InvalidArgument unless it is a plain identifier, PrinterRefused when denied.""" + if not _NAME.fullmatch(name): + raise InvalidArgument( + "macro", + repr(name), + ["an identifier: letters, digits, _ (starts with a letter or _)"], + command="printer", + extra="Newlines, spaces, `;` and quotes are never sent.", + ) + upper = name.upper() + refuse_denied(upper, "macro name") + return upper + + +def refuse_denied(token: str, what: str) -> None: + for word in _WORD.findall(token.upper()): + if word in DENIED: + raise PrinterRefused( + f"{what} contains {word}: 3d never sends the emergency stop", + command="printer", + remediation=[ + "Stop the printer with its own emergency-stop control (touchscreen, Fluidd, power)." + ], + ) + + +def check_value(key: str, value: str) -> str: + if not _VALUE.fullmatch(value): + raise InvalidArgument( + key, + repr(value), + ["1-64 characters from A-Z a-z 0-9 _ . , : + / -"], + command="printer", + extra="No spaces, quotes, `;`, braces or newlines: a value is one token.", + ) + refuse_denied(value, f"value of {key}") + return value + + +def build_line(name: str, params: list[str]) -> str: + """`NAME KEY=VALUE ...` from a macro name and `KEY=VALUE` words; every character is validated here.""" + out = [check_name(name)] + seen: set[str] = set() + for word in params: + key, sep, value = word.partition("=") + if not sep or not _NAME.fullmatch(key): + raise InvalidArgument( + "parameter", + repr(word), + ["KEY=VALUE, e.g. EXTRUDER=0"], + command="printer", + ) + key = key.upper() + if key in seen: + raise InvalidArgument( + "parameter", key, ["each key once"], command="printer" + ) + seen.add(key) + out.append(f"{key}={check_value(key, value)}") + return " ".join(out) + + +def command_of(line: str) -> str: + """First word of a G-code line, upper-cased (`G28 X` -> `G28`).""" + return line.split(None, 1)[0].upper() if line.strip() else "" + + +def classify(name: str) -> str: + n = name.upper() + if n in _DANGEROUS: + return DANGEROUS + if n in _JOB: + return JOB + if n in _HEATING: + return HEATING + if n in _INERT or n.startswith(_INERT_PREFIXES): + return INERT + return MOTION + + +@dataclass(frozen=True) +class Assessment: + kind: str # the strictest class among the lines + needs_idle: bool + yes_reasons: tuple[str, ...] # why `--yes` is needed; empty = not needed + + +_RANK: Final = {INERT: 0, JOB: 1, HEATING: 2, MOTION: 3, DANGEROUS: 4} + + +def assess( + lines: list[str], + trusted: frozenset[str], + *, + curated: bool, + kind: str | None = None, + entry_yes: bool = False, +) -> Assessment: + """Gates for a script. `curated` lines come from the alias table (the printer's own table or the + user's `aliases:`): they are vetted, so only a dangerous class or `yes: true` asks for --yes. + A `macro run` line (not curated) outside `trusted` asks for --yes. `kind` is an alias's declared class; + the strictest of it and each line's class wins.""" + worst = kind if kind in _RANK else INERT + reasons: list[str] = [] + for line in lines: + cmd = command_of(line) + refuse_denied(cmd, "command") + cls = classify(cmd) + if _RANK[cls] > _RANK[worst]: + worst = cls + if cls == DANGEROUS: + reasons.append( + f"{cmd} is dangerous (restart, config write or kinematics change)" + ) + elif not curated and cmd not in trusted: + reasons.append(f"{cmd} is not in this printer's trusted_macros") + if entry_yes: + reasons.append("this alias is declared to need confirmation") + return Assessment(worst, worst in IDLE_KINDS, tuple(dict.fromkeys(reasons))) + + +@dataclass(frozen=True) +class MacroCatalog: + """What a printer defines: `macros` = [gcode_macro X] names, `commands` = every extended command + with its help text (macros included).""" + + macros: frozenset[str] + commands: dict[str, str] = field(default_factory=dict, hash=False) + + def knows(self, name: str) -> bool: + n = name.upper() + return n in self.macros or n in self.commands + + def describe(self, name: str) -> str: + return self.commands.get(name.upper(), "") + + +def parse_catalog(objects: object, help_text: object) -> MacroCatalog: + """Moonraker's `objects/list` result (`{"objects": [...]}`) and `gcode/help` result (`{CMD: text}`).""" + names = objects.get("objects") if isinstance(objects, dict) else None + if not isinstance(names, list) or not isinstance(help_text, dict): + raise ValueError("objects/list or gcode/help has an unexpected shape") + macros = frozenset( + str(n).split(" ", 1)[1].strip().upper() + for n in names + if isinstance(n, str) and n.lower().startswith("gcode_macro ") and " " in n + ) + return MacroCatalog(macros, {str(k).upper(): str(v) for k, v in help_text.items()}) diff --git a/lib/printer_ops/moonraker.py b/lib/printer_ops/moonraker.py index 5ddb18c7..bc734594 100644 --- a/lib/printer_ops/moonraker.py +++ b/lib/printer_ops/moonraker.py @@ -71,7 +71,7 @@ from datetime import datetime, timezone from email.utils import parsedate_to_datetime from pathlib import Path -from typing import Any, TypeVar +from typing import TYPE_CHECKING, Any, TypeVar from urllib.parse import quote, urlsplit from printer_ops.base import ( @@ -104,6 +104,9 @@ ) from printer_ops.move_bounds import Limits +if TYPE_CHECKING: + from printer_ops.macros import MacroCatalog + _STATE = { "standby": IDLE, "printing": PRINTING, @@ -114,6 +117,8 @@ } _HEATERS = ("extruder", "extruder1", "extruder2", "extruder3", "heater_bed") _T = TypeVar("_T") +# Macro discovery: the `gcode_macro NAME` objects and every extended command with its help text. +_CATALOG_PATHS = ("/printer/objects/list", "/printer/gcode/help") _QUERY = ( "webhooks&print_stats&virtual_sdcard&display_status&" + "&".join(_HEATERS) @@ -209,22 +214,17 @@ def _differs(after: Slot, want: HeadFilament) -> list[str]: ] -def parse_status(status: dict[str, Any]) -> PrinterStatus: - """Moonraker `objects/query` status dict -> normalized PrinterStatus.""" - hooks = status.get("webhooks") or {} - ps = status.get("print_stats") or {} +def _job_state(hooks: dict[str, Any], ps: dict[str, Any]) -> tuple[str, str, Any]: + """(normalized state, raw state, message) of the job: Klipper not ready (startup/shutdown/error) + means nothing about the job can be trusted.""" klippy = hooks.get("state", "ready") raw_state = str(ps.get("state", "")) or "?" if klippy != "ready": - # Klipper not ready (startup/shutdown/error): nothing about the job can be trusted. - state, raw_state = UNKNOWN, f"klippy:{klippy}" - message = hooks.get("state_message") - else: - state = _STATE.get(raw_state, UNKNOWN) - message = ps.get("message") or None - progress = (status.get("virtual_sdcard") or {}).get("progress") - if progress is None: - progress = (status.get("display_status") or {}).get("progress") + return UNKNOWN, f"klippy:{klippy}", hooks.get("state_message") + return _STATE.get(raw_state, UNKNOWN), raw_state, ps.get("message") or None + + +def _temps(status: dict[str, Any]) -> dict[str, tuple[float | None, float | None]]: temps: dict[str, tuple[float | None, float | None]] = {} for h in _HEATERS: obj = status.get(h) @@ -233,12 +233,22 @@ def parse_status(status: dict[str, Any]) -> PrinterStatus: obj.get("temperature"), obj.get("target"), ) + return temps + + +def parse_status(status: dict[str, Any]) -> PrinterStatus: + """Moonraker `objects/query` status dict -> normalized PrinterStatus.""" + ps = status.get("print_stats") or {} + state, raw_state, message = _job_state(status.get("webhooks") or {}, ps) + progress = (status.get("virtual_sdcard") or {}).get("progress") + if progress is None: + progress = (status.get("display_status") or {}).get("progress") return PrinterStatus( state=state, raw_state=raw_state, filename=ps.get("filename") or None, progress=float(progress) if progress is not None else None, - temps=temps, + temps=_temps(status), message=message, slots=parse_slots(status.get("print_task_config") or {}), raw=status, @@ -577,10 +587,31 @@ def run_gcode(self, script: str, *, timeout: float = 120.0) -> None: STEPPER_BUZZ takes ~10 s), so the read timeout is at least `timeout`.""" saved, self.timeout = self.timeout, max(self.timeout, timeout) try: - self._post("/printer/gcode/script?script=" + quote(script, safe="")) + self._post(self._script_path(script)) finally: self.timeout = saved + @staticmethod + def _script_path(script: str) -> str: + return "/printer/gcode/script?script=" + quote(script, safe="") + + def gcode_request(self, script: str) -> str: + return f"POST {self.base}{self._script_path(script)}" + + def catalog_requests(self) -> list[str]: + return [f"GET {self.base}{path}" for path in _CATALOG_PATHS] + + def macro_catalog(self) -> MacroCatalog: + """`gcode_macro NAME` objects (`/printer/objects/list`) and every extended command with its help text + (`/printer/gcode/help`, macros included). Read-only.""" + from printer_ops.macros import parse_catalog + + objects, help_text = (self._get(path) for path in _CATALOG_PATHS) + try: + return parse_catalog(objects, help_text) + except ValueError as exc: + raise self._comm_error("macro discovery", exc) from exc + def file_metadata(self, filename: str) -> dict[str, Any]: """Moonraker's parsed metadata of a stored G-code file (`object_height`, `layer_count`, ...); read-only.""" result = self._get("/server/files/metadata?filename=" + quote(filename, safe="")) diff --git a/pyproject.toml b/pyproject.toml index d347e931..20afe6c2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "3d-cli" -version = "0.78.0" +version = "0.79.0" description = "Scriptable, AI-assisted CLI + web toolkit for parametric 3D FDM modeling, verification, matching and print prep (OpenSCAD-first)." readme = "README.md" requires-python = ">=3.10" diff --git a/tests/e2e/test_printer_macros_workflows.py b/tests/e2e/test_printer_macros_workflows.py new file mode 100644 index 00000000..c5b7db74 --- /dev/null +++ b/tests/e2e/test_printer_macros_workflows.py @@ -0,0 +1,176 @@ +"""Real `bin/3d printer -n NAME macro|aliases|ALIAS` runs against the loopback fake Moonraker. + +The fleet is a copy of the repo's fleet/printers.yaml (`u1`, `k1se`) with the address swapped for the +fake's (tests/printer_fakes.py `write_fleet`); no real printer is reachable. Issue #718. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +from printer_fakes import FakeMoonraker, u1_status +from printer_macro_fakes import FakeMoonrakerMacros, write_fleet + +from .workflow_helper import run_cli + +U1_MACROS = { + n: n + for n in ( + "PAUSE", + "RESUME", + "M600", + "AUTO_FEEDING", + "SM_PRINT_AUTO_FEED", + "ROUGHLY_CLEAN_NOZZLE", + "BED_MESH_CALIBRATE", + ) +} +U1_COMMANDS = { + n: n for n in ("PARK_EXTRUDER", "SET_HEATER_TEMPERATURE", "TURN_OFF_HEATERS") +} + + +def test_a_user_lists_macros_then_runs_one_and_a_dry_run_sends_nothing( + tmp_path: Path, +) -> None: + """A user lists the macros a U1 defines, runs one with typed parameters and sees the exact line sent; a dry run sends nothing.""" + u1 = FakeMoonrakerMacros(macros=dict(U1_MACROS), commands=dict(U1_COMMANDS)) + with u1.serve(): + env = {"THREED_FLEET": str(write_fleet(tmp_path, u1=u1))} + listed = run_cli( + tmp_path, "printer", "-n", "u1", "macro", "list", env_extra=env + ) + dry = run_cli( + tmp_path, + "printer", + "-n", + "u1", + "macro", + "run", + "AUTO_FEEDING", + "EXTRUDER=1", + "--yes", + "--dry-run", + env_extra=env, + ) + writes_after_dry = u1.writes() + ran = run_cli( + tmp_path, + "printer", + "-n", + "u1", + "macro", + "run", + "SM_PRINT_AUTO_FEED", + "EXTRUDER=1", + env_extra=env, + ) + + assert listed.returncode == 0, listed.stderr + assert ( + "u1: 7 macros defined by the printer" in listed.stdout + and "SM_PRINT_AUTO_FEED" in listed.stdout + ) + assert dry.returncode == 0, dry.stderr + assert ( + "AUTO_FEEDING EXTRUDER=1" in dry.stdout + and "script=AUTO_FEEDING%20EXTRUDER%3D1" in dry.stdout + ) + assert writes_after_dry == [] + assert ran.returncode == 0, ran.stderr + assert u1.scripts == ["SM_PRINT_AUTO_FEED EXTRUDER=1"] + + +def test_a_user_runs_aliases_and_gets_a_clear_refusal_for_a_busy_printer_and_for_injection( + tmp_path: Path, +) -> None: + """A user runs aliases such as home and preheat; a busy printer and an injected value are refused with a reason and nothing is sent.""" + u1 = FakeMoonrakerMacros(macros=dict(U1_MACROS), commands=dict(U1_COMMANDS)) + with u1.serve(): + env = {"THREED_FLEET": str(write_fleet(tmp_path, u1=u1))} + load = run_cli(tmp_path, "printer", "-n", "u1", "load", "T2", env_extra=env) + injected = run_cli( + tmp_path, + "printer", + "-n", + "u1", + "macro", + "run", + "AUTO_FEEDING", + "EXTRUDER=1;M112", + "--yes", + env_extra=env, + ) + emergency = run_cli( + tmp_path, "printer", "-n", "u1", "macro", "run", "M112", env_extra=env + ) + scripts_before = list(u1.scripts) + u1.status = u1_status("printing", "part.gcode", 0.3) + busy = run_cli(tmp_path, "printer", "-n", "u1", "park", env_extra=env) + pause = run_cli(tmp_path, "printer", "-n", "u1", "pause", env_extra=env) + + assert load.returncode == 0, load.stderr + assert scripts_before == ["AUTO_FEEDING EXTRUDER=1 LOAD=1"] + assert injected.returncode == 2 and "EXTRUDER" in injected.stderr + assert ( + emergency.returncode == 1 + and "3d never sends the emergency stop" in emergency.stderr + ) + assert ( + busy.returncode == 1 + and "is printing" in busy.stderr + and "needs an idle printer" in busy.stderr + ) + assert pause.returncode == 0, pause.stderr + assert u1.scripts == [*scripts_before, "PAUSE"] + + +def test_read_only_and_offline_printers_are_refused_and_never_contacted( + tmp_path: Path, +) -> None: + """A user points `macro run` at a read-only and at an offline printer: both are refused and the printer is never contacted.""" + k1 = FakeMoonrakerMacros(macros={"PAUSE": "p"}) + with k1.serve(): + env = { + "THREED_FLEET": str( + write_fleet(tmp_path, access={"k1se": "read-only"}, k1se=k1) + ) + } + read_only = run_cli(tmp_path, "printer", "-n", "k1se", "home", env_extra=env) + offline = run_cli( + tmp_path, "printer", "-n", "p1s", "macro", "run", "M620", env_extra=env + ) + requests = list(k1.requests) + + assert read_only.returncode == 1 and "is read-only" in read_only.stderr + assert offline.returncode == 1 and "is offline" in offline.stderr + assert requests == [] + + +def test_aliases_listing_and_help_per_model(tmp_path: Path) -> None: + """A user reads `aliases` and one alias's --help per printer model and sees what each alias sends.""" + k1 = FakeMoonraker() + with k1.serve(): + env = {"THREED_FLEET": str(write_fleet(tmp_path, k1se=k1))} + unknown = run_cli( + tmp_path, "printer", "-n", "k1se", "purifier", "hot", env_extra=env + ) + table = run_cli( + tmp_path, "printer", "-n", "k1se", "aliases", "--json", env_extra=env + ) + helped = run_cli(tmp_path, "printer", "preheat", "--help", env_extra=env) + top = run_cli(tmp_path, "printer", "--help", env_extra=env) + + assert ( + unknown.returncode == 2 + and "alias `purifier` is not defined for k1se" in unknown.stderr + ) + assert json.loads(table.stdout)["load"]["gcode"] == ["LOAD_MATERIAL"] + assert helped.returncode == 0 and "SET_HEATER_TEMPERATURE" in helped.stdout + assert ( + top.returncode == 0 + and "macro list|run" in top.stdout + and "level-heated" in top.stdout + ) + assert k1.requests == [] diff --git a/tests/e2e/test_printops_maint_workflows.py b/tests/e2e/test_printops_maint_workflows.py index 7758d9bb..8f88c111 100644 --- a/tests/e2e/test_printops_maint_workflows.py +++ b/tests/e2e/test_printops_maint_workflows.py @@ -4,6 +4,8 @@ from __future__ import annotations import json +import time +from datetime import date, timedelta from pathlib import Path from typing import Any @@ -45,11 +47,11 @@ def _history(url: str) -> Any: - """Moonraker GET /server/history/*: 2240 m in total, 120 m in the last two weeks.""" + """Moonraker GET /server/history/*: 2240 m in total, 120 m five days ago (after the maintenance logged ten days ago).""" if url.endswith("/server/history/totals"): return {"result": {"job_totals": {"total_jobs": 88.0, "total_print_time": 1191293.4, "total_filament_used": 2240300.0}}} - return {"result": {"count": 1, "jobs": [{"start_time": 1791900000.0, "print_duration": 7200.0, + return {"result": {"count": 1, "jobs": [{"start_time": time.time() - 5 * 86400, "print_duration": 7200.0, "filament_used": 120000.0}]}} @@ -99,7 +101,7 @@ def test_user_tracks_a_broken_head_a_nozzle_swap_and_due_maintenance(maint: Any) assert "[days ?/30, filament_m 2240.3/1000; never logged]" in out assert "done? 3d printops maint log u1 --rule u1-linear-shafts" in out - code, out, err = maint("log", "u1", "--rule", "u1-linear-shafts", "--at", "2026-10-10", + code, out, err = maint("log", "u1", "--rule", "u1-linear-shafts", "--at", (date.today() - timedelta(days=10)).isoformat(), # relative dates: fixed ones age "ok" into "soon" "--notes", "IPA + white lithium grease") assert code == 0, err assert "logged #2 kinematics lube on u1" in out and "(rule u1-linear-shafts)" in out @@ -120,8 +122,8 @@ def test_user_tracks_a_broken_head_a_nozzle_swap_and_due_maintenance(maint: Any) assert code == 0, err lines = out.splitlines() assert lines[0].startswith("#3 ") and "u1/T2 ptfe replace" in lines[0] and "invalidated: ptfe" in lines[0] - assert "u1/T3 nozzle replace" in lines[1] and "invalidated: nozzle_swap" in lines[1] - assert "kinematics lube" in lines[2] and "rule u1-linear-shafts" in lines[2] + assert "kinematics lube" in lines[1] and "rule u1-linear-shafts" in lines[1] # ten days ago, after the 2026-10-14 swap + assert "u1/T3 nozzle replace" in lines[2] and "invalidated: nozzle_swap" in lines[2] code, out, err = maint("fault", "open", "u1/T1", "--symptom", "grinding", "--component", "extruder") assert code == 0, err diff --git a/tests/printer_macro_fakes.py b/tests/printer_macro_fakes.py new file mode 100644 index 00000000..0af186e0 --- /dev/null +++ b/tests/printer_macro_fakes.py @@ -0,0 +1,76 @@ +"""Fakes for `3d printer macro|aliases|ALIAS` tests: a fake Moonraker that also answers macro discovery, and a +fleet file whose contactable printers point at loopback fakes (never at a real address). Issue #718.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from http.server import BaseHTTPRequestHandler +from pathlib import Path +from urllib.parse import parse_qs, urlsplit + +from printer_fakes import FakeMoonraker, Recorded + +REPO = Path(__file__).resolve().parent.parent + + +@dataclass +class FakeMoonrakerMacros(FakeMoonraker): + """FakeMoonraker + `/printer/objects/list` and `/printer/gcode/help` (Moonraker's macro discovery).""" + + macros: dict[str, str] = field( + default_factory=dict + ) # `[gcode_macro NAME]` -> description + commands: dict[str, str] = field( + default_factory=dict + ) # other extended commands -> help text + + def handler(self) -> type[BaseHTTPRequestHandler]: + fake = self + base = super().handler() + + class H(base): # type: ignore[valid-type, misc] + def do_GET(self) -> None: + u = urlsplit(self.path) + if u.path == "/printer/objects/list": + fake.requests.append(Recorded("GET", u.path, parse_qs(u.query))) + objects = [ + "webhooks", + "print_stats", + "toolhead", + "extruder", + "heater_bed", + *(f"gcode_macro {m}" for m in fake.macros), + ] + self._reply({"result": {"objects": objects}}) # type: ignore[attr-defined] + elif u.path == "/printer/gcode/help": + fake.requests.append(Recorded("GET", u.path, parse_qs(u.query))) + self._reply({"result": {**fake.commands, **fake.macros}}) # type: ignore[attr-defined] + else: + super().do_GET() + + return H + + +def write_fleet( + tmp_path: Path, access: dict[str, str] | None = None, **fakes: FakeMoonraker +) -> Path: + """A copy of the repo's fleet/printers.yaml whose contactable printers are the named fakes (`u1=moon`): their + host/port point at the loopback fake and every other contactable entry is dropped, so no test can reach a real + address. Offline fleet printers stay as they are (never contacted); `access` overrides the access per printer.""" + import yaml + + data = yaml.safe_load( + (REPO / "fleet" / "printers.yaml").read_text(encoding="utf-8") + ) + for name in list(data): + if data[name].get("access") == "offline": + continue + if name not in fakes: + del data[name] + continue + data[name].update(host="127.0.0.1", port=fakes[name].port) + if access and name in access: + data[name]["access"] = access[name] + path = tmp_path / "fleet.yaml" + path.write_text(yaml.safe_dump(data), encoding="utf-8") + return path diff --git a/tests/test_printer_macros.py b/tests/test_printer_macros.py new file mode 100644 index 00000000..5b9b8582 --- /dev/null +++ b/tests/test_printer_macros.py @@ -0,0 +1,866 @@ +"""`3d printer -n NAME macro list|run`, `aliases` and the alias subcommands (home, load, park, ...). + +Every printer is a loopback fake Moonraker (tests/printer_fakes.py) built from the REAL fleet entries of +`u1` and `k1se` (fleet/printers.yaml with the address swapped for the fake's); a refused command is proven +by the fake having received NO write (often no request at all). Issue #718. +""" + +from __future__ import annotations + +import json +from collections.abc import Iterator +from pathlib import Path +from typing import Any + +import pytest +import yaml +from printer_fakes import FakeCreality, u1_status, write_devices +from printer_macro_fakes import FakeMoonrakerMacros, write_fleet + +from commands.printer import USAGE, run +from errors import InvalidArgument, ThreeDError, UsageError +from printer_ops import aliases, macros +from printer_ops.base import PrinterRefused, UnsupportedOperation + +REPO = Path(__file__).resolve().parent.parent + +U1_MACROS = { + name: f"{name} macro" + for name in ( + "PAUSE", + "RESUME", + "CANCEL_PRINT", + "M600", + "T0", + "T1", + "BED_MESH_CALIBRATE", + "ROUGHLY_CLEAN_NOZZLE", + "AUTO_FEEDING", + "SM_PRINT_AUTO_FEED", + "AUTO_BED_MESH_CALIBRATE", + "SM_PRINT_FLOW_CALIBRATE", + "_CLIENT_VARIABLE", + ) +} +U1_COMMANDS = { + name: f"{name} command" + for name in ( + "SET_HEATER_TEMPERATURE", + "TURN_OFF_HEATERS", + "PARK_EXTRUDER", + "PARK_EXTRUDER1", + "PARK_EXTRUDER2", + "MOVE_TO_XY_IDLE_POSITION_EXTRUDER", + "SET_PURIFIER_MODE", + "FIRMWARE_RESTART", + "GET_POSITION", + "BED_MESH_CLEAR", + ) +} +K1_MACROS = { + name: f"{name} macro" + for name in ( + "PAUSE", + "RESUME", + "LOAD_MATERIAL", + "QUIT_MATERIAL", + "G29", + "START_PRINT", + "END_PRINT", + ) +} +K1_COMMANDS = { + name: f"{name} command" for name in ("SET_HEATER_TEMPERATURE", "TURN_OFF_HEATERS") +} + + +class Shop: + """The two shop printers as fakes, plus a way to run `3d printer ...` and read what it printed.""" + + def __init__( + self, u1: FakeMoonrakerMacros, k1se: FakeMoonrakerMacros, capsys: pytest.CaptureFixture[str] + ) -> None: + self.u1, self.k1se, self.capsys = u1, k1se, capsys + + def out(self, *argv: str) -> str: + assert run(list(argv)) == 0 + return self.capsys.readouterr().out + + def fails(self, exc: type[ThreeDError], *argv: str) -> ThreeDError: + with pytest.raises(exc) as info: + run(list(argv)) + return info.value + + +@pytest.fixture() +def shop( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> Iterator[Shop]: + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "xdg")) + monkeypatch.chdir(tmp_path) + u1 = FakeMoonrakerMacros(macros=dict(U1_MACROS), commands=dict(U1_COMMANDS)) + k1 = FakeMoonrakerMacros(macros=dict(K1_MACROS), commands=dict(K1_COMMANDS)) + with u1.serve(), k1.serve(): + monkeypatch.setenv("THREED_FLEET", str(write_fleet(tmp_path, u1=u1, k1se=k1))) + yield Shop(u1, k1, capsys) + + +def use_fleet(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, **kw: Any) -> None: + monkeypatch.setenv("THREED_FLEET", str(write_fleet(tmp_path, **kw))) + + +# ---- macro list ----------------------------------------------------------------------------- + + +def test_macro_list_reads_the_printer_and_marks_trusted_macros(shop: Shop) -> None: + text = shop.out("-n", "u1", "macro", "list") + + assert "u1: 12 macros defined by the printer, 1 private (_NAME) hidden" in text + assert any( + line.split()[:2] == ["SM_PRINT_AUTO_FEED", "*"] for line in text.splitlines() + ) # in trusted_macros + assert any( + line.split()[:1] == ["AUTO_FEEDING"] and "*" not in line.split()[:2] + for line in text.splitlines() + ) + assert "_CLIENT_VARIABLE" not in text + assert [r.path for r in shop.u1.requests] == [ + "/printer/objects/list", + "/printer/gcode/help", + ] + assert shop.u1.writes() == [] + + +def test_macro_list_all_json_includes_private_macros(shop: Shop) -> None: + rows = json.loads(shop.out("-n", "k1se", "macro", "list", "--all", "--json")) + + assert {r["name"] for r in rows} == set(K1_MACROS) + start = next(r for r in rows if r["name"] == "START_PRINT") + assert start["trusted"] is True and start["description"] == "START_PRINT macro" + assert next(r for r in rows if r["name"] == "LOAD_MATERIAL")["trusted"] is False + + +def test_macro_list_offline_uses_registry_data_and_contacts_nothing(shop: Shop) -> None: + text = shop.out("-n", "u1", "macro", "list", "--offline", "--json") + names = {r["name"] for r in json.loads(text)} + + assert { + "SM_PRINT_AUTO_FEED", + "DETECT_BED_PLATE", + } <= names # trusted_macros of fleet/printers.yaml + assert {"AUTO_FEEDING", "ROUGHLY_CLEAN_NOZZLE", "G28"} <= names # alias targets + assert shop.u1.requests == [] + + +def test_macro_list_dry_run_prints_the_requests_and_sends_none(shop: Shop) -> None: + text = shop.out("-n", "u1", "macro", "list", "--dry-run") + + assert f"GET http://127.0.0.1:{shop.u1.port}/printer/objects/list" in text + assert f"GET http://127.0.0.1:{shop.u1.port}/printer/gcode/help" in text + assert "dry-run: nothing was sent" in text + assert shop.u1.requests == [] + + +def test_macro_list_of_an_offline_fleet_printer_uses_the_registry(shop: Shop) -> None: + text = shop.out("-n", "p1s", "macro", "list") + + assert ( + "macros known from the registry (nothing contacted" in text and "M620" in text + ) + assert shop.u1.requests == [] and shop.k1se.requests == [] + + +# ---- macro run ------------------------------------------------------------------------------ + + +def test_macro_run_passes_params_and_prints_exactly_the_line_sent(shop: Shop) -> None: + text = shop.out("-n", "u1", "macro", "run", "sm_print_auto_feed", "extruder=1") + + assert shop.u1.scripts == ["SM_PRINT_AUTO_FEED EXTRUDER=1"] + assert "gcode:\n SM_PRINT_AUTO_FEED EXTRUDER=1\n" in text + assert "sent to u1" in text + + +def test_macro_run_dry_run_prints_the_request_and_sends_nothing(shop: Shop) -> None: + text = shop.out( + "-n", "u1", "macro", "run", "SM_PRINT_AUTO_FEED", "EXTRUDER=1", "--dry-run" + ) + + assert ( + f"POST http://127.0.0.1:{shop.u1.port}/printer/gcode/script?script=SM_PRINT_AUTO_FEED%20EXTRUDER%3D1" + in text + ) + assert "dry-run: nothing was sent" in text + assert shop.u1.scripts == [] and shop.u1.writes() == [] + + +def test_macro_outside_trusted_macros_needs_yes(shop: Shop) -> None: + err = shop.fails( + UsageError, "-n", "u1", "macro", "run", "AUTO_FEEDING", "EXTRUDER=0" + ) + dry = shop.out( + "-n", "u1", "macro", "run", "AUTO_FEEDING", "EXTRUDER=0", "--dry-run" + ) + assert shop.u1.writes() == [] + + shop.out("-n", "u1", "macro", "run", "AUTO_FEEDING", "EXTRUDER=0", "--yes") + + assert ( + "needs --yes" in err.message + and "AUTO_FEEDING is not in this printer's trusted_macros" in err.message + ) + assert "would need --yes" in dry + assert shop.u1.scripts == ["AUTO_FEEDING EXTRUDER=0"] + + +def test_dangerous_macro_needs_yes_even_when_trusted( + shop: Shop, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + cfg = tmp_path / "xdg" / "3d-cli" + cfg.mkdir(parents=True) + (cfg / "printer-devices.yaml").write_text( + f"u1: {{adapter: moonraker, host: 127.0.0.1, port: {shop.u1.port}, model: Snapmaker U1, " + "trusted_macros: [FIRMWARE_RESTART]}\n", + encoding="utf-8", + ) + + err = shop.fails(UsageError, "-n", "u1", "macro", "run", "FIRMWARE_RESTART") + + assert "FIRMWARE_RESTART is dangerous" in err.message + assert shop.u1.writes() == [] + + +def test_motion_macro_is_refused_while_printing_but_pause_style_is_not( + shop: Shop, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + shop.u1.status = u1_status("printing", "part.gcode", 0.4) + + err = shop.fails( + PrinterRefused, + "-n", + "u1", + "macro", + "run", + "AUTO_FEEDING", + "EXTRUDER=0", + "--yes", + ) + dry = shop.fails( + PrinterRefused, + "-n", + "u1", + "macro", + "run", + "AUTO_FEEDING", + "EXTRUDER=0", + "--yes", + "--dry-run", + ) + assert shop.u1.writes() == [] + shop.out( + "-n", "u1", "macro", "run", "PAUSE", "--yes" + ) # PAUSE is a job macro, but not in this printer's trusted list + shop.out( + "-n", "u1", "macro", "run", "M600" + ) # trusted + job class: no --yes, no idle printer needed + + for e in (err, dry): + assert "is printing" in e.message and "needs an idle printer" in e.message + assert shop.u1.scripts == ["PAUSE", "M600"] + + +def test_paused_and_unreadable_printers_count_as_busy(shop: Shop) -> None: + shop.u1.status = u1_status("paused", "part.gcode", 0.4) + assert ( + "is paused" + in shop.fails( + PrinterRefused, "-n", "u1", "macro", "run", "AUTO_FEEDING", "--yes" + ).message + ) + shop.u1.status = u1_status("error") + shop.u1.status["webhooks"]["state"] = ( + "shutdown" # Klippy shut down: the state reads unknown + ) + assert shop.fails(PrinterRefused, "-n", "u1", "home").message.endswith( + "needs an idle printer" + ) + assert shop.u1.writes() == [] + + +def test_read_only_printer_refuses_every_write_but_can_be_listed( + shop: Shop, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + use_fleet(monkeypatch, tmp_path, access={"u1": "read-only"}, u1=shop.u1) + + run_err = shop.fails( + PrinterRefused, "-n", "u1", "macro", "run", "SM_PRINT_AUTO_FEED", "EXTRUDER=0" + ) + alias_err = shop.fails(PrinterRefused, "-n", "u1", "home") + dry_err = shop.fails(PrinterRefused, "-n", "u1", "home", "--dry-run") + listed = shop.out("-n", "u1", "macro", "list") + + for e in (run_err, alias_err, dry_err): + assert "is read-only" in e.message and "refused" in e.message + assert shop.u1.writes() == [] and "SM_PRINT_AUTO_FEED" in listed + + +def test_offline_fleet_printer_refuses_run_and_aliases_and_is_never_contacted( + shop: Shop, +) -> None: + errs = [ + shop.fails(PrinterRefused, "-n", "p1s", *cmd) + for cmd in (("macro", "run", "M620"), ("home",), ("pause", "--dry-run")) + ] + + assert all("is offline" in e.message for e in errs) + assert shop.u1.requests == [] and shop.k1se.requests == [] + + +def test_macro_names_and_values_that_could_inject_a_second_command_are_rejected_before_any_contact( + shop: Shop, +) -> None: + bad_names = [ + "PAUSE\nM112", + "PAUSE;M112", + "PAUSE M112", + "G1 X1 Y1", + "", + "1PAUSE", + "PAUSE\r", + "SM-PRINT", + "ÅÄÖ", + "A" * 65, + ] + bad_values = [ + "1;M112", + "1\nM112", + "a b", + "'x'", + '"x"', + "{1}", + "%25", + "1#c", + "1*3", + "", + "x" * 65, + ] + for name in bad_names: + with pytest.raises(InvalidArgument): + run(["-n", "u1", "macro", "run", name]) + for value in bad_values: + with pytest.raises(InvalidArgument): + run(["-n", "u1", "macro", "run", "SM_PRINT_AUTO_FEED", f"EXTRUDER={value}"]) + for word in ("EXTRUDER", "=1", "1EXTRUDER=1", "EXT RUDER=1"): + with pytest.raises(InvalidArgument): + run(["-n", "u1", "macro", "run", "SM_PRINT_AUTO_FEED", word]) + with pytest.raises(InvalidArgument): + run( + [ + "-n", + "u1", + "macro", + "run", + "SM_PRINT_AUTO_FEED", + "EXTRUDER=1", + "extruder=2", + ] + ) # a key twice + + assert shop.u1.requests == [] + + +@pytest.mark.parametrize( + "argv", + [ + ["macro", "run", "M112"], + ["macro", "run", "m112"], + ["macro", "run", "EMERGENCY_STOP"], + ["macro", "run", "SM_PRINT_AUTO_FEED", "EXTRUDER=M112"], + ["macro", "run", "SM_PRINT_AUTO_FEED", "MSG=a,m112"], + ["macro", "run", "SM_PRINT_AUTO_FEED", "MSG=EMERGENCY_STOP", "--yes"], + ], +) +def test_the_emergency_stop_is_never_sent_whatever_the_flags( + shop: Shop, argv: list[str] +) -> None: + err = shop.fails(PrinterRefused, "-n", "u1", *argv, "--yes") + + assert "3d never sends the emergency stop" in err.message + assert shop.u1.requests == [] + + +def test_a_standard_gcode_is_not_a_macro(shop: Shop) -> None: + err = shop.fails(PrinterRefused, "-n", "u1", "macro", "run", "G28", "--yes") + shop.fails(PrinterRefused, "-n", "u1", "macro", "run", "M104", "S=200", "--yes") + + assert "standard G-code, not a macro u1 defines" in err.message + assert shop.u1.writes() == [] + + +def test_a_gcode_macro_that_overrides_a_standard_code_runs_as_a_macro( + shop: Shop, +) -> None: + shop.out( + "-n", "k1se", "macro", "run", "G29", "BED_TEMP=60", "--yes" + ) # the K1's stock bed-mesh macro + + assert shop.k1se.scripts == ["G29 BED_TEMP=60"] + + +def test_an_undefined_name_is_refused_with_a_hint(shop: Shop) -> None: + err = shop.fails(PrinterRefused, "-n", "u1", "macro", "run", "PAUSEE", "--yes") + + assert err.message == "u1 does not define PAUSEE" and "PAUSE" in err.remediation[0] + assert shop.u1.writes() == [] + + +def test_offline_run_prints_the_script_and_contacts_nothing(shop: Shop) -> None: + text = shop.out( + "-n", "u1", "macro", "run", "AUTO_FEEDING", "EXTRUDER=0", "--offline" + ) + + assert ( + "AUTO_FEEDING EXTRUDER=0" in text and "offline: nothing was contacted" in text + ) + assert shop.u1.requests == [] + + +def test_there_is_no_raw_gcode_passthrough(shop: Shop) -> None: + err = shop.fails(UsageError, "-n", "u1", "gcode", "G28") + + assert "unknown subcommand 'gcode'" in err.message or "alias `gcode`" in err.message + assert shop.u1.requests == [] + + +# ---- aliases -------------------------------------------------------------------------------- + + +def test_home_sends_g28_or_only_the_named_axes(shop: Shop) -> None: + shop.out("-n", "k1se", "home") + shop.out("-n", "k1se", "home", "xy") + shop.out("-n", "k1se", "home", "--axes", "Z") + + assert shop.k1se.scripts == ["G28", "G28 X Y", "G28 Z"] + for bad in ("XX", "A", "X Y"): + with pytest.raises(ThreeDError): + run(["-n", "k1se", "home", bad]) + assert len(shop.k1se.scripts) == 3 + + +def test_alias_dry_run_checks_everything_and_sends_nothing(shop: Shop) -> None: + text = shop.out("-n", "k1se", "level", "--dry-run") + + assert "alias level" in text and " G29\n" in text + assert ( + f"POST http://127.0.0.1:{shop.k1se.port}/printer/gcode/script?script=G29" + in text + ) + assert shop.k1se.writes() == [] + assert [r.path for r in shop.k1se.requests] == [ + "/printer/objects/list", + "/printer/gcode/help", + "/printer/objects/query", + ] # reads only: macros, then the idle check + + +def test_preheat_material_and_explicit_temperatures(shop: Shop) -> None: + shop.out("-n", "k1se", "preheat", "PETG") + shop.out("-n", "k1se", "preheat", "--nozzle", "200") + shop.out("-n", "k1se", "preheat", "pla", "--bed", "55.5") + shop.out("-n", "u1", "preheat", "ABS", "--head", "T2") + + assert shop.k1se.scripts == [ + "SET_HEATER_TEMPERATURE HEATER=extruder TARGET=235\nSET_HEATER_TEMPERATURE HEATER=heater_bed TARGET=80", + "SET_HEATER_TEMPERATURE HEATER=extruder TARGET=200", + "SET_HEATER_TEMPERATURE HEATER=extruder TARGET=210\nSET_HEATER_TEMPERATURE HEATER=heater_bed TARGET=55.5", + ] + assert shop.u1.scripts == [ + "SET_HEATER_TEMPERATURE HEATER=extruder1 TARGET=250\nSET_HEATER_TEMPERATURE HEATER=heater_bed TARGET=100" + ] + + +@pytest.mark.parametrize( + "argv", + [ + ["preheat"], + ["preheat", "NYLON"], + ["preheat", "--nozzle", "999"], + ["preheat", "--bed", "-1"], + ["preheat", "--nozzle", "hot"], + ["preheat", "--nozzle", "nan"], + ["preheat", "PLA", "--head", "T9"], + ["preheat", "PLA", "--head", "2"], + ["preheat", "PLA", "extra", "words"], + ["preheat", "--colour", "red"], + ["preheat", "--nozzle", "200; M112"], + ], +) +def test_preheat_rejects_bad_arguments_before_contacting_the_printer( + shop: Shop, argv: list[str] +) -> None: + with pytest.raises(ThreeDError): + run(["-n", "k1se", *argv]) + + assert shop.k1se.requests == [] + + +def test_cooldown_pause_resume_alias_scripts(shop: Shop) -> None: + shop.out("-n", "k1se", "cooldown") + shop.out("-n", "k1se", "pause") + shop.out("-n", "k1se", "resume") + + assert shop.k1se.scripts == ["TURN_OFF_HEATERS", "PAUSE", "RESUME"] + + +def test_pause_and_resume_aliases_work_while_a_job_runs_but_cooldown_does_not( + shop: Shop, +) -> None: + shop.k1se.status = u1_status("printing", "part.gcode", 0.4) + + shop.out("-n", "k1se", "pause") + err = shop.fails(PrinterRefused, "-n", "k1se", "cooldown") + + assert shop.k1se.scripts == ["PAUSE"] and "needs an idle printer" in err.message + + +def test_u1_specific_aliases_map_to_the_firmware_macros(shop: Shop) -> None: + shop.out("-n", "u1", "load", "T2") + shop.out("-n", "u1", "unload", "t1") + shop.out("-n", "u1", "clean-nozzle") + shop.out("-n", "u1", "park") + shop.out("-n", "u1", "park", "T3") + shop.out("-n", "u1", "purifier", "hot") + shop.out("-n", "u1", "level-heated") + shop.out("-n", "u1", "level-heated", "--bed", "60") + shop.out("-n", "u1", "level") + + assert shop.u1.scripts == [ + "AUTO_FEEDING EXTRUDER=1 LOAD=1", + "AUTO_FEEDING EXTRUDER=0 UNLOAD=1 STAGE=prepare\nAUTO_FEEDING EXTRUDER=0 UNLOAD=1 STAGE=doing", + "ROUGHLY_CLEAN_NOZZLE", + "PARK_EXTRUDER", + "PARK_EXTRUDER2", + "SET_PURIFIER_MODE MODE=3", + "AUTO_BED_MESH_CALIBRATE TEMP=55", + "AUTO_BED_MESH_CALIBRATE TEMP=60", + "BED_MESH_CALIBRATE", + ] + + +def test_u1_head_arguments_follow_the_fleet_heads(shop: Shop) -> None: + for bad in ( + ["load"], + ["load", "T5"], + ["load", "2"], + ["load", "0"], + ["park", "T0"], + ["purifier", "turbo"], + ): + with pytest.raises(ThreeDError): + run(["-n", "u1", *bad]) + + assert shop.u1.requests == [] + + +def test_an_unusable_head_is_refused_with_its_reason( + shop: Shop, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + data = yaml.safe_load((tmp_path / "fleet.yaml").read_text(encoding="utf-8")) + data["u1"]["heads"][1].update(usable=False, reason="clogged") + (tmp_path / "fleet.yaml").write_text(yaml.safe_dump(data), encoding="utf-8") + + err = shop.fails(PrinterRefused, "-n", "u1", "load", "T2") + + assert ( + "head T2 is marked unusable: clogged" in err.message and shop.u1.requests == [] + ) + + +def test_k1se_filament_aliases_use_the_stock_creality_macros(shop: Shop) -> None: + shop.out("-n", "k1se", "load") + shop.out("-n", "k1se", "unload") + with pytest.raises(UsageError): + run(["-n", "k1se", "load", "T1"]) # the K1 macro has no head argument + + assert shop.k1se.scripts == ["LOAD_MATERIAL", "QUIT_MATERIAL"] + + +@pytest.mark.parametrize( + ("printer", "alias", "words"), + [ + ("k1se", "clean-nozzle", "the stock K1 SE config has no nozzle-clean macro"), + ("k1se", "park", "park"), + ("k1se", "purifier", "purifier"), + ], +) +def test_an_alias_a_model_does_not_have_says_so_and_lists_what_it_has( + shop: Shop, printer: str, alias: str, words: str +) -> None: + err = shop.fails(UsageError, "-n", printer, alias) + + assert ( + f"alias `{alias}` is not defined for k1se (Creality K1 SE, adapter moonraker)" + in err.message + ) + assert ( + "Defined for it: " in err.remediation[0] + and "home" in err.remediation[0] + and "park" not in err.remediation[0] + ) + assert shop.k1se.requests == [] + + +def test_an_alias_a_model_marks_unsupported_gives_the_reason(shop: Shop) -> None: + err = shop.fails(UsageError, "-n", "k1se", "qgl") + + assert ( + "not supported on k1se" in err.message + and "defines no quad_gantry_level" in err.message + ) + assert shop.k1se.requests == [] + + +def test_an_alias_whose_macro_the_firmware_lacks_is_refused_not_guessed(shop: Shop) -> None: + del shop.k1se.macros["LOAD_MATERIAL"] # a K1 without the stock Creality macro + + err = shop.fails(PrinterRefused, "-n", "k1se", "load") + + assert err.message == "k1se does not define LOAD_MATERIAL" + assert shop.k1se.writes() == [] + + +def test_u1_has_no_gantry_to_level(shop: Shop) -> None: + err = shop.fails(UsageError, "-n", "u1", "qgl") + + assert "not supported on u1" in err.message and "defines no quad_gantry_level" in err.message + assert shop.u1.requests == [] + + +def test_alias_help_and_listing_name_what_each_alias_sends(shop: Shop) -> None: + one = shop.out("-n", "u1", "unload", "--help") + table = json.loads(shop.out("-n", "u1", "aliases", "--json")) + k1 = shop.out("-n", "k1se", "aliases") + generic = shop.out("home", "--help") + builtin = shop.out("aliases") + + assert ( + "AUTO_FEEDING EXTRUDER={slot} UNLOAD=1 STAGE=prepare; AUTO_FEEDING EXTRUDER={slot} UNLOAD=1 STAGE=doing" + in one + ) + assert ( + table["load"]["gcode"] == ["AUTO_FEEDING EXTRUDER={slot} LOAD=1"] + and table["qgl"]["unsupported"] is not None + ) + assert ( + "level" in k1 + and "G29" in k1 + and "qgl: NOT supported - " in k1 + and "park" not in k1 + ) + assert ( + "adapter moonraker" in generic + and "model snapmaker u1" not in generic.split("home")[0] + ) + assert "model snapmaker u1:" in builtin and "model creality k1:" in builtin + assert shop.u1.requests == [] + + +def test_an_alias_needs_the_printer_and_name_is_not_accepted_for_positional_subcommands( + shop: Shop, +) -> None: + needs = shop.fails(UsageError, "home") + named = shop.fails(UsageError, "-n", "u1", "status", "u1") + unknown = shop.fails(UsageError, "frobnicate") + unknown_on_printer = shop.fails(UsageError, "-n", "u1", "frobnicate") + + assert "is an alias and needs the printer: 3d printer -n NAME home" in needs.message + assert "takes the printer as NAME" in named.message + assert "unknown subcommand 'frobnicate'" in unknown.message + assert "alias `frobnicate` is not defined for u1" in unknown_on_printer.message + + +def test_a_printers_own_aliases_extend_replace_and_remove_the_table( + shop: Shop, tmp_path: Path +) -> None: + cfg = tmp_path / "xdg" / "3d-cli" + cfg.mkdir(parents=True) + (cfg / "printer-aliases.yaml").write_text( + yaml.safe_dump( + { + "u1": { + "home": {"summary": "home XY only", "kind": "motion", "gcode": ["G28 X Y"]}, + "nozzle-wipe": {"summary": "wipe", "kind": "motion", "gcode": ["ROUGHLY_CLEAN_NOZZLE"]}, + "park": None, + "cooldown": {"unsupported": "the chamber heater needs a manual step first"}, + }, + "k1se": {"home": {"summary": "k1 home", "kind": "motion", "gcode": ["G28 Z"]}}, + } + ), + encoding="utf-8", + ) + # the project file wins over the user file, per alias + (tmp_path / "printer-aliases.yaml").write_text( + yaml.safe_dump({"u1": {"home": {"summary": "project home", "kind": "motion", "gcode": ["G28 Y"]}}}), + encoding="utf-8", + ) + + shop.out("-n", "u1", "home") + shop.out("-n", "u1", "nozzle-wipe") + shop.out("-n", "k1se", "home") + parked = shop.fails(UsageError, "-n", "u1", "park") + cooled = shop.fails(UsageError, "-n", "u1", "cooldown") + + assert shop.u1.scripts == ["G28 Y", "ROUGHLY_CLEAN_NOZZLE"] and shop.k1se.scripts == ["G28 Z"] + assert "disabled for this printer" in parked.message and "chamber heater" in cooled.message + + +def test_a_broken_printer_aliases_file_is_a_config_error(shop: Shop, tmp_path: Path) -> None: + (tmp_path / "printer-aliases.yaml").write_text("u1: [not, a, mapping]\n", encoding="utf-8") + + err = shop.fails(ThreeDError, "-n", "u1", "aliases") + + assert err.exit_code == 2 and "must be a mapping of alias name" in err.message + assert shop.u1.requests == [] + + +@pytest.mark.parametrize( + ("entry", "why"), + [ + ({"summary": "x", "kind": "fast", "gcode": ["G28"]}, "kind"), + ({"summary": "x", "kind": "motion", "gcode": []}, "gcode"), + ( + {"summary": "x", "kind": "motion", "gcode": ["G28 {nope}"]}, + "no arg provides", + ), + ({"summary": "x", "kind": "motion", "gcode": ["G28; M84"]}, "without `;`"), + ({"summary": "x", "kind": "motion", "gcode": ["M112"]}, "emergency stop"), + ( + { + "summary": "x", + "kind": "motion", + "gcode": ["G28"], + "args": {"a": {"type": "text"}}, + }, + "type", + ), + ( + {"summary": "x", "kind": "motion", "gcode": ["G28"], "timeout": 99999}, + "timeout", + ), + ({"summary": "x", "kind": "motion", "gcode": ["G28"], "extra": 1}, "keys"), + ({"kind": "motion", "gcode": ["G28"]}, "summary"), + ], +) +def test_a_broken_alias_entry_is_a_config_error( + entry: dict[str, Any], why: str +) -> None: + with pytest.raises(ThreeDError) as info: + aliases.parse_entries({"mine": entry}, "printer-aliases.yaml u1") + + assert why in info.value.message + + +def test_the_builtin_table_parses_and_every_template_is_valid() -> None: + names = aliases.builtin_names() + + assert { + "home", + "level", + "clean-nozzle", + "load", + "unload", + "preheat", + "cooldown", + "pause", + "resume", + "park", + } <= names + for _label, entries in aliases.builtin_layers(): + for spec in entries.values(): + assert spec.unsupported or spec.kind in macros.KINDS + + +def test_a_creality_stock_printer_has_no_gcode_channel( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "xdg")) + monkeypatch.chdir(tmp_path) + k1 = FakeCreality() + with k1.serve(): + write_devices(tmp_path / "xdg", k1=k1) + for argv in ( + ["macro", "run", "PAUSE", "--yes"], + ["pause"], + ["macro", "list"], + ["load"], + ): + with pytest.raises(UnsupportedOperation) as info: + run(["-n", "k1se", *argv]) + assert ( + "does not support macros and aliases (no G-code channel)" + in info.value.message + ) + + assert k1.writes() == [] + capsys.readouterr() + + +# ---- the rules in isolation ------------------------------------------------------------------ + + +def test_classification_and_gates() -> None: + trusted = frozenset({"PAUSE", "START_PRINT"}) + + assert [ + macros.classify(n) + for n in ( + "pause", + "G28", + "SET_HEATER_TEMPERATURE", + "FIRMWARE_RESTART", + "QUERY_ENDSTOPS", + ) + ] == [macros.JOB, macros.MOTION, macros.HEATING, macros.DANGEROUS, macros.INERT] + job = macros.assess(["PAUSE"], trusted, curated=False) + other = macros.assess(["AUTO_FEEDING EXTRUDER=0"], trusted, curated=False) + curated = macros.assess( + ["AUTO_FEEDING EXTRUDER=0"], trusted, curated=True, kind=macros.MOTION + ) + restart = macros.assess(["FIRMWARE_RESTART"], trusted, curated=True) + assert (job.needs_idle, job.yes_reasons) == (False, ()) + assert other.needs_idle and other.yes_reasons == ( + "AUTO_FEEDING is not in this printer's trusted_macros", + ) + assert curated.needs_idle and curated.yes_reasons == () + assert restart.kind == macros.DANGEROUS and restart.yes_reasons + + +def test_parse_catalog_reads_moonraker_replies_and_rejects_other_shapes() -> None: + cat = macros.parse_catalog( + {"objects": ["webhooks", "gcode_macro Pause", "gcode_macro _x y"]}, + {"G29": "bed", "pause": "p"}, + ) + + assert ( + cat.macros == {"PAUSE", "_X Y"} + and cat.knows("g29") + and cat.describe("PAUSE") == "p" + ) + with pytest.raises(ValueError, match="unexpected shape"): + macros.parse_catalog([], {}) + + +def test_usage_text_names_every_builtin_alias() -> None: + for name in aliases.builtin_names(): + assert name in USAGE, f"`3d printer --help` does not mention the alias {name}" + assert "macro list|run" in USAGE and "aliases [--json]" in USAGE + + +def test_docs_name_every_builtin_alias_and_the_gcode_decision() -> None: + doc = (REPO / "docs" / "commands" / "printer.md").read_text(encoding="utf-8") + + for name in aliases.builtin_names(): + assert f"`{name}" in doc, ( + f"docs/commands/printer.md does not document the alias {name}" + ) + assert "no raw `gcode`" in doc and "3d printer -n u1 macro run" in doc diff --git a/uv.lock b/uv.lock index 0b5f28ba..55378003 100644 --- a/uv.lock +++ b/uv.lock @@ -10,7 +10,7 @@ resolution-markers = [ [[package]] name = "3d-cli" -version = "0.78.0" +version = "0.79.0" source = { virtual = "." } dependencies = [ { name = "manifold3d" },