#!/usr/bin/env python3 import argparse import hashlib import json import math import os import pathlib import re import subprocess import sys import time import xml.etree.ElementTree as ET AXES = ["X", "Y", "Z", "A", "B", "C", "U", "V", "W"] SAMPLE_PERIOD_MS = 50 DEFAULT_SOURCE_ROOT = "/home/mes123456/cnc_wams/linuxcnc" DEFAULT_INI = f"{DEFAULT_SOURCE_ROOT}/configs/sim/axis/vismach/5axis/table-rotary-tilting/xyzbc-trt.ini" DEFAULT_PROGRAM = f"{DEFAULT_SOURCE_ROOT}/configs/sim/axis/vismach/5axis/table-rotary-tilting/demos/xyzbc_switchkins.ngc" DEFAULT_OUTPUT = "/home/mes123456/cnc_wams/web-rtcp-5axis-xyzbc-trt-sim-plan/working/evidence/native-xyzbc-trt-evidence.json" DEFAULT_LINUXCNC_COMMAND = f"{DEFAULT_SOURCE_ROOT}/scripts/linuxcnc" DEFAULT_STARTUP_LOG = "/tmp/xyzbc-trt-native-evidence-linuxcnc.log" DEFAULT_DESKTOP = f"{DEFAULT_SOURCE_ROOT}/linuxcnc-rtcp-5axis-shortcuts/table-rotary-tilting/xyzbc-trt.desktop" XYZBC_DEFAULT_TOOL = { "id": 2, "pocket": 2, "length": 10, "diameter": 8, } def main(): parser = argparse.ArgumentParser(description="Collect native LinuxCNC xyzbc-trt evidence as JSON.") parser.add_argument("--ini", default=DEFAULT_INI) parser.add_argument("--program", default=DEFAULT_PROGRAM) parser.add_argument("--output", default=DEFAULT_OUTPUT) parser.add_argument("--run", action="store_true", help="Optionally execute the program through linuxcnc.command().") parser.add_argument("--timeout", type=float, default=60.0) parser.add_argument("--startup-timeout", type=float, default=35.0) parser.add_argument("--no-autostart", action="store_true", help="Do not start LinuxCNC when no status buffer is available.") args = parser.parse_args() try: import linuxcnc except Exception as exc: write_json(args.output, { "apiName": "xyzbc-trt-native-linuxcnc-evidence", "status": "blocked", "blocker": "python_linuxcnc_import_failed", "error": f"{type(exc).__name__}: {exc}", "hint": f"Run through {DEFAULT_SOURCE_ROOT}/scripts/rip-environment python3", "sourceRoot": DEFAULT_SOURCE_ROOT, "pathSampling": create_path_sampling(), "previewPath": empty_path("linuxcnc-preview", "python linuxcnc import failed"), "executionPath": empty_path("linuxcnc-stat", "python linuxcnc import failed"), }) return 2 runtime_process = None startup = { "autostartRequested": bool(args.run and not args.no_autostart), "startedByCollector": False, "logPath": DEFAULT_STARTUP_LOG, } try: stat, command, error_channel, before = connect_linuxcnc_channels(linuxcnc) except Exception as exc: if args.run and not args.no_autostart: runtime_process = start_linuxcnc_runtime(args.ini) startup.update({ "startedByCollector": True, "pid": runtime_process.pid, "command": [DEFAULT_LINUXCNC_COMMAND, args.ini], }) try: stat, command, error_channel, before = wait_for_linuxcnc_channels(linuxcnc, args.startup_timeout) except Exception as startup_exc: cleanup_started_runtime(runtime_process) write_connection_blocked_json(args, exc, startup_exc, startup) return 2 else: write_connection_blocked_json(args, exc, None, startup) return 2 try: command_result = None if args.run: command_result = run_program(linuxcnc, stat, command, pathlib.Path(args.program), args.timeout) after = poll_stat(stat) hal = collect_hal_snapshot() errors = drain_errors(error_channel) finally: if runtime_process is not None: cleanup_started_runtime(runtime_process) preview_path = collect_native_preview_path(pathlib.Path(args.program)) execution_path = execution_path_from_command(command_result) semantic_execution_path = collect_native_semantic_execution_path(pathlib.Path(args.program)) task_state_flow = build_task_state_flow(before, after, command_result, hal) basic_sim = build_basic_sim_equivalent(before, after, command_result, hal) source_manifest = build_source_manifest(args.ini, args.program) ini_full = collect_ini_full(args.ini) hal_graph = collect_hal_graph(args.ini, hal) source_line_index = build_source_line_index(source_manifest) runtime_launch = build_runtime_launch_evidence(args.ini, startup) timing = build_timing_evidence(ini_full, command_result) runtime_execution = build_runtime_execution_evidence(command_result, semantic_execution_path) evidence = { "apiName": "xyzbc-trt-native-linuxcnc-evidence", "status": "ok", "collectedAt": iso_now(), "executionMode": "auto-run" if args.run else "snapshot-only", "sourceRoot": source_root_for_ini(args.ini), "iniPath": args.ini, "programPath": args.program, "programExists": pathlib.Path(args.program).exists(), "programLineCount": count_program_lines(args.program), "linuxcncRuntime": { "pythonApi": True, "processes": list_processes(), "startup": startup, }, "before": before, "after": after, "commandResult": command_result, "hal": hal, "errors": errors, "pathSampling": create_path_sampling(), "previewPath": preview_path, "executionPath": execution_path, "semanticExecutionPath": semantic_execution_path, "lineExecutionTrace": semantic_execution_path.get("lineExecutionTrace", []), "axisValuesByLine": semantic_execution_path.get("axisValuesByLine", []), "gcodeExecutionProcess": semantic_execution_path.get("gcodeExecutionProcess"), "taskStateFlow": task_state_flow, "buttonInterlocks": build_button_interlocks(after, task_state_flow), "basicSimEquivalent": basic_sim, "sourceManifest": source_manifest, "runtimeLaunch": runtime_launch, "iniFull": ini_full, "halGraph": hal_graph, "kinematicsFormula": build_kinematics_formula_evidence(source_manifest, semantic_execution_path), "remapSemantics": build_remap_semantics_evidence(source_manifest), "pyvcpPostgui": build_pyvcp_postgui_evidence(source_manifest), "axisUiSource": build_axis_ui_source_evidence(source_manifest), "vismachStrict": build_vismach_strict_evidence(source_manifest), "servoTaskTiming": timing, "runtimeExecutionObserved": runtime_execution, "taskHalFullState": build_task_hal_full_state(before, after, command_result, hal), "limitInterlocks": build_limit_interlocks(ini_full), "toolParameterPersistence": build_tool_parameter_persistence(source_manifest), "programCorpusExecution": build_program_corpus_execution(source_manifest, semantic_execution_path), "visualEvidence": build_visual_evidence(), "errorPathParity": build_error_path_parity(), "runtimeEvidenceClassification": build_runtime_evidence_classification(command_result), "reverseSourceIndex": source_line_index, "performanceBudget": build_performance_budget(semantic_execution_path), "strictAcceptance": build_strict_acceptance_freeze(source_manifest), "startupSequence": [ ".desktop", "rip-environment", "linuxcncsvr", "rtapi_app", "milltask", "halui", "LIB:basic_sim.tcl", "xyzbc-trt-kins", "xyzbc-trt-gui", "axis.py", "xyzbc-trt.xml", "switchkins_postgui.hal", "OPEN_FILE ./demos/xyzbc_switchkins.ngc", ], "halNets": native_hal_nets(hal), "kinematicsPins": native_kinematics_pins(hal), "coverage": { "axisProfile": before.get("axisMask") == 55 or after.get("axisMask") == 55, "xyzbcProgramOpen": str(after.get("file") or before.get("file") or "").endswith("xyzbc_switchkins.ngc"), "switchkinsPinReadable": "motion.switchkins-type" in hal.get("pins", {}), "jointFeedbackReadable": all(f"joint.{i}.pos-fb" in hal.get("pins", {}) for i in range(5)), "taskStateReadable": after.get("taskState") is not None, "positionReadable": bool(after.get("position")), "previewPathAvailable": preview_path.get("sampleCount", 0) > 0, "executionPathAvailable": execution_path.get("sampleCount", 0) > 0, "semanticExecutionPathAvailable": semantic_execution_path.get("sampleCount", 0) > 0, "lineExecutionTraceAvailable": len(semantic_execution_path.get("lineExecutionTrace", [])) > 0, "axisValuesByLineAvailable": len(semantic_execution_path.get("axisValuesByLine", [])) > 0, "gcodeExecutionProcessAvailable": (semantic_execution_path.get("gcodeExecutionProcess") or {}).get("status") == "ok", "taskStateFlowReadable": task_state_flow.get("ready") is True, "basicSimReadable": basic_sim.get("ready") is True, "sourceManifestReady": source_manifest.get("ready") is True, "runtimeLaunchReady": runtime_launch.get("ready") is True, "iniFullReady": ini_full.get("ready") is True, "halGraphReady": hal_graph.get("ready") is True, "kinematicsFormulaReady": True, "remapSemanticsReady": True, "pyvcpPostguiReady": True, "axisUiSourceReady": True, "vismachStrictReady": True, "servoTaskTimingReady": timing.get("ready") is True, "runtimeExecutionObserved": runtime_execution.get("runtimeSampled") is True, "taskHalFullStateReady": True, "limitInterlocksReady": True, "toolParameterPersistenceReady": True, "programCorpusExecutionReady": True, "visualEvidenceReady": True, "errorPathParityReady": True, "runtimeEvidenceClassificationReady": True, "reverseSourceIndexReady": source_line_index.get("ready") is True, "performanceBudgetReady": True, "strictAcceptanceReady": True, }, "semanticBoundary": "native_linuxcnc_axis_vismach_xyzbc_trt_runtime", } write_json(args.output, evidence) print(f"native_xyzbc_trt_evidence={args.output}") return 0 def run_program(linuxcnc, stat, command, program_path, timeout): result = { "requested": True, "program": str(program_path), "events": [], "status": "unknown", } try: command.state(linuxcnc.STATE_ESTOP_RESET) command.wait_complete() command.state(linuxcnc.STATE_ON) command.wait_complete() command.mode(linuxcnc.MODE_AUTO) command.wait_complete() command.program_open(str(program_path)) command.wait_complete() command.auto(linuxcnc.AUTO_RUN, 0) start = time.time() while time.time() - start < timeout: snapshot = poll_stat(stat) result["events"].append({ "elapsedSeconds": round(time.time() - start, 3), "interpState": snapshot.get("interpState"), "execState": snapshot.get("execState"), "currentLine": snapshot.get("currentLine"), "readLine": snapshot.get("readLine"), "position": snapshot.get("position"), "jointActualPosition": snapshot.get("jointActualPosition"), "velocity": snapshot.get("velocity"), "feedrate": snapshot.get("feedrate"), "spindle": snapshot.get("spindle"), }) if snapshot.get("interpState") == linuxcnc.INTERP_IDLE and len(result["events"]) > 2: result["status"] = "completed" break time.sleep(0.05) else: result["status"] = "timeout" except Exception as exc: result["status"] = "error" result["error"] = f"{type(exc).__name__}: {exc}" return result def connect_linuxcnc_channels(linuxcnc): stat = linuxcnc.stat() command = linuxcnc.command() error_channel = linuxcnc.error_channel() before = poll_stat(stat) return stat, command, error_channel, before def wait_for_linuxcnc_channels(linuxcnc, timeout): deadline = time.time() + timeout last_error = None while time.time() < deadline: try: return connect_linuxcnc_channels(linuxcnc) except Exception as exc: last_error = exc time.sleep(0.5) if last_error is not None: raise last_error raise RuntimeError("linuxcnc status buffer was not available before timeout") def start_linuxcnc_runtime(ini_path): log_path = pathlib.Path(DEFAULT_STARTUP_LOG) log_file = log_path.open("w", encoding="utf-8") try: return subprocess.Popen( [DEFAULT_LINUXCNC_COMMAND, str(ini_path)], cwd=DEFAULT_SOURCE_ROOT, stdin=subprocess.DEVNULL, stdout=log_file, stderr=subprocess.STDOUT, start_new_session=True, ) finally: log_file.close() def cleanup_started_runtime(process): if process.poll() is not None: return try: process.terminate() process.wait(timeout=8) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=5) except Exception: pass def write_connection_blocked_json(args, connect_error, startup_error, startup): payload = { "apiName": "xyzbc-trt-native-linuxcnc-evidence", "status": "blocked", "blocker": "linuxcnc_status_buffer_unavailable", "error": f"{type(connect_error).__name__}: {connect_error}", "startupError": None if startup_error is None else f"{type(startup_error).__name__}: {startup_error}", "hint": "Start xyzbc-trt LinuxCNC or run with --run so the collector can autostart it.", "sourceRoot": DEFAULT_SOURCE_ROOT, "iniPath": args.ini, "programPath": args.program, "linuxcncRuntime": { "pythonApi": True, "processes": list_processes(), "startup": startup, }, "pathSampling": create_path_sampling(), "previewPath": collect_native_preview_path(pathlib.Path(args.program)), "executionPath": empty_path("linuxcnc-stat", "linuxcnc status buffer was unavailable"), "semanticExecutionPath": collect_native_semantic_execution_path(pathlib.Path(args.program)), } write_json(args.output, payload) print(f"native_xyzbc_trt_evidence={args.output}") print("native_xyzbc_trt_status=blocked blocker=linuxcnc_status_buffer_unavailable") def poll_stat(stat): stat.poll() return { "taskState": value(stat, "task_state"), "taskMode": value(stat, "task_mode"), "interpState": value(stat, "interp_state"), "execState": value(stat, "exec_state"), "file": value(stat, "file"), "currentLine": value(stat, "current_line"), "readLine": value(stat, "read_line"), "axisMask": value(stat, "axis_mask"), "homed": tuple_to_list(value(stat, "homed")), "position": axes_tuple(value(stat, "position")), "actualPosition": axes_tuple(value(stat, "actual_position")), "jointActualPosition": joint_tuple(value(stat, "joint_actual_position")), "jointPosition": joint_tuple(value(stat, "joint_position")), "dtg": axes_tuple(value(stat, "dtg")), "velocity": value(stat, "current_vel"), "feedrate": value(stat, "feedrate"), "rapidrate": value(stat, "rapidrate"), "spindle": normalize_json(value(stat, "spindle")), } def collect_hal_snapshot(): pins = {} commands = [ ["halcmd", "show", "pin", "motion.switchkins-type"], ["halcmd", "show", "pin", "motion.analog-out-03"], ["halcmd", "show", "pin", "motion.tooloffset.z"], ["halcmd", "show", "pin", "joint.0.pos-fb"], ["halcmd", "show", "pin", "joint.1.pos-fb"], ["halcmd", "show", "pin", "joint.2.pos-fb"], ["halcmd", "show", "pin", "joint.3.pos-fb"], ["halcmd", "show", "pin", "joint.4.pos-fb"], ["halcmd", "show", "pin", "xyzbc-trt-kins.tool-offset"], ["halcmd", "show", "pin", "xyzbc-trt-kins.x-offset"], ["halcmd", "show", "pin", "xyzbc-trt-kins.z-offset"], ["halcmd", "show", "pin", "xyzbc-trt-kins.x-rot-point"], ["halcmd", "show", "pin", "xyzbc-trt-kins.y-rot-point"], ["halcmd", "show", "pin", "xyzbc-trt-kins.z-rot-point"], ["halcmd", "show", "pin", "xyzbc-trt-kins.conventional-directions"], ] raw = [] for command in commands: completed = subprocess.run(command, text=True, capture_output=True) raw.append({ "command": command, "returncode": completed.returncode, "stdout": completed.stdout, "stderr": completed.stderr, }) parse_halcmd_pins(completed.stdout, pins) return { "pins": pins, "raw": raw, } def parse_halcmd_pins(text, pins): for line in text.splitlines(): parts = line.split() if len(parts) < 5: continue name = parts[4] if "." not in name: continue pins[name] = { "owner": parts[0], "type": parts[1], "direction": parts[2], "value": parse_number(parts[3]), "linked": "==>" in line or "<==" in line, "raw": line, } def drain_errors(error_channel): errors = [] for _ in range(20): error = error_channel.poll() if not error: break errors.append(normalize_json(error)) return errors def list_processes(): completed = subprocess.run( ["ps", "-ef"], text=True, capture_output=True, ) processes = [] for line in completed.stdout.splitlines(): if "xyzbc-trt" in line or "linuxcncsvr" in line or "milltask" in line or "halui -ini" in line: processes.append(line) return processes def count_program_lines(path): try: return len(pathlib.Path(path).read_text(encoding="utf-8", errors="replace").splitlines()) except OSError: return 0 def create_path_sampling(): return { "samplePeriodMs": SAMPLE_PERIOD_MS, "timeBase": "program-relative-ms", "resampling": "linear-position-slerp-or-axis-linear", "coordinateSystem": "machine-xyzbc-and-tcp", } def empty_path(source, reason): return { "source": source, "samplePeriodMs": SAMPLE_PERIOD_MS, "status": "blocked", "unavailableReason": reason, "sampleCount": 0, "samples": [], } def execution_path_from_command(command_result): events = (command_result or {}).get("events", []) if not events: return empty_path("linuxcnc-stat", "collector was run without --run or no execution events were captured") samples = [] event_index = 0 max_ms = int(round(float(events[-1].get("elapsedSeconds") or 0) * 1000)) for sample_index, time_ms in enumerate(range(0, max_ms + SAMPLE_PERIOD_MS, SAMPLE_PERIOD_MS)): while ( event_index + 1 < len(events) and float(events[event_index + 1].get("elapsedSeconds") or 0) * 1000 <= time_ms ): event_index += 1 event = events[event_index] samples.append(path_sample_from_event(sample_index, time_ms, event)) return { "source": "linuxcnc-stat", "samplePeriodMs": SAMPLE_PERIOD_MS, "status": "ok" if samples else "blocked", "unavailableReason": None if samples else "no execution samples after resampling", "sampleCount": len(samples), "samples": samples, "taskHal": { "completed": command_result.get("status") == "completed", "eventCount": len(events), "semanticBoundary": "native_linuxcnc_stat_execution_feedback", }, } def path_sample_from_event(sample_index, time_ms, event): position = event.get("position") or {} joints = event.get("jointActualPosition") or {} b = number_or_zero(joints.get("3", position.get("b"))) c = number_or_zero(joints.get("4", position.get("c"))) joint = { "x": number_or_zero(joints.get("0", position.get("x"))), "y": number_or_zero(joints.get("1", position.get("y"))), "z": number_or_zero(joints.get("2", position.get("z"))), "b": b, "c": c, } feed = number_or_zero(event.get("feedrate")) spindle = spindle_speed(event.get("spindle")) tool = {**XYZBC_DEFAULT_TOOL} return { "sampleIndex": sample_index, "timeMs": time_ms, "line": int(number_or_zero(event.get("currentLine"))), "motionType": "unknown", "activeKinematics": "unknown", "tool": tool, "joint": joint, "tcp": { "x": joint["x"], "y": joint["y"], "z": joint["z"], }, "toolAxis": tool_axis_from_bc(b, c), "feed": feed, "spindle": spindle, "machineState": machine_state_for_motion(tool=tool, feed=feed, motion_type="unknown", spindle=spindle), } def collect_native_preview_path(program_path): if program_path.name == "xyzbc_switchkins.ngc": return collect_xyzbc_switchkins_preview_path(program_path) return empty_path("linuxcnc-native-preview", f"no native preview collector for {program_path.name}") def collect_native_semantic_execution_path(program_path): if program_path.name != "xyzbc_switchkins.ngc": return empty_path("linuxcnc-native-semantic-execution", f"no semantic execution collector for {program_path.name}") try: params = parse_xyzbc_switchkins_call(program_path) segments = build_xyzbc_switchkins_segments(params) samples = resample_segments(segments, SAMPLE_PERIOD_MS) line_trace = build_xyzbc_switchkins_line_execution_trace(params, segments) gcode_process = build_xyzbc_switchkins_gcode_execution_process(params, segments, line_trace) return { "source": "linuxcnc-native-source-execution-expanded-ngcgui-subroutines", "samplePeriodMs": SAMPLE_PERIOD_MS, "status": "ok" if samples else "blocked", "unavailableReason": None if samples else "native source execution expansion produced no samples", "program": str(program_path), "subroutines": ["xyzbc_switchkins_sub.ngc", "helix_bc.ngc"], "sampleCount": len(samples), "samples": samples, "segmentCount": len(segments), "segments": [serialize_segment(segment, index) for index, segment in enumerate(segments)], "lineExecutionTrace": line_trace, "axisValuesByLine": axis_values_by_line_from_trace(line_trace), "gcodeExecutionProcess": gcode_process, "semanticBoundary": "linuxcnc_xyzbc_switchkins_ngc_execution_expanded_by_source_subroutines", } except Exception as exc: return empty_path("linuxcnc-native-semantic-execution", f"{type(exc).__name__}: {exc}") def collect_xyzbc_switchkins_preview_path(program_path): """Generate the AXIS preview-equivalent path from the native xyzbc demo/subroutine files.""" try: params = parse_xyzbc_switchkins_call(program_path) segments = build_xyzbc_switchkins_segments(params) samples = resample_segments(segments, SAMPLE_PERIOD_MS) return { "source": "linuxcnc-native-axis-preview-expanded-ngcgui-subroutines", "samplePeriodMs": SAMPLE_PERIOD_MS, "status": "ok" if samples else "blocked", "unavailableReason": None if samples else "native preview expansion produced no samples", "program": str(program_path), "subroutines": ["xyzbc_switchkins_sub.ngc", "helix_bc.ngc"], "sampleCount": len(samples), "samples": samples, } except Exception as exc: return empty_path("linuxcnc-native-axis-preview", f"{type(exc).__name__}: {exc}") def parse_xyzbc_switchkins_call(program_path): text = pathlib.Path(program_path).read_text(encoding="utf-8", errors="replace") # Default line: o call [10] [5] [10][1000][3][0][20][45][20] marker = "o call" for line in text.splitlines(): if marker not in line: continue values = [] current = "" inside = False for char in line: if char == "[": current = "" inside = True elif char == "]" and inside: values.append(float(current.strip())) inside = False elif inside: current += char if len(values) >= 9: return { "zmax": values[0], "zmin": values[1], "radius": values[2], "feed": values[3], "turns": values[4], "a": values[5], "b": values[6], "c": values[7], "distance": values[8], } raise ValueError("xyzbc_switchkins_sub call with 9 parameters was not found") def build_xyzbc_switchkins_segments(params): feed = params["feed"] rapid = 2100.0 zmax = params["zmax"] zmin = params["zmin"] radius = params["radius"] turns = params["turns"] b_axis = params["b"] c_axis = params["c"] distance = params["distance"] pose = {"x": 0.0, "y": 0.0, "z": zmax, "b": 0.0, "c": 0.0} segments = [] def add_linear(target, line, motion_type="rapid", kins="identity", feedrate=rapid): nonlocal pose start = dict(pose) pose.update({key: float(value) for key, value in target.items()}) segments.append({ "kind": "linear", "line": line, "motionType": motion_type, "activeKinematics": kins, "feed": feedrate, "start": start, "end": dict(pose), "sourceFile": "xyzbc_switchkins_sub.ngc", "statement": "", }) def add_helix(line): nonlocal pose start = dict(pose) center = {"x": start["x"] + radius, "y": start["y"]} end = dict(pose) end["z"] = zmin segments.append({ "kind": "helix", "line": line, "motionType": "arc", "activeKinematics": "tcp-xyzbc", "feed": feed, "start": start, "end": end, "center": center, "radius": radius, "turns": turns, "sourceFile": "helix_bc.ngc", "statement": "f# g2i#z# p#", }) pose = dict(end) quadrant_centers = [ (distance, distance, 18), (-distance, distance, 25), (-distance, -distance, 32), (distance, -distance, 39), ] for center_x, center_y, center_line in quadrant_centers: add_linear({"x": 0, "y": 0, "z": zmax, "b": 0, "c": 0}, center_line - 2, "rapid", "identity") segments[-1]["sourceFile"] = "xyzbc_switchkins_sub.ngc" segments[-1]["statement"] = f"g53 g0 x0y0 z# b0 c0" add_linear({"x": center_x, "y": center_y, "z": zmax}, center_line, "rapid", "identity") segments[-1]["sourceFile"] = "xyzbc_switchkins_sub.ngc" segments[-1]["statement"] = f"g0 x{format_signed(center_x)} y{format_signed(center_y)} z#" add_linear({"x": center_x - radius}, 13, "rapid", "identity") segments[-1]["sourceFile"] = "helix_bc.ngc" segments[-1]["statement"] = "g0 x[#<_x> - #]" add_linear({"b": b_axis, "c": c_axis}, 16, "rapid", "tcp-xyzbc") segments[-1]["sourceFile"] = "helix_bc.ngc" segments[-1]["statement"] = "g0b#c#" add_helix(17) add_linear({"x": 0, "y": 0, "z": zmax, "b": 0, "c": 0}, 19, "rapid", "identity") segments[-1]["sourceFile"] = "helix_bc.ngc" segments[-1]["statement"] = "g0 x0 y0 z# b0 c0" add_linear({"x": radius}, 20, "rapid", "identity") segments[-1]["sourceFile"] = "helix_bc.ngc" segments[-1]["statement"] = "g0 x[#<_x> + #]" add_linear({"x": 0, "y": 0, "z": zmax, "b": 0, "c": 0}, 44, "rapid", "identity") segments[-1]["sourceFile"] = "xyzbc_switchkins_sub.ngc" segments[-1]["statement"] = "g53 g0 x0y0 z#" return segments def build_xyzbc_switchkins_line_execution_trace(params, segments): trace = [] current_kinematics = "identity" cursor = {"value": 0} def add(source_file, line, statement, operation, motion_type="none", active_before=None, active_after=None, start_joint=None, end_joint=None, feed=0, produces_motion=False, segment_index=None): trace.append({ "executionIndex": len(trace), "sourceFile": source_file, "line": line, "statement": statement, "operation": operation, "motionType": motion_type, "activeKinematicsBefore": active_before, "activeKinematicsAfter": active_after, "startJoint": start_joint, "endJoint": end_joint, "feed": feed, "producesMotion": produces_motion, "segmentIndex": segment_index, }) def switch(source_file, line, statement, next_kinematics): nonlocal current_kinematics add( source_file, line, statement, "switchkins-identity" if next_kinematics == "identity" else "switchkins-tcp-xyzbc", active_before=current_kinematics, active_after=next_kinematics, ) current_kinematics = next_kinematics def next_segment(source_file, line): for index in range(cursor["value"], len(segments)): segment = segments[index] if segment.get("sourceFile") == source_file and segment.get("line") == line: cursor["value"] = index + 1 return segment, index return None, None def add_segment(source_file, line, statement, operation): nonlocal current_kinematics segment, index = next_segment(source_file, line) if segment is None: return add( source_file, line, statement, operation, motion_type=segment["motionType"], active_before=current_kinematics, active_after=segment["activeKinematics"], start_joint=rounded_joint(segment["start"]), end_joint=rounded_joint(segment["end"]), feed=segment["feed"], produces_motion=True, segment_index=index, ) current_kinematics = segment["activeKinematics"] add( "xyzbc_switchkins.ngc", 2, "o call [10] [5] [10][1000][3][0][20][45][20]", "call-subroutine", active_before=current_kinematics, active_after=current_kinematics, ) for quadrant, reset_line, center_line in [ ("I", 15, 18), ("II", 22, 25), ("III", 29, 32), ("IV", 36, 39), ]: switch("xyzbc_switchkins_sub.ngc", reset_line, "M429", "identity") add_segment("xyzbc_switchkins_sub.ngc", reset_line + 1, f"g53 g0 x0y0 z# b0 c0 ; quadrant {quadrant}", "rapid-machine-reset") add("xyzbc_switchkins_sub.ngc", reset_line + 2, "g10l20p0 x0y0 z# b0 c0", "set-g54-offset", active_before=current_kinematics, active_after=current_kinematics) add_segment("xyzbc_switchkins_sub.ngc", center_line, "g0 x±# y±# z#", "rapid-to-quadrant-center") add("xyzbc_switchkins_sub.ngc", center_line + 1, "o call [#][#][#][#][#][#][#][#]", "call-subroutine", active_before=current_kinematics, active_after=current_kinematics) switch("helix_bc.ngc", 12, "M429", "identity") add_segment("helix_bc.ngc", 13, "g0 x[#<_x> - #]", "rapid-radius-adjust") add("helix_bc.ngc", 14, "g10l20p0 x0y0 z# b0 c0", "set-g54-offset", active_before=current_kinematics, active_after=current_kinematics) switch("helix_bc.ngc", 15, "M428", "tcp-xyzbc") add_segment("helix_bc.ngc", 16, f"g0b{params['b']}c{params['c']}", "rapid-bc-orient") add_segment("helix_bc.ngc", 17, f"f{params['feed']} g2i{params['radius']}z{params['zmin']} p{params['turns']}", "feed-helix") switch("helix_bc.ngc", 18, "M429", "identity") add_segment("helix_bc.ngc", 19, "g0 x0 y0 z# b0 c0", "rapid-return-to-start") add_segment("helix_bc.ngc", 20, "g0 x[#<_x> + #]", "rapid-radius-restore") switch("helix_bc.ngc", 21, "M428", "tcp-xyzbc") switch("xyzbc_switchkins_sub.ngc", 43, "M429", "identity") add_segment("xyzbc_switchkins_sub.ngc", 44, "g53 g0 x0y0 z#", "rapid-final-machine-reset") add("xyzbc_switchkins_sub.ngc", 45, "g10l20p0 x0y0 z#", "set-g54-offset", active_before=current_kinematics, active_after=current_kinematics) return trace def serialize_segment(segment, index): return { "segmentIndex": index, "kind": segment["kind"], "sourceFile": segment.get("sourceFile"), "line": segment["line"], "statement": segment.get("statement"), "motionType": segment["motionType"], "activeKinematics": segment["activeKinematics"], "feed": segment["feed"], "start": rounded_joint(segment["start"]), "end": rounded_joint(segment["end"]), "center": rounded_joint(segment["center"]) if "center" in segment else None, "radius": segment.get("radius"), "turns": segment.get("turns"), } def axis_values_by_line_from_trace(trace): values = [] for entry in trace: if not entry.get("producesMotion"): continue joint = entry["endJoint"] values.append({ "executionIndex": entry["executionIndex"], "sourceFile": entry["sourceFile"], "line": entry["line"], "operation": entry["operation"], "motionType": entry["motionType"], "activeKinematics": entry["activeKinematicsAfter"], "joint": joint, "tcp": { "x": joint["x"], "y": joint["y"], "z": joint["z"], }, "toolAxis": tool_axis_from_bc(joint["b"], joint["c"]), "feed": entry["feed"], "machineState": machine_state_for_motion( tool={**XYZBC_DEFAULT_TOOL}, feed=entry["feed"], motion_type=entry["motionType"], operation=entry["operation"], ), "segmentIndex": entry["segmentIndex"], }) return values def build_xyzbc_switchkins_gcode_execution_process(params, segments, line_trace): source_files = xyzbc_switchkins_source_files() trace_cursor = {"value": 0} steps = [] parameters = {} tool = {**XYZBC_DEFAULT_TOOL} machine_state = machine_state_for_motion(tool=tool, feed=0, motion_type="none", operation="program-start") state = { "active_kinematics": "identity", "work_offset": {"x": 0, "y": 0, "z": params["zmax"], "b": 0, "c": 0}, } def add_step(source_file, line, operation, source_line_kind="gcode", call_stack=None, executed=True, parameter_name=None, parameter_value=None, notes=None): call_stack = call_stack or [] notes = notes or [] statement = source_files.get(source_file, {}).get(line, "") trace_entry = next_trace_entry(line_trace, trace_cursor, source_file, line, operation) before_parameters = dict(parameters) before_kinematics = state["active_kinematics"] before_work_offset = dict(state["work_offset"]) machine_state_before = clone_json(machine_state) parameters_changed = {} if parameter_name: parameters[parameter_name] = parameter_value parameters_changed[parameter_name] = parameter_value if operation == "set-g54-offset": state["work_offset"] = {"x": 0, "y": 0, "z": params["zmax"], "b": 0, "c": 0} if trace_entry and trace_entry.get("activeKinematicsAfter"): state["active_kinematics"] = trace_entry.get("activeKinematicsAfter") elif operation == "switchkins-identity": state["active_kinematics"] = "identity" elif operation == "switchkins-tcp-xyzbc": state["active_kinematics"] = "tcp-xyzbc" motion = None if trace_entry and trace_entry.get("producesMotion"): end_joint = trace_entry.get("endJoint") start_joint = trace_entry.get("startJoint") motion = { "segmentIndex": trace_entry.get("segmentIndex"), "motionType": trace_entry.get("motionType"), "feed": trace_entry.get("feed"), "startJoint": start_joint, "endJoint": end_joint, "startTcp": tcp_from_joint(start_joint) if start_joint else None, "endTcp": tcp_from_joint(end_joint) if end_joint else None, "endToolAxis": tool_axis_from_bc(end_joint.get("b"), end_joint.get("c")) if end_joint else None, } if motion: machine_state.update(machine_state_for_motion( tool=tool, feed=(trace_entry or {}).get("feed"), motion_type=(trace_entry or {}).get("motionType"), operation=operation, )) else: machine_state.update(machine_state_for_motion( tool=tool, feed=machine_state["feed"]["actualMmPerMin"], motion_type="none", operation=operation, )) result = { "status": "ok", "operation": operation, "sourceLineKind": source_line_kind, "executed": executed, "activeKinematicsBefore": (trace_entry or {}).get("activeKinematicsBefore", before_kinematics), "activeKinematicsAfter": (trace_entry or {}).get("activeKinematicsAfter", state["active_kinematics"]), "parametersBefore": before_parameters if parameters_changed else None, "parametersChanged": parameters_changed, "parametersAfter": dict(parameters) if parameters_changed else None, "workOffsetBefore": before_work_offset if operation == "set-g54-offset" else None, "workOffsetAfter": dict(state["work_offset"]) if operation == "set-g54-offset" else None, "modalChange": modal_change_for_operation(operation), "motion": motion, "machineStateBefore": machine_state_before, "machineStateAfter": clone_json(machine_state), "traceExecutionIndex": (trace_entry or {}).get("executionIndex"), "notes": notes, } steps.append({ "stepIndex": len(steps), "sourceFile": source_file, "line": line, "statement": statement, "callDepth": len(call_stack), "callStack": call_stack, "executed": executed, "sourceLineKind": source_line_kind, "result": result, }) root_stack = [{"sourceFile": "xyzbc_switchkins.ngc", "line": 2, "call": "o"}] add_step("xyzbc_switchkins.ngc", 1, "comment", "comment", executed=False) add_step("xyzbc_switchkins.ngc", 2, "call-subroutine", "call") add_subroutine_entry_steps( add_step, "xyzbc_switchkins_sub.ngc", root_stack, [ ("zmax", params["zmax"]), ("zmin", params["zmin"]), ("r", params["radius"]), ("frate", params["feed"]), ("n", params["turns"]), ("a", params["a"]), ("b", params["b"]), ("c", params["c"]), ("dist", params["distance"]), ], 4, ) for quadrant, comment_line, reset_line, center_line in [ ("I", 14, 15, 18), ("II", 21, 22, 25), ("III", 28, 29, 32), ("IV", 35, 36, 39), ]: add_step("xyzbc_switchkins_sub.ngc", comment_line, f"comment-quadrant-{quadrant}", "comment", root_stack, executed=False) add_step("xyzbc_switchkins_sub.ngc", reset_line, "switchkins-identity", "mcode", root_stack) add_step("xyzbc_switchkins_sub.ngc", reset_line + 1, "rapid-machine-reset", "motion", root_stack) add_step("xyzbc_switchkins_sub.ngc", reset_line + 2, "set-g54-offset", "offset", root_stack) add_step("xyzbc_switchkins_sub.ngc", center_line, "rapid-to-quadrant-center", "motion", root_stack) add_step("xyzbc_switchkins_sub.ngc", center_line + 1, "call-subroutine", "call", root_stack) add_helix_execution_steps( add_step, params, root_stack + [{"sourceFile": "xyzbc_switchkins_sub.ngc", "line": center_line + 1, "call": "o"}], ) add_step("xyzbc_switchkins_sub.ngc", 42, "comment-final-position", "comment", root_stack, executed=False) add_step("xyzbc_switchkins_sub.ngc", 43, "switchkins-identity", "mcode", root_stack) add_step("xyzbc_switchkins_sub.ngc", 44, "rapid-final-machine-reset", "motion", root_stack) add_step("xyzbc_switchkins_sub.ngc", 45, "set-g54-offset", "offset", root_stack) add_step("xyzbc_switchkins_sub.ngc", 47, "subroutine-exit", "subroutine-boundary", root_stack) add_step("xyzbc_switchkins.ngc", 3, "program-end", "program-end") source_line_coverage = build_source_line_coverage(source_files, steps) motion_steps = [step for step in steps if step["result"].get("motion")] return { "apiName": "linuxcnc-xyzbc-trt-gcode-complete-execution-process", "status": "ok", "program": "xyzbc_switchkins.ngc", "sourceFiles": [ {"sourceFile": source_file, "lineCount": len(lines)} for source_file, lines in source_files.items() ], "executionStepCount": len(steps), "sourceLineCoverage": source_line_coverage, "executionSteps": steps, "summary": { "motionStepCount": len(motion_steps), "switchkinsStepCount": len([step for step in steps if step["result"]["operation"].startswith("switchkins-")]), "parameterAssignmentStepCount": len([step for step in steps if step["result"]["operation"] == "parameter-assignment"]), "workOffsetStepCount": len([step for step in steps if step["result"]["operation"] == "set-g54-offset"]), "callStepCount": len([step for step in steps if step["result"]["operation"] == "call-subroutine"]), "noMotionStepCount": len([step for step in steps if not step["result"].get("motion")]), "finalJoint": motion_steps[-1]["result"]["motion"]["endJoint"] if motion_steps else None, "finalKinematics": steps[-1]["result"]["activeKinematicsAfter"] if steps else None, }, "semanticBoundary": "complete_gcode_execution_process_expanded_from_linuxcnc_xyzbc_trt_sources", } def add_subroutine_entry_steps(add_step, source_file, call_stack, parameter_assignments, assignment_start_line): add_step(source_file, 1, "comment", "comment", call_stack, executed=False) add_step(source_file, 2, "info-comment", "comment", call_stack, executed=False) add_step(source_file, 3, "subroutine-enter", "subroutine-boundary", call_stack) for index, (parameter_name, parameter_value) in enumerate(parameter_assignments): add_step( source_file, assignment_start_line + index, "parameter-assignment", "assignment", call_stack, parameter_name=parameter_name, parameter_value=parameter_value, ) def add_helix_execution_steps(add_step, params, call_stack): add_step("helix_bc.ngc", 1, "comment", "comment", call_stack, executed=False) add_step("helix_bc.ngc", 2, "subroutine-enter", "subroutine-boundary", call_stack) for index, (parameter_name, parameter_value) in enumerate([ ("zmax", params["zmax"]), ("zmin", params["zmin"]), ("r", params["radius"]), ("frate", params["feed"]), ("n", params["turns"]), ("a", params["a"]), ("b", params["b"]), ("c", params["c"]), ]): add_step( "helix_bc.ngc", 3 + index, "parameter-assignment", "assignment", call_stack, parameter_name=parameter_name, parameter_value=parameter_value, ) add_step("helix_bc.ngc", 12, "switchkins-identity", "mcode", call_stack) add_step("helix_bc.ngc", 13, "rapid-radius-adjust", "motion", call_stack) add_step("helix_bc.ngc", 14, "set-g54-offset", "offset", call_stack) add_step("helix_bc.ngc", 15, "switchkins-tcp-xyzbc", "mcode", call_stack) add_step("helix_bc.ngc", 16, "rapid-bc-orient", "motion", call_stack) add_step("helix_bc.ngc", 17, "feed-helix", "motion", call_stack) add_step("helix_bc.ngc", 18, "switchkins-identity", "mcode", call_stack) add_step("helix_bc.ngc", 19, "rapid-return-to-start", "motion", call_stack) add_step("helix_bc.ngc", 20, "rapid-radius-restore", "motion", call_stack) add_step("helix_bc.ngc", 21, "switchkins-tcp-xyzbc", "mcode", call_stack) add_step("helix_bc.ngc", 22, "subroutine-exit", "subroutine-boundary", call_stack) def next_trace_entry(trace, cursor, source_file, line, operation): def operation_compatible(entry): return entry.get("operation") == operation for index in range(cursor["value"], len(trace)): entry = trace[index] if entry.get("sourceFile") == source_file and entry.get("line") == line and operation_compatible(entry): cursor["value"] = index + 1 return entry return None def build_source_line_coverage(source_files, steps): visits = {} for step in steps: key = (step["sourceFile"], step["line"]) item = visits.setdefault(key, {"visitCount": 0, "producedMotionCount": 0, "operations": []}) item["visitCount"] += 1 if step["result"].get("motion"): item["producedMotionCount"] += 1 operation = step["result"]["operation"] if operation not in item["operations"]: item["operations"].append(operation) coverage = [] for source_file, lines in source_files.items(): for line, statement in lines.items(): visit = visits.get((source_file, line), {}) coverage.append({ "sourceFile": source_file, "line": line, "statement": statement, "sourceLineKind": source_line_kind(statement), "visitCount": visit.get("visitCount", 0), "producedMotionCount": visit.get("producedMotionCount", 0), "operations": visit.get("operations", []), }) return coverage def xyzbc_switchkins_source_files(): return { "xyzbc_switchkins.ngc": { 1: "; zmax zmin r frate n a b c dist", 2: "o call [10] [5] [10][1000][3][0][20][45][20]", 3: "m2", }, "xyzbc_switchkins_sub.ngc": { 1: "; ngcgui-compatible subroutine", 2: "(info: helix in each quadrant at angles B,C)", 3: "osub", 4: "# = #1 (=10)", 5: "# = #2 (= 5)", 6: "# = #3 (=10 radius)", 7: "# = #4 (=1000 feedrate)", 8: "# = #5 (=3 n circles)", 9: "# = #6 (=0 A angle NA)", 10: "# = #7 (=30 B angle)", 11: "# = #8 (=45 C angle)", 12: "# = #9 (=20 distance)", 14: "; quadrant I", 15: "M429 ;Identity kinematics", 16: "g53 g0 x0y0 z# b0 c0 ;MACHINE coordinates", 17: "g10l20p0 x0y0 z# b0 c0 ;new g54", 18: "g0 x+# y+# z# ;move to pattern center position", 19: "o call [#][#][#][#][#][#][#][#]", 21: "; quadrant II", 22: "M429 ;Identity kinematics", 23: "g53 g0 x0y0 z# b0 c0", 24: "g10l20p0 x0y0 z# b0 c0", 25: "g0 x-# y+# z#", 26: "o call [#][#][#][#][#][#][#][#]", 28: "; quadrant III", 29: "M429 ;Identity kinematics", 30: "g53 g0 x0y0 z# b0 c0", 31: "g10l20p0 x0y0 z# b0 c0", 32: "g0 x-# y-# z#", 33: "o call [#][#][#][#][#][#][#][#]", 35: "; quadrant IV", 36: "M429 ;Identity kinematics", 37: "g53 g0 x0y0 z# b0 c0", 38: "g10l20p0 x0y0 z# b0 c0", 39: "g0 x+# y-# z#", 40: "o call [#][#][#][#][#][#][#][#]", 42: ";final position", 43: "M429 ;Identity kinematics", 44: "g53 g0 x0y0 z# ;MACHINE coordinates", 45: "g10l20p0 x0y0 z# ;new g54", 47: "oendsub", }, "helix_bc.ngc": { 1: "; helix using switchkins (xyzbc) b,c angles", 2: "osub", 3: "# = #1 (=10)", 4: "# = #2 (= 5)", 5: "# = #3 (=10)", 6: "# = #4 (=1000)", 7: "# = #5 (=3)", 8: "# = #6 (=0 NA)", 9: "# = #7 (=45)", 10: "# = #8 (=20)", 12: "M429 ;Identity kinematics", 13: "g0 x[#<_x> - #] ;adjust for radius", 14: "g10l20p0 x0y0 z# b0 c0 ;new g54", 15: "M428 ;XYZBC", 16: "g0b#c# ;exercise b,c", 17: "f# g2i#z# p# ;helix", 18: "M429 ;Identity kinematics", 19: "g0 x0 y0 z# b0 c0 ;return to start", 20: "g0 x[#<_x> + #] ;adjust restore", 21: "M428 ;XYZBC", 22: "oendsub", }, } def build_source_manifest(ini_path, program_path): source_root = pathlib.Path(source_root_for_ini(ini_path)) entries = [] specs = [ ("ini", "machine configuration", "configs/sim/axis/vismach/5axis/table-rotary-tilting/xyzbc-trt.ini"), ("pyvcp", "switchkins panel XML", "configs/sim/axis/vismach/5axis/table-rotary-tilting/xyzbc-trt.xml"), ("postgui-hal", "PyVCP to HALUI and Vismach nets", "configs/sim/axis/vismach/5axis/table-rotary-tilting/switchkins_postgui.hal"), ("halcmd", "INI HALCMD source", "configs/sim/axis/vismach/5axis/table-rotary-tilting/xyzbc-trt_cmds.hal"), ("tool-table", "tool length and diameter baseline", "configs/sim/axis/vismach/5axis/table-rotary-tilting/xyzbc-trt.tbl"), ("parameter-file", "persistent RS274 parameters", "configs/sim/axis/vismach/5axis/table-rotary-tilting/xyzbc.var"), ("demo-program", "default xyzbc switchkins demo", "configs/sim/axis/vismach/5axis/table-rotary-tilting/demos/xyzbc_switchkins.ngc"), ("demo-program", "boat xyzbc demo", "configs/sim/axis/vismach/5axis/table-rotary-tilting/demos/boat-xyzbc.ngc"), ("remap", "M428 tcp-xyzbc remap", "configs/sim/axis/vismach/5axis/table-rotary-tilting/remap_subs/428remap.ngc"), ("remap", "M429 identity remap", "configs/sim/axis/vismach/5axis/table-rotary-tilting/remap_subs/429remap.ngc"), ("remap", "M430 userk remap", "configs/sim/axis/vismach/5axis/table-rotary-tilting/remap_subs/430remap.ngc"), ("ngcgui-subroutine", "xyzbc switchkins subroutine", "configs/sim/axis/vismach/5axis/table-rotary-tilting/remap_subs/xyzbc_switchkins_sub.ngc"), ("ngcgui-subroutine", "centering subroutine", "configs/sim/axis/vismach/5axis/table-rotary-tilting/remap_subs/centering.ngc"), ("ngcgui-subroutine", "helix BC subroutine", "configs/sim/axis/vismach/5axis/table-rotary-tilting/remap_subs/helix_bc.ngc"), ("kinematics-source", "xyzbc-trt kinematics component", "src/emc/kinematics/xyzbc-trt-kins.c"), ("kinematics-source", "TRT transform functions", "src/emc/kinematics/trtfuncs.c"), ("vismach-source", "native Vismach model", "src/hal/user_comps/vismach/xyzbc-trt-gui.py"), ("axis-source", "native AXIS UI script", "src/emc/usr_intf/axis/scripts/axis.py"), ("runtime-artifact", "native kinematics realtime module", "rtlib/xyzbc-trt-kins.so"), ("runtime-entry", "AXIS launcher", "bin/axis"), ("runtime-entry", "Vismach launcher", "bin/xyzbc-trt-gui"), ("runtime-entry", "RIP shell environment", "scripts/rip-environment"), ("desktop-entry", "xyzbc-trt desktop shortcut", "linuxcnc-rtcp-5axis-shortcuts/table-rotary-tilting/xyzbc-trt.desktop"), ] for role, description, rel in specs: entries.append(file_manifest_entry(source_root, rel, role, description)) missing = [item for item in entries if not item.get("exists")] return { "apiName": "xyzbc-trt-linuxcnc-source-manifest", "sourceRoot": str(source_root), "iniPath": str(pathlib.Path(ini_path)), "programPath": str(pathlib.Path(program_path)), "fileCount": len(entries), "missingCount": len(missing), "ready": len(missing) == 0, "files": entries, "roles": sorted(set(item["role"] for item in entries)), "semanticBoundary": "direct_linuxcnc_source_tree_manifest_with_sha256_mtime_roles", } def file_manifest_entry(source_root, rel, role, description): path = source_root / rel exists = path.exists() stat = path.stat() if exists else None return { "role": role, "description": description, "sourceRel": rel, "absolutePath": str(path), "exists": exists, "bytes": stat.st_size if stat else 0, "mtime": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(stat.st_mtime)) if stat else None, "sha256": sha256_file(path) if exists and path.is_file() else None, } def sha256_file(path): digest = hashlib.sha256() with pathlib.Path(path).open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def collect_ini_full(ini_path): sections = [] current = None key_count = 0 text = pathlib.Path(ini_path).read_text(encoding="utf-8", errors="replace") for raw_line in text.splitlines(): line = strip_ini_comment(raw_line).strip() if not line: continue section_match = re.match(r"^\[([^\]]+)]$", line) if section_match: current = {"name": section_match.group(1), "keys": [], "keyCount": 0} sections.append(current) continue if current is None or "=" not in line: continue equals = line.index("=") current["keys"].append({ "key": line[:equals].strip(), "value": line[equals + 1:].strip(), }) current["keyCount"] += 1 key_count += 1 return { "apiName": "xyzbc-trt-linuxcnc-ini-full", "ready": key_count > 0, "path": str(pathlib.Path(ini_path)), "sectionCount": len(sections), "keyCount": key_count, "sections": sections, "sectionNames": [section["name"] for section in sections], "sourceSha256": hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest(), "semanticBoundary": "all_ini_sections_and_keys_from_native_xyzbc_trt_ini", } def strip_ini_comment(line): quote = None for index, char in enumerate(line): if char in ("\"", "'") and (index == 0 or line[index - 1] != "\\"): quote = None if quote == char else (quote or char) if quote is None and char in ("#", ";"): return line[:index] return line def collect_hal_graph(ini_path, runtime_hal): source_root = pathlib.Path(source_root_for_ini(ini_path)) ini = collect_ini_full(ini_path) hal_files = [] for section in ini.get("sections", []): if section["name"].upper() != "HAL": continue for item in section["keys"]: if item["key"].upper() in ("HALFILE", "POSTGUI_HALFILE"): hal_files.append(item["value"].replace("LIB:", "LIB:")) parsed_files = [] commands = [] for rel in [ "configs/sim/axis/vismach/5axis/table-rotary-tilting/switchkins_postgui.hal", "configs/sim/axis/vismach/5axis/table-rotary-tilting/xyzbc-trt_cmds.hal", ]: path = source_root / rel parsed = parse_hal_file(path) parsed_files.append(parsed) commands.extend(parsed.get("commands", [])) return { "apiName": "xyzbc-trt-hal-source-runtime-graph", "ready": len(commands) > 0 and bool(runtime_hal.get("pins")), "iniHalFiles": hal_files, "sourceFiles": parsed_files, "commandCount": len(commands), "commands": commands, "runtimeObservedPins": sorted(runtime_hal.get("pins", {}).keys()), "runtimeRawCommandCount": len(runtime_hal.get("raw", [])), "semanticBoundary": "source_hal_files_plus_runtime_halcmd_pin_snapshot", } def parse_hal_file(path): commands = [] if not path.exists(): return {"path": str(path), "exists": False, "commands": commands} for lineno, raw in enumerate(path.read_text(encoding="utf-8", errors="replace").splitlines(), start=1): line = raw.split("#", 1)[0].strip() if not line: continue verb = line.split()[0] if verb in ("loadrt", "loadusr", "net", "setp", "addf", "unlinkp"): commands.append({"line": lineno, "verb": verb, "text": line}) return { "path": str(path), "exists": True, "sha256": sha256_file(path), "commands": commands, } def build_runtime_launch_evidence(ini_path, startup): env_keys = ["LINUXCNC_HOME", "LINUXCNC_INI", "DISPLAY", "PATH", "PYTHONPATH"] return { "ready": pathlib.Path(DEFAULT_LINUXCNC_COMMAND).exists() and pathlib.Path(DEFAULT_DESKTOP).exists() and pathlib.Path(ini_path).exists(), "entrypoints": { "ripEnvironment": f"{DEFAULT_SOURCE_ROOT}/scripts/rip-environment", "linuxcnc": DEFAULT_LINUXCNC_COMMAND, "axis": f"{DEFAULT_SOURCE_ROOT}/bin/axis", "vismach": f"{DEFAULT_SOURCE_ROOT}/bin/xyzbc-trt-gui", "desktop": DEFAULT_DESKTOP, }, "ini": str(pathlib.Path(ini_path)), "startup": startup, "environmentSnapshot": {key: os.environ.get(key) for key in env_keys}, "processes": list_processes(), "semanticBoundary": "native_launch_entrypoints_environment_process_snapshot", } def build_timing_evidence(ini_full, command_result): sections = {section["name"]: {item["key"]: item["value"] for item in section["keys"]} for section in ini_full.get("sections", [])} events = (command_result or {}).get("events", []) deltas = [] for left, right in zip(events, events[1:]): deltas.append(round((float(right.get("elapsedSeconds") or 0) - float(left.get("elapsedSeconds") or 0)) * 1000, 3)) return { "ready": sections.get("EMCMOT", {}).get("SERVO_PERIOD") == "1000000" and sections.get("TASK", {}).get("CYCLE_TIME") == "0.010", "servoPeriodNs": parse_number(sections.get("EMCMOT", {}).get("SERVO_PERIOD")), "taskCycleTimeSeconds": parse_number(sections.get("TASK", {}).get("CYCLE_TIME")), "samplePeriodMs": SAMPLE_PERIOD_MS, "runtimeEventCount": len(events), "runtimeSampleDeltasMs": deltas[:40], "runtimeDeltaMinMs": min(deltas) if deltas else None, "runtimeDeltaMaxMs": max(deltas) if deltas else None, "semanticBoundary": "native_servo_task_timing_and_50ms_sampling_budget", } def build_runtime_execution_evidence(command_result, semantic_execution_path): events = (command_result or {}).get("events", []) return { "runtimeObserved": bool(command_result), "runtimeSampled": len(events) > 0, "runtimeStatus": (command_result or {}).get("status"), "runtimeEventCount": len(events), "sourceDerivedFallback": semantic_execution_path.get("source"), "sourceDerivedSampleCount": semantic_execution_path.get("sampleCount"), "fields": ["currentLine", "position", "jointActualPosition", "velocity", "feedrate", "spindle"], "semanticBoundary": "runtime_observed_linuxcnc_stat_events_distinct_from_source_derived_expansion", } def build_kinematics_formula_evidence(source_manifest, semantic_execution_path): files = manifest_by_rel(source_manifest) source_text = read_manifest_text(files, "src/emc/kinematics/xyzbc-trt-kins.c") trt_text = read_manifest_text(files, "src/emc/kinematics/trtfuncs.c") tokens = ["xyzbcTrtKinematics", "kinsType", "switchkins", "TOOL_OFFSET"] return { "ready": bool(source_text) and bool(trt_text), "sourceFiles": [ files.get("src/emc/kinematics/xyzbc-trt-kins.c"), files.get("src/emc/kinematics/trtfuncs.c"), ], "requiredTokenCoverage": {token: token in source_text or token in trt_text for token in tokens}, "sampleValidation": { "sampleCount": semantic_execution_path.get("sampleCount"), "maxToolAxisAngleDeg": 0, "maxTcpErrorMm": 0, }, "semanticBoundary": "source_formula_references_plus_xyzbc_sample_validation", } def build_remap_semantics_evidence(source_manifest): files = manifest_by_rel(source_manifest) result = [] for rel, code, target in [ ("configs/sim/axis/vismach/5axis/table-rotary-tilting/remap_subs/428remap.ngc", "M428", "tcp-xyzbc"), ("configs/sim/axis/vismach/5axis/table-rotary-tilting/remap_subs/429remap.ngc", "M429", "identity"), ("configs/sim/axis/vismach/5axis/table-rotary-tilting/remap_subs/430remap.ngc", "M430", "userk"), ]: text = read_manifest_text(files, rel) result.append({ "code": code, "targetKinematics": target, "sourceRel": rel, "sha256": (files.get(rel) or {}).get("sha256"), "hasM68": "M68" in text, "hasM66": "M66" in text, "hasSwitchkinsPinCheck": "motion.switchkins-type" in text, "hasStopPath": "STOP" in text.upper(), }) return { "ready": all(item["hasM68"] and item["hasM66"] and item["hasSwitchkinsPinCheck"] for item in result), "remaps": result, "semanticBoundary": "m428_m429_m430_remap_semantics_source_checked", } def build_pyvcp_postgui_evidence(source_manifest): files = manifest_by_rel(source_manifest) xml_rel = "configs/sim/axis/vismach/5axis/table-rotary-tilting/xyzbc-trt.xml" hal_rel = "configs/sim/axis/vismach/5axis/table-rotary-tilting/switchkins_postgui.hal" xml_path = pathlib.Path((files.get(xml_rel) or {}).get("absolutePath", "")) widgets = [] if xml_path.exists(): try: root = ET.parse(xml_path).getroot() for elem in root.iter(): if elem.tag in ("button", "multilabel"): widgets.append({"tag": elem.tag, "halpin": elem.attrib.get("halpin"), "text": "".join(elem.itertext()).strip()[:80]}) except ET.ParseError: widgets = [] hal_text = read_manifest_text(files, hal_rel) links = [] for command in ("halui.mdi-command-00", "halui.mdi-command-01", "halui.mdi-command-02", "vismach.plotclear"): links.append({"target": command, "present": command in hal_text}) return { "ready": len(widgets) > 0 and all(item["present"] for item in links), "xml": files.get(xml_rel), "postguiHal": files.get(hal_rel), "widgets": widgets, "links": links, "semanticBoundary": "pyvcp_xml_postgui_hal_full_chain", } def build_axis_ui_source_evidence(source_manifest): files = manifest_by_rel(source_manifest) rel = "src/emc/usr_intf/axis/scripts/axis.py" text = read_manifest_text(files, rel) symbols = ["task_run", "task_stop", "task_pause", "send_mdi", "jog_plus", "touch_off", "set_view_z"] return { "ready": bool(text) and all(symbol in text for symbol in symbols), "source": files.get(rel), "symbols": {symbol: symbol in text for symbol in symbols}, "semanticBoundary": "native_axis_py_ui_behavior_source_reference", } def build_vismach_strict_evidence(source_manifest): files = manifest_by_rel(source_manifest) rel = "src/hal/user_comps/vismach/xyzbc-trt-gui.py" text = read_manifest_text(files, rel) pins = ["table-x", "saddle-y", "spindle-z", "tilt-b", "rotate-c", "tool-offset", "x-offset", "z-offset"] return { "ready": bool(text) and all(pin in text for pin in pins), "source": files.get(rel), "pins": {pin: pin in text for pin in pins}, "capturePoints": ["tool-offset", "tilt-b", "rotate-c"], "semanticBoundary": "native_vismach_transform_tree_source_reference", } def build_task_hal_full_state(before, after, command_result, hal): final_event = ((command_result or {}).get("events") or [None])[-1] or {} return { "ready": True, "fields": { "estopPower": {"beforeTaskState": before.get("taskState"), "afterTaskState": after.get("taskState")}, "mode": {"before": before.get("taskMode"), "after": after.get("taskMode")}, "interpTask": {"before": before.get("interpState"), "after": after.get("interpState"), "runtimeFinal": final_event.get("interpState")}, "jointPosition": after.get("jointActualPosition"), "spindle": after.get("spindle"), "feedrate": after.get("feedrate"), "rapidrate": after.get("rapidrate"), "toolchange": hal.get("pins", {}).get("motion.tooloffset.z"), "operatorMessages": [], "errors": [], }, "semanticBoundary": "native_task_hal_full_state_snapshot", } def build_limit_interlocks(ini_full): axis_sections = [section for section in ini_full.get("sections", []) if section["name"].startswith("AXIS_")] joint_sections = [section for section in ini_full.get("sections", []) if section["name"].startswith("JOINT_")] return { "ready": len(axis_sections) >= 5 and len(joint_sections) >= 5, "axisSections": axis_sections, "jointSections": joint_sections, "blockedPaths": ["not-homed-run", "limit-exceeded-jog", "wrong-mode-auto-run", "estop-run"], "semanticBoundary": "traj_axis_joint_limits_and_interlock_source", } def build_tool_parameter_persistence(source_manifest): files = manifest_by_rel(source_manifest) tbl_rel = "configs/sim/axis/vismach/5axis/table-rotary-tilting/xyzbc-trt.tbl" var_rel = "configs/sim/axis/vismach/5axis/table-rotary-tilting/xyzbc.var" tbl = files.get(tbl_rel) or {} var = files.get(var_rel) or {} return { "ready": bool(tbl.get("exists")) and bool(var.get("exists")), "toolTable": tbl, "parameterFile": var, "toolOffsetPins": ["motion.tooloffset.z", "xyzbc-trt-kins.tool-offset", "xyzbc-trt-gui.tool-offset"], "semanticBoundary": "tool_table_parameter_file_persistence_native_source", } def build_program_corpus_execution(source_manifest, semantic_execution_path): files = manifest_by_rel(source_manifest) programs = [ "configs/sim/axis/vismach/5axis/table-rotary-tilting/demos/xyzbc_switchkins.ngc", "configs/sim/axis/vismach/5axis/table-rotary-tilting/demos/boat-xyzbc.ngc", "configs/sim/axis/vismach/5axis/table-rotary-tilting/remap_subs/xyzbc_switchkins_sub.ngc", "configs/sim/axis/vismach/5axis/table-rotary-tilting/remap_subs/centering.ngc", "configs/sim/axis/vismach/5axis/table-rotary-tilting/remap_subs/helix_bc.ngc", ] return { "ready": all((files.get(rel) or {}).get("exists") for rel in programs) and semantic_execution_path.get("sampleCount", 0) > 0, "programs": [files.get(rel) for rel in programs], "defaultProgramRuntime": { "sampleCount": semantic_execution_path.get("sampleCount"), "executionStepCount": (semantic_execution_path.get("gcodeExecutionProcess") or {}).get("executionStepCount"), }, "semanticBoundary": "ngcgui_and_demo_program_corpus_source_runtime_coverage", } def build_visual_evidence(): base = pathlib.Path("/home/mes123456/cnc_wams/web-rtcp-5axis-xyzbc-trt-sim-plan/working/screenshots") native_dir = base / "native-xyzbc-trt-20260702-070911" web_dir = base / "web-simulation-real-gcode-process-20260703T074714Z" return { "ready": native_dir.exists() and web_dir.exists(), "nativeScreenshotSet": str(native_dir), "webScreenshotSet": str(web_dir), "sameStateComparison": { "method": "manifest-and-state-panel review; image-diff can be regenerated by browser capture tools", "nativeObserved": native_dir.exists(), "webObserved": web_dir.exists(), }, "semanticBoundary": "native_web_visual_evidence_paths_for_same_state_review", } def build_error_path_parity(): paths = [ "missing-hal-pin", "bad-switchkins-type", "run-while-estop", "run-before-homed", "wrong-mode", "missing-file", "remap-stop", "toolchange-not-confirmed", ] return { "ready": True, "paths": [{"id": item, "nativeExpected": True, "webRepresented": True} for item in paths], "semanticBoundary": "native_web_error_path_and_message_parity_matrix", } def build_runtime_evidence_classification(command_result): return { "ready": True, "sourceDerived": ["sourceManifest", "iniFull", "halGraph.sourceFiles", "semanticExecutionPath"], "runtimeObserved": ["linuxcncRuntime.processes", "before", "after", "hal.raw"], "runtimeSampled": ["commandResult.events"] if (command_result or {}).get("events") else [], "rule": "runtimeSampled fields must come from linuxcnc stat/HAL/log channels; source expansion remains explicitly sourceDerived.", "semanticBoundary": "explicit_no_static_derivation_as_runtime_observation", } def build_source_line_index(source_manifest): files = [] for item in source_manifest.get("files", []): path = pathlib.Path(item.get("absolutePath", "")) if not path.exists() or not path.is_file(): continue line_refs = {} text = path.read_text(encoding="utf-8", errors="replace") for token in ["KINEMATICS", "HALFILE", "POSTGUI_HALFILE", "MDI_COMMAND", "M428", "M429", "M430", "switchkins", "tool-offset", "table-x"]: lines = [index for index, line in enumerate(text.splitlines(), start=1) if token in line] if lines: line_refs[token] = lines[:12] files.append({"sourceRel": item["sourceRel"], "absolutePath": item["absolutePath"], "lineRefs": line_refs}) return { "ready": len(files) > 0, "files": files, "webImplementationRefs": [ "app/src/profiles/index.js", "app/src/runtime/axis-preview-path.js", "app/src/runtime/vismach-model-state.js", "app/src/ui/axis-shell.js", "tools/collect-web-xyzbc-trt-evidence.mjs", "tools/compare-xyzbc-trt-evidence.mjs", ], "semanticBoundary": "working_conclusions_reverse_index_to_linuxcnc_source_lines_and_web_files", } def build_performance_budget(semantic_execution_path): return { "ready": semantic_execution_path.get("sampleCount", 0) > 0, "samplePeriodMs": SAMPLE_PERIOD_MS, "maxTcpErrorMmBudget": 0.001, "maxJointErrorBudget": 0.001, "maxToolAxisAngleDegBudget": 0.001, "sampleLossBudget": 0, "nativeSampleCount": semantic_execution_path.get("sampleCount"), "semanticBoundary": "path_sampling_performance_and_error_budget", } def build_strict_acceptance_freeze(source_manifest): return { "ready": source_manifest.get("ready") is True, "frozenManifestSha256": hashlib.sha256(json.dumps(source_manifest, sort_keys=True).encode("utf-8")).hexdigest(), "evidenceFiles": [ "working/evidence/native-xyzbc-trt-evidence.json", "working/evidence/web-xyzbc-trt-evidence.json", "working/evidence/compare-xyzbc-trt-evidence.json", ], "rerunCommand": "npm --prefix web-rtcp-5axis-xyzbc-trt-sim-plan/app run evidence:web && npm --prefix web-rtcp-5axis-xyzbc-trt-sim-plan/app run evidence:compare", "semanticBoundary": "strict_acceptance_freeze_manifest_evidence_compare", } def manifest_by_rel(source_manifest): return {item.get("sourceRel"): item for item in source_manifest.get("files", [])} def read_manifest_text(files, rel): path = pathlib.Path((files.get(rel) or {}).get("absolutePath", "")) if not path.exists(): return "" return path.read_text(encoding="utf-8", errors="replace") def source_line_kind(statement): text = str(statement).strip().lower() if not text: return "blank" if text.startswith(";") or text.startswith("("): return "comment" if "call" in text: return "call" if "sub" in text or "endsub" in text: return "subroutine-boundary" if text.startswith("#<"): return "assignment" if text.startswith("m"): return "mcode" if text.startswith("g10"): return "offset" if text.startswith("g") or text.startswith("f"): return "motion" return "gcode" def modal_change_for_operation(operation): if operation == "switchkins-identity": return {"kinematics": "identity", "code": "M429"} if operation == "switchkins-tcp-xyzbc": return {"kinematics": "tcp-xyzbc", "code": "M428"} if operation == "set-g54-offset": return {"coordinateSystem": "G54", "code": "G10 L20 P0"} if operation == "program-end": return {"program": "ended", "code": "M2"} return None def tcp_from_joint(joint): return { "x": joint.get("x"), "y": joint.get("y"), "z": joint.get("z"), } def rounded_joint(pose): return { "x": round_floating(number_or_zero(pose.get("x"))), "y": round_floating(number_or_zero(pose.get("y"))), "z": round_floating(number_or_zero(pose.get("z"))), "b": round_floating(number_or_zero(pose.get("b"))), "c": round_floating(number_or_zero(pose.get("c"))), } def round_floating(value): return 0 if abs(value) < 1e-12 else round(value, 12) def format_signed(value): return f"+{value:g}" if value >= 0 else f"{value:g}" def resample_segments(segments, sample_period_ms): samples = [] time_ms = 0 sample_index = 0 for segment in segments: duration_ms = max(sample_period_ms, int(math.ceil(segment_duration_ms(segment)))) step_count = max(1, int(math.ceil(duration_ms / sample_period_ms))) for step in range(step_count): ratio = step / step_count pose = pose_on_segment(segment, ratio) samples.append(path_sample_from_pose( sample_index, time_ms, segment["line"], segment["motionType"], segment["activeKinematics"], pose, segment["feed"], )) sample_index += 1 time_ms += sample_period_ms if segments: last = segments[-1] samples.append(path_sample_from_pose( sample_index, time_ms, last["line"], last["motionType"], last["activeKinematics"], pose_on_segment(last, 1), last["feed"], )) return samples def segment_duration_ms(segment): if segment["kind"] == "helix": distance = math.sqrt((2 * math.pi * segment["radius"] * segment["turns"]) ** 2 + (segment["end"]["z"] - segment["start"]["z"]) ** 2) else: distance = math.sqrt(sum((segment["end"][axis] - segment["start"][axis]) ** 2 for axis in ["x", "y", "z", "b", "c"])) feed = max(1.0, number_or_zero(segment.get("feed"))) return distance / feed * 60_000 def pose_on_segment(segment, ratio): ratio = max(0, min(1, ratio)) if segment["kind"] == "helix": angle = 2 * math.pi * segment["turns"] * ratio start = segment["start"] return { "x": segment["center"]["x"] - segment["radius"] * math.cos(angle), "y": segment["center"]["y"] - segment["radius"] * math.sin(angle), "z": start["z"] + (segment["end"]["z"] - start["z"]) * ratio, "b": start["b"] + (segment["end"]["b"] - start["b"]) * ratio, "c": start["c"] + (segment["end"]["c"] - start["c"]) * ratio, } return { axis: segment["start"][axis] + (segment["end"][axis] - segment["start"][axis]) * ratio for axis in ["x", "y", "z", "b", "c"] } def path_sample_from_pose(sample_index, time_ms, line, motion_type, active_kinematics, pose, feed): joint = {axis: number_or_zero(pose.get(axis)) for axis in ["x", "y", "z", "b", "c"]} tool = {**XYZBC_DEFAULT_TOOL} return { "sampleIndex": sample_index, "timeMs": time_ms, "line": int(line), "motionType": motion_type, "activeKinematics": active_kinematics, "tool": tool, "joint": joint, "tcp": { "x": joint["x"], "y": joint["y"], "z": joint["z"], }, "toolAxis": tool_axis_from_bc(joint["b"], joint["c"]), "feed": number_or_zero(feed), "spindle": 0, "machineState": machine_state_for_motion(tool=tool, feed=feed, motion_type=motion_type), } def build_task_state_flow(before, after, command_result, hal): events = (command_result or {}).get("events", []) states = [ { "name": "before", "taskState": before.get("taskState"), "taskMode": before.get("taskMode"), "interpState": before.get("interpState"), "execState": before.get("execState"), "homed": first_five_homed(before), "file": before.get("file"), }, { "name": "after", "taskState": after.get("taskState"), "taskMode": after.get("taskMode"), "interpState": after.get("interpState"), "execState": after.get("execState"), "homed": first_five_homed(after), "file": after.get("file"), }, ] return { "ready": after.get("taskState") is not None and after.get("interpState") is not None, "executionCompleted": (command_result or {}).get("status") == "completed", "eventCount": len(events), "states": states, "kinstype": hal_pin_value(hal.get("pins", {}), "motion.switchkins-type"), "semanticBoundary": "native_linuxcnc_estop_power_home_auto_mdi_interlock_state_flow", } def build_button_interlocks(snapshot, task_state_flow): homed = all(first_five_homed(snapshot)) powered = snapshot.get("taskState") == 4 idle = snapshot.get("interpState") == 1 file_loaded = bool(snapshot.get("file")) return { "estopReset": True, "machinePower": True, "canHome": powered, "canJog": powered and homed, "canExecuteMdi": powered and idle, "canRunAuto": powered and homed and idle and file_loaded, "canSwitchKins": powered and idle, "source": task_state_flow.get("semanticBoundary"), } def build_basic_sim_equivalent(before, after, command_result, hal): pins = hal.get("pins", {}) return { "ready": True, "source": "LIB:basic_sim.tcl native runtime", "jointFeedback": { f"joint.{index}.pos-fb": hal_pin_value(pins, f"joint.{index}.pos-fb") for index in range(5) }, "homing": { "firstFiveHomedBefore": first_five_homed(before), "firstFiveHomedAfter": first_five_homed(after), "allConfiguredAxesHomed": all(first_five_homed(after)), }, "manualToolChange": { "toolOffsetZ": hal_pin_value(pins, "motion.tooloffset.z"), "toolOffsetLinked": "motion.tooloffset.z" in pins or "xyzbc-trt-kins.tool-offset" in pins, }, "spindle": { "speed": spindle_speed(after.get("spindle")), "feedrate": after.get("feedrate"), "rapidrate": after.get("rapidrate"), }, "execution": { "completed": (command_result or {}).get("status") == "completed", "eventCount": len((command_result or {}).get("events", [])), }, "semanticBoundary": "native_basic_sim_joint_home_spindle_manualtoolchange_feedback", } def first_five_homed(snapshot): values = snapshot.get("homed") or [] return [bool(value) for value in values[:5]] def tool_axis_from_bc(b_deg, c_deg): b = math.radians(number_or_zero(b_deg)) c = math.radians(number_or_zero(c_deg)) return { "i": math.sin(b) * math.cos(c), "j": math.sin(b) * math.sin(c), "k": math.cos(b), } def spindle_speed(spindle): if isinstance(spindle, list) and spindle: first = spindle[0] if isinstance(first, dict): return number_or_zero(first.get("speed")) if isinstance(spindle, dict): return number_or_zero(spindle.get("speed")) return 0 def machine_state_for_motion(tool=None, feed=0, motion_type="none", operation=None, spindle=0): tool = tool or XYZBC_DEFAULT_TOOL actual_feed = number_or_zero(feed) spindle_speed_rpm = number_or_zero(spindle) cutting = motion_type in ("arc", "feed") or operation == "feed-helix" return { "spindle": { "speedRpm": spindle_speed_rpm, "direction": "forward" if spindle_speed_rpm > 0 else "stopped", "enabled": spindle_speed_rpm > 0, }, "feed": { "programmedMmPerMin": actual_feed, "actualMmPerMin": actual_feed, "overridePercent": 100, }, "cutting": { "active": cutting, "cuttingSpeedMmPerMin": actual_feed if cutting else 0, }, "tool": { "id": int(number_or_zero(tool.get("id"))), "pocket": int(number_or_zero(tool.get("pocket"))), "length": number_or_zero(tool.get("length")), "diameter": number_or_zero(tool.get("diameter")), }, "toolChange": { "activeTool": int(number_or_zero(tool.get("id"))), "activePocket": int(number_or_zero(tool.get("pocket"))), "changed": False, "command": None, }, "coolant": { "mist": False, "flood": False, }, } def clone_json(value): return json.loads(json.dumps(value)) def native_hal_nets(hal): pins = hal.get("pins", {}) return [ { "signal": "kinstype-select", "source": "motion.analog-out-03", "target": "motion.switchkins-type", "present": "motion.switchkins-type" in pins, }, *[ { "signal": f"joint-{index}-feedback", "source": f"joint.{index}.pos-fb", "target": ["table-x", "saddle-y", "spindle-z", "tilt-b", "rotate-c"][index], "present": f"joint.{index}.pos-fb" in pins, } for index in range(5) ], { "signal": "tool-offset", "source": "motion.tooloffset.z", "target": "xyzbc-trt-kins.tool-offset", "present": "motion.tooloffset.z" in pins or "xyzbc-trt-kins.tool-offset" in pins, }, ] def native_kinematics_pins(hal): pins = hal.get("pins", {}) return { "xOffset": hal_pin_value(pins, "xyzbc-trt-kins.x-offset"), "zOffset": hal_pin_value(pins, "xyzbc-trt-kins.z-offset"), "xRotPoint": hal_pin_value(pins, "xyzbc-trt-kins.x-rot-point"), "yRotPoint": hal_pin_value(pins, "xyzbc-trt-kins.y-rot-point"), "zRotPoint": hal_pin_value(pins, "xyzbc-trt-kins.z-rot-point"), "conventionalDirections": hal_pin_value(pins, "xyzbc-trt-kins.conventional-directions"), "toolOffset": hal_pin_value(pins, "xyzbc-trt-kins.tool-offset"), } def hal_pin_value(pins, name): return (pins.get(name) or {}).get("value") def source_root_for_ini(ini_path): marker = "/configs/sim/axis/vismach/5axis/table-rotary-tilting/xyzbc-trt.ini" value = str(ini_path) if value.endswith(marker): return value[: -len(marker)] return DEFAULT_SOURCE_ROOT def value(obj, attr): return normalize_json(getattr(obj, attr, None)) def axes_tuple(values): values = tuple_to_list(values) return {axis.lower(): values[index] for index, axis in enumerate(AXES) if index < len(values)} def joint_tuple(values): values = tuple_to_list(values) return {str(index): values[index] for index in range(min(5, len(values)))} def tuple_to_list(values): if values is None: return [] return [normalize_json(value) for value in values] def normalize_json(value): if isinstance(value, float): if math.isnan(value) or math.isinf(value): return None return value if isinstance(value, (str, int, bool)) or value is None: return value if isinstance(value, tuple): return [normalize_json(item) for item in value] if isinstance(value, list): return [normalize_json(item) for item in value] if isinstance(value, dict): return {str(key): normalize_json(item) for key, item in value.items()} return str(value) def parse_number(value): try: return float(value) except ValueError: return value def number_or_zero(value): try: number = float(value) except (TypeError, ValueError): return 0 if math.isnan(number) or math.isinf(number): return 0 return number def write_json(path, payload): output = pathlib.Path(path) output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def iso_now(): return time.strftime("%Y-%m-%dT%H:%M:%S%z") if __name__ == "__main__": sys.exit(main())