"""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 ":\" 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