Unblock native XYZAC TP oracle
This commit is contained in:
@@ -16,16 +16,22 @@ import os
|
||||
import re
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
# The oracle must bind to the same source build that run-5axis-rtcp.sh starts.
|
||||
# In particular, importing the distro linuxcnc extension would silently bind to
|
||||
# a different NML ABI/version and produce plausible but unusable status values.
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SOURCE_PYTHON_DIR = ROOT / "linuxcnc/lib/python"
|
||||
sys.path.insert(0, str(SOURCE_PYTHON_DIR))
|
||||
import linuxcnc
|
||||
|
||||
from run_axis3_tutorial_native import cleanup_linuxcnc_services
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
CASE = "linuxcnc/configs/sim/axis/vismach/5axis/table-rotary-tilting/xyzac-trt.ini"
|
||||
PROGRAM = ROOT / "linuxcnc/configs/sim/axis/vismach/5axis/table-rotary-tilting/demos/xyzac_switchkins.ngc"
|
||||
SCHEMA = "xyzac-native-linuxcnc-tick-trace.v1"
|
||||
@@ -58,10 +64,13 @@ def scalar(value: Any) -> float | int | bool | str | None:
|
||||
|
||||
def stat_row(stat: Any, sample_index: int, previous_taskbeat: int | None) -> dict[str, Any]:
|
||||
stat.poll()
|
||||
taskbeat = scalar(getattr(stat, "taskbeat", None))
|
||||
taskbeat = scalar(getattr(stat, "task_heartbeat", None))
|
||||
commanded_pose = finite_list(getattr(stat, "position", None))
|
||||
distance_to_go = finite_list(getattr(stat, "dtg", None))
|
||||
return {
|
||||
"sample_index": sample_index,
|
||||
"taskbeat": taskbeat,
|
||||
"task_heartbeat": taskbeat,
|
||||
"taskbeat_changed": previous_taskbeat is None or taskbeat != previous_taskbeat,
|
||||
"task_state": scalar(getattr(stat, "task_state", None)),
|
||||
"task_mode": scalar(getattr(stat, "task_mode", None)),
|
||||
@@ -71,7 +80,7 @@ def stat_row(stat: Any, sample_index: int, previous_taskbeat: int | None) -> dic
|
||||
"motion_line": scalar(getattr(stat, "motion_line", None)),
|
||||
"read_line": scalar(getattr(stat, "read_line", None)),
|
||||
"actual_pose": finite_list(getattr(stat, "actual_position", None)),
|
||||
"commanded_pose": finite_list(getattr(stat, "command_position", None)),
|
||||
"commanded_pose": commanded_pose,
|
||||
"joint_actual": finite_list(getattr(stat, "joint_actual_position", None)),
|
||||
"joint_commanded": finite_list(getattr(stat, "joint_position", None)),
|
||||
"velocity": scalar(getattr(stat, "current_vel", None)),
|
||||
@@ -86,7 +95,7 @@ def stat_row(stat: Any, sample_index: int, previous_taskbeat: int | None) -> dic
|
||||
"active_segment_id": scalar(getattr(stat, "motion_id", None)),
|
||||
"active_segment_geometry": None,
|
||||
"active_segment_geometry_status": "unavailable_linuxcnc_stat_api",
|
||||
"distance_to_go": finite_list(getattr(stat, "distance_to_go", None)),
|
||||
"distance_to_go": distance_to_go,
|
||||
"position": finite_list(getattr(stat, "position", None)),
|
||||
"homed": list(getattr(stat, "homed", [])) if getattr(stat, "homed", None) is not None else None,
|
||||
"limits": list(getattr(stat, "limit", [])) if getattr(stat, "limit", None) is not None else None,
|
||||
@@ -150,6 +159,7 @@ def run(duration: float) -> dict[str, Any]:
|
||||
output_lines: list[str] = []
|
||||
rows: list[dict[str, Any]] = []
|
||||
errors: list[str] = []
|
||||
stage = "launch"
|
||||
stat: Any = None
|
||||
control: Any = None
|
||||
connected = False
|
||||
@@ -161,34 +171,56 @@ def run(duration: float) -> dict[str, Any]:
|
||||
if not window:
|
||||
time.sleep(0.1)
|
||||
continue
|
||||
try:
|
||||
stat = linuxcnc.stat()
|
||||
control = linuxcnc.command()
|
||||
stat.poll()
|
||||
if getattr(stat, "task_state", None) in (linuxcnc.STATE_ESTOP, linuxcnc.STATE_ESTOP_RESET, linuxcnc.STATE_OFF, linuxcnc.STATE_ON):
|
||||
# The AXIS toplevel is created only after linuxcncsvr/milltask have
|
||||
# opened the NML buffers. Creating linuxcnc.stat before that point
|
||||
# permanently binds an invalid channel, so wait for the real
|
||||
# window first and create fresh channels afterwards.
|
||||
time.sleep(0.1)
|
||||
if stat is None or control is None or not connected:
|
||||
if not window:
|
||||
raise RuntimeError("native_linuxcnc_window_unavailable")
|
||||
stat = None
|
||||
control = None
|
||||
attach_deadline = time.monotonic() + 10
|
||||
while time.monotonic() < attach_deadline and process.poll() is None:
|
||||
try:
|
||||
candidate_stat = linuxcnc.stat()
|
||||
candidate_stat.poll()
|
||||
stat = candidate_stat
|
||||
control = linuxcnc.command()
|
||||
connected = True
|
||||
break
|
||||
except Exception:
|
||||
time.sleep(0.1)
|
||||
if stat is None or control is None or not connected:
|
||||
raise RuntimeError("native_linuxcnc_stat_unavailable")
|
||||
except BaseException:
|
||||
stat = None
|
||||
control = None
|
||||
time.sleep(0.1)
|
||||
if stat is None or control is None or not connected:
|
||||
raise RuntimeError("native_linuxcnc_stat_unavailable")
|
||||
|
||||
stage = "state_estop_reset"
|
||||
control.state(linuxcnc.STATE_ESTOP_RESET)
|
||||
control.wait_complete(5)
|
||||
wait_stat(stat, lambda item: item.task_state != linuxcnc.STATE_ESTOP, 5)
|
||||
stage = "state_on"
|
||||
control.state(linuxcnc.STATE_ON)
|
||||
control.wait_complete(5)
|
||||
wait_stat(stat, lambda item: item.task_state == linuxcnc.STATE_ON, 5)
|
||||
stage = "mode_manual"
|
||||
control.mode(linuxcnc.MODE_MANUAL)
|
||||
control.wait_complete(5)
|
||||
for joint in range(5):
|
||||
stage = f"home_joint_{joint}"
|
||||
control.home(joint)
|
||||
control.wait_complete(5)
|
||||
stage = "wait_home_all"
|
||||
wait_stat(stat, lambda item: all(bool(value) for value in list(getattr(item, "homed", []))[:5]), 10)
|
||||
stage = "program_open"
|
||||
control.program_open(str(PROGRAM))
|
||||
control.wait_complete(5)
|
||||
stage = "mode_auto"
|
||||
control.mode(linuxcnc.MODE_AUTO)
|
||||
control.wait_complete(5)
|
||||
stage = "program_run"
|
||||
control.auto(linuxcnc.AUTO_RUN, 0)
|
||||
|
||||
start = time.monotonic()
|
||||
@@ -208,10 +240,11 @@ def run(duration: float) -> dict[str, Any]:
|
||||
sample_index += 1
|
||||
time.sleep(0.005)
|
||||
final = stat_row(stat, sample_index + 1, previous_taskbeat)
|
||||
final["native_elapsed_ms"] = round((time.monotonic() - start) * 1000.0, 3)
|
||||
rows.append(final)
|
||||
if final["taskbeat_changed"]:
|
||||
final["native_elapsed_ms"] = round((time.monotonic() - start) * 1000.0, 3)
|
||||
rows.append(final)
|
||||
except Exception as error:
|
||||
errors.append(f"{type(error).__name__}:{error}")
|
||||
errors.append(f"{stage}:{type(error).__name__}:{error}")
|
||||
finally:
|
||||
terminate_process(process)
|
||||
try:
|
||||
@@ -222,12 +255,19 @@ def run(duration: float) -> dict[str, Any]:
|
||||
cleanup_linuxcnc_services()
|
||||
|
||||
source_bytes = PROGRAM.read_bytes()
|
||||
ini_path = ROOT / CASE
|
||||
launcher_path = ROOT / "run-5axis-rtcp.sh"
|
||||
heartbeat_values = [row.get("task_heartbeat") for row in rows]
|
||||
valid_heartbeats = [value for value in heartbeat_values if isinstance(value, int)]
|
||||
heartbeat_monotonic = all(left < right for left, right in zip(valid_heartbeats, valid_heartbeats[1:]))
|
||||
trace_hash = sha256_bytes(canonical(rows).encode())
|
||||
python_module = getattr(linuxcnc, "__file__", None)
|
||||
source_python_module = str(ROOT / "linuxcnc/lib/python/linuxcnc.so")
|
||||
if python_module != source_python_module:
|
||||
errors.append("NATIVE_PYTHON_MODULE_MISMATCH")
|
||||
status = "verified" if rows and not errors else "blocked"
|
||||
status = "verified" if rows and not errors and heartbeat_monotonic and len(valid_heartbeats) == len(set(valid_heartbeats)) else "blocked"
|
||||
if rows and not heartbeat_monotonic:
|
||||
errors.append("NATIVE_TASK_HEARTBEAT_NON_MONOTONIC")
|
||||
return {
|
||||
"schema": SCHEMA,
|
||||
"status": status,
|
||||
@@ -235,8 +275,15 @@ def run(duration: float) -> dict[str, Any]:
|
||||
"launcher": command,
|
||||
"native_axis_window": window,
|
||||
"program": {"path": PROGRAM.relative_to(ROOT).as_posix(), "sha256": sha256_bytes(source_bytes)},
|
||||
"native_linuxcnc": {"python_module": python_module, "expected_source_python_module": source_python_module, "display": environment["DISPLAY"]},
|
||||
"tick_source": "linuxcnc.stat.taskbeat",
|
||||
"native_linuxcnc": {"python_module": python_module, "expected_source_python_module": source_python_module, "display": environment["DISPLAY"], "python_module_sha256": sha256_bytes(Path(python_module).read_bytes()) if python_module and Path(python_module).is_file() else None},
|
||||
"inputs": {
|
||||
"case_ini": {"path": CASE, "sha256": sha256_bytes(ini_path.read_bytes())},
|
||||
"launcher": {"path": "run-5axis-rtcp.sh", "sha256": sha256_bytes(launcher_path.read_bytes())},
|
||||
"program": {"path": PROGRAM.relative_to(ROOT).as_posix(), "sha256": sha256_bytes(source_bytes)},
|
||||
"source_python_module": {"path": source_python_module, "sha256": sha256_bytes(Path(source_python_module).read_bytes()) if Path(source_python_module).is_file() else None},
|
||||
},
|
||||
"tick_source": "linuxcnc.stat.task_heartbeat",
|
||||
"tick_integrity": {"first": valid_heartbeats[0] if valid_heartbeats else None, "last": valid_heartbeats[-1] if valid_heartbeats else None, "row_count": len(valid_heartbeats), "unique_count": len(set(valid_heartbeats)), "strictly_monotonic": heartbeat_monotonic},
|
||||
"rows": rows,
|
||||
"row_count": len(rows),
|
||||
"trace_sha256": trace_hash,
|
||||
|
||||
Reference in New Issue
Block a user