#!/usr/bin/env python3 """Record a real LinuxCNC command/stat sequence for the selected case.""" from __future__ import annotations import argparse import hashlib import json import os import shutil import signal import subprocess import tempfile import time from pathlib import Path from typing import Any import linuxcnc import hal from run_axis3_tutorial_native import FILES, LINUXCNC_ROOT, ROOT, SOURCE, axis_window, cleanup_case_processes, cleanup_linuxcnc_services, sha256 CASE = "linuxcnc/configs/sim/axis/vismach/3axis-tutorial" SCHEMA = "axis3-tutorial-native-sequence.v1" def canonical(value: Any) -> str: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def finite_list(value: Any) -> list[float] | None: try: return [float(item) for item in value] except (TypeError, ValueError): return None def snapshot(stat: Any) -> dict[str, Any]: stat.poll() return { "task_state": getattr(stat, "task_state", None), "task_mode": getattr(stat, "task_mode", None), "interp_state": getattr(stat, "interp_state", None), "exec_state": getattr(stat, "exec_state", None), "motion_line": getattr(stat, "motion_line", None), "current_line": getattr(stat, "current_line", None), "actual_position": finite_list(getattr(stat, "actual_position", None)), "command_position": finite_list(getattr(stat, "command_position", None)), "joint_actual": finite_list(getattr(stat, "joint_actual_position", None)), "homed": list(getattr(stat, "homed", [])) if getattr(stat, "homed", None) is not None else None, "current_tool": getattr(stat, "tool_in_spindle", None), "spindle_speed": getattr(stat, "spindle_speed", None), } def wait_complete(command: Any, timeout: float = 8.0) -> None: command.wait_complete(int(timeout)) def wait_stat(stat: Any, predicate: Any, timeout: float = 8.0) -> None: deadline = time.monotonic() + timeout while time.monotonic() < deadline: stat.poll() if predicate(stat): return time.sleep(0.05) raise RuntimeError("native_stat_transition_timeout") def run(output: Path) -> dict[str, Any]: if not SOURCE.is_dir(): raise SystemExit(f"selected_case_missing:{SOURCE}") service_cleanup = cleanup_linuxcnc_services() with tempfile.TemporaryDirectory(prefix="linuxcnc-3axis-tutorial-sequence-") as directory: case_dir = Path(directory) / "3axis-tutorial" case_dir.mkdir() for name in FILES: shutil.copy2(SOURCE / name, case_dir / name) (case_dir / "3axis-tutorial").chmod(0o755) update = subprocess.run(["update_ini", "-f", str(case_dir / "3axis-tutorial.ini")], cwd=LINUXCNC_ROOT, env=os.environ.copy(), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False) source = case_dir / "sequence.ngc" source.write_text("G21 G90\nG0 X0 Y0 Z0\nG1 X1 Y1 F60\nG91\nG1 Z-1 F30\nG90\nM2\n", encoding="utf-8") environment = os.environ.copy() environment["DISPLAY"] = environment.get("DISPLAY", ":1") environment["AXIS_OPEN_FILE"] = str(source) log_path = case_dir / "native-sequence.log" with log_path.open("w", encoding="utf-8") as log: process = subprocess.Popen(["linuxcnc", str(case_dir / "3axis-tutorial.ini")], cwd=LINUXCNC_ROOT, env=environment, stdout=log, stderr=subprocess.STDOUT, start_new_session=True) window = None try: deadline = time.monotonic() + 20 while time.monotonic() < deadline and process.poll() is None: window = axis_window(environment) if window: break time.sleep(0.25) if not window: return {"schema": SCHEMA, "status": "blocked", "reason": "axis_window_missing", "process_returncode": process.poll(), "service_cleanup": service_cleanup} command = linuxcnc.command() stat = linuxcnc.stat() rows: list[dict[str, Any]] = [] def record(operation: str, action: Any, predicate: Any | None = None) -> None: before = snapshot(stat) error = None try: action() wait_complete(command) if predicate is not None: wait_stat(stat, predicate) except Exception as cause: # native API reports errors asynchronously error = f"{type(cause).__name__}:{cause}" after = snapshot(stat) rows.append({"operation": operation, "before": before, "after": after, "error": error}) def toolchange() -> None: before = snapshot(stat) error = None button_set = False try: command.mode(linuxcnc.MODE_MDI) wait_complete(command) command.mdi("T1 M6") deadline = time.monotonic() + 8.0 while time.monotonic() < deadline: result = subprocess.run(["halcmd", "setp", "hal_manualtoolchange.change_button", "1"], cwd=LINUXCNC_ROOT, env=os.environ.copy(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False) if result.returncode == 0: button_set = True break time.sleep(0.05) if not button_set: raise RuntimeError("manual_toolchange_ack_pin_unavailable") wait_stat(stat, lambda item: item.tool_in_spindle == 1) subprocess.run(["halcmd", "setp", "hal_manualtoolchange.change_button", "0"], cwd=LINUXCNC_ROOT, env=os.environ.copy(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False) wait_complete(command) except Exception as cause: error = f"{type(cause).__name__}:{cause}" finally: if button_set: try: hal.set_p("hal_manualtoolchange.change_button", "0") except Exception: pass rows.append({"operation": "TOOL_CHANGE_T1", "before": before, "after": snapshot(stat), "error": error, "ack": "hal_manualtoolchange.change_button"}) record("STATE_ESTOP_RESET", lambda: command.state(linuxcnc.STATE_ESTOP_RESET), lambda item: item.task_state != linuxcnc.STATE_ESTOP) record("STATE_ON", lambda: command.state(linuxcnc.STATE_ON), lambda item: item.task_state == linuxcnc.STATE_ON) record("MODE_MANUAL", lambda: command.mode(linuxcnc.MODE_MANUAL), lambda item: item.task_mode == linuxcnc.MODE_MANUAL) record("HOME_Z", lambda: command.home(2), lambda item: bool(item.homed[2])) record("HOME_X", lambda: command.home(0), lambda item: bool(item.homed[0])) record("HOME_Y", lambda: command.home(1), lambda item: bool(item.homed[1])) record("JOG_X_INCREMENT", lambda: command.jog(linuxcnc.JOG_INCREMENT, 0, 0, 5.0, 1.0), lambda item: item.actual_position is not None and abs(item.actual_position[0]) > 1e-6) record("JOG_Y_INCREMENT", lambda: command.jog(linuxcnc.JOG_INCREMENT, 0, 1, -5.0, 1.0), lambda item: item.actual_position is not None and abs(item.actual_position[1]) > 1e-6) record("JOG_Z_INCREMENT", lambda: command.jog(linuxcnc.JOG_INCREMENT, 0, 2, -2.0, 0.5), lambda item: item.actual_position is not None and abs(item.actual_position[2]) > 1e-6) record("MODE_MDI", lambda: (command.teleop_enable(False), command.mode(linuxcnc.MODE_MDI)), lambda item: item.task_mode == linuxcnc.MODE_MDI) record("MDI_G21_G90", lambda: command.mdi("G21 G90")) record("MDI_G0", lambda: command.mdi("G0 X0 Y0 Z0")) record("MDI_G1", lambda: command.mdi("G1 X1 Y1 F60")) record("PROGRAM_OPEN", lambda: command.program_open(str(source))) record("MODE_AUTO", lambda: command.mode(linuxcnc.MODE_AUTO)) record("PROGRAM_RUN", lambda: command.auto(linuxcnc.AUTO_RUN, 0)) record("PROGRAM_PAUSE", lambda: command.auto(linuxcnc.AUTO_PAUSE)) record("PROGRAM_RESUME", lambda: command.auto(linuxcnc.AUTO_RESUME)) record("PROGRAM_ABORT", lambda: command.abort()) toolchange() sequence_hash = hashlib.sha256(canonical(rows).encode()).hexdigest() return { "schema": SCHEMA, "status": "verified" if all(row.get("error") is None for row in rows if "error" in row) else "blocked", "case": CASE, "entrypoint": f"{CASE}/3axis-tutorial.ini", "command": ["linuxcnc", "/3axis-tutorial.ini"], "cwd": "linuxcnc_source_root", "axis_panel": window, "update_ini": {"returncode": update.returncode, "stdout_sha256": hashlib.sha256(update.stdout.encode()).hexdigest()}, "exact_source_files": [{"path": f"{CASE}/{name}", "sha256": sha256(SOURCE / name)} for name in FILES], "program_source_sha256": sha256(source), "operations": rows, "sequence_sha256": sequence_hash, "host_boundaries": [], "service_cleanup": service_cleanup, "native_log_sha256": hashlib.sha256(log_path.read_bytes()).hexdigest(), "process_returncode": process.returncode, } finally: if process.poll() is None: try: os.killpg(process.pid, signal.SIGTERM) except ProcessLookupError: pass try: process.wait(timeout=5) except subprocess.TimeoutExpired: try: os.killpg(process.pid, signal.SIGKILL) except ProcessLookupError: pass process.wait(timeout=2) cleanup_case_processes(case_dir) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--output", type=Path, default=Path("coverage/axis3-tutorial-native-sequence.report.json")) args = parser.parse_args() report = run(args.output) unsigned = {key: value for key, value in report.items() if key != "report_sha256"} report["report_sha256"] = hashlib.sha256(canonical(unsigned).encode()).hexdigest() destination = args.output if args.output.is_absolute() else ROOT / args.output destination.parent.mkdir(parents=True, exist_ok=True) destination.write_text(json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n", encoding="utf-8") print(json.dumps(report, ensure_ascii=False, sort_keys=True)) return 0 if report["status"] == "verified" else 1 if __name__ == "__main__": raise SystemExit(main())