1012 lines
46 KiB
Python
1012 lines
46 KiB
Python
"""Macro and node data models."""
|
|
|
|
import uuid
|
|
|
|
|
|
def new_id():
|
|
return str(uuid.uuid4())[:8]
|
|
|
|
|
|
def _ps_quote(s: str) -> str:
|
|
"""Wrap ``s`` as a PowerShell single-quoted literal.
|
|
|
|
Single-quoted strings in PowerShell are literal except the single quote
|
|
itself, which doubles as the escape — so we just replace ``'`` with ``''``.
|
|
Newlines and other control chars are passed through; the user shouldn't
|
|
embed those in a Check's expected/path value, but if they do PowerShell
|
|
handles them inside the quotes.
|
|
"""
|
|
return "'" + (s or "").replace("'", "''") + "'"
|
|
|
|
|
|
SETTLE_MS_DEFAULT = 800 # delay inside the block before any toggle fires
|
|
|
|
|
|
def _render_semi_auto_sequence(sequence: list, settle_ms: int = SETTLE_MS_DEFAULT):
|
|
"""Convert a Semi-Auto step list into a PowerShell script + outcomes.
|
|
|
|
Each ``check`` step in the sequence becomes one outcome (toggle_count =
|
|
1-based step number, value = step['value']). On match, the script
|
|
invokes ``t N`` to fire N Scroll Lock toggles and ``return``s out of
|
|
the enclosing ``& { ... }`` block so the shell stays open and any
|
|
subsequent checks are skipped.
|
|
|
|
Critical timing detail: ``sleep -m {settle_ms}`` runs as the FIRST
|
|
statement inside the ``& { ... }`` block. The block evaluates the
|
|
instant the device finishes typing its closing ``}``, but the device's
|
|
Scroll-Lock listen window doesn't start until ~520 ms later (the
|
|
extra-Enter + pre-listen-ms gap). Without this settle, the toggles fire
|
|
BEFORE the device begins listening — script "instantly finishes" with
|
|
no observed toggle. 800 ms covers the gap with comfortable margin.
|
|
|
|
Compact form: a single ``Add-Type`` with ``uint dwExtraInfo`` (so we
|
|
can pass literal ``0`` instead of ``[System.UIntPtr]::Zero``), a
|
|
one-line ``t($n)`` helper. Output is NOT redirected — stderr stays
|
|
visible, captured stdout is echoed via ``$o`` so the user sees what
|
|
the command returned.
|
|
"""
|
|
body: list[str] = []
|
|
outcomes: list[dict] = []
|
|
check_count = 0
|
|
needs_wsh = False # only emit WScript.Shell if a step actually uses it
|
|
|
|
for step in sequence:
|
|
kind = step.get("kind", "")
|
|
if kind == "wait":
|
|
ms = int(step.get("ms", 500) or 0)
|
|
body.append(f" sleep -m {ms}")
|
|
elif kind == "send_keys":
|
|
keys = step.get("keys", "")
|
|
body.append(f" $wsh.SendKeys({_ps_quote(keys)})")
|
|
needs_wsh = True
|
|
elif kind == "run":
|
|
cmd = (step.get("command", "") or "").rstrip()
|
|
if cmd:
|
|
body.append(" & {")
|
|
for ln in cmd.splitlines():
|
|
body.append(" " + ln)
|
|
body.append(" }")
|
|
elif kind == "check":
|
|
check_count += 1
|
|
n = check_count
|
|
outcomes.append({
|
|
"toggle_count": n,
|
|
"value": step.get("value", ""),
|
|
})
|
|
op = step.get("operator", "equals")
|
|
expected = step.get("expected", "") or ""
|
|
cmd = (step.get("command", "") or "").rstrip()
|
|
body.append(f" # Check {n} ({op}) -> on match: t {n}")
|
|
if op == "exists":
|
|
# `expected` holds the path; command field unused.
|
|
body.append(f" if (Test-Path {_ps_quote(expected)}) {{ t {n}; return }}")
|
|
else:
|
|
# Capture stdout, leave stderr visible (no 2>&1).
|
|
if cmd:
|
|
if "\n" in cmd:
|
|
body.append(" $o = ((& {")
|
|
for ln in cmd.splitlines():
|
|
body.append(" " + ln)
|
|
body.append(" }) | Out-String).Trim()")
|
|
else:
|
|
body.append(f" $o = (({cmd.strip()}) | Out-String).Trim()")
|
|
else:
|
|
body.append(" $o = ''")
|
|
body.append(' "out=[$o]"') # echo so the user sees the captured value
|
|
if op == "equals":
|
|
body.append(f" if ($o -eq {_ps_quote(expected)}) {{ t {n}; return }}")
|
|
elif op == "contains":
|
|
pattern = "*" + (expected or "").replace("*", "`*").replace("?", "`?") + "*"
|
|
body.append(f" if ($o -like {_ps_quote(pattern)}) {{ t {n}; return }}")
|
|
elif op == "regex":
|
|
body.append(f" if ($o -match {_ps_quote(expected)}) {{ t {n}; return }}")
|
|
elif op == "numeric_gt":
|
|
body.append(
|
|
f" try {{ if ([double]$o -gt [double]{_ps_quote(expected)}) "
|
|
f"{{ t {n}; return }} }} catch {{}}"
|
|
)
|
|
elif op == "numeric_lt":
|
|
body.append(
|
|
f" try {{ if ([double]$o -lt [double]{_ps_quote(expected)}) "
|
|
f"{{ t {n}; return }} }} catch {{}}"
|
|
)
|
|
else:
|
|
body.append(f" # unknown operator {op!r} — skipped")
|
|
body.append("")
|
|
# Unknown step kinds silently skipped.
|
|
|
|
# Compact PInvoke wrapper. `uint e` (instead of System.UIntPtr) lets us
|
|
# pass literal 0 from PowerShell — the high bits of dwExtraInfo are
|
|
# unused on x64 in practice for keybd_event so the size mismatch is OK.
|
|
# 0x91 = VK_SCROLL, flag 0 = keydown, flag 2 = KEYEVENTF_KEYUP.
|
|
# GetKeyState is included so the normalization line below can read the
|
|
# current Scroll Lock state without loading System.Windows.Forms.
|
|
head: list[str] = [
|
|
"Add-Type -Name K -Namespace W -MemberDefinition '"
|
|
"[System.Runtime.InteropServices.DllImport(\"user32\")]"
|
|
"public static extern void keybd_event(byte v,byte s,uint f,uint e);"
|
|
"[System.Runtime.InteropServices.DllImport(\"user32\")]"
|
|
"public static extern short GetKeyState(int v);' -EA 0",
|
|
"function t($n){1..$n|%{[W.K]::keybd_event(0x91,0,0,0);"
|
|
"[W.K]::keybd_event(0x91,0,2,0);sleep -m 200}}",
|
|
# Force Scroll Lock to OFF before the main block runs. probeScroll
|
|
# LockSequence counts state transitions, so a stray ON state at boot
|
|
# would either be a no-op (firmware tracks current state) or, worse,
|
|
# be paired with the device's own listening start at an unpredictable
|
|
# time. Normalizing here gives the script a known starting point.
|
|
"if (([W.K]::GetKeyState(0x91) -band 1) -eq 1) "
|
|
"{ [W.K]::keybd_event(0x91,0,0,0); [W.K]::keybd_event(0x91,0,2,0) }",
|
|
]
|
|
if needs_wsh:
|
|
head.append("$wsh = New-Object -ComObject WScript.Shell")
|
|
|
|
lines: list[str] = list(head)
|
|
lines.append("& {")
|
|
# Settle delay runs INSIDE the block at evaluation time, after the
|
|
# device has finished typing and started listening.
|
|
lines.append(f" sleep -m {int(settle_ms)}")
|
|
lines.extend(body)
|
|
lines.append("}")
|
|
return ("\n".join(lines) + "\n", outcomes)
|
|
|
|
|
|
def _render_run_script_check(run_script: dict,
|
|
settle_ms: int = SETTLE_MS_DEFAULT) -> str:
|
|
"""Render a 'Run Script + Check' invocation to PowerShell.
|
|
|
|
The author's .ps1 lives on disk (locally, or on a USB drive identified by
|
|
label). The host's job is just to find it and run it — the script itself
|
|
is responsible for emitting Scroll Lock toggles, so the outcomes table is
|
|
matched against whatever the script produces just like Manual mode.
|
|
|
|
The wrapper mirrors `_render_semi_auto_sequence`'s preamble (the same
|
|
`Add-Type` + `t($n)` helper and the same settle delay inside `& { ... }`)
|
|
so the device's listen-window timing contract is identical across all
|
|
three sub-modes. We still emit `t($n)` even though the user's script
|
|
won't see it — it's a no-op overhead but keeps the prefix uniform and
|
|
cheap, and leaves room for future host-injected checks.
|
|
"""
|
|
run_script = run_script or {}
|
|
location = run_script.get("location", "local")
|
|
args = (run_script.get("args", "") or "").strip()
|
|
args_tail = (" " + args) if args else ""
|
|
|
|
lines: list[str] = [
|
|
"Add-Type -Name K -Namespace W -MemberDefinition '"
|
|
"[System.Runtime.InteropServices.DllImport(\"user32\")]"
|
|
"public static extern void keybd_event(byte v,byte s,uint f,uint e);"
|
|
"[System.Runtime.InteropServices.DllImport(\"user32\")]"
|
|
"public static extern short GetKeyState(int v);' -EA 0",
|
|
"function t($n){1..$n|%{[W.K]::keybd_event(0x91,0,0,0);"
|
|
"[W.K]::keybd_event(0x91,0,2,0);sleep -m 200}}",
|
|
# Force Scroll Lock to OFF before the main block runs (see the
|
|
# matching comment in _render_semi_auto_sequence).
|
|
"if (([W.K]::GetKeyState(0x91) -band 1) -eq 1) "
|
|
"{ [W.K]::keybd_event(0x91,0,0,0); [W.K]::keybd_event(0x91,0,2,0) }",
|
|
"& {",
|
|
f" sleep -m {int(settle_ms)}",
|
|
]
|
|
|
|
if location == "usb":
|
|
label = run_script.get("drive_label", "") or ""
|
|
rel = (run_script.get("relative_path", "") or "").lstrip("\\/")
|
|
err_msg = _ps_quote(f"USB drive '{label}' not mounted")
|
|
lines.append(
|
|
f" $d = (Get-Volume -FileSystemLabel {_ps_quote(label)} -EA 0 | "
|
|
f"Where-Object DriveLetter | Select-Object -First 1).DriveLetter"
|
|
)
|
|
lines.append(f" if (-not $d) {{ Write-Error {err_msg}; return }}")
|
|
# Build the path as "<letter>:\<relative>" with -f formatting so the
|
|
# relative segment (which may contain backslashes) doesn't need any
|
|
# double-quote escaping gymnastics.
|
|
rel_literal = _ps_quote("{0}:\\" + rel)
|
|
lines.append(f" $p = {rel_literal} -f $d")
|
|
lines.append(f" & $p{args_tail}")
|
|
else:
|
|
path = run_script.get("local_path", "") or ""
|
|
lines.append(f" & {_ps_quote(path)}{args_tail}")
|
|
|
|
lines.append("}")
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
class NodeData:
|
|
"""Represents a single node in a macro's node graph."""
|
|
|
|
def __init__(self, node_type: str, node_id: str = None, x: int = 100, y: int = 100,
|
|
data: dict = None, flipped: bool = False):
|
|
self.id = node_id or new_id()
|
|
self.type = node_type
|
|
self.x = x
|
|
self.y = y
|
|
self.data = data or self.default_data()
|
|
# Visual flag: when True, the node renders with its input(s) on the
|
|
# right side and output(s) on the left. Purely cosmetic — does not
|
|
# affect flatten/execution behavior.
|
|
self.flipped = flipped
|
|
|
|
def default_data(self) -> dict:
|
|
defaults = {
|
|
"start": {},
|
|
"text": {
|
|
"text": "",
|
|
# Editor-only syntax-highlight hint ("none", "cmd", "powershell");
|
|
# the device ignores this field.
|
|
"language": "none",
|
|
},
|
|
"combo": {
|
|
"mods": [], "key": "",
|
|
# When custom_timings is True, the 4 ms values below override
|
|
# the device's global combo timing settings for THIS combo only.
|
|
# Defaults are tuned at 3x faster than the device defaults so
|
|
# enabling them maps to the legacy "fast" behavior.
|
|
"custom_timings": False,
|
|
"custom_pre_ms": 167,
|
|
"custom_post_ms": 167,
|
|
"custom_key_pre_ms": 3,
|
|
"custom_key_post_ms": 8,
|
|
},
|
|
"pause": {"wait": "click", "text": "Press to continue", "font_size": 12},
|
|
"branch": {
|
|
# mode: "manual" — user picks on-device dropdown (legacy default)
|
|
# mode: "by_variable" — device reads `var_name` and routes to the
|
|
# first choice whose `match_value` equals the variable's value;
|
|
# if no match, falls through to the LAST choice (else branch).
|
|
"mode": "manual",
|
|
"var_name": "",
|
|
"var_scope": "auto", # "auto" (device→universal), "device", "universal"
|
|
"choices": [
|
|
{"label": "Option A", "next": -1, "match_value": ""},
|
|
{"label": "Option B", "next": -1, "match_value": ""},
|
|
],
|
|
},
|
|
"delay": {"ms": 500},
|
|
"repeat": {"count": 2, "start_idx": 0, "use_selector": False},
|
|
"loop_selector": {
|
|
"min": 1, "max": 10, "step": 1, "default": 1,
|
|
"prompt": "Loop count?",
|
|
"ask_start": False,
|
|
},
|
|
"iteration_branch": {
|
|
"loop_node_id": "", # ID of the repeat node this is tied to
|
|
"choices": [
|
|
{"label": "Path A"},
|
|
{"label": "Path B"},
|
|
],
|
|
# When True, skip the branch on the final iteration of the
|
|
# tied loop. Useful when paths represent "transition to the
|
|
# next iteration" (e.g. switch KVM device) — on the last pass
|
|
# there's no next to transition to, so the branch no-ops.
|
|
"skip_final_iteration": False,
|
|
},
|
|
"note": {
|
|
"text": "Note",
|
|
"font_size": 14,
|
|
"color": "white",
|
|
"width": 220, # canvas width in world units
|
|
},
|
|
"aggregator": {"input_count": 2},
|
|
"mouse": {"button": "left", "action": "click"},
|
|
"media": {"action": "play_pause"},
|
|
"bluetooth": {
|
|
# Mode dispatcher. The on-device interpreter switches on this.
|
|
# pull_ble — host pushes encrypted vars to device (BLE)
|
|
# push_ble — device pushes its on-device vars up (BLE; device scope only)
|
|
# request_ble — device asks host to prompt for value(s) (BLE)
|
|
# set_local — device writes static (name, value) pairs locally
|
|
# get_local — device runs a script, decodes Scroll Lock toggles, sets a var
|
|
"mode": "pull_ble",
|
|
|
|
# Used by pull_ble / set_local: which on-device store to target.
|
|
# "device" — per-device store (keyed by eFuse MAC)
|
|
# "universal" — shared across all devices
|
|
"scope": "universal",
|
|
|
|
# request_ble: which variable names to prompt for on host.
|
|
"names": [],
|
|
# If true, host plays a Windows notification sound when prompt opens.
|
|
"play_sound": True,
|
|
|
|
# set_local: list of {name, value} pairs to assign on-device.
|
|
"assignments": [],
|
|
|
|
# get_local fields:
|
|
"script": "",
|
|
"script_language": "powershell",
|
|
"var_name": "",
|
|
"pre_listen_ms": 500, # delay between Enter and start of listening
|
|
"listen_window_ms": 5000, # total time to wait for Num Lock toggles
|
|
"outcomes": [
|
|
# {"toggle_count": 1, "value": "true"},
|
|
# {"toggle_count": 3, "value": "false"},
|
|
],
|
|
|
|
# get_local sub-mode:
|
|
# "manual" — author writes the PowerShell script directly
|
|
# "semi_auto" — author builds a sequence of steps; the host
|
|
# renders them to PowerShell + outcomes at upload
|
|
# "run_script_check" — author points at an existing .ps1 file
|
|
# (local path or USB-by-label); host emits a
|
|
# thin wrapper that invokes it. The .ps1 itself
|
|
# is responsible for firing Scroll Lock toggles.
|
|
"script_mode": "manual",
|
|
|
|
# run_script_check fields:
|
|
"run_script": {
|
|
"location": "local", # "local" | "usb"
|
|
"local_path": "", # absolute path; used when location == "local"
|
|
"drive_label": "", # USB volume label; used when location == "usb"
|
|
"relative_path": "", # path from drive root, e.g. "scripts\\check.ps1"
|
|
"args": "", # optional arguments appended after the path
|
|
},
|
|
|
|
# Optional Win+R launcher run BEFORE typing the script. Used
|
|
# to bring up an elevated terminal so the script has admin
|
|
# rights and a clean focus target.
|
|
"elevated_launch": {
|
|
"enabled": False,
|
|
"command": "powershell -Command \"Start-Process wt -Verb RunAs\"",
|
|
"win_r_wait_ms": 5000,
|
|
"post_type_wait_ms": 15000,
|
|
# When uac_accept is True, after pressing Enter on the
|
|
# Run box the device waits uac_wait_ms for the UAC
|
|
# dialog to appear, then sends Left+Enter to click Yes.
|
|
# Default Yes-button focus on Win10/11 UAC is on the
|
|
# left, so Left moves focus to it (or keeps it there
|
|
# if already focused) and Enter accepts.
|
|
"uac_accept": False,
|
|
"uac_wait_ms": 10000,
|
|
},
|
|
|
|
# Semi-Auto step list. Each step is a dict with a `kind` field
|
|
# plus kind-specific config. See widgets/sequence_editor_dialog.py
|
|
# for the schema.
|
|
"sequence": [],
|
|
},
|
|
"rs232": {
|
|
"baud": 9600, "data_bits": 8, "stop_bits": "1", "parity": "none",
|
|
"message": "", "line_ending": "none",
|
|
"wait_response": False, "expected_response": "", "timeout_ms": 5000,
|
|
"post_send_delay_ms": 0,
|
|
},
|
|
"subroutine": {"name": ""},
|
|
"pc_alive_check": {
|
|
"condition": "pc_response", # "numlock_on", "numlock_off", "pc_response"
|
|
"loop": True,
|
|
"poll_delay_ms": 500,
|
|
},
|
|
"macro": {
|
|
# "events" is a list of [t_ms, action, hid_code] triples:
|
|
# t_ms — wall-clock ms since recording start
|
|
# action — 0 = press, 1 = release
|
|
# hid_code — raw USB HID usage code (see TKKEYSYM_TO_HID)
|
|
# Raw HID codes (not ASCII) let us reproduce chords exactly —
|
|
# modifiers stay held across other keys instead of being
|
|
# stripped by the keyboard library's ASCII→shift mapping.
|
|
"events": [],
|
|
"name": "", # optional human-readable label
|
|
},
|
|
}
|
|
return defaults.get(self.type, {}).copy()
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"id": self.id,
|
|
"type": self.type,
|
|
"x": self.x,
|
|
"y": self.y,
|
|
"data": self.data.copy(),
|
|
"flipped": self.flipped,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, d: dict) -> "NodeData":
|
|
node_type = d["type"]
|
|
data = d.get("data")
|
|
|
|
# Legacy migration: combo nodes used to have a single "fast" boolean.
|
|
# Map fast=True to custom_timings=True with the 3x-faster defaults so
|
|
# existing macros behave identically after an upgrade.
|
|
if node_type == "combo" and isinstance(data, dict) and "fast" in data:
|
|
was_fast = bool(data.pop("fast"))
|
|
if was_fast and not data.get("custom_timings"):
|
|
data["custom_timings"] = True
|
|
data.setdefault("custom_pre_ms", 167)
|
|
data.setdefault("custom_post_ms", 167)
|
|
data.setdefault("custom_key_pre_ms", 3)
|
|
data.setdefault("custom_key_post_ms", 8)
|
|
|
|
return cls(
|
|
node_type=node_type,
|
|
node_id=d.get("id"),
|
|
x=d.get("x", 100),
|
|
y=d.get("y", 100),
|
|
data=data,
|
|
flipped=d.get("flipped", False),
|
|
)
|
|
|
|
|
|
class ConnectionData:
|
|
"""Represents a connection between two nodes."""
|
|
|
|
def __init__(self, from_id: str, from_port: str, to_id: str, to_port: str):
|
|
self.from_id = from_id
|
|
self.from_port = from_port
|
|
self.to_id = to_id
|
|
self.to_port = to_port
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"from": self.from_id,
|
|
"from_port": self.from_port,
|
|
"to": self.to_id,
|
|
"to_port": self.to_port,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, d: dict) -> "ConnectionData":
|
|
return cls(d["from"], d["from_port"], d["to"], d["to_port"])
|
|
|
|
|
|
class Macro:
|
|
"""Represents a complete macro with its node graph."""
|
|
|
|
def __init__(self, name: str = "New Routine", image_path: str = None, label_color: str = "white"):
|
|
self.name = name
|
|
self.image_path = image_path
|
|
self.label_color = label_color
|
|
self.nodes: list[NodeData] = []
|
|
self.connections: list[ConnectionData] = []
|
|
|
|
def add_node(self, node: NodeData):
|
|
self.nodes.append(node)
|
|
|
|
def remove_node(self, node_id: str):
|
|
self.nodes = [n for n in self.nodes if n.id != node_id]
|
|
self.connections = [
|
|
c for c in self.connections
|
|
if c.from_id != node_id and c.to_id != node_id
|
|
]
|
|
|
|
def add_connection(self, conn: ConnectionData):
|
|
# An input port can only have one source — drop any existing one first
|
|
self.connections = [
|
|
c for c in self.connections
|
|
if not (c.to_id == conn.to_id and c.to_port == conn.to_port)
|
|
]
|
|
self.connections.append(conn)
|
|
|
|
def remove_connection(self, from_id: str, from_port: str, to_id: str, to_port: str):
|
|
self.connections = [
|
|
c for c in self.connections
|
|
if not (c.from_id == from_id and c.from_port == from_port
|
|
and c.to_id == to_id and c.to_port == to_port)
|
|
]
|
|
|
|
def get_node(self, node_id: str) -> NodeData | None:
|
|
for n in self.nodes:
|
|
if n.id == node_id:
|
|
return n
|
|
return None
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"name": self.name,
|
|
"label_color": self.label_color,
|
|
"image_path": self.image_path,
|
|
"nodes": [n.to_dict() for n in self.nodes],
|
|
"connections": [c.to_dict() for c in self.connections],
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, d: dict) -> "Macro":
|
|
m = cls(name=d.get("name", "Unnamed"), image_path=d.get("image_path"),
|
|
label_color=d.get("label_color", "white"))
|
|
m.nodes = [NodeData.from_dict(nd) for nd in d.get("nodes", [])]
|
|
m.connections = [ConnectionData.from_dict(cd) for cd in d.get("connections", [])]
|
|
return m
|
|
|
|
def clone(self, new_name: str = None) -> "Macro":
|
|
"""Deep-copy this macro with fresh node IDs.
|
|
|
|
Node IDs only need to be unique within a single macro's graph,
|
|
but regenerating them on duplicate avoids any chance of a stale
|
|
cross-macro reference (e.g. a future global node index) and
|
|
matches what users expect from a "duplicate" action.
|
|
Connections are remapped to point at the new IDs.
|
|
"""
|
|
id_map: dict[str, str] = {n.id: new_id() for n in self.nodes}
|
|
|
|
cloned_nodes: list[NodeData] = []
|
|
for n in self.nodes:
|
|
d = n.to_dict()
|
|
d["id"] = id_map[n.id]
|
|
cloned_nodes.append(NodeData.from_dict(d))
|
|
|
|
cloned_conns: list[ConnectionData] = [
|
|
ConnectionData(
|
|
id_map.get(c.from_id, c.from_id),
|
|
c.from_port,
|
|
id_map.get(c.to_id, c.to_id),
|
|
c.to_port,
|
|
)
|
|
for c in self.connections
|
|
]
|
|
|
|
copy = Macro(
|
|
name=new_name if new_name is not None else f"{self.name} (copy)",
|
|
image_path=self.image_path,
|
|
label_color=self.label_color,
|
|
)
|
|
copy.nodes = cloned_nodes
|
|
copy.connections = cloned_conns
|
|
return copy
|
|
|
|
def _auto_linked_connections(self) -> list["ConnectionData"]:
|
|
"""Synthesize a linear chain of connections when none are defined.
|
|
|
|
A common source of "the sub-routine runs but does nothing" bugs is
|
|
a macro / sub-routine whose GUI graph has nodes but no edges —
|
|
either because the user laid out nodes in the editor and forgot
|
|
to wire them, or because an older config-format upgrade dropped
|
|
the connections array. With no connections, walk_chain has
|
|
nothing to follow from the start node and the flatten returns
|
|
empty; the device shows the macro name and immediately exits.
|
|
|
|
Here we paper over that case: when ``self.connections`` is empty
|
|
but we have at least two non-note nodes, build a sensible linear
|
|
chain by sorting nodes left-to-right (and top-to-bottom as a
|
|
tiebreaker), putting the explicit ``start`` node first if one
|
|
exists, and connecting each via the "out" → "in" port pair
|
|
(the convention used by every simple node type).
|
|
|
|
This is intentionally conservative: branch / repeat /
|
|
iteration_branch / aggregator graphs need named ports
|
|
(out_0, loop_body, in_0, etc.) and can't be auto-wired
|
|
meaningfully — for those, the chain still ends up linear via
|
|
"out", which the firmware ignores for those node types. The net
|
|
effect is "do what the user obviously intended for a linear
|
|
sequence; gracefully degrade for advanced graphs."
|
|
"""
|
|
non_note = [n for n in self.nodes if n.type != "note"]
|
|
if len(non_note) < 2:
|
|
return []
|
|
# Stable sort by (y, x) so the chain matches the visual top-to-bottom,
|
|
# left-to-right reading order. Start node is forced to position 0 so
|
|
# the flatten's "find start" logic picks it up naturally.
|
|
start_node = next((n for n in non_note if n.type == "start"), None)
|
|
rest = [n for n in non_note if n is not start_node]
|
|
rest.sort(key=lambda n: (round(n.y / 40.0), n.x))
|
|
ordered = ([start_node] if start_node else []) + rest
|
|
return [
|
|
ConnectionData(ordered[i].id, "out", ordered[i + 1].id, "in")
|
|
for i in range(len(ordered) - 1)
|
|
]
|
|
|
|
def flatten_for_device(self) -> list[dict]:
|
|
"""Convert the visual node graph into a linear node array for the device.
|
|
|
|
Performs topological sort starting from the node with no incoming
|
|
connections. Branch nodes produce multiple chains laid out sequentially.
|
|
"""
|
|
if not self.nodes:
|
|
return []
|
|
|
|
# Auto-link orphan graphs (see _auto_linked_connections). We use the
|
|
# synthesized edges only when the user actually left connections empty
|
|
# — never override explicit wiring.
|
|
active_connections = (
|
|
self.connections
|
|
if self.connections
|
|
else self._auto_linked_connections()
|
|
)
|
|
|
|
# Build adjacency: node_id -> {port: target_node_id}
|
|
outgoing = {}
|
|
incoming_ids = set()
|
|
for c in active_connections:
|
|
if c.from_id not in outgoing:
|
|
outgoing[c.from_id] = {}
|
|
outgoing[c.from_id][c.from_port] = c.to_id
|
|
incoming_ids.add(c.to_id)
|
|
|
|
# Find start node (explicit "start" type, or fallback to no incoming connections).
|
|
# Note nodes are GUI-only annotations — never treat one as a start candidate.
|
|
start_nodes = [n for n in self.nodes if n.type == "start"]
|
|
if not start_nodes:
|
|
start_nodes = [n for n in self.nodes
|
|
if n.id not in incoming_ids and n.type != "note"]
|
|
if not start_nodes and self.nodes:
|
|
non_note = [n for n in self.nodes if n.type != "note"]
|
|
if non_note:
|
|
start_nodes = [non_note[0]]
|
|
|
|
# Assign a stable loop_id (small integer) to each repeat node so that
|
|
# iteration_branch nodes can cross-reference them on the device.
|
|
loop_ids = {}
|
|
for n in self.nodes:
|
|
if n.type == "repeat":
|
|
loop_ids[n.id] = len(loop_ids)
|
|
|
|
result = []
|
|
visited = set()
|
|
# aggregator node_id -> its position in ``result`` once emitted.
|
|
# Lets a later branch path that loops BACK to an already-emitted
|
|
# aggregator be patched to jump directly to that known position
|
|
# (i.e. "retry" the preceding sub-graph), instead of falling through
|
|
# to ``end_target`` — which inside a loop body would hit
|
|
# ``_loop_back`` and spuriously bump the outer iteration counter
|
|
# (e.g. a nested-loop BIOS-setup routine).
|
|
agg_positions: dict[str, int] = {}
|
|
|
|
def walk_branch_paths(branch_entry, ports, port_name_fn, default_label_fn):
|
|
"""Shared helper for branch + iteration_branch path walking.
|
|
|
|
Walks each choice's chain with aggregator-merge semantics: if a path
|
|
reaches an Aggregator node, it stops there and is patched to jump to
|
|
the aggregator's position. After all paths are walked, each unique
|
|
NEW aggregator is emitted once as a passthrough, and its output
|
|
chain is walked once. Paths whose chains loop BACK to an aggregator
|
|
that was emitted earlier (e.g. a retry loop) are patched to jump
|
|
directly to that earlier position. Paths that don't reach any
|
|
aggregator fall back to the classic "jump past all branches" merge
|
|
point.
|
|
|
|
Also writes ``skip_target`` into ``branch_entry['data']`` pointing
|
|
to the "natural next node" if the branch were skipped entirely.
|
|
For a branch with an aggregator merge point, that's the aggregator
|
|
(so post-merge body nodes still run); otherwise it's end_target.
|
|
iteration_branch's runtime uses this when ``skip_final_iteration``
|
|
is set so the last pass can bypass the path logic and merge
|
|
straight into the rest of the body.
|
|
"""
|
|
choices = branch_entry["data"]["choices"]
|
|
end_markers = [] # `_jump` indexes that should land on end_target
|
|
agg_markers = {} # NEW agg_node_id -> list of `_jump` indexes
|
|
back_edges = [] # (jump_idx, existing_agg_pos) — patch directly
|
|
new_agg_positions: dict[str, int] = {} # agg_id -> position emitted here
|
|
|
|
for i, _ in enumerate(choices):
|
|
target = ports.get(port_name_fn(i))
|
|
choices[i]["next"] = len(result)
|
|
|
|
stopped_agg = None
|
|
if target:
|
|
stopped_agg = walk_chain(target, stop_at_aggregator=True)
|
|
|
|
jump_idx = len(result)
|
|
result.append({"type": "_jump", "data": {"target": -1}})
|
|
|
|
if stopped_agg:
|
|
# Aggregator already emitted earlier? Back-edge: patch
|
|
# straight to its known position. Otherwise group it so
|
|
# we emit the aggregator once below.
|
|
if stopped_agg in agg_positions:
|
|
back_edges.append((jump_idx, agg_positions[stopped_agg]))
|
|
else:
|
|
agg_markers.setdefault(stopped_agg, []).append(jump_idx)
|
|
else:
|
|
end_markers.append(jump_idx)
|
|
|
|
# Back-edges point at already-emitted positions; patch directly.
|
|
for jump_idx, agg_pos in back_edges:
|
|
result[jump_idx]["data"]["target"] = agg_pos
|
|
|
|
# Emit each NEW aggregator once and walk its output chain.
|
|
for agg_id, jump_indices in agg_markers.items():
|
|
agg_pos = len(result)
|
|
for idx in jump_indices:
|
|
result[idx]["data"]["target"] = agg_pos
|
|
result.append({"type": "aggregator", "data": {}})
|
|
visited.add(agg_id)
|
|
agg_positions[agg_id] = agg_pos
|
|
new_agg_positions[agg_id] = agg_pos
|
|
agg_out = outgoing.get(agg_id, {}).get("out")
|
|
if agg_out:
|
|
walk_chain(agg_out)
|
|
|
|
# Non-aggregator paths collapse to the final position.
|
|
end_target = len(result)
|
|
for idx in end_markers:
|
|
result[idx]["data"]["target"] = end_target
|
|
|
|
# Publish where a "skipped" branch rejoins normal flow.
|
|
# Prefer the first NEW aggregator we emitted: its position sits
|
|
# BEFORE any post-merge body nodes added by the agg chain, so
|
|
# those body nodes still run on skip. Back-edge aggregators
|
|
# aren't candidates — they'd skip to an earlier-in-time node.
|
|
# Fallback: end_target, where non-aggregator paths go anyway.
|
|
if new_agg_positions:
|
|
branch_entry["data"]["skip_target"] = min(new_agg_positions.values())
|
|
else:
|
|
branch_entry["data"]["skip_target"] = end_target
|
|
|
|
def walk_chain(node_id, stop_at_aggregator=False):
|
|
"""Walk a linear chain of nodes, emitting flattened entries.
|
|
|
|
If ``stop_at_aggregator`` is True, returns the aggregator's
|
|
node_id when the walk reaches one (first-time or already-visited).
|
|
The caller decides whether to emit it anew or patch a back-edge
|
|
to its existing position. Aggregators encountered this way are
|
|
not emitted here and not added to ``visited`` by this function.
|
|
Returns None if the walk ends for any other reason.
|
|
"""
|
|
# If the starting node itself is an already-visited aggregator,
|
|
# the while-loop would exit immediately without reporting it.
|
|
if stop_at_aggregator and node_id and node_id in visited:
|
|
node = self.get_node(node_id)
|
|
if node and node.type == "aggregator":
|
|
return node_id
|
|
|
|
while node_id and node_id not in visited:
|
|
node = self.get_node(node_id)
|
|
if not node:
|
|
break
|
|
|
|
# Branch-walking mode: stop at the first aggregator we reach.
|
|
# The caller emits it and walks its tail once.
|
|
if stop_at_aggregator and node.type == "aggregator":
|
|
return node_id
|
|
|
|
visited.add(node_id)
|
|
|
|
# Start node is a visual marker; don't emit it.
|
|
if node.type == "start":
|
|
ports = outgoing.get(node_id, {})
|
|
node_id = ports.get("out")
|
|
continue
|
|
|
|
# Notes are GUI-only annotations; never send to the device.
|
|
if node.type == "note":
|
|
break
|
|
|
|
if node.type == "branch":
|
|
branch_entry = {"type": "branch", "data": {
|
|
"mode": node.data.get("mode", "manual"),
|
|
"var_name": node.data.get("var_name", ""),
|
|
"var_scope": node.data.get("var_scope", "auto"),
|
|
"choices": [],
|
|
}}
|
|
result.append(branch_entry)
|
|
for i, choice in enumerate(node.data.get("choices", [])):
|
|
branch_entry["data"]["choices"].append({
|
|
"label": choice.get("label", f"Option {i+1}"),
|
|
"next": 0,
|
|
"match_value": choice.get("match_value", ""),
|
|
})
|
|
walk_branch_paths(
|
|
branch_entry,
|
|
outgoing.get(node_id, {}),
|
|
lambda i: f"out_{i}",
|
|
lambda i: f"Option {i+1}",
|
|
)
|
|
return None
|
|
|
|
if node.type == "iteration_branch":
|
|
tied_loop = node.data.get("loop_node_id", "")
|
|
loop_id = loop_ids.get(tied_loop, -1)
|
|
it_entry = {"type": "iteration_branch", "data": {
|
|
"loop_id": loop_id,
|
|
"skip_final_iteration": bool(node.data.get("skip_final_iteration", False)),
|
|
"choices": [],
|
|
}}
|
|
result.append(it_entry)
|
|
for i, choice in enumerate(node.data.get("choices", [])):
|
|
it_entry["data"]["choices"].append({
|
|
"label": choice.get("label", f"Path {i+1}"),
|
|
"next": 0,
|
|
})
|
|
walk_branch_paths(
|
|
it_entry,
|
|
outgoing.get(node_id, {}),
|
|
lambda i: f"out_{i}",
|
|
lambda i: f"Path {i+1}",
|
|
)
|
|
return None
|
|
|
|
if node.type == "aggregator":
|
|
# Normal encounter (not inside a stop_at_aggregator walk):
|
|
# emit as passthrough and continue through "out". Record
|
|
# its position so a later back-edge (e.g. a Retry path
|
|
# looping back here) can patch directly to it.
|
|
agg_positions[node_id] = len(result)
|
|
result.append({"type": "aggregator", "data": {}})
|
|
ports = outgoing.get(node_id, {})
|
|
node_id = ports.get("out")
|
|
continue
|
|
|
|
if node.type == "repeat":
|
|
# 2 inputs (in, loop_back), 2 outputs (loop_body, done).
|
|
count = node.data.get("count", 2)
|
|
loop_start_idx = len(result)
|
|
loop_id = loop_ids.get(node_id, 0)
|
|
result.append({"type": "_loop_start", "data": {
|
|
"count": count,
|
|
"body": loop_start_idx + 1,
|
|
"done": -1,
|
|
"use_selector": node.data.get("use_selector", False),
|
|
"loop_id": loop_id,
|
|
}})
|
|
|
|
ports = outgoing.get(node_id, {})
|
|
body_target = ports.get("loop_body")
|
|
if body_target:
|
|
walk_chain(body_target)
|
|
|
|
# End of body jumps back to loop_start.
|
|
result.append({"type": "_loop_back", "data": {"target": loop_start_idx}})
|
|
|
|
# done target is the position after the loop_back.
|
|
result[loop_start_idx]["data"]["done"] = len(result)
|
|
|
|
done_target = ports.get("done")
|
|
if done_target:
|
|
walk_chain(done_target)
|
|
|
|
return
|
|
|
|
if node.type == "pc_alive_check":
|
|
# Two outputs (true / false). Looping is handled internally
|
|
# by the firmware when loop=True.
|
|
check_entry = {"type": "pc_alive_check", "data": {
|
|
"condition": node.data.get("condition", "pc_response"),
|
|
"loop": node.data.get("loop", True),
|
|
"poll_delay_ms": node.data.get("poll_delay_ms", 500),
|
|
"true_target": -1,
|
|
"false_target": -1,
|
|
}}
|
|
check_idx = len(result)
|
|
result.append(check_entry)
|
|
|
|
ports = outgoing.get(node_id, {})
|
|
end_markers = []
|
|
agg_markers = {} # NEW agg_id -> [_jump indices]
|
|
|
|
for port_name in ("true", "false"):
|
|
target = ports.get(port_name)
|
|
result[check_idx]["data"][f"{port_name}_target"] = len(result)
|
|
|
|
stopped_agg = None
|
|
if target:
|
|
stopped_agg = walk_chain(target, stop_at_aggregator=True)
|
|
|
|
jump_idx = len(result)
|
|
result.append({"type": "_jump", "data": {"target": -1}})
|
|
|
|
if stopped_agg:
|
|
if stopped_agg in agg_positions:
|
|
# Back-edge to an aggregator already emitted by
|
|
# the other port's chain. Patch directly.
|
|
result[jump_idx]["data"]["target"] = agg_positions[stopped_agg]
|
|
else:
|
|
agg_markers.setdefault(stopped_agg, []).append(jump_idx)
|
|
else:
|
|
end_markers.append(jump_idx)
|
|
|
|
# Emit each NEW aggregator once and walk its output chain.
|
|
for agg_id, jump_indices in agg_markers.items():
|
|
agg_pos = len(result)
|
|
for idx in jump_indices:
|
|
result[idx]["data"]["target"] = agg_pos
|
|
result.append({"type": "aggregator", "data": {}})
|
|
visited.add(agg_id)
|
|
agg_positions[agg_id] = agg_pos
|
|
agg_out = outgoing.get(agg_id, {}).get("out")
|
|
if agg_out:
|
|
walk_chain(agg_out)
|
|
|
|
end_target = len(result)
|
|
for idx in end_markers:
|
|
result[idx]["data"]["target"] = end_target
|
|
|
|
return
|
|
|
|
if node.type == "bluetooth" and node.data.get("mode") == "get_local":
|
|
# Two outputs (pass / fail). Same shape as pc_alive_check.
|
|
bt_entry = {"type": "bluetooth", "data": node.data.copy()}
|
|
bt_entry["data"]["pass_target"] = -1
|
|
bt_entry["data"]["fail_target"] = -1
|
|
|
|
# Semi-Auto mode: render the visual sequence into a real
|
|
# PowerShell script + auto-derive the outcomes table. The
|
|
# device-side path is identical to manual mode after this.
|
|
if bt_entry["data"].get("script_mode") == "semi_auto":
|
|
rendered_script, rendered_outcomes = _render_semi_auto_sequence(
|
|
bt_entry["data"].get("sequence") or [])
|
|
bt_entry["data"]["script"] = rendered_script
|
|
bt_entry["data"]["outcomes"] = rendered_outcomes
|
|
elif bt_entry["data"].get("script_mode") == "run_script_check":
|
|
# Outcomes pass through — the author wrote them in the UI
|
|
# and the .ps1 owns the toggle emission.
|
|
bt_entry["data"]["script"] = _render_run_script_check(
|
|
bt_entry["data"].get("run_script") or {})
|
|
|
|
bt_idx = len(result)
|
|
result.append(bt_entry)
|
|
|
|
ports = outgoing.get(node_id, {})
|
|
end_markers = []
|
|
agg_markers = {} # NEW agg_id -> [_jump indices]
|
|
|
|
for port_name in ("pass", "fail"):
|
|
target = ports.get(port_name)
|
|
result[bt_idx]["data"][f"{port_name}_target"] = len(result)
|
|
|
|
stopped_agg = None
|
|
if target:
|
|
stopped_agg = walk_chain(target, stop_at_aggregator=True)
|
|
|
|
jump_idx = len(result)
|
|
result.append({"type": "_jump", "data": {"target": -1}})
|
|
|
|
if stopped_agg:
|
|
if stopped_agg in agg_positions:
|
|
# Back-edge to an aggregator already emitted by
|
|
# the other port's chain (e.g. fail wire merging
|
|
# back into the pass-chain's KVM-reset path).
|
|
result[jump_idx]["data"]["target"] = agg_positions[stopped_agg]
|
|
else:
|
|
agg_markers.setdefault(stopped_agg, []).append(jump_idx)
|
|
else:
|
|
end_markers.append(jump_idx)
|
|
|
|
# Emit each NEW aggregator once and walk its output chain.
|
|
for agg_id, jump_indices in agg_markers.items():
|
|
agg_pos = len(result)
|
|
for idx in jump_indices:
|
|
result[idx]["data"]["target"] = agg_pos
|
|
result.append({"type": "aggregator", "data": {}})
|
|
visited.add(agg_id)
|
|
agg_positions[agg_id] = agg_pos
|
|
agg_out = outgoing.get(agg_id, {}).get("out")
|
|
if agg_out:
|
|
walk_chain(agg_out)
|
|
|
|
end_target = len(result)
|
|
for idx in end_markers:
|
|
result[idx]["data"]["target"] = end_target
|
|
|
|
return
|
|
|
|
# Default linear node: emit and follow "out".
|
|
entry = {"type": node.type, "data": node.data.copy()}
|
|
result.append(entry)
|
|
|
|
ports = outgoing.get(node_id, {})
|
|
node_id = ports.get("out")
|
|
|
|
# Loop exited because node_id is None or already visited. In
|
|
# branch-walking mode, if we fell through onto a visited
|
|
# aggregator (back-edge to an earlier merge point), report it
|
|
# so the caller patches a direct jump to its known position
|
|
# instead of treating this path as a "fell-off" end-marker.
|
|
# That's what makes a "Retry" path looping back to the
|
|
# top-of-body aggregator actually retry, instead of falling
|
|
# through to the enclosing loop's _loop_back and spuriously
|
|
# incrementing the outer iteration counter.
|
|
if stop_at_aggregator and node_id:
|
|
tail = self.get_node(node_id)
|
|
if tail and tail.type == "aggregator":
|
|
return node_id
|
|
|
|
# Linear walk fell through to an already-emitted aggregator.
|
|
# Emit an explicit back-edge so execution actually jumps to
|
|
# the known merge point instead of running off the end of the
|
|
# current chain into whatever node happens to be emitted next
|
|
# (e.g. the enclosing loop's _loop_back, which silently
|
|
# consumes an iteration). This is the path a Retry chain
|
|
# walked via an aggregator's "out" takes when it loops back
|
|
# to the top-of-body aggregator.
|
|
if node_id and node_id in agg_positions:
|
|
tail = self.get_node(node_id)
|
|
if tail and tail.type == "aggregator":
|
|
result.append({"type": "_jump",
|
|
"data": {"target": agg_positions[node_id]}})
|
|
|
|
return None
|
|
|
|
for start in start_nodes:
|
|
walk_chain(start.id)
|
|
|
|
return result
|