Spaces:
Running
Running
File size: 6,783 Bytes
d646dd7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 | """Helpers for treating the two SenseGlove channels as one input device.
The recorder stores the left and right channels separately because they are
two data streams. That storage detail must not leak into the workflow UI:
one glove device can still carry both ``left_glove`` and ``right_glove``
source keys in its node configuration.
"""
from __future__ import annotations
import re
_SIDE_TOKEN = re.compile(r"(^|[_\-.])(left|right)(?=[_\-.]|$)", re.IGNORECASE)
def _glove_base(key: str) -> str | None:
"""Return a stable pair key for a left/right glove channel."""
value = str(key or "").strip()
lower = value.lower()
if "glove" not in lower:
return None
match = _SIDE_TOKEN.search(value)
if not match:
return None
# Remove the separator together with the side token. For example,
# left_glove_joint/right_glove_joint both become ``glove_joint``.
base = _SIDE_TOKEN.sub("_", value, count=1).strip("_-. ").lower()
return base or "glove"
def group_glove_source_keys(values) -> list[list[str]]:
"""Group paired glove channels while preserving singleton channels.
The result is deterministic and ordered left, right. Non-glove sensor
names are returned as singleton groups so this helper is safe to use on
a complete ``info.json`` sensor list.
"""
unique: list[str] = []
seen: set[str] = set()
for raw in values or []:
key = str(raw or "").strip()
if key and key.lower() not in seen:
unique.append(key)
seen.add(key.lower())
grouped: dict[str, dict[str, str]] = {}
singleton_order: list[str] = []
for key in unique:
base = _glove_base(key)
match = _SIDE_TOKEN.search(key) if base else None
side = match.group(2).lower() if match else ""
if not base or side not in {"left", "right"}:
singleton_order.append(key)
continue
bucket = grouped.setdefault(base, {})
bucket.setdefault(side, key)
result: list[list[str]] = []
result.extend([key] for key in singleton_order)
# Pair ordering is based on the first appearance of each base, while the
# channels inside a pair always use left then right.
bases: list[str] = []
for key in unique:
base = _glove_base(key)
if base and base not in bases:
bases.append(base)
paired: list[list[str]] = []
for base in bases:
bucket = grouped.get(base, {})
keys = [bucket[side] for side in ("left", "right") if side in bucket]
if keys:
paired.append(keys)
# Keep original singleton order, followed by grouped glove channels. In
# the normal recorder layout this gives left_glove,right_glove together.
return result + paired
def is_paired_glove_source(keys: list[str]) -> bool:
"""Whether a group contains both left and right glove channels."""
sides = {
(match.group(2).lower() if match else "")
for key in keys
for match in [_SIDE_TOKEN.search(str(key))]
if match and "glove" in str(key).lower()
}
return sides == {"left", "right"}
def merge_paired_glove_nodes(graph: dict) -> bool:
"""Collapse legacy left/right glove input nodes in a workflow graph.
Older auto-generated workflows contain one node per channel. The first
node is kept, both source keys are written to its config, and outgoing
edges from the removed node are redirected to it. Returns whether the
graph changed.
"""
if not isinstance(graph, dict):
return False
nodes = list(graph.get("nodes") or [])
by_base: dict[str, list[dict]] = {}
for node in nodes:
data = node.get("data") or {}
if data.get("nodeType") != "glove_sensor":
continue
config = data.get("config") or {}
raw = config.get("source_keys") or config.get("source_key") or config.get("position")
values = raw.split(",") if isinstance(raw, str) else raw or []
keys = [str(value).strip() for value in values if str(value).strip()]
bases = {_glove_base(key) for key in keys}
for base in bases:
if base:
by_base.setdefault(base, []).append(node)
changed = False
redirects: dict[str, str] = {}
remove_ids: set[str] = set()
for duplicate_nodes in by_base.values():
if len(duplicate_nodes) < 2:
continue
keeper = duplicate_nodes[0]
keeper_data = keeper.setdefault("data", {})
keeper_config = keeper_data.setdefault("config", {})
all_keys: list[str] = []
for node in duplicate_nodes:
config = (node.get("data") or {}).get("config") or {}
raw = config.get("source_keys") or config.get("source_key") or config.get("position")
values = raw.split(",") if isinstance(raw, str) else raw or []
for key in values:
key = str(key).strip()
if key and key.lower() not in {item.lower() for item in all_keys}:
all_keys.append(key)
merged = group_glove_source_keys(all_keys)
paired = next((group for group in merged if is_paired_glove_source(group)), None)
if not paired:
continue
keeper_config["source_key"] = paired[0]
keeper_config["source_keys"] = ",".join(paired)
keeper_config.pop("position", None)
keeper_data["label"] = keeper_data.get("label") or "Glove Sensor"
for node in duplicate_nodes[1:]:
node_id = str(node.get("id") or "")
if node_id:
redirects[node_id] = str(keeper.get("id") or "")
remove_ids.add(node_id)
changed = True
if not changed:
return False
graph["nodes"] = [node for node in nodes
if str(node.get("id") or "") not in remove_ids]
seen_edges: set[tuple[str, str, str, str]] = set()
rewritten: list[dict] = []
for edge in graph.get("edges") or []:
source = redirects.get(str(edge.get("source") or ""), edge.get("source"))
target = redirects.get(str(edge.get("target") or ""), edge.get("target"))
if not source or not target:
continue
edge = dict(edge)
edge["source"] = source
edge["target"] = target
identity = (str(source), str(edge.get("sourceHandle") or ""),
str(target), str(edge.get("targetHandle") or ""))
if identity in seen_edges:
continue
seen_edges.add(identity)
edge["id"] = (
f"xy-edge__{source}{edge.get('sourceHandle') or ''}-"
f"{target}{edge.get('targetHandle') or ''}")
rewritten.append(edge)
graph["edges"] = rewritten
return True
|