diff --git a/README.md b/README.md index 4645d8c2..470a746c 100644 --- a/README.md +++ b/README.md @@ -8,3 +8,32 @@ The official Cartographer3D plugin. Documentation can be found at [https://docs.cartographer3d.com/](https://docs.cartographer3d.com/) + +## Kalico multiple probes + +Support is capability-based: Kalico with the central probe registry from +[Kalico PR #972](https://github.com/KalicoCrew/kalico/pull/972) always registers Cartographer as `cartographer`. +`register_as_probe: true` (default) makes it the default probe; `false` registers it as named-only. +Older Kalico and other firmware retain their existing behavior. + +The supported standard mux commands are `PROBE`, `PROBE_ACCURACY`, `QUERY_PROBE`, and `Z_OFFSET_APPLY_PROBE`; +Cartographer extends rather than overrides other probes' handlers. Probe selectors are case-sensitive. + +```gcode +PROBE PROBE=cartographer +BED_MESH_CALIBRATE PROBE=cartographer METHOD=scan +``` + +For bed mesh, omitting `PROBE` follows the default probe. Selected Cartographer uses optimized scanning by default; +other selected or default probes use native automatic meshing. `METHOD=scan` requires Cartographer; +`METHOD=manual` remains native. +Unknown `PROBE` selectors are rejected before movement even with `METHOD=manual`; without a default probe, +nonmanual meshing requires an explicit `PROBE`, while manual meshing remains available. + +Endstop chip names are unchanged: `probe:z_virtual_endstop` with `register_as_probe: true`, +`cartographer_probe:z_virtual_endstop` with `false`. There is no `cartographer:` pin-chip alias. + +Native `CALIBRATE_Z` and `PROBE_Z_ACCURACY` cannot use Cartographer; use the existing +`CARTOGRAPHER_SCAN_CALIBRATE` / `CARTOGRAPHER_TOUCH_CALIBRATE` commands instead. +`[nozzle_cleanup]` selecting Cartographer is rejected at configuration time; if Cartographer is the default, +explicitly select another available probe. Cartographer does not provide nozzle-cleaning support. diff --git a/src/cartographer/adapters/kalico/integrator.py b/src/cartographer/adapters/kalico/integrator.py new file mode 100644 index 00000000..f2df6329 --- /dev/null +++ b/src/cartographer/adapters/kalico/integrator.py @@ -0,0 +1,174 @@ +from __future__ import annotations + +from importlib import import_module +from typing import TYPE_CHECKING, Callable, Protocol, cast, final + +from typing_extensions import override + +from cartographer.adapters.kalico.probe import KalicoCartographerProbe +from cartographer.adapters.klipper.endstop import KlipperEndstop +from cartographer.adapters.klipper_like.integrator import ( + FallbackMacroAdapter, + KlipperLikeAdapters, + KlipperLikeIntegrator, + catch_macro_errors, +) +from cartographer.interfaces.printer import SupportsFallbackMacro + +if TYPE_CHECKING: + from configfile import ConfigWrapper + from gcode import GCodeCommand + from klippy import Printer + + from cartographer.core import MacroRegistration, PrinterCartographer + + +class _ProbeRegistry(Protocol): + def add_probe_object(self, obj: KalicoCartographerProbe, config: ConfigWrapper) -> object: ... + def get_command_probe(self, gcmd: GCodeCommand, default: object = ...) -> object | None: ... + + +class _ProbeList(Protocol): + def get_list(self, printer: Printer) -> _ProbeRegistry: ... + + +class _NativeZCalibration(Protocol): + load_probe: Callable[[object], object] + + +@final +class KalicoIntegrator(KlipperLikeIntegrator): + def __init__(self, adapters: KlipperLikeAdapters) -> None: + super().__init__(adapters, KalicoCartographerProbe) + self._probe_list: _ProbeList | None = None + self._registry_probe: KalicoCartographerProbe | None = None + try: + probe_module = import_module("extras.probe") + except ModuleNotFoundError as error: + if error.name not in ("extras", "extras.probe"): + raise + else: + probe_list = getattr(probe_module, "ProbeList", None) + if all( + callable(getattr(probe_list, method, None)) + for method in ("get_list", "add_probe_object", "get_command_probe") + ): + self._probe_list = cast("_ProbeList", probe_list) + + @override + def setup(self) -> None: + super().setup() + if self._probe_list is not None: + self._printer.register_event_handler("klippy:connect", self._guard_native_z_calibration) + + def _guard_native_z_calibration(self) -> None: + if not self._config.wrapper.has_section("z_calibration"): + return + lookup = cast("Callable[[str, object], object]", self._printer.lookup_object) + helper = lookup("z_calibration", None) + if not callable(getattr(helper, "load_probe", None)): + msg = "Configured [z_calibration] helper requires a callable load_probe" + raise self._config.wrapper.error(msg) + native = cast("_NativeZCalibration", helper) + original = native.load_probe + probe = self._registry_probe + + def load_probe(selected: object) -> object: + if probe is not None and selected is probe: + msg = "Cartographer native Z calibration is unsupported; use Cartographer calibration instead" + raise self._printer.command_error(msg) + return original(selected) + + # Guard the shared helper boundary, including renamed commands and direct callers. + native.load_probe = load_probe + + @override + def register_probe(self, cartographer: PrinterCartographer) -> None: + if self._probe_list is None: + super().register_probe(cartographer) + return + + probe = KalicoCartographerProbe( + self._toolhead, + cartographer.scan_mode, + cartographer.probe_macro, + cartographer.query_probe_macro, + cartographer.config.general, + printer=self._printer, + mcu_probe=KlipperEndstop(self._mcu, cartographer.scan_mode), + ) + if self._config.wrapper.has_section("nozzle_cleanup"): + section = self._config.wrapper.getsection("nozzle_cleanup") + selected = section.get("probe", None) + if selected == probe.probe_name or (selected is None and probe.is_default_probe): + msg = "[nozzle_cleanup] is unsupported with Cartographer; select another probe" + raise section.error(msg) + registry = self._probe_list.get_list(self._printer) + _ = registry.add_probe_object(probe, self._config.wrapper) + self._registry_probe = probe + if not probe.is_default_probe: + for registration in cartographer.probe_macros: + self.register_macro(registration) + + @override + def register_macro(self, registration: MacroRegistration) -> None: + probe = self._registry_probe + if probe is not None and registration.name == "BED_MESH_CALIBRATE": + assert self._probe_list is not None + registry = self._probe_list.get_list(self._printer) + macro = registration.macro + original = self._gcode.register_command(registration.name, None) + if original is not None and isinstance(macro, SupportsFallbackMacro): + macro.set_fallback_macro(FallbackMacroAdapter(registration.name, original)) + + def route_mesh(gcmd: GCodeCommand) -> None: + selected = registry.get_command_probe(gcmd, None) + method = gcmd.get("METHOD", None) + if method is not None: + method = method.lower() + if selected is None and method != "manual": + # Ask the registry for its native missing-default error before any work. + selected = registry.get_command_probe(gcmd) + if selected is probe and method in (None, "scan"): + params = {key: value for key, value in gcmd.get_command_parameters().items() if key != "PROBE"} + command = self._gcode.create_gcode_command(gcmd.get_command(), gcmd.get_commandline(), params) + macro.run(command) + return + if method == "scan": + msg = "METHOD=scan requires the Cartographer probe" + raise gcmd.error(msg) + if original is None: + msg = "The native BED_MESH_CALIBRATE handler is unavailable" + raise gcmd.error(msg) + if selected is probe and method != "manual": + macro.run(gcmd) + else: + original(gcmd) + + self._gcode.register_command(registration.name, catch_macro_errors(route_mesh), desc=macro.description) + return + + if probe is None or registration.name not in ( + "PROBE", + "PROBE_ACCURACY", + "QUERY_PROBE", + "Z_OFFSET_APPLY_PROBE", + ): + super().register_macro(registration) + return + + def run(gcmd: GCodeCommand) -> None: + params = {key: value for key, value in gcmd.get_command_parameters().items() if key != "PROBE"} + command = self._gcode.create_gcode_command(gcmd.get_command(), gcmd.get_commandline(), params) + registration.macro.run(command) + + handler = catch_macro_errors(run) + keys = (probe.probe_name, None) if probe.is_default_probe else (probe.probe_name,) + for key in keys: + self._gcode.register_mux_command( + registration.name, + "PROBE", + key, + handler, + desc=registration.macro.description, + ) diff --git a/src/cartographer/adapters/kalico/probe.py b/src/cartographer/adapters/kalico/probe.py index 0bd5b653..8655af24 100644 --- a/src/cartographer/adapters/kalico/probe.py +++ b/src/cartographer/adapters/kalico/probe.py @@ -6,7 +6,9 @@ if TYPE_CHECKING: from gcode import GCodeCommand + from klippy import Printer + from cartographer.adapters.klipper.endstop import KlipperEndstop from cartographer.interfaces.configuration import GeneralConfig from cartographer.interfaces.printer import ProbeMode, Toolhead from cartographer.macros.probe import ProbeMacro, QueryProbeMacro @@ -21,7 +23,17 @@ def __init__( probe_macro: ProbeMacro, query_probe_macro: QueryProbeMacro, config: GeneralConfig, + *, + printer: Printer | None = None, + mcu_probe: KlipperEndstop | None = None, ) -> None: + self.name = "cartographer" + self.probe_name = "cartographer" + self.is_default_probe = config.register_as_probe + self.printer = printer + if mcu_probe is not None: + self.mcu_probe = mcu_probe + self.probe = probe self.probe_macro = probe_macro self.query_probe_macro = query_probe_macro diff --git a/src/cartographer/adapters/klipper_like/integrator.py b/src/cartographer/adapters/klipper_like/integrator.py index 2cc96b0a..06464d5c 100644 --- a/src/cartographer/adapters/klipper_like/integrator.py +++ b/src/cartographer/adapters/klipper_like/integrator.py @@ -66,7 +66,6 @@ def get_steppers(self) -> list[MCU_stepper]: ... def get_endstops(self) -> list[tuple[MCU_endstop, str]]: ... -@final class KlipperLikeIntegrator(Integrator): def __init__( self, @@ -77,7 +76,9 @@ def __init__( self._printer: Printer = adapters.printer self._mcu: CartographerMcu = adapters.mcu self._toolhead: Toolhead = adapters.toolhead - self._target_probe_class = target_probe_class + self._target_probe_class: Callable[ + [Toolhead, ProbeMode, ProbeMacro, QueryProbeMacro, GeneralConfig], object + ] = target_probe_class self._gcode: GCodeDispatch = self._printer.lookup_object("gcode") @@ -113,6 +114,8 @@ def register_macro(self, registration: MacroRegistration) -> None: @override def register_probe(self, cartographer: PrinterCartographer) -> None: + if not cartographer.config.general.register_as_probe: + return self._printer.add_object( "probe", self._target_probe_class( diff --git a/src/cartographer/core.py b/src/cartographer/core.py index 20ee647a..a60528f2 100644 --- a/src/cartographer/core.py +++ b/src/cartographer/core.py @@ -81,6 +81,8 @@ def __init__(self, adapters: Adapters) -> None: self.probe_macro = ProbeMacro(probe, toolhead) self.query_probe_macro = QueryProbeMacro(probe) + self.probe_macros = self._create_probe_macro_registrations(probe, toolhead) + # Register all macros self.macros = self._create_macro_registrations(probe, toolhead, adapters) @@ -151,7 +153,8 @@ def _create_macro_registrations( registrations: list[MacroRegistration] = [] # Core probe macros - registrations.extend(self._create_probe_macro_registrations(probe, toolhead)) + if self.config.general.register_as_probe: + registrations.extend(self.probe_macros) # Cartographer-specific macros registrations.extend(self._create_cartographer_macro_registrations(probe, toolhead, adapters)) @@ -171,31 +174,16 @@ def _create_macro_registrations( return registrations def _create_probe_macro_registrations(self, probe: Probe, toolhead: Toolhead) -> list[MacroRegistration]: - """Create standard probe macro registrations (only when register_as_probe is true).""" - if not self.config.general.register_as_probe: - return [] - return list( - chain.from_iterable( - [ - self._register_macro("PROBE", self.probe_macro, use_prefix=False), - self._register_macro( - "PROBE_ACCURACY", - ProbeAccuracyMacro(probe, toolhead, lift_speed=self.config.general.lift_speed), - use_prefix=False, - ), - self._register_macro( - "QUERY_PROBE", - self.query_probe_macro, - use_prefix=False, - ), - self._register_macro( - "Z_OFFSET_APPLY_PROBE", - ZOffsetApplyProbeMacro(probe, toolhead, self.config), - use_prefix=False, - ), - ] - ) - ) + """Create reusable standard probe macro registrations.""" + return [ + MacroRegistration("PROBE", self.probe_macro), + MacroRegistration( + "PROBE_ACCURACY", + ProbeAccuracyMacro(probe, toolhead, lift_speed=self.config.general.lift_speed), + ), + MacroRegistration("QUERY_PROBE", self.query_probe_macro), + MacroRegistration("Z_OFFSET_APPLY_PROBE", ZOffsetApplyProbeMacro(probe, toolhead, self.config)), + ] def _create_cartographer_macro_registrations( self, probe: Probe, toolhead: Toolhead, adapters: Adapters diff --git a/src/cartographer/extra.py b/src/cartographer/extra.py index 8e9f62e6..82a001bc 100644 --- a/src/cartographer/extra.py +++ b/src/cartographer/extra.py @@ -16,8 +16,7 @@ def load_config(config: object) -> object: cartographer = PrinterCartographer(adapters) - if cartographer.config.general.register_as_probe: - integrator.register_probe(cartographer) + integrator.register_probe(cartographer) for macro in cartographer.macros: integrator.register_macro(macro) diff --git a/src/cartographer/interfaces/configuration.py b/src/cartographer/interfaces/configuration.py index 22747871..15cc4707 100644 --- a/src/cartographer/interfaces/configuration.py +++ b/src/cartographer/interfaces/configuration.py @@ -175,7 +175,11 @@ class GeneralConfig: "When true, Cartographer registers as the 'probe' printer object and 'probe:' pin chip," " overriding PROBE/PROBE_ACCURACY/QUERY_PROBE/Z_OFFSET_APPLY_PROBE commands." " When false, Cartographer registers its endstop under the 'cartographer_probe:' pin chip" - " and does not claim the 'probe' object, allowing a separate [probe] section to coexist.", + " and does not claim the 'probe' object, allowing a separate [probe] section to coexist." + " On registry-capable Kalico, Cartographer always registers as the named 'cartographer' probe:" + " true makes it the default probe, while false registers it as named-only." + " There, standard probe mux commands extend rather than override other probes' handlers." + " Pin-chip names and existing behavior elsewhere are unchanged.", default=True, ) diff --git a/src/cartographer/macros/docs.py b/src/cartographer/macros/docs.py index 26d4fca5..23f5d43c 100644 --- a/src/cartographer/macros/docs.py +++ b/src/cartographer/macros/docs.py @@ -41,7 +41,8 @@ # Macros in the order they should appear in docs. # (macro_name, macro_class, params_dataclass) MACROS: list[tuple[str, type[Macro], type]] = [ - # Standard probe macros (no CARTOGRAPHER_ prefix) — only registered when register_as_probe is true + # Standard probe macros: default/no-selector entries require register_as_probe=true; + # registry-capable Kalico also registers named mux entries when false. ("PROBE", ProbeMacro, ProbeMacroParams), ("PROBE_ACCURACY", ProbeAccuracyMacro, ProbeAccuracyParams), ("QUERY_PROBE", QueryProbeMacro, QueryProbeMacroParams), diff --git a/src/cartographer/runtime/loader.py b/src/cartographer/runtime/loader.py index c35ba11e..787868cf 100644 --- a/src/cartographer/runtime/loader.py +++ b/src/cartographer/runtime/loader.py @@ -36,11 +36,10 @@ def init_runtime(config: object) -> tuple[Adapters, Integrator]: if env == Environment.Kalico: from cartographer.adapters.kalico.adapters import KalicoAdapters - from cartographer.adapters.kalico.probe import KalicoCartographerProbe - from cartographer.adapters.klipper_like.integrator import KlipperLikeIntegrator + from cartographer.adapters.kalico.integrator import KalicoIntegrator adapters = KalicoAdapters(cast("KlipperConfigWrapper", config)) - return adapters, KlipperLikeIntegrator(adapters, KalicoCartographerProbe) + return adapters, KalicoIntegrator(adapters) msg = f"Unsupported environment: {env}" raise RuntimeError(msg) diff --git a/tests/klipper/test_integrator.py b/tests/klipper/test_integrator.py index c6496643..ae83008a 100644 --- a/tests/klipper/test_integrator.py +++ b/tests/klipper/test_integrator.py @@ -61,6 +61,17 @@ def test_non_probe_chip_registers_plain_endstop(self, integrator: KlipperLikeInt class TestRegisterProbe: + def test_named_only_skips_legacy_registration( + self, integrator: KlipperLikeIntegrator, adapters: Mock, probe_class: Mock + ) -> None: + cartographer = Mock() + cartographer.config.general.register_as_probe = False + + integrator.register_probe(cartographer) + + probe_class.assert_not_called() + adapters.printer.add_object.assert_not_called() + def test_invokes_probe_callable_with_expected_deps( self, integrator: KlipperLikeIntegrator, adapters: Mock, probe_class: Mock ) -> None: diff --git a/tests/klipper/test_kalico_integrator.py b/tests/klipper/test_kalico_integrator.py new file mode 100644 index 00000000..8c8ded96 --- /dev/null +++ b/tests/klipper/test_kalico_integrator.py @@ -0,0 +1,417 @@ +from __future__ import annotations + +from types import ModuleType +from typing import TYPE_CHECKING, Callable, cast, final +from unittest.mock import Mock + +import pytest + +from cartographer.adapters.kalico.integrator import KalicoIntegrator +from cartographer.adapters.kalico.probe import KalicoCartographerProbe +from cartographer.adapters.klipper.endstop import KlipperEndstop +from cartographer.core import MacroRegistration +from cartographer.extra import load_config +from cartographer.interfaces.printer import Position +from cartographer.macros.bed_mesh.scan_mesh import BedMeshCalibrateMacro +from tests.mocks.params import MockParams + +if TYPE_CHECKING: + from gcode import GCodeCommand + from pytest_mock import MockerFixture + + +@final +class Command(MockParams): + error = ValueError + + def __init__(self, command: str, commandline: str, params: dict[str, str]) -> None: + super().__init__() + self.command = command + self.commandline = commandline + self.params = params + + def get_command(self) -> str: + return self.command + + def get_commandline(self) -> str: + return self.commandline + + def get_command_parameters(self) -> dict[str, str]: + return self.params + + +PROBE_COMMANDS = ("PROBE", "PROBE_ACCURACY", "QUERY_PROBE", "Z_OFFSET_APPLY_PROBE") + + +@pytest.fixture +def host(mocker: MockerFixture) -> Mock: + host = Mock() + host.objects = {"gcode": host.gcode} + + def lookup(name: str, default: object = None) -> object: + return host.objects.get(name, default) + + host.printer.lookup_object.side_effect = lookup + host.printer.command_error = ValueError + + def config_error(message: str) -> ValueError: + return ValueError(message) + + host.config.wrapper.error.side_effect = config_error + host.config.wrapper.has_section.return_value = False + host.gcode.create_gcode_command.side_effect = Command + host.gcode.register_command.return_value = None + host.device.config.general.register_as_probe = True + host.device.config.general.lift_speed = 6.0 + host.device.macros = [] + host.device.probe_macros = [MacroRegistration(name, Mock(spec=["run", "description"])) for name in PROBE_COMMANDS] + module = ModuleType("extras.probe") + host.api.get_list.return_value = host.registry + module.__dict__["ProbeList"] = host.api + host.module = module + _ = mocker.patch("cartographer.adapters.kalico.integrator.import_module", return_value=module) + return host + + +def register(host: Mock, is_default: bool = True) -> tuple[KalicoIntegrator, KalicoCartographerProbe]: + host.device.config.general.register_as_probe = is_default + integrator = KalicoIntegrator(host) + integrator.register_probe(host.device) + probe = host.registry.add_probe_object.call_args.args[0] + assert type(probe) is KalicoCartographerProbe + return integrator, probe + + +def command_handler(host: Mock) -> Callable[[Command], None]: + return host.gcode.register_command.call_args.args[1] + + +def assert_copy(host: Mock, original: Command, received: Command) -> None: + assert received is not original + assert received.params is not original.params + assert received.params == {key: value for key, value in original.params.items() if key != "PROBE"} + assert received.command == original.command + assert received.commandline == original.commandline + host.gcode.create_gcode_command.assert_called_once_with(original.command, original.commandline, received.params) + + +@pytest.mark.parametrize("is_default", [True, False]) +def test_config_registers_real_probe(mocker: MockerFixture, host: Mock, is_default: bool) -> None: + host.device.config.general.register_as_probe = is_default + integrator = KalicoIntegrator(host) + _ = mocker.patch("cartographer.extra.init_runtime", return_value=(host, integrator)) + _ = mocker.patch("cartographer.extra.PrinterCartographer", return_value=host.device) + for method in ("setup", "register_coil_temperature_sensor", "register_endstop_pin"): + _ = mocker.patch.object(integrator, method) + assert load_config(host.config.wrapper) is host.device + probe = host.registry.add_probe_object.call_args.args[0] + assert type(probe) is KalicoCartographerProbe + assert probe.name == probe.probe_name == "cartographer" + assert probe.printer is host.printer + assert probe.is_default_probe is is_default + assert probe.probe is host.device.scan_mode + assert type(probe.mcu_probe) is KlipperEndstop + assert probe.mcu_probe.mcu is host.mcu + assert probe.mcu_probe.endstop is host.device.scan_mode + host.api.get_list.assert_called_once_with(host.printer) + host.registry.add_probe_object.assert_called_once_with(probe, host.config.wrapper) + host.printer.add_object.assert_not_called() + + +@pytest.mark.parametrize("missing", ["module", "class", "get_list", "add_probe_object", "get_command_probe"]) +@pytest.mark.parametrize("is_default", [True, False]) +def test_legacy_fallback(mocker: MockerFixture, host: Mock, missing: str, is_default: bool) -> None: + if missing == "module": + _ = mocker.patch( + "cartographer.adapters.kalico.integrator.import_module", + side_effect=ModuleNotFoundError(name="extras.probe"), + ) + elif missing == "class": + del host.module.ProbeList + else: + setattr(host.api, missing, None) + host.device.config.general.register_as_probe = is_default + KalicoIntegrator(host).register_probe(host.device) + if is_default: + name, probe = host.printer.add_object.call_args.args + assert name == "probe" and type(probe) is KalicoCartographerProbe + assert not hasattr(probe, "mcu_probe") + host.printer.add_object.assert_called_once() + else: + host.printer.add_object.assert_not_called() + host.registry.add_probe_object.assert_not_called() + + +def test_dependency_and_registry_errors_propagate(mocker: MockerFixture, host: Mock) -> None: + error = ModuleNotFoundError(name="dependency") + importer = mocker.patch("cartographer.adapters.kalico.integrator.import_module", side_effect=error) + with pytest.raises(ModuleNotFoundError) as caught: + _ = KalicoIntegrator(host) + assert caught.value is error + importer.side_effect = None + importer.return_value = host.module + conflict = ValueError("native registry conflict") + host.registry.add_probe_object.side_effect = conflict + with pytest.raises(ValueError) as registry_error: + _ = register(host) + assert registry_error.value is conflict + host.printer.add_object.assert_not_called() + + +@pytest.mark.parametrize("is_default", [True, False]) +def test_four_mux_commands_copy_only_selector(host: Mock, is_default: bool) -> None: + integrator, _ = register(host, is_default) + if is_default: + for registration in host.device.probe_macros: + integrator.register_macro(registration) + calls = host.gcode.register_mux_command.call_args_list + assert [(c.args[0], c.args[1], c.args[2]) for c in calls] == [ + (name, "PROBE", key) + for name in PROBE_COMMANDS + for key in (["cartographer", None] if is_default else ["cartographer"]) + ] + for registration in host.device.probe_macros: + call = next(c for c in calls if c.args[0] == registration.name) + assert call.kwargs["desc"] == registration.macro.description + if is_default: + assert next(c.args[3] for c in calls if c.args[:3] == (registration.name, "PROBE", None)) is call.args[3] + original = Command(registration.name, "original line", {"PROBE": "cartographer", "EXTRA": "retained"}) + call.args[3](original) + assert_copy(host, original, registration.macro.run.call_args.args[0]) + assert original.params == {"PROBE": "cartographer", "EXTRA": "retained"} + host.gcode.create_gcode_command.reset_mock() + host.gcode.register_command.assert_not_called() + + +@pytest.mark.parametrize("mesh", [False, True]) +def test_registered_callbacks_catch_macro_errors(host: Mock, mesh: bool) -> None: + integrator, probe = register(host) + host.registry.get_command_probe.return_value = probe + name = "BED_MESH_CALIBRATE" if mesh else "PROBE" + macro = Mock(spec=["run", "description"]) + macro.run.side_effect = RuntimeError("macro failure") + integrator.register_macro(MacroRegistration(name, macro)) + callback = command_handler(host) if mesh else host.gcode.register_mux_command.call_args.args[3] + command = Command(name, name, {"PROBE": "cartographer", "METHOD": "scan"}) + with pytest.raises(command.error, match="^macro failure$"): + _ = callback(command) + + +@pytest.mark.parametrize("other_first", [True, False]) +def test_named_mux_preserves_other_and_default_callbacks(host: Mock, other_first: bool) -> None: + other = Mock() + existing = {(name, key): other for name in PROBE_COMMANDS for key in ("other", None)} + callbacks: dict[tuple[str, str | None], Callable[[Command], None]] = dict(existing) if other_first else {} + + def capture(name: str, key: str, value: str | None, handler: Callable[[Command], None], **_kwargs: object) -> None: + assert key == "PROBE" + assert (name, value) not in callbacks + callbacks[name, value] = handler + + host.gcode.register_mux_command.side_effect = capture + _ = register(host, False) + if not other_first: + assert not callbacks.keys() & existing.keys() + callbacks.update(existing) + for name in PROBE_COMMANDS: + assert callbacks[name, "other"] is callbacks[name, None] is other + host.gcode.register_command.assert_not_called() + + +@pytest.mark.parametrize(("legacy", "name"), [(True, "PROBE"), (False, "CARTOGRAPHER_SCAN_PROBE")]) +def test_direct_commands_use_shared_route(host: Mock, legacy: bool, name: str) -> None: + if legacy: + host.api.get_command_probe = None + integrator, macro = KalicoIntegrator(host), Mock(spec=["run", "description"]) + integrator.register_probe(host.device) + integrator.register_macro(MacroRegistration(name, macro)) + original = Command(name, name, {"PROBE": "retained"}) + command_handler(host)(original) + macro.run.assert_called_once_with(original) + host.gcode.register_mux_command.assert_not_called() + host.gcode.create_gcode_command.assert_not_called() + + +@pytest.mark.parametrize( + ("selection", "params", "route"), + [ + ("cartographer", {"PROBE": "cartographer"}, "copy"), + ("cartographer", {"PROBE": "cartographer", "METHOD": "scan"}, "copy"), + ("cartographer", {}, "copy"), + ("cartographer", {"METHOD": "scan"}, "copy"), + ("cartographer", {"PROBE": "cartographer", "METHOD": "automatic"}, "fallback"), + ("other", {"PROBE": "other"}, "native"), + ("other", {"PROBE": "other", "METHOD": "automatic"}, "native"), + ("other", {}, "native"), + ("other", {"METHOD": "automatic"}, "native"), + ("none", {"METHOD": "manual"}, "native"), + ("cartographer", {"PROBE": "cartographer", "METHOD": "manual"}, "native"), + ("other", {"PROBE": "other", "METHOD": "scan"}, "scan error"), + ("unknown", {"PROBE": "unknown"}, "unknown"), + ("unknown", {"PROBE": "unknown", "METHOD": "manual"}, "unknown"), + ("none", {}, "No default probe"), + ], +) +def test_mesh_routes_original_or_copy( + mocker: MockerFixture, host: Mock, selection: str, params: dict[str, str], route: str +) -> None: + integrator, probe = register(host, selection == "cartographer" and "PROBE" not in params) + macro, native = Mock(spec=BedMeshCalibrateMacro, adapter=host.mesh), Mock(side_effect=host.toolhead.move) + host.gcode.register_command.return_value = native + _ = mocker.patch("cartographer.adapters.klipper_like.integrator.GCodeCommand", Command) + integrator.register_macro(MacroRegistration("BED_MESH_CALIBRATE", macro)) + assert host.gcode.register_command.call_args_list[0].args == ("BED_MESH_CALIBRATE", None) + fallback = macro.set_fallback_macro.call_args.args[0] + macro.run.side_effect = fallback.run if route == "fallback" else host.mesh.clear_mesh + selected = {"cartographer": probe, "other": object(), "none": None, "unknown": None}[selection] + host.registry.get_command_probe.return_value = selected + if selection == "unknown": + host.registry.get_command_probe.side_effect = ValueError("unknown") + elif route == "No default probe": + host.registry.get_command_probe.side_effect = [None, ValueError(route)] + original = Command("BED_MESH_CALIBRATE", "original mesh line", {**params, "PROFILE": "retained"}) + before = original.params.copy() + if route in ("scan error", "unknown", "No default probe"): + with pytest.raises(ValueError, match="requires the Cartographer" if route == "scan error" else route): + command_handler(host)(original) + macro.run.assert_not_called() + native.assert_not_called() + host.gcode.create_gcode_command.assert_not_called() + host.mesh.clear_mesh.assert_not_called() + host.mesh.apply_mesh.assert_not_called() + host.toolhead.move.assert_not_called() + else: + command_handler(host)(original) + if route == "copy": + assert_copy(host, original, macro.run.call_args.args[0]) + native.assert_not_called() + else: + native.assert_called_once_with(original) + assert native.call_args.args[0] is original + host.gcode.create_gcode_command.assert_not_called() + if route == "fallback": + macro.run.assert_called_once_with(original) + else: + macro.run.assert_not_called() + assert host.registry.get_command_probe.call_args_list[0].args == (original, None) + assert original.params == before + + +def test_missing_native_errors_but_named_scan_works(host: Mock) -> None: + integrator, probe = register(host, False) + host.registry.get_command_probe.return_value = probe + macro = Mock(spec=BedMeshCalibrateMacro, adapter=host.mesh) + macro.run.side_effect = host.mesh.clear_mesh + integrator.register_macro(MacroRegistration("BED_MESH_CALIBRATE", macro)) + original = Command("BED_MESH_CALIBRATE", "mesh", {"PROBE": "cartographer", "METHOD": "automatic"}) + with pytest.raises(ValueError, match="native BED_MESH_CALIBRATE.*unavailable"): + command_handler(host)(original) + macro.run.assert_not_called() + host.toolhead.move.assert_not_called() + macro.adapter.clear_mesh.assert_not_called() + original.params["METHOD"] = "scan" + command_handler(host)(original) + assert_copy(host, original, macro.run.call_args.args[0]) + assert original.params == {"PROBE": "cartographer", "METHOD": "scan"} + macro.set_fallback_macro.assert_not_called() + + +def test_real_probe_native_rapid_diagnostic_and_point_contract(host: Mock) -> None: + _, probe = register(host) + assert probe.name == "cartographer" + assert not hasattr(probe.mcu_probe, "begin_collect_distance") + command = Command("BED_MESH_CALIBRATE", "rapid scan", {"METHOD": "rapid_scan", "LIFT_SPEED": "8"}) + host.device.scan_mode.perform_probe.assert_not_called() + host.toolhead.move.assert_not_called() + host.toolhead.get_position.return_value = Position(10, 20, 5) + host.device.scan_mode.perform_probe.return_value = 1.5 + host.device.scan_mode.offset.as_tuple.return_value = (1.0, 2.0, 3.0) + native_command = cast("GCodeCommand", cast("object", command)) + assert probe.run_probe(native_command, object()) == [10, 20, 1.5] + host.device.scan_mode.perform_probe.assert_called_once_with() + assert probe.get_offsets() == (1.0, 2.0, 3.0) + assert probe.get_lift_speed(native_command) == 8 and probe.get_lift_speed() == 6 + for triggered in (True, False): + host.device.scan_mode.query_is_triggered.return_value = triggered + assert probe.mcu_probe.query_endstop(12.5) == int(triggered) + host.device.scan_mode.query_is_triggered.assert_called_with(12.5) + + +@pytest.mark.parametrize("selection", ["selected", "default", "other", "missing"]) +def test_z_guard_connect_shared_boundary(mocker: MockerFixture, host: Mock, selection: str) -> None: + integrator, probe = register(host) + _ = mocker.patch.object(integrator, "_configure_macro_logger") + integrator.setup() + host.config.wrapper.has_section.return_value = True + original, motion = Mock(), host.toolhead.move + helper = Mock(load_probe=original) + host.objects["z_calibration"] = None if selection == "missing" else helper + callback = next( + c.args[1] for c in host.printer.register_event_handler.call_args_list if c.args[0] == "klippy:connect" + ) + if selection == "missing": + with pytest.raises(ValueError, match="z_calibration.*load_probe"): + callback() + else: + callback() + selected = object() if selection == "other" else probe + host.registry.get_command_probe.return_value = selected + + def native_command(command: Command) -> object: + result = helper.load_probe(host.registry.get_command_probe(command)) + motion() + return result + + params = {} if selection == "default" else {"PROBE": "other" if selection == "other" else "cartographer"} + command = Command("RENAMED_CALIBRATE_Z", "native alias", {**params, "OPTIONAL_OVERRIDE": "7"}) + if selection == "other": + assert native_command(command) is original.return_value + original.assert_called_once_with(selected) + motion.assert_called_once_with() + else: + with pytest.raises(ValueError, match="native Z calibration.*unsupported.*Cartographer calibration"): + _ = native_command(command) + if selection != "other": + original.assert_not_called() + motion.assert_not_called() + + +@pytest.mark.parametrize("legacy", [False, True]) +def test_z_guard_inactive(mocker: MockerFixture, host: Mock, legacy: bool) -> None: + if legacy: + host.api.get_command_probe = None + integrator = KalicoIntegrator(host) + _ = mocker.patch.object(integrator, "_configure_macro_logger") + integrator.setup() + helper = Mock() + original = helper.load_probe + host.objects["z_calibration"] = helper + callbacks = [c.args[1] for c in host.printer.register_event_handler.call_args_list if c.args[0] == "klippy:connect"] + if legacy: + assert not callbacks + else: + assert len(callbacks) == 1 + callbacks[0]() + assert helper.load_probe is original + + +@pytest.mark.parametrize( + ("is_default", "selector", "reject"), + [(True, None, True), (False, None, False), (False, "cartographer", True), (True, "other", False)], +) +def test_nozzle_config(host: Mock, is_default: bool, selector: str | None, reject: bool) -> None: + host.config.wrapper.has_section.return_value = True + section = host.config.wrapper.getsection.return_value + section.get.return_value = selector + section.error.side_effect = host.config.wrapper.error.side_effect + if reject: + with pytest.raises(ValueError, match="nozzle_cleanup.*unsupported.*Cartographer"): + _ = register(host, is_default) + host.api.get_list.assert_not_called() + host.registry.add_probe_object.assert_not_called() + host.printer.add_object.assert_not_called() + else: + _ = register(host, is_default) + host.registry.add_probe_object.assert_called_once() + section.get.assert_called_once_with("probe", None) + section.getfloat.assert_not_called() diff --git a/tests/test_core.py b/tests/test_core.py index 459f7cfc..e5859b76 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -1,12 +1,17 @@ +from __future__ import annotations + from dataclasses import replace +from typing import TYPE_CHECKING from unittest.mock import Mock import pytest from cartographer.core import PrinterCartographer -from cartographer.interfaces.configuration import Configuration from cartographer.macros.message import MessageMacro -from cartographer.runtime.adapters import Adapters + +if TYPE_CHECKING: + from cartographer.interfaces.configuration import Configuration + from cartographer.runtime.adapters import Adapters @pytest.fixture @@ -36,6 +41,21 @@ def test_core_probe_macros_registered(self, mock_adapters: Adapters): expected = {"PROBE", "PROBE_ACCURACY", "QUERY_PROBE", "Z_OFFSET_APPLY_PROBE"} assert expected.issubset(registered_names), f"Missing core probe macros: {expected - registered_names}" + @pytest.mark.parametrize("is_default", [True, False]) + def test_reusable_probe_macros(self, mock_adapters: Adapters, is_default: bool): + mock_adapters.config.general = replace(mock_adapters.config.general, register_as_probe=is_default) + cartographer = PrinterCartographer(mock_adapters) + group = {reg.name: reg.macro for reg in cartographer.probe_macros} + assert set(group) == {"PROBE", "PROBE_ACCURACY", "QUERY_PROBE", "Z_OFFSET_APPLY_PROBE"} + assert group["PROBE"] is cartographer.probe_macro + assert group["QUERY_PROBE"] is cartographer.query_probe_macro + visible = {reg.name: reg.macro for reg in cartographer.macros} + if is_default: + assert all(visible[name] is macro for name, macro in group.items()) + else: + assert set(group).isdisjoint(visible) + assert visible["CARTOGRAPHER_SCAN_PROBE"] is cartographer.probe_macro + def test_cartographer_prefixed_macros_registered(self, mock_adapters: Adapters): """Verify all CARTOGRAPHER_ prefixed macros are registered.""" cartographer = PrinterCartographer(mock_adapters) diff --git a/typings/gcode.pyi b/typings/gcode.pyi index 1c399ca4..73b486a2 100644 --- a/typings/gcode.pyi +++ b/typings/gcode.pyi @@ -10,6 +10,8 @@ class GCodeCommand: def respond_raw(self, msg: str) -> None: ... def respond_info(self, msg: str, log: bool = True) -> None: ... def get_command_parameters(self) -> dict[str, str]: ... + def get_command(self) -> str: ... + def get_commandline(self) -> str: ... @overload def get( self, @@ -96,6 +98,14 @@ class GCodeDispatch: when_not_ready: bool = False, desc: str | None = None, ) -> Callable[[GCodeCommand], None] | None: ... + def register_mux_command( + self, + cmd: str, + key: str, + value: str | None, + func: Callable[[GCodeCommand], None], + desc: str | None = None, + ) -> None: ... def create_gcode_command( self, command: str,