#!/usr/bin/env python3 """Run the single source-owned 3axis-tutorial case and require AXIS panel evidence.""" from __future__ import annotations import argparse import hashlib import json import os import re import shutil import signal import subprocess import tempfile import time from pathlib import Path ROOT = Path(__file__).resolve().parents[1] LINUXCNC_ROOT = ROOT / "linuxcnc" SOURCE = ROOT / "linuxcnc/configs/sim/axis/vismach/3axis-tutorial" CASE = "linuxcnc/configs/sim/axis/vismach/3axis-tutorial" PROGRAM = Path("linuxcnc/configs/nc_files/3dtest.ngc") FILES = ("3axis-tutorial.ini", "3axis-tutorial.hal", "3axis-tutorial", "tool.tbl") SCHEMA = "axis3-tutorial-native-run.v1" def sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def axis_window(environment: dict[str, str]) -> dict[str, object] | None: result = subprocess.run(["xwininfo", "-root", "-tree"], env=environment, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, check=False) pattern = re.compile(r'^\s*(0x[0-9a-f]+)\s+"([^"\n]*3axis-tutorial[^"\n]*)": \("axis" "Axis"\)\s+(\d+)x(\d+)\+', re.I) for line in result.stdout.splitlines(): match = pattern.search(line) if match and int(match.group(3)) >= 600 and int(match.group(4)) >= 400: return {"window_id": match.group(1), "title": match.group(2), "class": "axis", "instance": "Axis", "width": int(match.group(3)), "height": int(match.group(4)), "raw": line.strip()} return None def process_rows() -> list[tuple[int, str]]: result = subprocess.run(["ps", "-eo", "pid=,args="], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, check=False) rows = [] for line in result.stdout.splitlines(): match = re.match(r"\s*(\d+)\s+(.*)$", line) if match: rows.append((int(match.group(1)), match.group(2))) return rows def cleanup_case_processes(case_dir: Path | None = None) -> list[int]: """Stop only LinuxCNC services created by this runner's temporary case dirs.""" marker = str(case_dir) if case_dir is not None else "/tmp/linuxcnc-3axis-tutorial-run-" killed: list[int] = [] for pid, command in process_rows(): if pid == os.getpid() or marker not in command or not any(token in command for token in ("linuxcnc", "milltask", "halui", "axis")): continue try: os.kill(pid, signal.SIGTERM) killed.append(pid) except ProcessLookupError: continue deadline = time.monotonic() + 1.5 while time.monotonic() < deadline: if not any(pid in {row[0] for row in process_rows()} for pid in killed): break time.sleep(0.05) for pid in killed: try: os.kill(pid, signal.SIGKILL) except ProcessLookupError: pass return killed def cleanup_linuxcnc_services() -> dict[str, object]: """Mirror linuxcnc.in Cleanup before launching another case.""" commands: list[dict[str, object]] = [] for command in (["axis-remote", "--quit"], ["halcmd", "stop"], ["halcmd", "unload", "all"], ["/usr/lib/linuxcnc/realtime", "stop"]): result = subprocess.run(command, cwd=LINUXCNC_ROOT, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False) commands.append({"command": command, "returncode": result.returncode, "stdout_sha256": hashlib.sha256(result.stdout.encode()).hexdigest()}) service_pattern = re.compile(r"(?:^|/)(?:linuxcncsvr|milltask|motion-logger|rtapi_app|halui|iocontrol|axis)(?:\s|$)") killed: list[int] = [] for pid, command in process_rows(): if pid == os.getpid() or not service_pattern.search(command): continue try: os.kill(pid, signal.SIGTERM) killed.append(pid) except ProcessLookupError: continue deadline = time.monotonic() + 2.0 while time.monotonic() < deadline and any(pid in {row[0] for row in process_rows()} for pid in killed): time.sleep(0.05) for pid in killed: try: os.kill(pid, signal.SIGKILL) except ProcessLookupError: pass lock = Path("/tmp/linuxcnc.lock") lock_removed = False if lock.exists() and not any(service_pattern.search(command) for _, command in process_rows()): lock.unlink() lock_removed = True return {"commands": commands, "killed_pids": killed, "lock_path": lock.as_posix(), "lock_removed": lock_removed, "remaining_services": [command for _, command in process_rows() if service_pattern.search(command)]} def run(screenshot_path: Path | None = None) -> dict[str, object]: if not SOURCE.is_dir(): raise SystemExit(f"selected_case_missing:{SOURCE}") environment = os.environ.copy() environment["DISPLAY"] = environment.get("DISPLAY", ":1") service_cleanup = cleanup_linuxcnc_services() stale_services = cleanup_case_processes() with tempfile.TemporaryDirectory(prefix="linuxcnc-3axis-tutorial-run-") 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) environment["AXIS_OPEN_FILE"] = str(LINUXCNC_ROOT / "configs/nc_files/3dtest.ngc") update = subprocess.run(["update_ini", "-f", str(case_dir / "3axis-tutorial.ini")], cwd=LINUXCNC_ROOT, env=environment, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False) log_path = case_dir / "native.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() + 15 while time.monotonic() < deadline and process.poll() is None: window = axis_window(environment) if window: if "(no file)" in str(window.get("title", "")).lower() and time.monotonic() + 0.25 < deadline: time.sleep(0.25) continue if screenshot_path is not None: screenshot_path.parent.mkdir(parents=True, exist_ok=True) capture = subprocess.run(["import", "-window", str(window["window_id"]), str(screenshot_path)], env=environment, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False) window["screenshot"] = {"path": screenshot_path.as_posix(), "returncode": capture.returncode, "bytes": screenshot_path.stat().st_size if screenshot_path.is_file() else 0, "sha256": sha256(screenshot_path) if screenshot_path.is_file() else None, "stderr_sha256": hashlib.sha256(capture.stderr).hexdigest()} break time.sleep(0.25) 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) log_text = log_path.read_text(encoding="utf-8", errors="replace") generated = sorted(path.name for path in case_dir.iterdir() if path.name not in (*FILES, "native.log")) return { "schema": SCHEMA, "status": "verified" if window else "blocked", "case": CASE, "command": ["linuxcnc", "/3axis-tutorial.ini"], "cwd": "linuxcnc_source_root", "display": environment["DISPLAY"], "ini_update": {"command": ["update_ini", "-f", "/3axis-tutorial.ini"], "returncode": update.returncode, "output_sha256": hashlib.sha256(update.stdout.encode()).hexdigest()}, "exact_source_files": [{"path": f"{CASE}/{name}", "sha256": sha256(SOURCE / name)} for name in FILES] + [{"path": PROGRAM.as_posix(), "sha256": sha256(ROOT / PROGRAM)}], "axis_panel": window, "process_returncode": process.returncode, "generated_files": generated, "native_log_sha256": hashlib.sha256(log_text.encode()).hexdigest(), "native_log_excerpt": "\n".join(line for line in log_text.splitlines() if any(token in line for token in ("Starting LinuxCNC", "Found file", "MAXV", "jog_order", "task:"))), "boundaries": ["POSIX fallback is native simulation execution, not hardware validation", "AXIS window evidence is required; timeout/task markers alone are insufficient", "installed LinuxCNC package version is recorded separately from locked source"], "termination": "intentional_diagnostic_cleanup_after_visible_window", "service_cleanup": service_cleanup, "stale_services_cleaned": stale_services, } if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--output", type=Path, default=Path("coverage/axis3-tutorial-native-run.report.json")) parser.add_argument("--screenshot", type=Path) args = parser.parse_args() screenshot = args.screenshot if args.screenshot is None or args.screenshot.is_absolute() else ROOT / args.screenshot report = run(screenshot) report["report_sha256"] = hashlib.sha256(json.dumps(report, ensure_ascii=False, sort_keys=True, separators=(",", ":")).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)) raise SystemExit(0 if report["status"] == "verified" else 1)