1546 lines
60 KiB
Python
Executable File
1546 lines
60 KiB
Python
Executable File
#!/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 = 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"
|
|
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)
|
|
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,
|
|
"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,
|
|
},
|
|
"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<xyzbc_switchkins_sub> call [10] [5] [10][1000][3][0][20][45][20]
|
|
marker = "o<xyzbc_switchkins_sub> 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#<frate> g2i#<r>z#<zmin> p#<n>",
|
|
})
|
|
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#<zmax> 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#<zmax>"
|
|
add_linear({"x": center_x - radius}, 13, "rapid", "identity")
|
|
segments[-1]["sourceFile"] = "helix_bc.ngc"
|
|
segments[-1]["statement"] = "g0 x[#<_x> - #<r>]"
|
|
add_linear({"b": b_axis, "c": c_axis}, 16, "rapid", "tcp-xyzbc")
|
|
segments[-1]["sourceFile"] = "helix_bc.ngc"
|
|
segments[-1]["statement"] = "g0b#<b>c#<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#<zmax> b0 c0"
|
|
add_linear({"x": radius}, 20, "rapid", "identity")
|
|
segments[-1]["sourceFile"] = "helix_bc.ngc"
|
|
segments[-1]["statement"] = "g0 x[#<_x> + #<r>]"
|
|
|
|
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#<zmax>"
|
|
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<xyzbc_switchkins_sub> 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#<zmax> b0 c0 ; quadrant {quadrant}", "rapid-machine-reset")
|
|
add("xyzbc_switchkins_sub.ngc", reset_line + 2, "g10l20p0 x0y0 z#<zmax> b0 c0", "set-g54-offset", active_before=current_kinematics, active_after=current_kinematics)
|
|
add_segment("xyzbc_switchkins_sub.ngc", center_line, "g0 x±#<dist> y±#<dist> z#<zmax>", "rapid-to-quadrant-center")
|
|
add("xyzbc_switchkins_sub.ngc", center_line + 1, "o<helix_bc> call [#<zmax>][#<zmin>][#<r>][#<frate>][#<n>][#<a>][#<b>][#<c>]", "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> - #<r>]", "rapid-radius-adjust")
|
|
add("helix_bc.ngc", 14, "g10l20p0 x0y0 z#<zmax> 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#<zmax> b0 c0", "rapid-return-to-start")
|
|
add_segment("helix_bc.ngc", 20, "g0 x[#<_x> + #<r>]", "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#<zmax>", "rapid-final-machine-reset")
|
|
add("xyzbc_switchkins_sub.ngc", 45, "g10l20p0 x0y0 z#<zmax>", "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<xyzbc_switchkins_sub>"}]
|
|
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<helix_bc>"}],
|
|
)
|
|
|
|
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<xyzbc_switchkins_sub> 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: "o<xyzbc_switchkins_sub>sub",
|
|
4: "#<zmax> = #1 (=10)",
|
|
5: "#<zmin> = #2 (= 5)",
|
|
6: "#<r> = #3 (=10 radius)",
|
|
7: "#<frate> = #4 (=1000 feedrate)",
|
|
8: "#<n> = #5 (=3 n circles)",
|
|
9: "#<a> = #6 (=0 A angle NA)",
|
|
10: "#<b> = #7 (=30 B angle)",
|
|
11: "#<c> = #8 (=45 C angle)",
|
|
12: "#<dist> = #9 (=20 distance)",
|
|
14: "; quadrant I",
|
|
15: "M429 ;Identity kinematics",
|
|
16: "g53 g0 x0y0 z#<zmax> b0 c0 ;MACHINE coordinates",
|
|
17: "g10l20p0 x0y0 z#<zmax> b0 c0 ;new g54",
|
|
18: "g0 x+#<dist> y+#<dist> z#<zmax> ;move to pattern center position",
|
|
19: "o<helix_bc> call [#<zmax>][#<zmin>][#<r>][#<frate>][#<n>][#<a>][#<b>][#<c>]",
|
|
21: "; quadrant II",
|
|
22: "M429 ;Identity kinematics",
|
|
23: "g53 g0 x0y0 z#<zmax> b0 c0",
|
|
24: "g10l20p0 x0y0 z#<zmax> b0 c0",
|
|
25: "g0 x-#<dist> y+#<dist> z#<zmax>",
|
|
26: "o<helix_bc> call [#<zmax>][#<zmin>][#<r>][#<frate>][#<n>][#<a>][#<b>][#<c>]",
|
|
28: "; quadrant III",
|
|
29: "M429 ;Identity kinematics",
|
|
30: "g53 g0 x0y0 z#<zmax> b0 c0",
|
|
31: "g10l20p0 x0y0 z#<zmax> b0 c0",
|
|
32: "g0 x-#<dist> y-#<dist> z#<zmax>",
|
|
33: "o<helix_bc> call [#<zmax>][#<zmin>][#<r>][#<frate>][#<n>][#<a>][#<b>][#<c>]",
|
|
35: "; quadrant IV",
|
|
36: "M429 ;Identity kinematics",
|
|
37: "g53 g0 x0y0 z#<zmax> b0 c0",
|
|
38: "g10l20p0 x0y0 z#<zmax> b0 c0",
|
|
39: "g0 x+#<dist> y-#<dist> z#<zmax>",
|
|
40: "o<helix_bc> call [#<zmax>][#<zmin>][#<r>][#<frate>][#<n>][#<a>][#<b>][#<c>]",
|
|
42: ";final position",
|
|
43: "M429 ;Identity kinematics",
|
|
44: "g53 g0 x0y0 z#<zmax> ;MACHINE coordinates",
|
|
45: "g10l20p0 x0y0 z#<zmax> ;new g54",
|
|
47: "o<xyzbc_switchkins_sub>endsub",
|
|
},
|
|
"helix_bc.ngc": {
|
|
1: "; helix using switchkins (xyzbc) b,c angles",
|
|
2: "o<helix_bc>sub",
|
|
3: "#<zmax> = #1 (=10)",
|
|
4: "#<zmin> = #2 (= 5)",
|
|
5: "#<r> = #3 (=10)",
|
|
6: "#<frate> = #4 (=1000)",
|
|
7: "#<n> = #5 (=3)",
|
|
8: "#<a> = #6 (=0 NA)",
|
|
9: "#<b> = #7 (=45)",
|
|
10: "#<c> = #8 (=20)",
|
|
12: "M429 ;Identity kinematics",
|
|
13: "g0 x[#<_x> - #<r>] ;adjust for radius",
|
|
14: "g10l20p0 x0y0 z#<zmax> b0 c0 ;new g54",
|
|
15: "M428 ;XYZBC",
|
|
16: "g0b#<b>c#<c> ;exercise b,c",
|
|
17: "f#<frate> g2i#<r>z#<zmin> p#<n> ;helix",
|
|
18: "M429 ;Identity kinematics",
|
|
19: "g0 x0 y0 z#<zmax> b0 c0 ;return to start",
|
|
20: "g0 x[#<_x> + #<r>] ;adjust restore",
|
|
21: "M428 ;XYZBC",
|
|
22: "o<helix_bc>endsub",
|
|
},
|
|
}
|
|
|
|
|
|
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())
|