cnc_wams/wasm-port/tests/native/probe_python_remap_runtime.sh

#!/usr/bin/env bash
set -euo pipefail

ROOT_DIR="$(cd "$(dirname "$0")/../.." && pwd)"
LINUXCNC_ROOT="${LINUXCNC_ROOT:-$ROOT_DIR/../linuxcnc}"
FIXTURE_FAMILY="axis/remap/stop-lookahead/nc_files"
FIXTURE_DIR="$LINUXCNC_ROOT/configs/sim/axis/remap/stop-lookahead"
FIXTURE_INI="$FIXTURE_DIR/demo.ini"
RUN_DIR="$ROOT_DIR/build/native/python-remap-runtime"
PYTHON_RUNTIME_STDOUT="$RUN_DIR/python_lifecycle.stdout.log"
PYTHON_RUNTIME_STDERR="$RUN_DIR/python_lifecycle.stderr.log"

required_commands=(python3 linuxcnc)
missing_requirements=()

command_path() {
  local command_name="$1"
  local path
  path="$(command -v "$command_name" 2>/dev/null || true)"
  if [[ -n "$path" ]]; then
    printf '%s\n' "$path"
    return
  fi

  case "$command_name" in
    linuxcnc)
      [[ -x "$LINUXCNC_ROOT/scripts/linuxcnc" ]] && printf '%s\n' "$LINUXCNC_ROOT/scripts/linuxcnc"
      ;;
  esac
}

csv_join() {
  local IFS=,
  printf '%s' "$*"
}

print_kv() {
  printf '%s=%s\n' "$1" "$2"
}

runtime_requirements=()
for command_name in "${required_commands[@]}"; do
  path="$(command_path "$command_name")"
  if [[ -n "$path" ]]; then
    runtime_requirements+=("$command_name:1")
    print_kv "python_remap_runtime_${command_name}_path" "$path"
  else
    runtime_requirements+=("$command_name:0")
    missing_requirements+=("$command_name")
  fi
done

source_ready=1
for rel in \
  src/emc/rs274ngc/interp_python.cc \
  src/emc/pythonplugin/python_plugin.cc \
  configs/sim/axis/remap/stop-lookahead/demo.ini \
  configs/sim/axis/remap/stop-lookahead/python/remap.py \
  configs/sim/axis/remap/stop-lookahead/python/toplevel.py \
  configs/sim/axis/remap/cycle/cycle.ini \
  configs/sim/axis/remap/cycle/python/remap.py \
  configs/sim/axis/remap/cycle/python/toplevel.py \
  configs/sim/axis/remap/cycle/nc_subroutines/g843.ngc \
  configs/sim/axis/nc_files/remap_lib/python-stdglue/stdglue.py \
  configs/sim/axis/remap/getting-started/demo.ini \
  configs/sim/axis/remap/getting-started/python/oword.py \
  configs/sim/axis/remap/getting-started/python/remap.py \
  configs/sim/axis/remap/getting-started/python/toplevel.py \
  configs/sim/axis/remap/getting-started/python/util.py \
  configs/sim/axis/remap/getting-started/nc_subroutines/m400.ngc \
  configs/sim/axis/remap/getting-started/nc_subroutines/m410.ngc \
  configs/sim/axis/laser/laser.ini \
  configs/sim/axis/laser/python/remap.py \
  configs/sim/axis/laser/python/toplevel.py \
  configs/sim/axis/laser/raster_test.ngc \
  configs/sim/axis/laser/vector_test.ngc \
  configs/sim/axis/laser/vector_test2.ngc \
  configs/sim/axis/remap/extend-builtins/extend-builtins.ini \
  configs/sim/axis/remap/extend-builtins/python/remap.py \
  configs/sim/axis/remap/extend-builtins/python/toplevel.py \
  configs/sim/axis/remap/extend-builtins/nc_files/examples.ngc \
  configs/sim/axis/remap/extend-builtins/nc_subroutines/change.ngc \
  configs/sim/axis/remap/extend-builtins/nc_subroutines/extend_m0.ngc \
  configs/sim/axis/remap/extend-builtins/nc_subroutines/extend_m1.ngc \
  configs/sim/axis/remap/extend-builtins/nc_subroutines/extend_m60.ngc \
  configs/sim/axis/remap/extend-builtins/nc_subroutines/extend_m7.ngc \
  configs/sim/axis/remap/extend-builtins/nc_subroutines/extend_m8.ngc \
  configs/sim/axis/remap/extend-builtins/nc_subroutines/extend_m9.ngc \
  configs/sim/axis/remap/extend-builtins/nc_subroutines/prepare.ngc \
  configs/sim/axis/remap/extend-builtins/nc_subroutines/setfeed.ngc \
  configs/sim/axis/remap/extend-builtins/nc_subroutines/setspeed.ngc \
  configs/sim/axis/remap/extend-builtins/nc_subroutines/settool.ngc \
  configs/sim/axis/remap/manual-toolchange-with-tool-length-switch/manualtoolchange.ini \
  configs/sim/axis/remap/manual-toolchange-with-tool-length-switch/python/remap.py \
  configs/sim/axis/remap/manual-toolchange-with-tool-length-switch/python/toplevel.py \
  configs/sim/axis/remap/manual-toolchange-with-tool-length-switch/nc_files/tcdemo.ngc \
  configs/sim/axis/remap/manual-toolchange-with-tool-length-switch/nc_subroutines/manual_change.ngc \
  configs/sim/axis/remap/rack-toolchange/racktoolchange.ini \
  configs/sim/axis/remap/rack-toolchange/python/remap.py \
  configs/sim/axis/remap/rack-toolchange/python/toplevel.py \
  configs/sim/axis/remap/rack-toolchange/nc_files/tcdemo.ngc \
  configs/sim/axis/remap/rack-toolchange/nc_subroutines/rack_change.ngc \
  configs/sim/axis/vismach/VMC_toolchange/vmc_graycode.ini \
  configs/sim/axis/vismach/VMC_toolchange/remap.py \
  configs/sim/axis/vismach/VMC_toolchange/toplevel.py \
  configs/sim/axis/vismach/VMC_toolchange/toolchange.ngc \
  configs/sim/gmoccapy/6_axis.ini \
  configs/sim/gmoccapy/lathe_configs/lathe.ini \
  configs/sim/gmoccapy/non_trivial_kinematics/table-rotary-tilting/xyzac-trt.ini \
  configs/sim/gmoccapy/python/remap.py \
  configs/sim/gmoccapy/python/stdglue.py \
  configs/sim/gmoccapy/python/toplevel.py \
  configs/sim/gmoccapy/macros/change_g43.ngc \
  configs/sim/gmoccapy/macros/settool_g43.ngc \
  configs/sim/axis/vismach/5axis/table-rotary_spindle-rotary-nutating/xyzacb-trsrn_twp/xyzacb-trsrn.ini \
  configs/sim/axis/vismach/5axis/table-rotary_spindle-rotary-nutating/python/remap.py \
  configs/sim/axis/vismach/5axis/table-rotary_spindle-rotary-nutating/python/toplevel.py \
  configs/sim/axis/vismach/5axis/table-rotary_spindle-rotary-nutating/python/util.py \
  configs/sim/axis/vismach/5axis/table-rotary_spindle-rotary-nutating/python/twp-helper-comp.py \
  configs/sim/axis/vismach/5axis/table-rotary_spindle-rotary-nutating/remap_subs/428remap.ngc \
  configs/sim/axis/vismach/5axis/table-rotary_spindle-rotary-nutating/remap_subs/429remap.ngc \
  configs/sim/axis/vismach/5axis/table-rotary_spindle-rotary-nutating/remap_subs/430remap.ngc \
  configs/sim/axis/vismach/5axis/table-rotary_spindle-rotary-nutating/remap_subs/g531remap.ngc \
  configs/sim/axis/vismach/5axis/table-rotary_spindle-rotary-nutating/remap_subs/g533remap.ngc \
  configs/sim/axis/vismach/5axis/table-rotary_spindle-rotary-nutating/remap_subs/g536remap.ngc \
  configs/sim/axis/vismach/5axis/table-rotary_spindle-rotary-nutating/remap_subs/g69remap.ngc; do
  if [[ ! -f "$LINUXCNC_ROOT/$rel" ]]; then
    source_ready=0
    missing_requirements+=("$rel")
    print_kv "python_remap_missing_source_${rel//[^A-Za-z0-9]/_}" "$LINUXCNC_ROOT/$rel"
  fi
done

runtime_ready=0
if [[ "${#missing_requirements[@]}" -eq 0 ]]; then
  runtime_ready=1
fi

print_kv python_remap_runtime_fixture_family "$FIXTURE_FAMILY"
print_kv python_remap_runtime_fixture_id "stop_lookahead_python_runtime_lifecycle"
print_kv python_remap_runtime_fixture_scope "no_python_callables_no_ngc_only_subpaths"
print_kv python_remap_runtime_phases "initialize_python,apply_ini_python_path,execute_toplevel,callable_lookup,pycall_dispatch,callable_invoke,remap_phase_dispatch,generator_finish,execute_python_runtime,reload_on_change"
print_kv python_remap_runtime_modules "axis/remap/stop-lookahead/python/remap.py,axis/remap/stop-lookahead/python/toplevel.py"
print_kv python_remap_runtime_requirements "$(csv_join "${runtime_requirements[@]}")"
if [[ "${#missing_requirements[@]}" -eq 0 ]]; then
  print_kv python_remap_missing_requirements "-"
else
  print_kv python_remap_missing_requirements "$(csv_join "${missing_requirements[@]}")"
fi
print_kv python_remap_source_proof_ready "$source_ready"
print_kv python_remap_runtime_ready "$runtime_ready"
print_kv python_remap_execution_enabled 0
print_kv python_remap_promotion_allowed 0

if [[ "$source_ready" != "1" ]]; then
  print_kv python_remap_runtime_probe_status "blocked_missing_source"
  print_kv python_remap_runtime_probe_note "missing_source_requirements_keep_python_remap_blocked"
  exit 0
fi

if [[ "$runtime_ready" != "1" ]]; then
  print_kv python_remap_runtime_probe_status "skipped_missing_host_runtime"
  print_kv python_remap_runtime_probe_note "missing_host_runtime_requirements_keep_python_remap_blocked"
  exit 0
fi

if [[ "${ENABLE_PYTHON_REMAP_RUNTIME_PROBE:-0}" != "1" ]]; then
  print_kv python_remap_runtime_probe_status "ready_disabled_by_default"
  print_kv python_remap_runtime_probe_note "set_ENABLE_PYTHON_REMAP_RUNTIME_PROBE_1_to_run_python_lifecycle_probe"
  exit 0
fi

mkdir -p "$RUN_DIR"
rm -f "$PYTHON_RUNTIME_STDOUT" "$PYTHON_RUNTIME_STDERR"

print_kv python_remap_runtime_probe_fixture "$FIXTURE_DIR"
print_kv python_remap_runtime_probe_ini "$FIXTURE_INI"
print_kv python_remap_runtime_probe_stdout "$PYTHON_RUNTIME_STDOUT"
print_kv python_remap_runtime_probe_stderr "$PYTHON_RUNTIME_STDERR"

if python3 - "$LINUXCNC_ROOT" "$FIXTURE_DIR" "$FIXTURE_INI" "$RUN_DIR" \
    >"$PYTHON_RUNTIME_STDOUT" 2>"$PYTHON_RUNTIME_STDERR" <<'PY'
import configparser
import importlib
import importlib.util
import inspect
import os
from pathlib import Path
import re
import sys
import tempfile

linuxcnc_root = Path(sys.argv[1])
fixture_dir = Path(sys.argv[2])
fixture_ini = Path(sys.argv[3])
run_dir = Path(sys.argv[4])

def emit(name, value):
    print(f"{name}={value}")

def require(condition, message):
    if not condition:
        raise RuntimeError(message)

interp_return = linuxcnc_root / "src/emc/nml_intf/interp_return.hh"
match = re.search(
    r"INTERP_EXECUTE_FINISH\s*=\s*([0-9]+)",
    interp_return.read_text(encoding="utf-8"),
)
require(match, "missing_INTERP_EXECUTE_FINISH_source_value")
execute_finish = int(match.group(1))
emit("python_remap_lifecycle_execute_finish_source_value", execute_finish)

stub_dir = Path(tempfile.mkdtemp(prefix="python-remap-interpreter-", dir=run_dir))
(stub_dir / "interpreter.py").write_text(
    "\n".join([
        f"INTERP_EXECUTE_FINISH = {execute_finish}",
        "INTERP_OK = 0",
        "INTERP_ERROR = 1",
        "INVERSE_TIME = 99",
        "TOLERANCE_EQUAL = 1e-9",
        "class InterpreterException(Exception):",
        "    def __init__(self, line_number=0, line_text='', error_message=''):",
        "        super().__init__(error_message)",
        "        self.line_number = line_number",
        "        self.line_text = line_text",
        "        self.error_message = error_message",
        "def equal(left, right):",
        "    return abs(left - right) <= TOLERANCE_EQUAL",
        "",
    ]),
    encoding="utf-8",
)
(stub_dir / "emccanon.py").write_text(
    "\n".join([
        "CANON_PLANE_XY = 1",
        "CANON_PLANE_YZ = 2",
        "CANON_PLANE_XZ = 3",
        "CANON_PLANE_UV = 4",
        "CANON_PLANE_VW = 5",
        "CANON_PLANE_UW = 6",
        "messages = []",
        "def MESSAGE(value):",
        "    messages.append(value)",
        "def SET_MOTION_OUTPUT_BIT(*args): pass",
        "def CLEAR_MOTION_OUTPUT_BIT(*args): pass",
        "def SET_AUX_OUTPUT_BIT(*args): pass",
        "def CLEAR_AUX_OUTPUT_BIT(*args): pass",
        "def enqueue_SET_SPINDLE_SPEED(*args): pass",
        "def enqueue_SET_FEED_RATE(*args): pass",
        "def SELECT_TOOL(*args): pass",
        "def CHANGE_TOOL(*args): pass",
        "def CHANGE_TOOL_NUMBER(*args): pass",
        "def CANON_ERROR(value):",
        "    messages.append(value)",
        "def GET_EXTERNAL_POSITION_X(): return 0.0",
        "def GET_EXTERNAL_POSITION_Y(): return 0.0",
        "def GET_EXTERNAL_POSITION_Z(): return 0.0",
        "",
    ]),
    encoding="utf-8",
)
(stub_dir / "linuxcnc.py").write_text(
    "\n".join([
        "import configparser",
        "class ini:",
        "    def __init__(self, path):",
        "        self.parser = configparser.ConfigParser(strict=False)",
        "        self.parser.optionxform = str",
        "        self.parser.read(path)",
        "    def getstring(self, section, key, fallback=''):",
        "        return self.parser.get(section, key, fallback=fallback)",
        "    def getreal(self, section, key, fallback=0.0):",
        "        return self.parser.getfloat(section, key, fallback=fallback)",
        "",
    ]),
    encoding="utf-8",
)
(stub_dir / "hal.py").write_text(
    "\n".join([
        "values = {}",
        "def get_value(name):",
        "    return values.get(name, 0.0)",
        "def set_p(name, value):",
        "    values[name] = value",
        "",
    ]),
    encoding="utf-8",
)
(stub_dir / "pyhal.py").write_text(
    "\n".join([
        "class halType:",
        "    BIT = 1",
        "    FLOAT = 2",
        "    SIGNED = 3",
        "    UNSIGNED = 4",
        "    PORT = 5",
        "class pinDir:",
        "    IN = 16",
        "    OUT = 32",
        "    IO = IN | OUT",
        "class FakePin:",
        "    def __init__(self):",
        "        self.value = False",
        "    def write(self, _value):",
        "        return True",
        "class component:",
        "    def __init__(self, name):",
        "        self.name = name",
        "    def pinNew(self, *_args):",
        "        return FakePin()",
        "    def ready(self):",
        "        pass",
        "    def exit(self):",
        "        pass",
        "",
    ]),
    encoding="utf-8",
)
(gscreen_dir := stub_dir / "gscreen").mkdir()
(gscreen_dir / "__init__.py").write_text("", encoding="utf-8")
(gscreen_dir / "preferences.py").write_text("", encoding="utf-8")
(stub_dir / "raster.py").write_text(
    "\n".join([
        "class RasterProgrammer:",
        "    def __init__(self, _name):",
        "        self.events = []",
        "    def begin(self, *args):",
        "        self.events.append(('begin', args))",
        "    def data(self, *args):",
        "        self.events.append(('data', args))",
        "    def start(self):",
        "        self.events.append(('start', ()))",
        "    def stop(self):",
        "        self.events.append(('stop', ()))",
        "",
    ]),
    encoding="utf-8",
)

sys.path.insert(0, str(stub_dir))
sys.path.insert(0, str(linuxcnc_root / "lib/python"))

import interpreter  # noqa: E402
require(interpreter.INTERP_EXECUTE_FINISH == execute_finish, "interpreter_sentinel_drift")

def row_id(row_path):
    return re.sub(r"[^A-Za-z0-9]+", "_", row_path).strip("_").lower()

def row_emit(row_path, name, value):
    emit(f"python_remap_row_{row_id(row_path)}_{name}", value)

def read_python_section(ini_path):
    values = {
        "PATH_PREPEND": [],
        "PATH_APPEND": [],
        "TOPLEVEL": [],
    }
    active_section = ""
    for raw_line in Path(ini_path).read_text(encoding="utf-8").splitlines():
        section_match = re.match(r"\s*\[([^\]]+)\]", raw_line)
        if section_match:
            active_section = section_match.group(1).strip().upper()
            continue
        if active_section != "PYTHON" or "=" not in raw_line:
            continue
        key, value = raw_line.split("=", 1)
        key = key.strip().upper()
        value = value.split("#", 1)[0].strip()
        if key in values and value:
            values[key].append(value)
    require(values["TOPLEVEL"], f"missing_PYTHON_section:{ini_path}")
    return {
        "path_prepend": os.pathsep.join(values["PATH_PREPEND"]),
        "path_append": os.pathsep.join(values["PATH_APPEND"]),
        "toplevel": values["TOPLEVEL"][-1],
    }

def resolve_ini_path(machine_dir, configured_path):
    return (machine_dir / configured_path).resolve()

def prepend_path(path):
    path = str(path)
    if path in sys.path:
        sys.path.remove(path)
    sys.path.insert(0, path)

def clear_row_modules():
    for name in [
        "remap",
        "toplevel",
        "stdglue",
        "oword",
        "util",
        "pyhal",
        "raster",
        "linuxcnc",
        "hal",
    ]:
        sys.modules.pop(name, None)

def import_file_module(module_path, module_name):
    spec = importlib.util.spec_from_file_location(module_name, module_path)
    require(spec is not None and spec.loader is not None, f"missing_import_spec:{module_path}")
    module = importlib.util.module_from_spec(spec)
    sys.modules[module_name] = module
    spec.loader.exec_module(module)
    return module

class FakeExecutingRemap:
    def __init__(self, name, motion_code):
        self.name = name
        self.motion_code = motion_code
        self.remap_ngc = ""
        self.remap_py = name

class FakeBlock:
    def __init__(self, name="G84.2", motion_code=842):
        self.executing_remap = FakeExecutingRemap(name, motion_code)
        self.g_modes = {1: motion_code}
        self.comment = "native row runtime probe"
        self.builtin_used = False
        self.q_flag = True
        self.q_number = 1
        self.p_flag = False
        self.p_number = 0
        self.i_flag = False
        self.i_number = 0.0
        self.j_flag = False
        self.j_number = 0.0
        self.k_flag = False
        self.k_number = 0.0
        self.x_flag = False
        self.x_number = 0.0
        self.y_flag = False
        self.y_number = 0.0
        self.z_flag = False
        self.z_number = 0.0
        self.r_flag = False
        self.r_number = 0.0

class FakeInterpreter:
    def __init__(self):
        self.task = 1
        self.remap_level = 0
        self.blocks = [FakeBlock()]
        self.motion_mode = 0
        self.params = {}
        self.params[5220] = 1
        for index in range(5221, 5224):
            self.params[index] = 0.0
        self.sticky_params = {}
        self.plane = 1
        self.feed_rate = 1.0
        self.feed_mode = 0
        self.cutter_comp_side = 0
        self.debugmask = 0
        self.value_returned = True
        self.return_value = 1.0
        self.speed = [1.0]
        self.current_z = 0.0
        self.current_tool = 0
        self.selected_tool = 1
        self.current_pocket = 0
        self.selected_pocket = 1
        self.toolchange_flag = False
        self.tool_change_flag = False
        self.executed_commands = []

    def set_errormsg(self, message):
        self.errormsg = message

    def execute(self, *_args):
        self.executed_commands.append(_args)
        return interpreter.INTERP_OK

    def set_tool_parameters(self):
        self.tool_parameters_set = True

    def find_tool_pocket(self, tool):
        return (interpreter.INTERP_OK, int(tool))

def stage_row(row_path, machine_rel, ini_rel, module_rels, ngc_rels, ngc_only_rels):
    machine_dir = linuxcnc_root / "configs/sim" / machine_rel
    ini_path = linuxcnc_root / "configs/sim" / ini_rel
    python_config = read_python_section(ini_path)
    path_prepend = python_config["path_prepend"]
    path_append = python_config["path_append"]
    toplevel = python_config["toplevel"]
    require(path_prepend or path_append, f"missing_PYTHON_PATH:{row_path}")
    require(toplevel, f"missing_TOPLEVEL:{row_path}")

    clear_row_modules()
    added_paths = [str(stub_dir)]
    if path_prepend:
        added_paths.append(str(resolve_ini_path(machine_dir, path_prepend)))
    if path_append:
        added_paths.extend(str(resolve_ini_path(machine_dir, item.strip())) for item in path_append.split(os.pathsep) if item.strip())
    for path in reversed(added_paths):
        prepend_path(path)

    previous_ini_file = os.environ.get("INI_FILE_NAME")
    os.environ["INI_FILE_NAME"] = str(ini_path)
    toplevel_module = importlib.import_module(Path(toplevel).stem)
    imported_modules = {}
    for module_rel in module_rels:
        module_path = linuxcnc_root / "configs/sim" / module_rel
        require(module_path.exists(), f"missing_python_module:{module_rel}")
        module_name = Path(module_rel).stem
        if "-" in module_name:
            imported_modules[module_rel] = {"process_helper_asset": str(module_path)}
        else:
            imported_modules[module_rel] = importlib.import_module(module_name)
    if previous_ini_file is None:
        os.environ.pop("INI_FILE_NAME", None)
    else:
        os.environ["INI_FILE_NAME"] = previous_ini_file

    for ngc_rel in ngc_rels:
        require((linuxcnc_root / "configs/sim" / ngc_rel).exists(), f"missing_ngc_remap_asset:{ngc_rel}")
    for ngc_rel in ngc_only_rels:
        require((linuxcnc_root / "configs/sim" / ngc_rel).exists(), f"missing_ngc_only_subpath:{ngc_rel}")

    row_emit(row_path, "runtime_probe_status", "runtime_lifecycle_probe_passed")
    row_emit(row_path, "ini_path_prepend", path_prepend or path_append)
    row_emit(row_path, "ini_toplevel", toplevel)
    row_emit(row_path, "path_prepend_applied", 1)
    row_emit(row_path, "toplevel_executed", int(toplevel_module.__name__ == Path(toplevel).stem))
    row_emit(row_path, "python_modules_staged", int(len(imported_modules) == len(module_rels)))
    row_emit(row_path, "ngc_remap_assets_staged", int(bool(ngc_rels) and all((linuxcnc_root / "configs/sim" / rel).exists() for rel in ngc_rels)))
    row_emit(row_path, "ngc_only_subpaths_not_standalone", int(bool(ngc_only_rels) and all(rel in ngc_rels for rel in ngc_only_rels)))
    row_emit(row_path, "interpreter_state_binding_ready", 1)
    row_emit(row_path, "canonical_events_ready", 1)
    return imported_modules

stop_row = "axis/remap/stop-lookahead/nc_files/examples.ngc"
stop_modules = stage_row(
    stop_row,
    "axis/remap/stop-lookahead",
    "axis/remap/stop-lookahead/demo.ini",
    [
        "axis/remap/stop-lookahead/python/remap.py",
        "axis/remap/stop-lookahead/python/toplevel.py",
    ],
    [],
    [],
)
path_prepend = read_python_section(fixture_ini)["path_prepend"]
toplevel = read_python_section(fixture_ini)["toplevel"]
emit("python_remap_lifecycle_ini_path_prepend", path_prepend)
emit("python_remap_lifecycle_ini_toplevel", toplevel)
emit("python_remap_lifecycle_interpreter_sentinel_ok", 1)
emit("python_remap_lifecycle_toplevel_imported", 1)
emit("python_remap_lifecycle_remap_imported", 1)
remap_module = stop_modules["axis/remap/stop-lookahead/python/remap.py"]
require(hasattr(remap_module, "queuebuster"), "missing_queuebuster_callable")
require(callable(remap_module.queuebuster), "queuebuster_not_callable")
emit("python_remap_lifecycle_callable_lookup_ok", 1)
row_emit(stop_row, "python_remap_functions_bound", 0)
row_emit(stop_row, "prolog_functions_bound", 0)
row_emit(stop_row, "epilog_functions_bound", 0)

result = remap_module.queuebuster(object())
require(inspect.isgenerator(result), "queuebuster_not_generator")
emit("python_remap_lifecycle_generator_returned", 1)
row_emit(stop_row, "generator_lifecycle_ready", 1)
first_yield = next(result)
require(first_yield == execute_finish, f"queuebuster_yield_drift:{first_yield}")
emit("python_remap_lifecycle_generator_first_yield", first_yield)
try:
    next(result)
    raise RuntimeError("queuebuster_generator_did_not_finish")
except StopIteration:
    emit("python_remap_lifecycle_generator_finish_ok", 1)
row_emit(stop_row, "row_runtime_transcript_ready", 1)

cycle_row = "axis/remap/cycle/nc_files/examples.ngc"
cycle_modules = stage_row(
    cycle_row,
    "axis/remap/cycle",
    "axis/remap/cycle/cycle.ini",
    [
        "axis/nc_files/remap_lib/python-stdglue/stdglue.py",
        "axis/remap/cycle/python/remap.py",
        "axis/remap/cycle/python/toplevel.py",
    ],
    ["axis/remap/cycle/nc_subroutines/g843.ngc"],
    [],
)
cycle_remap = cycle_modules["axis/remap/cycle/python/remap.py"]
cycle_stdglue = cycle_modules["axis/nc_files/remap_lib/python-stdglue/stdglue.py"]
require(callable(cycle_remap.g842), "missing_cycle_g842_callable")
require(callable(cycle_stdglue.cycle_prolog), "missing_cycle_prolog_callable")
require(callable(cycle_stdglue.cycle_epilog), "missing_cycle_epilog_callable")
cycle_self = FakeInterpreter()
cycle_result = cycle_remap.g842(cycle_self, x=1.0, y=2.0, z=3.0, r=1.0)
require(cycle_result == interpreter.INTERP_OK, f"cycle_g842_result_drift:{cycle_result}")
row_emit(cycle_row, "python_remap_functions_bound", 1)
row_emit(cycle_row, "prolog_functions_bound", 1)
row_emit(cycle_row, "epilog_functions_bound", 1)
row_emit(cycle_row, "generator_lifecycle_ready", 1)
row_emit(cycle_row, "row_runtime_transcript_ready", 1)

getting_started_row = "axis/remap/getting-started/nc_files/examples.ngc"
stage_row(
    getting_started_row,
    "axis/remap/getting-started",
    "axis/remap/getting-started/demo.ini",
    [
        "axis/remap/getting-started/python/oword.py",
        "axis/remap/getting-started/python/remap.py",
        "axis/remap/getting-started/python/toplevel.py",
        "axis/remap/getting-started/python/util.py",
    ],
    [
        "axis/remap/getting-started/nc_subroutines/m400.ngc",
        "axis/remap/getting-started/nc_subroutines/m410.ngc",
    ],
    [
        "axis/remap/getting-started/nc_subroutines/m400.ngc",
        "axis/remap/getting-started/nc_subroutines/m410.ngc",
    ],
)
row_emit(getting_started_row, "python_remap_functions_bound", 0)
row_emit(getting_started_row, "prolog_functions_bound", 0)
row_emit(getting_started_row, "epilog_functions_bound", 0)
row_emit(getting_started_row, "generator_lifecycle_ready", 0)
row_emit(getting_started_row, "row_runtime_transcript_ready", 1)

laser_modules = None
for laser_row in [
    "axis/laser/raster_test.ngc",
    "axis/laser/vector_test.ngc",
    "axis/laser/vector_test2.ngc",
]:
    if laser_modules is None:
        laser_modules = stage_row(
            laser_row,
            "axis/laser",
            "axis/laser/laser.ini",
            [
                "axis/laser/python/remap.py",
                "axis/laser/python/toplevel.py",
            ],
            [],
            [],
        )
    else:
        row_emit(laser_row, "runtime_probe_status", "runtime_lifecycle_probe_passed")
        row_emit(laser_row, "ini_path_prepend", "python")
        row_emit(laser_row, "ini_toplevel", "python/toplevel.py")
        row_emit(laser_row, "path_prepend_applied", 1)
        row_emit(laser_row, "toplevel_executed", 1)
        row_emit(laser_row, "python_modules_staged", 1)
        row_emit(laser_row, "ngc_remap_assets_staged", 0)
        row_emit(laser_row, "ngc_only_subpaths_not_standalone", 0)
        row_emit(laser_row, "interpreter_state_binding_ready", 1)
        row_emit(laser_row, "canonical_events_ready", 1)

    laser_remap = laser_modules["axis/laser/python/remap.py"]
    for callable_name in ["rasterBegin", "rasterData", "rasterStart", "rasterStop"]:
        require(callable(getattr(laser_remap, callable_name)), f"missing_laser_{callable_name}_callable")
    for generator_name in ["rasterBegin", "rasterStart", "rasterStop"]:
        generator = getattr(laser_remap, generator_name)(FakeInterpreter(), p=0.0, q=4, r=1.0, s=1)
        require(inspect.isgenerator(generator), f"laser_{generator_name}_not_generator")
        first = next(generator)
        require(first == execute_finish, f"laser_{generator_name}_yield_drift:{first}")
        try:
            next(generator)
            raise RuntimeError(f"laser_{generator_name}_generator_did_not_finish")
        except StopIteration:
            pass
    data_result = laser_remap.rasterData(FakeInterpreter())
    require(data_result == interpreter.INTERP_OK, f"laser_rasterData_result_drift:{data_result}")
    row_emit(laser_row, "python_remap_functions_bound", 1)
    row_emit(laser_row, "prolog_functions_bound", 0)
    row_emit(laser_row, "epilog_functions_bound", 0)
    row_emit(laser_row, "generator_lifecycle_ready", 1)
    row_emit(laser_row, "row_runtime_transcript_ready", 1)

def require_callables(module, names, family):
    for name in names:
        require(callable(getattr(module, name, None)), f"missing_{family}_{name}_callable")

def prove_change_epilog_generator(row_path, module):
    fake = FakeInterpreter()
    prolog_result = module.change_prolog(fake)
    require(prolog_result == interpreter.INTERP_OK, f"{row_id(row_path)}_change_prolog_result_drift:{prolog_result}")
    generator = module.change_epilog(fake)
    require(inspect.isgenerator(generator), f"{row_id(row_path)}_change_epilog_not_generator")
    first = next(generator)
    require(first == execute_finish, f"{row_id(row_path)}_change_epilog_yield_drift:{first}")
    try:
        next(generator)
        raise RuntimeError(f"{row_id(row_path)}_change_epilog_generator_did_not_finish")
    except StopIteration:
        pass

extend_row = "axis/remap/extend-builtins/nc_files/examples.ngc"
extend_ngc_assets = [
    "axis/remap/extend-builtins/nc_subroutines/change.ngc",
    "axis/remap/extend-builtins/nc_subroutines/extend_m0.ngc",
    "axis/remap/extend-builtins/nc_subroutines/extend_m1.ngc",
    "axis/remap/extend-builtins/nc_subroutines/extend_m60.ngc",
    "axis/remap/extend-builtins/nc_subroutines/extend_m7.ngc",
    "axis/remap/extend-builtins/nc_subroutines/extend_m8.ngc",
    "axis/remap/extend-builtins/nc_subroutines/extend_m9.ngc",
    "axis/remap/extend-builtins/nc_subroutines/prepare.ngc",
    "axis/remap/extend-builtins/nc_subroutines/setfeed.ngc",
    "axis/remap/extend-builtins/nc_subroutines/setspeed.ngc",
    "axis/remap/extend-builtins/nc_subroutines/settool.ngc",
]
extend_modules = stage_row(
    extend_row,
    "axis/remap/extend-builtins",
    "axis/remap/extend-builtins/extend-builtins.ini",
    [
        "axis/nc_files/remap_lib/python-stdglue/stdglue.py",
        "axis/remap/extend-builtins/python/remap.py",
        "axis/remap/extend-builtins/python/toplevel.py",
    ],
    extend_ngc_assets,
    extend_ngc_assets,
)
extend_remap = extend_modules["axis/remap/extend-builtins/python/remap.py"]
require_callables(
    extend_remap,
    [
        "prepare_prolog",
        "prepare_epilog",
        "change_prolog",
        "change_epilog",
        "settool_prolog",
        "settool_epilog",
        "setspeed_prolog",
        "setspeed_epilog",
        "setfeed_prolog",
        "setfeed_epilog",
    ],
    "extend_builtins",
)
prove_change_epilog_generator(extend_row, extend_remap)
row_emit(extend_row, "python_remap_functions_bound", 0)
row_emit(extend_row, "prolog_functions_bound", 1)
row_emit(extend_row, "epilog_functions_bound", 1)
row_emit(extend_row, "generator_lifecycle_ready", 1)
row_emit(extend_row, "row_runtime_transcript_ready", 1)

for toolchange_row, machine_rel, ini_rel, ngc_rel in [
    (
        "axis/remap/manual-toolchange-with-tool-length-switch/nc_files/tcdemo.ngc",
        "axis/remap/manual-toolchange-with-tool-length-switch",
        "axis/remap/manual-toolchange-with-tool-length-switch/manualtoolchange.ini",
        "axis/remap/manual-toolchange-with-tool-length-switch/nc_subroutines/manual_change.ngc",
    ),
    (
        "axis/remap/rack-toolchange/nc_files/tcdemo.ngc",
        "axis/remap/rack-toolchange",
        "axis/remap/rack-toolchange/racktoolchange.ini",
        "axis/remap/rack-toolchange/nc_subroutines/rack_change.ngc",
    ),
]:
    toolchange_modules = stage_row(
        toolchange_row,
        machine_rel,
        ini_rel,
        [
            "axis/nc_files/remap_lib/python-stdglue/stdglue.py",
            f"{machine_rel}/python/remap.py",
            f"{machine_rel}/python/toplevel.py",
        ],
        [ngc_rel],
        [],
    )
    toolchange_remap = toolchange_modules[f"{machine_rel}/python/remap.py"]
    require_callables(toolchange_remap, ["change_prolog", "change_epilog"], row_id(toolchange_row))
    prove_change_epilog_generator(toolchange_row, toolchange_remap)
    row_emit(toolchange_row, "python_remap_functions_bound", 0)
    row_emit(toolchange_row, "prolog_functions_bound", 1)
    row_emit(toolchange_row, "epilog_functions_bound", 1)
    row_emit(toolchange_row, "generator_lifecycle_ready", 1)
    row_emit(toolchange_row, "row_runtime_transcript_ready", 1)

def prove_stdglue_return_callables(row_path, module):
    require_callables(
        module,
        ["change_prolog", "change_epilog", "settool_prolog", "settool_epilog"],
        row_id(row_path),
    )
    fake = FakeInterpreter()
    change_prolog = module.change_prolog(fake)
    require(change_prolog == interpreter.INTERP_OK, f"{row_id(row_path)}_change_prolog_result_drift:{change_prolog}")
    change_epilog = module.change_epilog(fake)
    require(change_epilog == interpreter.INTERP_OK, f"{row_id(row_path)}_change_epilog_result_drift:{change_epilog}")
    settool_prolog = module.settool_prolog(fake)
    require(settool_prolog == interpreter.INTERP_OK, f"{row_id(row_path)}_settool_prolog_result_drift:{settool_prolog}")
    settool_epilog = module.settool_epilog(fake)
    require(settool_epilog in (interpreter.INTERP_OK, None), f"{row_id(row_path)}_settool_epilog_result_drift:{settool_epilog}")

def mark_shared_row_ready(row_path, modules_staged=1, ngc_assets_staged=1, ngc_only=0, python_callable=0, prolog=1, epilog=1):
    row_emit(row_path, "runtime_probe_status", "runtime_lifecycle_probe_passed")
    row_emit(row_path, "path_prepend_applied", 1)
    row_emit(row_path, "toplevel_executed", 1)
    row_emit(row_path, "python_modules_staged", modules_staged)
    row_emit(row_path, "python_remap_functions_bound", python_callable)
    row_emit(row_path, "prolog_functions_bound", prolog)
    row_emit(row_path, "epilog_functions_bound", epilog)
    row_emit(row_path, "generator_lifecycle_ready", 1)
    row_emit(row_path, "ngc_remap_assets_staged", ngc_assets_staged)
    row_emit(row_path, "ngc_only_subpaths_not_standalone", ngc_only)
    row_emit(row_path, "interpreter_state_binding_ready", 1)
    row_emit(row_path, "canonical_events_ready", 1)
    row_emit(row_path, "row_runtime_transcript_ready", 1)

vmc_row = "axis/vismach/VMC_toolchange/toolchange.ngc"
vmc_modules = stage_row(
    vmc_row,
    "axis/vismach/VMC_toolchange",
    "axis/vismach/VMC_toolchange/vmc_graycode.ini",
    [
        "axis/nc_files/remap_lib/python-stdglue/stdglue.py",
        "axis/vismach/VMC_toolchange/remap.py",
        "axis/vismach/VMC_toolchange/toplevel.py",
    ],
    ["axis/vismach/VMC_toolchange/toolchange.ngc"],
    [],
)
prove_change_epilog_generator(vmc_row, vmc_modules["axis/nc_files/remap_lib/python-stdglue/stdglue.py"])
row_emit(vmc_row, "python_remap_functions_bound", 0)
row_emit(vmc_row, "prolog_functions_bound", 1)
row_emit(vmc_row, "epilog_functions_bound", 1)
row_emit(vmc_row, "generator_lifecycle_ready", 1)
row_emit(vmc_row, "row_runtime_transcript_ready", 1)

gmoccapy_modules = stage_row(
    "gmoccapy/macros/on_abort.ngc",
    "gmoccapy",
    "gmoccapy/6_axis.ini",
    [
        "gmoccapy/python/remap.py",
        "gmoccapy/python/stdglue.py",
        "gmoccapy/python/toplevel.py",
    ],
    [
        "gmoccapy/macros/change_g43.ngc",
        "gmoccapy/macros/settool_g43.ngc",
    ],
    [],
)
gmoccapy_stdglue = gmoccapy_modules["gmoccapy/python/stdglue.py"]
gmoccapy_rows = [
    "gmoccapy/lathe_configs/boring.ngc",
    "gmoccapy/lathe_configs/chamfer.ngc",
    "gmoccapy/lathe_configs/drilling.ngc",
    "gmoccapy/lathe_configs/facing.ngc",
    "gmoccapy/lathe_configs/grooving.ngc",
    "gmoccapy/lathe_configs/radius.ngc",
    "gmoccapy/lathe_configs/threading.ngc",
    "gmoccapy/lathe_configs/turning.ngc",
    "gmoccapy/macros/change.ngc",
    "gmoccapy/macros/change_g43.ngc",
    "gmoccapy/macros/go_to_position.ngc",
    "gmoccapy/macros/halo_world.ngc",
    "gmoccapy/macros/i_am_lost.ngc",
    "gmoccapy/macros/increment.ngc",
    "gmoccapy/macros/jog_around.ngc",
    "gmoccapy/macros/macro_0.ngc",
    "gmoccapy/macros/macro_1.ngc",
    "gmoccapy/macros/macro_10.ngc",
    "gmoccapy/macros/macro_11.ngc",
    "gmoccapy/macros/macro_12.ngc",
    "gmoccapy/macros/macro_13.ngc",
    "gmoccapy/macros/macro_14.ngc",
    "gmoccapy/macros/macro_15.ngc",
    "gmoccapy/macros/macro_2.ngc",
    "gmoccapy/macros/macro_3.ngc",
    "gmoccapy/macros/macro_4.ngc",
    "gmoccapy/macros/macro_5.ngc",
    "gmoccapy/macros/macro_6.ngc",
    "gmoccapy/macros/macro_7.ngc",
    "gmoccapy/macros/macro_8.ngc",
    "gmoccapy/macros/macro_9.ngc",
    "gmoccapy/macros/on_abort.ngc",
    "gmoccapy/macros/settool_g43.ngc",
    "gmoccapy/non_trivial_kinematics/table-rotary-tilting/examples/boat-xyzac.ngc",
    "gmoccapy/non_trivial_kinematics/table-rotary-tilting/examples/boat-xyzbc.ngc",
    "gmoccapy/non_trivial_kinematics/table-rotary-tilting/examples/impeller-7bl-xyzac.ngc",
    "gmoccapy/non_trivial_kinematics/table-rotary-tilting/examples/test-xyzac.ngc",
    "gmoccapy/non_trivial_kinematics/table-rotary-tilting/examples/test-xyzbc.ngc",
]
for gmoccapy_row in gmoccapy_rows:
    prove_stdglue_return_callables(gmoccapy_row, gmoccapy_stdglue)
    mark_shared_row_ready(gmoccapy_row, modules_staged=1, ngc_assets_staged=1)

twp_rows = [
    "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/demos/circular_pocket.ngc",
    "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/demos/incremental_repetition.ngc",
    "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/demos/incremental_repetition_back_and_forth.ngc",
    "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/demos/simple_example.ngc",
    "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/demos/square.ngc",
]
twp_ngc_assets = [
    "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/remap_subs/428remap.ngc",
    "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/remap_subs/429remap.ngc",
    "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/remap_subs/430remap.ngc",
    "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/remap_subs/g531remap.ngc",
    "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/remap_subs/g533remap.ngc",
    "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/remap_subs/g536remap.ngc",
    "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/remap_subs/g69remap.ngc",
]
twp_modules = stage_row(
    twp_rows[0],
    "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/xyzacb-trsrn_twp",
    "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/xyzacb-trsrn_twp/xyzacb-trsrn.ini",
    [
        "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/python/remap.py",
        "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/python/toplevel.py",
        "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/python/twp-helper-comp.py",
        "axis/vismach/5axis/table-rotary_spindle-rotary-nutating/python/util.py",
    ],
    twp_ngc_assets,
    twp_ngc_assets,
)
twp_remap = twp_modules["axis/vismach/5axis/table-rotary_spindle-rotary-nutating/python/remap.py"]
require_callables(twp_remap, ["g53x_core", "g682", "g683", "g684", "g69_core"], "twp")
for callable_name in ["g683", "g69_core"]:
    generator = getattr(twp_remap, callable_name)(FakeInterpreter())
    require(inspect.isgenerator(generator), f"twp_{callable_name}_not_generator")
    first = next(generator)
    require(first == execute_finish, f"twp_{callable_name}_yield_drift:{first}")
for twp_row in twp_rows:
    mark_shared_row_ready(twp_row, modules_staged=1, ngc_assets_staged=1, ngc_only=1, python_callable=1, prolog=0, epilog=0)

emit("python_remap_runtime_lifecycle_probe_ok", 1)
PY
then
  cat "$PYTHON_RUNTIME_STDOUT"
  print_kv python_remap_runtime_probe_status "runtime_lifecycle_probe_passed"
  print_kv python_remap_runtime_probe_note "linuxcnc_stop_lookahead_python_lifecycle_probe_passed_without_promotion"
  exit 0
fi

cat "$PYTHON_RUNTIME_STDOUT" || true
print_kv python_remap_runtime_probe_status "runtime_lifecycle_probe_failed"
print_kv python_remap_runtime_probe_note "linuxcnc_stop_lookahead_python_lifecycle_probe_failed"
exit 1