From 953063d02ac6fd00c015e9a43e92f3e2f34fd151 Mon Sep 17 00:00:00 2001 From: MFA-X-AI Date: Sun, 5 Oct 2025 00:32:58 +0900 Subject: [PATCH 1/2] update base to allow dynaports to be subscriptable, update compiler to assign [x] for every dyna var --- xai_components/base.py | 46 +++++++++++++++++++++ xircuits/compiler/generator.py | 74 ++++++++++++++++++++-------------- 2 files changed, 89 insertions(+), 31 deletions(-) diff --git a/xai_components/base.py b/xai_components/base.py index ad14a2f3..681bafb3 100644 --- a/xai_components/base.py +++ b/xai_components/base.py @@ -72,6 +72,43 @@ def __deepcopy__(self, memo): memo[id_self] = _copy return _copy + class _DynaIndex: + def __init__(self, parent, idx): + self._p = parent + self._i = idx + + def _assign(self, obj): + v = self._p._value + # Start from empty container if needed + if v is None: + v = [] + # Internally first use list for convenience + is_tuple = isinstance(v, tuple) + if is_tuple: + v = list(v) + + while len(v) <= self._i: + v.append(None) + v[self._i] = obj + + # Convert back to tuple if the port uses dynatuple semantics + try_is_dynatuple = (self._p._getter is dynatuple.getter) + self._p._value = tuple(v) if try_is_dynatuple else v + + def connect(self, ref): + # Accept InArg/OutArg or literal-like objects + self._assign(ref) + + def set(self, value): + self._assign(value) + + def __getitem__(self, idx): + # Allow indexing when the port holds a list/tuple (dynaports) + return InArg._DynaIndex(self, idx) + + def __setitem__(self, idx, value): + # Support literal assignment: port[i] = 'a' + self[idx].set(value) class InCompArg(Generic[T]): def __init__(self, value: T = None, getter: Callable[[T], any] = lambda x: x) -> None: @@ -265,6 +302,10 @@ def getter(x): return [] return [item.value if isinstance(item, (InArg, OutArg)) else item for item in x] + @staticmethod + def initial_value(): + return [] + class dynatuple(tuple): def __init__(self, *args): @@ -282,6 +323,11 @@ def resolve(item): return item return tuple(resolve(item) for item in x) + @staticmethod + def initial_value(): + return tuple() + + def parse_bool(value): if value is None: return None diff --git a/xircuits/compiler/generator.py b/xircuits/compiler/generator.py index 79f20999..6ceb6649 100644 --- a/xircuits/compiler/generator.py +++ b/xircuits/compiler/generator.py @@ -199,42 +199,54 @@ def connect_args(target, source): # Handle dynamic connections dynaports = [p for p in node.ports if - p.direction == 'in' and p.type != 'triangle-link' and p.dataType in DYNAMIC_PORTS] - ports_by_varName = {} - - RefOrValue = namedtuple('RefOrValue', ['value', 'is_ref']) # Renamed to RefOrValue + p.direction == 'in' and p.type != 'triangle-link' and p.dataType in DYNAMIC_PORTS] - # Group ports by varName - for port in dynaports: - if port.varName not in ports_by_varName: - ports_by_varName[port.varName] = [] - ports_by_varName[port.varName].append(port) + # Capture index N from port names like 'parameter-dynalist-dlist-2' + _name_idx_re = re.compile(r'-(\d+)\s*$') - for varName, ports in ports_by_varName.items(): - dynaport_values = [] + # Map: varName -> {index: port} + ports_by_varName = {} - for port in ports: - if port.source.id not in named_nodes: - value = _get_value_from_literal_port(port) - dynaport_values.append(RefOrValue(value, False)) - else: - # Handle named node references - value = "%s.%s" % (named_nodes[port.source.id], port.sourceLabel) # Variable reference - dynaport_values.append(RefOrValue(value, True)) - - if ports[0].dataType == 'dynatuple': - tuple_elements = [item.value if item.is_ref else repr(item.value) for item in dynaport_values] - if len(tuple_elements) == 1: - assignment_value = '(' + tuple_elements[0] + ',)' + for port in dynaports: + var_name = port.varName + name = port.name + # Extract index from name; default to 0 if no '-N' suffix + m = _name_idx_re.search(name) + idx = int(m.group(1)) if m else 0 + + ports_by_varName.setdefault(var_name, {}) + ports_by_varName[var_name].setdefault(idx, port) + + # Emit code per element in numeric order + for var_name, mapping in ports_by_varName.items(): + for i in sorted(mapping.keys()): + port = mapping[i] + target_indexed = f"{named_nodes[port.target.id]}.{var_name}[{i}]" + + if port.source.id in named_nodes: + # Component reference -> connect + source_ref = f"{named_nodes[port.source.id]}.{port.sourceLabel}" + init_code.append(ast.parse(f"{target_indexed}.connect({source_ref})")) else: - assignment_value = '(' + ', '.join(tuple_elements) + ')' - else: - list_elements = [item.value if item.is_ref else repr(item.value) for item in dynaport_values] - assignment_value = '[' + ', '.join(list_elements) + ']' + # Regex: matches e.g. 'Argument (string): argsName' + pattern = re.compile(r'^Argument \((.+?)\): (.+)$') + if port.source.file is None and port.source.name.startswith("Argument "): + match = pattern.match(port.source.name) + arg_type = type_mapping.get(match.group(1), 'any') + arg_name = match.group(2) + + if arg_name not in existing_args: + args_code.append(ast.parse(f"{arg_name}: InArg[{arg_type}]").body[0]) + existing_args.add(arg_name) + + init_code.append(ast.parse(f"{target_indexed}.connect(self.{arg_name})")) + else: + # Literal -> `[i] = ` + lit_value = _get_value_from_literal_port(port) + assign = ast.parse(f"{target_indexed} = 0") + assign.body[0].value = ast.parse(repr(lit_value)).body[0].value + init_code.append(assign) - assignment_target = "%s.%s" % (named_nodes[ports[0].target.id], ports[0].varName) - tpl = set_value(assignment_target, assignment_value) - init_code.append(tpl) # Handle output connections for i, port in enumerate(p for p in finish_node.ports if p.dataType == 'dynalist'): From aaff6407a643276db5aba28163fee559dbf120d9 Mon Sep 17 00:00:00 2001 From: MFA-X-AI Date: Mon, 6 Oct 2025 11:47:32 +0900 Subject: [PATCH 2/2] simplify indexing implementation --- xai_components/base.py | 46 ++++++++++++++++-------------------------- 1 file changed, 17 insertions(+), 29 deletions(-) diff --git a/xai_components/base.py b/xai_components/base.py index 681bafb3..7738a55e 100644 --- a/xai_components/base.py +++ b/xai_components/base.py @@ -72,43 +72,31 @@ def __deepcopy__(self, memo): memo[id_self] = _copy return _copy - class _DynaIndex: + def _put_at_index(self, idx: int, obj): + v = self._value + if v is None: + v = [] + if isinstance(v, tuple): + v = list(v) + if idx >= len(v): + v.extend([None] * (idx + 1 - len(v))) + v[idx] = obj + self._value = v + + class _IndexProxy: def __init__(self, parent, idx): self._p = parent self._i = idx - - def _assign(self, obj): - v = self._p._value - # Start from empty container if needed - if v is None: - v = [] - # Internally first use list for convenience - is_tuple = isinstance(v, tuple) - if is_tuple: - v = list(v) - - while len(v) <= self._i: - v.append(None) - v[self._i] = obj - - # Convert back to tuple if the port uses dynatuple semantics - try_is_dynatuple = (self._p._getter is dynatuple.getter) - self._p._value = tuple(v) if try_is_dynatuple else v - def connect(self, ref): - # Accept InArg/OutArg or literal-like objects - self._assign(ref) - - def set(self, value): - self._assign(value) + self._p._put_at_index(self._i, ref) def __getitem__(self, idx): - # Allow indexing when the port holds a list/tuple (dynaports) - return InArg._DynaIndex(self, idx) + # Enables: port[i].connect(other_port) + return InArg._IndexProxy(self, idx) def __setitem__(self, idx, value): - # Support literal assignment: port[i] = 'a' - self[idx].set(value) + # Enables: port[i] = + self._put_at_index(idx, value) class InCompArg(Generic[T]): def __init__(self, value: T = None, getter: Callable[[T], any] = lambda x: x) -> None: