@dataclass
class OpenHypergraph:
"""An open hypergraph with input and output nodes."""
nodes: list[Node] = field(default_factory=list)
edges: list[HyperEdge] = field(default_factory=list)
input_nodes: list[int] = field(default_factory=list)
output_nodes: list[int] = field(default_factory=list)
def flatten(self, vertex_label_map=None, edge_label_map=None):
"""Construct array representation used in data parallel approach"""
## Convert readable string labels to compact integer labels
if vertex_label_map is None:
vertex_label_map = constructLabelMap([v.label for v in self.nodes])
if edge_label_map is None:
edge_label_map = constructLabelMap([e.label for e in self.edges])
(
node_labels,
node_sources,
node_targets,
source_ports,
target_ports,
) = self.flatten_elements(
self.nodes, len(self.nodes), vertex_label_map, nodes=True
)
edge_labels, edge_sources, edge_targets, _, _ = self.flatten_elements(
self.edges, len(self.edges), edge_label_map, nodes=False
)
num_inputs = len(self.input_nodes)
global_interface = self.input_nodes + self.output_nodes
return FlatHypergraph(
node_labels,
node_sources,
source_ports,
node_targets,
target_ports,
edge_labels,
edge_sources,
edge_targets,
num_inputs,
global_interface,
)
def flatten_elements(self, elements, N, label_map, nodes=True):
labels = [0] * N
total_sources = sum([len(x.sources) for x in elements])
total_targets = sum([len(x.targets) for x in elements])
sources = SegmentedArray.reserving(N, total_sources)
targets = SegmentedArray.reserving(N, total_targets)
s_ports = SegmentedArray.reserving(N, total_sources)
t_ports = SegmentedArray.reserving(N, total_targets)
source_count = 0
target_count = 0
for i in range(N):
labels[i] = label_map[elements[i].label]
sources.initials[i] = source_count
targets.initials[i] = target_count
s_ports.initials[i] = source_count
t_ports.initials[i] = target_count
for j in range(len(elements[i].sources)):
src = elements[i].sources[j]
if nodes:
sources.elements[source_count] = src.index
s_ports.elements[source_count] = src.port
else:
sources.elements[source_count] = src
source_count += 1
for j in range(len(elements[i].targets)):
tgt = elements[i].targets[j]
if nodes:
targets.elements[target_count] = tgt.index
t_ports.elements[target_count] = tgt.port
else:
targets.elements[target_count] = tgt
target_count += 1
sources.sizes[i] = source_count - sources.initials[i]
targets.sizes[i] = target_count - targets.initials[i]
return labels, sources, targets, s_ports, t_ports
# TODO: Improve efficiency by caching results and invalidating on changes
def is_valid(self) -> bool:
"""Check if the hypergraph is valid."""
if not self.nodes:
return False
if not self.edges:
return False
return True
def check_nodes_in_graph(self, nodes) -> bool:
"""Check if all nodes are in the hypergraph."""
return all(node < len(self.nodes) for node in nodes)
def set_next_prev(self, edge: HyperEdge):
"""Set the next and previous edges for nodes based on edges in the hypergraph."""
for i, v in enumerate(edge.sources):
node = self.nodes[v]
node.targets.append(EdgeInfo(edge.index, i, edge.label))
# else:
# raise ValueError(
# f"Source node {node.label} of edge {edge.label} already has a next edge. This is not currently supported."
# )
for i, v in enumerate(edge.targets):
node = self.nodes[v]
node.sources.append(EdgeInfo(edge.index, i, edge.label))
# else:
# raise ValueError(
# f"Target node {node.label} of edge {edge.label} already has a previous edge. This is not currently supported."
# )
def __post_init__(self):
for edge in self.edges:
if not self.check_nodes_in_graph(edge.sources + edge.targets):
raise ValueError(f"Edge {edge.label} has nodes not in hypergraph nodes")
self.set_next_prev(edge)