Advance simulation parity and 3-axis tutorial coverage
This commit is contained in:
181
scripts/run_axis3_tutorial_native.py
Normal file
181
scripts/run_axis3_tutorial_native.py
Normal file
@@ -0,0 +1,181 @@
|
||||
#!/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"
|
||||
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)
|
||||
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 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", "<case_dir>/3axis-tutorial.ini"],
|
||||
"cwd": "linuxcnc_source_root",
|
||||
"display": environment["DISPLAY"],
|
||||
"ini_update": {"command": ["update_ini", "-f", "<case_dir>/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],
|
||||
"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)
|
||||
Reference in New Issue
Block a user