#!/usr/bin/env python3 import argparse import json import math import pathlib import subprocess import sys import time AXES = ["X", "Y", "Z", "A", "B", "C", "U", "V", "W"] SAMPLE_PERIOD_MS = 20 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" 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) task_state_flow = build_task_state_flow(before, after, command_result, hal) basic_sim = build_basic_sim_equivalent(before, after, command_result, hal) 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, "taskStateFlow": task_state_flow, "buttonInterlocks": build_button_interlocks(after, task_state_flow), "basicSimEquivalent": basic_sim, "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, "taskStateFlowReadable": task_state_flow.get("ready") is True, "basicSimReadable": basic_sim.get("ready") is 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"), } 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, } return { "sampleIndex": sample_index, "timeMs": time_ms, "line": int(number_or_zero(event.get("currentLine"))), "motionType": "unknown", "activeKinematics": "unknown", "tool": { **XYZBC_DEFAULT_TOOL, }, "joint": joint, "tcp": { "x": joint["x"], "y": joint["y"], "z": joint["z"], }, "toolAxis": tool_axis_from_bc(b, c), "feed": number_or_zero(event.get("feedrate")), "spindle": spindle_speed(event.get("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_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), }) 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, }) 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") add_linear({"x": center_x, "y": center_y, "z": zmax}, center_line, "rapid", "identity") add_linear({"x": center_x - radius}, 13, "rapid", "identity") add_linear({"b": b_axis, "c": c_axis}, 16, "rapid", "tcp-xyzbc") add_helix(17) add_linear({"x": 0, "y": 0, "z": zmax, "b": 0, "c": 0}, 19, "rapid", "identity") add_linear({"x": radius}, 20, "rapid", "identity") add_linear({"x": 0, "y": 0, "z": zmax, "b": 0, "c": 0}, 44, "rapid", "identity") return segments 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"]} return { "sampleIndex": sample_index, "timeMs": time_ms, "line": int(line), "motionType": motion_type, "activeKinematics": active_kinematics, "tool": {**XYZBC_DEFAULT_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, } 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 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())