diff --git a/src/components/port/CustomDynaPortModel.tsx b/src/components/port/CustomDynaPortModel.tsx
index db32b61a..b89f7939 100644
--- a/src/components/port/CustomDynaPortModel.tsx
+++ b/src/components/port/CustomDynaPortModel.tsx
@@ -3,7 +3,7 @@ import { CustomPortModel, CustomPortModelOptions } from '../port/CustomPortModel
import { CustomNodeModel } from '../node/CustomNodeModel';
export const DYNAMIC_PARAMETER_NODE_TYPES = [
- 'dynalist', 'dynadict', 'dynatuple'
+ 'dynalist', 'dynatuple'
];
export interface DynaPortRef {
@@ -59,11 +59,6 @@ export class CustomDynaPortModel extends CustomPortModel {
return true; // Accepts anything
}
- // if thisLinkedPortType is dynadict, accept only dict
- if (thisLinkedPortType === 'dynadict' && thisNodeModelType !== 'dict') {
- return false;
- }
-
// default check
return super.isTypeCompatible(thisNodeModelType, thisLinkedPortType);
}
diff --git a/src/components/port/CustomPortLabel.tsx b/src/components/port/CustomPortLabel.tsx
index fa129eb9..a914366a 100644
--- a/src/components/port/CustomPortLabel.tsx
+++ b/src/components/port/CustomPortLabel.tsx
@@ -79,7 +79,6 @@ export class CustomPortLabel extends React.Component<CustomPortLabelProps> {
"dict": '{ }',
"dynalist": '«[]»',
"dynatuple": '«()»',
- "dynadict": '«{}»',
"union": ' U',
"secret": '🗝️',
"chat": '🗨',
diff --git a/src/components/port/CustomPortModel.ts b/src/components/port/CustomPortModel.ts
index 89afd5b7..2b833946 100644
--- a/src/components/port/CustomPortModel.ts
+++ b/src/components/port/CustomPortModel.ts
@@ -154,19 +154,35 @@ export class CustomPortModel extends DefaultPortModel {
return PARAMETER_NODE_TYPES.includes(nodeModelType);
}
+ static typeCompatibilityMap = {
+ "chat": ["list"],
+ "secret": ["string", "int", "float"],
+ };
+
isTypeCompatible(thisNodeModelType, dataType) {
- if(thisNodeModelType !== dataType){
- // Skip 'any' type check
- if(dataType === 'any'){
- return true;
- }
- // if multiple types are accepted by target node port, check if source port type is among them
- if(dataType.includes(thisNodeModelType)) {
+ // Check for direct compatibility or 'any' type
+ if (thisNodeModelType === dataType || dataType === 'any') {
+ return true;
+ }
+
+ // Check if the thisNodeModelType exists in the compatibility map
+ if (CustomPortModel.typeCompatibilityMap.hasOwnProperty(thisNodeModelType)) {
+ // Get the array of compatible data types for thisNodeModelType
+ const compatibleDataTypes = CustomPortModel.typeCompatibilityMap[thisNodeModelType];
+
+ // Check if dataType is in the array of compatible types
+ if (compatibleDataTypes.includes(dataType)) {
return true;
}
- return false; // types are incompatible
}
- return true;
+
+ // If multiple types are accepted by target node port, check if source port type is among them
+ if (dataType.includes(thisNodeModelType)) {
+ return true;
+ }
+
+ // If none of the above checks pass, the types are incompatible
+ return false;
}
canTriangleLinkToTriangle = (thisPort, port) => {
diff --git a/xai_components/base.py b/xai_components/base.py
index 953e1321..d3810ea2 100644
--- a/xai_components/base.py
+++ b/xai_components/base.py
@@ -1,40 +1,46 @@
from argparse import Namespace
-from typing import TypeVar, Generic, Tuple, NamedTuple, List
+from typing import TypeVar, Generic, Tuple, NamedTuple, Callable, List
T = TypeVar('T')
-
class InArg(Generic[T]):
- value: T
+ def __init__(self, value: T = None, getter: Callable[[T], any] = lambda x: x) -> None:
+ self._value = value
+ self._getter = getter
- def __init__(self, value: T) -> None:
- self.value = value
-
- @classmethod
- def empty(cls):
- return InArg(None)
+ @property
+ def value(self):
+ return self._getter(self._value)
+ @value.setter
+ def value(self, value: T):
+ self._value = value
class OutArg(Generic[T]):
- value: T
-
- def __init__(self, value: T) -> None:
+ def __init__(self, value: T = None, getter: Callable[[T], any] = lambda x: x) -> None:
self.value = value
+ self._getter = getter
- @classmethod
- def empty(cls):
- return OutArg(None)
+ @property
+ def value(self):
+ return self._getter(self._value)
-class InCompArg(Generic[T]):
- value: T
+ @value.setter
+ def value(self, value: T):
+ self._value = value
- def __init__(self, value: T) -> None:
+class InCompArg(Generic[T]):
+ def __init__(self, value: T = None, getter: Callable[[T], any] = lambda x: x) -> None:
self.value = value
+ self._getter = getter
- @classmethod
- def empty(cls):
- return InCompArg(None)
+ @property
+ def value(self):
+ return self._getter(self._value)
+ @value.setter
+ def value(self, value: T):
+ self._value = value
def xai_component(*args, **kwargs):
# Passthrough element without any changes.
@@ -54,19 +60,24 @@ class ExecutionContext:
def __init__(self, args: Namespace):
self.args = args
-
class BaseComponent:
-
def __init__(self):
all_ports = self.__annotations__
for key, type_arg in all_ports.items():
- if isinstance(type_arg, InArg[any].__class__):
- setattr(self, key, InArg.empty())
- elif isinstance(type_arg, InCompArg[any].__class__):
- setattr(self, key, InCompArg.empty())
- elif isinstance(type_arg, OutArg[any].__class__):
- setattr(self, key, OutArg.empty())
- elif type_arg == str(self.__class__):
+ port_class = type_arg.__origin__
+ port_type = type_arg.__args__[0]
+ if port_class in (InArg, InCompArg, OutArg):
+ if hasattr(port_type, 'initial_value'):
+ port_value = port_type.initial_value()
+ else:
+ port_value = None
+
+ if hasattr(port_type, 'getter'):
+ port_getter = port_type.getter
+ else:
+ port_getter = lambda x: x
+ setattr(self, key, port_class(port_value, port_getter))
+ else:
setattr(self, key, None)
@classmethod
@@ -79,7 +90,6 @@ def execute(self, ctx) -> None:
def do(self, ctx) -> Tuple[bool, 'BaseComponent']:
pass
-
class Component(BaseComponent):
next: BaseComponent
@@ -125,14 +135,8 @@ def execute_graph(args: Namespace, start: BaseComponent, ctx) -> None:
class secret:
+ pass
- def __init__(self, value):
- self.__value = value
-
- def get_value(self):
- return self.__value
-
-
class message(NamedTuple):
role: str
content: str
@@ -140,14 +144,26 @@ class message(NamedTuple):
class chat(NamedTuple):
messages: List[message]
-class dynalist:
- def __init__(self, value):
- self.value = value
-
-class dynatuple:
- def __init__(self, value):
- self.value = value
-
-class dynadict:
- def __init__(self, value):
- self.value = value
\ No newline at end of file
+class dynalist(list):
+ def __init__(self, *args):
+ super().__init__(args)
+
+ @staticmethod
+ def getter(x):
+ if x is None:
+ return []
+ return [item.value if isinstance(item, (InArg, OutArg)) else item for item in x]
+
+class dynatuple(tuple):
+ def __init__(self, *args):
+ super().__init__(args)
+ @staticmethod
+ def getter(x):
+ if x is None:
+ return tuple()
+ def resolve(item):
+ if isinstance(item, (InArg, InCompArg,OutArg)):
+ return item.value
+ else:
+ return item
+ return tuple(resolve(item) for item in x)
\ No newline at end of file
diff --git a/xai_components/xai_utils/utils.py b/xai_components/xai_utils/utils.py
index 207eabbc..066bc6b7 100644
--- a/xai_components/xai_utils/utils.py
+++ b/xai_components/xai_utils/utils.py
@@ -1,4 +1,4 @@
-from xai_components.base import InArg, OutArg, InCompArg, Component, xai_component
+from xai_components.base import InArg, OutArg, InCompArg, Component, xai_component, dynalist, dynatuple
import os
import sys
@@ -168,6 +168,68 @@ def execute(self, ctx) -> None:
print("Sleeping for " + str(sleep_timer) + " seconds.")
time.sleep(sleep_timer)
+@xai_component(color="grey")
+class MakeList(Component):
+ """
+ A component that takes values from its dynamic list port and output as a normal list.
+ ##### inPorts:
+ - list_values: Dynamic list port that can take any vars and append it in a list.
+
+ ##### outPorts:
+ - output_list: The constructed list from the dynamic list inPorts.
+ """
+ list_values: InArg[dynalist]
+ output_list: OutArg[list]
+
+ def execute(self, ctx) -> None:
+
+ self.output_list.value = self.list_values.value
+ print("Constructed List:", self.output_list.value)
+
+@xai_component(color="grey")
+class MakeTuple(Component):
+ """
+ A component that takes values from its dynamic tuple port and output as a normal tuple.
+ ##### inPorts:
+ - tuple_values: Dynamic tuple port that can take any vars and append it in a tuple.
+
+ ##### outPorts:
+ - output_tuple: The constructed tuple from the dynamic tuple inPorts.
+ """
+ tuple_values: InArg[dynatuple]
+ output_tuple: OutArg[tuple]
+
+ def execute(self, ctx) -> None:
+
+ self.output_tuple.value = self.tuple_values.value
+ print("Constructed Tuple:", self.output_tuple.value)
+
+@xai_component(color="grey")
+class MakeDict(Component):
+ """
+ A component that takes two dynamic lists (dynalists) as inputs - one for keys and one for values,
+ and constructs a dictionary from these lists. If there are more keys than values, the unmatched keys
+ will have None as their value.
+
+ ##### inPorts:
+ - keys_list: Dynamic list of keys for the dictionary.
+ - values_list: Dynamic list of values for the dictionary.
+
+ ##### outPorts:
+ - output_dict: The constructed dictionary from the provided keys and values.
+ """
+ keys_list: InArg[dynalist]
+ values_list: InArg[dynalist]
+ output_dict: OutArg[dict]
+
+ def execute(self, ctx) -> None:
+ keys = self.keys_list.value
+ values = self.values_list.value
+
+ constructed_dict = {key: values[i] if i < len(values) else None for i, key in enumerate(keys)}
+
+ self.output_dict.value = constructed_dict
+ print("Constructed Dictionary:", self.output_dict.value)
@xai_component(color="orange")
class ExecuteNotebook(Component):
diff --git a/xircuits/compiler/generator.py b/xircuits/compiler/generator.py
index 54a84d10..f794366a 100644
--- a/xircuits/compiler/generator.py
+++ b/xircuits/compiler/generator.py
@@ -1,5 +1,6 @@
import ast
import itertools
+from collections import namedtuple
import re
import json
import sys
@@ -156,8 +157,9 @@ def main(args):
# 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 = {}
- ports_by_varName = {}
+ RefOrValue = namedtuple('RefOrValue', ['value', 'is_ref']) # Renamed to RefOrValue
# Group ports by varName
for port in dynaports:
@@ -165,14 +167,12 @@ def main(args):
ports_by_varName[port.varName] = []
ports_by_varName[port.varName].append(port)
- # Iterate through grouped ports and append values
for varName, ports in ports_by_varName.items():
- appended_list = []
- merged_dict = {}
- convert_to_tuple = False
+ dynaport_values = []
for port in ports:
if port.source.id not in named_nodes:
+ # Handle Literals
if port.source.type == "string":
value = port.sourceLabel
elif port.source.type == "list":
@@ -180,37 +180,31 @@ def main(args):
elif port.source.type == "dict":
value = json.loads("{" + port.sourceLabel + "}")
elif port.source.type == "tuple":
- value = list(eval(port.sourceLabel))
+ value = tuple(eval(port.sourceLabel))
else:
value = eval(port.sourceLabel)
-
+ dynaport_values.append(RefOrValue(value, False))
else:
- value = "%s.%s" % (named_nodes[port.source.id], port.sourceLabel)
+ # Handle named node references
+ value = "%s.%s" % (named_nodes[port.source.id], port.sourceLabel) # Variable reference
+ dynaport_values.append(RefOrValue(value, True))
- if port.dataType == 'dynadict':
- merged_dict.update(value)
- else:
- appended_list.append(value)
-
- if port.dataType == 'dynatuple':
- convert_to_tuple = True
-
- # Create a single AST node for each unique varName and append to code
assignment_target = "%s.%s.value" % (named_nodes[ports[0].target.id], ports[0].varName)
-
- assignment_value = ''
-
- if merged_dict:
- assignment_value = repr(merged_dict)
- elif appended_list:
- if convert_to_tuple:
- assignment_value = repr(tuple(appended_list))
+
+ 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] + ',)'
else:
- assignment_value = repr(appended_list)
-
+ 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) + ']'
+
tpl = ast.parse("%s = %s" % (assignment_target, assignment_value))
code.append(tpl)
+
# Set up control flow
for node in component_nodes:
has_next = False
diff --git a/xircuits/compiler/port.py b/xircuits/compiler/port.py
index 303f95ec..c99f5647 100644
--- a/xircuits/compiler/port.py
+++ b/xircuits/compiler/port.py
@@ -13,4 +13,4 @@ class Port:
sourceLabel: str
direction: str
-DYNAMIC_PORTS = ['dynalist', 'dynadict', 'dynatuple']
+DYNAMIC_PORTS = ['dynalist', 'dynatuple']