From 8e7c463567d630138004dd3e7a63b5742a41d89b Mon Sep 17 00:00:00 2001 From: Joppe Blondel Date: Thu, 23 Jul 2026 16:02:14 +0200 Subject: [PATCH] Bondgraph causality inference --- src/bedit_core/bondgraph/__init__.py | 14 ++ src/bedit_core/bondgraph/causality.py | 206 ++++++++++++++++++++++++++ src/bedit_core/bondgraph/flatten.py | 119 +++++++++++++++ src/bedit_core/bondgraph/network.py | 60 ++++++++ src/bedit_core/models.py | 2 + 5 files changed, 401 insertions(+) create mode 100644 src/bedit_core/bondgraph/__init__.py create mode 100644 src/bedit_core/bondgraph/causality.py create mode 100644 src/bedit_core/bondgraph/flatten.py create mode 100644 src/bedit_core/bondgraph/network.py diff --git a/src/bedit_core/bondgraph/__init__.py b/src/bedit_core/bondgraph/__init__.py new file mode 100644 index 0000000..a4e2dd1 --- /dev/null +++ b/src/bedit_core/bondgraph/__init__.py @@ -0,0 +1,14 @@ +"""Bond-graph flattening and network representations.""" + +from .flatten import FlattenError, flatten_bondgraph +from .network import BondGraphNetwork, NetworkBond, NetworkPort +from .causality import causality_inference + +__all__ = [ + "BondGraphNetwork", + "FlattenError", + "NetworkBond", + "NetworkPort", + "flatten_bondgraph", + "causality_inference" +] diff --git a/src/bedit_core/bondgraph/causality.py b/src/bedit_core/bondgraph/causality.py new file mode 100644 index 0000000..4de4e71 --- /dev/null +++ b/src/bedit_core/bondgraph/causality.py @@ -0,0 +1,206 @@ +from bedit_core.models import Component, BondCausality, BondPort, PortCausality, BondConnection +from .network import BondGraphNetwork, NetworkPort, NetworkBond +from .flatten import flatten_bondgraph + +class InferenceError(ValueError): + pass + +def causality_inference(root: Component) -> Component: + """Performs causality inference on a component and its subcomponents and returns with a constrained one""" + engine = _CausalityEngine(root) + return engine.inference() + +class _CausalityEngine(): + def __init__(self, root: Component) -> None: + self._root = root + self._network = flatten_bondgraph(root) + + def inference(self): + self.clear_causalities() + + # Phase 1: fixed causalities + for port in self._network.ports: + if not isinstance(port.port, BondPort): + continue + if port.port.causality_preference == PortCausality.FIXED_EFFORT_OUT: + self.propagate_from_port(port, BondCausality.EFFORT_OUT) + elif port.port.causality_preference == PortCausality.FIXED_FLOW_OUT: + self.propagate_from_port(port, BondCausality.FLOW_OUT) + + # Phase 2: preferred causalities + for port in self._network.ports: + if not isinstance(port.port, BondPort): + continue + if self.is_fully_assigned(port): + continue + if port.port.causality_preference == PortCausality.PREFERRED_EFFORT_OUT: + self.propagate_from_port(port, BondCausality.EFFORT_OUT) + elif port.port.causality_preference == PortCausality.PREFERRED_FLOW_OUT: + self.propagate_from_port(port, BondCausality.FLOW_OUT) + + # Phase 2: soft choiced causalities + # May need several passed when a junction port has multiple resolved bonds + # Stop as soon as a pass makes no progress + while True: + unresolved_before = sum( + isinstance(con.connection, BondConnection) and con.connection.causality == BondCausality.NONE + for con in self._network.bonds + ) + for port in self._network.ports: + if not isinstance(port.port, BondPort): + continue + if self.is_fully_assigned(port): + continue + if port.port.causality_preference == PortCausality.LIKES_EFFORT_OUT: + self.propagate_from_port(port, BondCausality.EFFORT_OUT) + elif port.port.causality_preference == PortCausality.LIKES_FLOW_OUT: + self.propagate_from_port(port, BondCausality.FLOW_OUT) + elif port.port.causality_preference == PortCausality.SINGLE_EFFORT_IN: + self.evaluate_junction_constraints(port) + if not self.is_fully_assigned(port): + self.propagate_from_port(port, BondCausality.FLOW_OUT) + elif port.port.causality_preference == PortCausality.SINGLE_FLOW_IN: + self.evaluate_junction_constraints(port) + if not self.is_fully_assigned(port): + self.propagate_from_port(port, BondCausality.EFFORT_OUT) + elif port.port.causality_preference == PortCausality.INDIFFERENT: + # Force an arbirary assignment on the first unassigned bond + self.propagate_from_port(port, BondCausality.EFFORT_OUT) + unresolved_after = sum( + isinstance(con.connection, BondConnection) and con.connection.causality == BondCausality.NONE + for con in self._network.bonds + ) + if unresolved_after == 0 or unresolved_after >= unresolved_before: + break + + # Last check + for con in self._network.bonds: + if con.connection.causality == BondCausality.NONE: + raise InferenceError("System under-constrained: unresolved causal loops or disconnected elements remain") + + return self._root + + def clear_causalities(self) -> None: + for bond in self._network.bonds: + bond.connection.causality = BondCausality.NONE + + def propagate_from_port(self, port: NetworkPort, causality: BondCausality) -> None: + attached_connections = self._network.bonds_for(port) + # Get first unassigned connection on this port + target_conn = None + for conn in attached_connections: + if conn.connection.causality == BondCausality.NONE: + target_conn = conn + break + if target_conn is None: + # Nothing to resolve + return + # Update causality based on target_conns perspective + if target_conn.source.component == port.component: + target_conn.connection.causality = BondCausality.EFFORT_OUT if (causality == BondCausality.EFFORT_OUT) else BondCausality.FLOW_OUT + else: + target_conn.connection.causality = BondCausality.FLOW_OUT if (causality == BondCausality.EFFORT_OUT) else BondCausality.EFFORT_OUT + # Continue on the other side of the bond + self.propagate_to_neighbor(port.component, target_conn) + + def propagate_to_neighbor(self, component: Component, connection: NetworkBond) -> None: + # Get neighor + is_source = (component == connection.source.component) + neighbor = connection.target if is_source else connection.target + # Direct the evaluation based on what type of port it is + if neighbor.port.causality_preference in [PortCausality.SINGLE_EFFORT_IN, PortCausality.SINGLE_FLOW_IN]: + self.evaluate_junction_constraints(neighbor) + else: + self.verify_component_compatibility(neighbor) + + def evaluate_junction_constraints(self, port: NetworkPort) -> None: + efforts_in = 0 + flows_in = 0 + unassigned_conns: list[NetworkBond] = [] + + for con in self._network.bonds_for(port): + if con.connection.causality == BondCausality.NONE: + unassigned_conns.append(con) + continue + is_source = (con.source.component == port.component) + if (is_source and con.connection.causality == BondCausality.FLOW_OUT) or (not is_source and con.connection.causality == BondCausality.EFFORT_OUT): + efforts_in += 1 + else: + flows_in += 1 + + # 0 junction port + if port.port.causality_preference == PortCausality.SINGLE_EFFORT_IN: + # Error check + if efforts_in > 1: + raise InferenceError(f"Critical causality conflict: multiple ports are dictating effort to 0-junction port: {port.component.name}") + + # Propagation A: 1 effort is comming in so remaining connections are effort out + if efforts_in == 1 and len(unassigned_conns) > 0: + # Force effort out to the rest + for con in unassigned_conns: + desired = BondCausality.EFFORT_OUT if (con.source.component == port.component) else BondCausality.FLOW_OUT + con.connection.causality = desired + self.propagate_to_neighbor(port.component, con) + + # Propagation B: no effort is coming in yet and only one remaining so must be effort in + elif efforts_in == 0 and len(unassigned_conns) == 1: + con = unassigned_conns[0] + desired = BondCausality.FLOW_OUT if (con.source.component == port.component) else BondCausality.EFFORT_OUT + con.connection.causality = desired + self.propagate_to_neighbor(port.component, con) + + # 1 junction port + elif port.port.causality_preference == PortCausality.SINGLE_FLOW_IN: + # Error check + if flows_in > 1: + raise InferenceError(f"Critical causality conflict: multiple ports are dictating flow to 1-junction port: {port.component.name}") + + # Propagation A: 1 flow is comming in so remaining connections are effort in + if flows_in == 1 and len(unassigned_conns) > 0: + # Force flow out to the rest + for con in unassigned_conns: + desired = BondCausality.FLOW_OUT if (con.source.component == port.component) else BondCausality.EFFORT_OUT + con.connection.causality = desired + self.propagate_to_neighbor(port.component, con) + + # Propagation B: no flow is coming in yet and only one remaining so must be flow in + elif flows_in == 0 and len(unassigned_conns) == 1: + con = unassigned_conns[0] + desired = BondCausality.EFFORT_OUT if (con.source.component == port.component) else BondCausality.FLOW_OUT + con.connection.causality = desired + self.propagate_to_neighbor(port.component, con) + + def verify_component_compatibility(self, port: NetworkPort) -> None: + conn = None + for c in self._network.bonds_for(port): + if c.connection.causality != BondCausality.NONE: + conn = c + break + if conn is None: + return + + # Figure out what causality was pushed onto this port from the outside world + state = conn.connection.causality + is_source = (conn.source.component == port.component) + actual_causality = BondCausality.NONE + if (is_source and state == BondCausality.EFFORT_OUT) or (not is_source and state == BondCausality.FLOW_OUT): + actual_causality = BondCausality.EFFORT_OUT + else: + actual_causality = BondCausality.FLOW_OUT + + if port.port.causality_preference == PortCausality.FIXED_EFFORT_OUT and actual_causality == BondCausality.FLOW_OUT: + raise InferenceError(f"Critical source conflict: Fixed effort source {port.component.name} forced into an input state") + if port.port.causality_preference == PortCausality.FIXED_FLOW_OUT and actual_causality == BondCausality.EFFORT_OUT: + raise InferenceError(f"Critical source conflict: Fixed flow source {port.component.name} forced into an input state") + + # Non-preferenced causality checks + if port.port.causality_preference == PortCausality.PREFERRED_EFFORT_OUT and actual_causality == BondCausality.FLOW_OUT: + conn.connection.undesired = True + if port.port.causality_preference == PortCausality.PREFERRED_FLOW_OUT and actual_causality == BondCausality.EFFORT_OUT: + conn.connection.undesired = True + + def is_fully_assigned(self, port: NetworkPort) -> bool: + for con in self._network.bonds_for(port): + if con.connection.causality == BondCausality.NONE: + return False + return True diff --git a/src/bedit_core/bondgraph/flatten.py b/src/bedit_core/bondgraph/flatten.py new file mode 100644 index 0000000..c2ac49b --- /dev/null +++ b/src/bedit_core/bondgraph/flatten.py @@ -0,0 +1,119 @@ +"""Flatten hierarchical graph implementations into one bond-graph network.""" + +from __future__ import annotations + +from bedit_core.models import ( + BondConnection, + BondPort, + Component, + ComponentID, + GraphImplementation, + PortID, +) + +from .network import BondGraphNetwork, ComponentPath, NetworkBond, NetworkPort + + +class FlattenError(ValueError): + """Raised when a hierarchical bond graph cannot be resolved.""" + + +def flatten_bondgraph(root: Component) -> BondGraphNetwork: + """Return a flat network containing every nested bond-graph element. + + Graph interface ports are represented once and reused by bonds on either + side of the component boundary, making the full hierarchy connected. + """ + return _NetworkBuilder(root).build() + + +class _NetworkBuilder: + def __init__(self, root: Component) -> None: + self._root = root + self._network = BondGraphNetwork(root=root) + self._ports: dict[tuple[ComponentPath, PortID], NetworkPort] = {} + + def build(self) -> BondGraphNetwork: + """Traverse the hierarchy and return the completed network.""" + self._visit_component(self._root, ()) + return self._network + + def _visit_component( + self, + component: Component, + path: ComponentPath, + ) -> None: + self._network.components.append(component) + self._register_ports(component, path) + + if not isinstance(component.implementation, GraphImplementation): + return + + graph = component.implementation.graph + for component_id, child in graph.components.items(): + self._visit_component(child, path + (component_id,)) + + for connection_id, connection in graph.connections.items(): + if not isinstance(connection, BondConnection): + continue + self._network.bonds.append( + NetworkBond( + component_path=path, + connection_id=connection_id, + connection=connection, + source=self._resolve_port( + path, + graph.components, + connection.source, + ), + target=self._resolve_port( + path, + graph.components, + connection.target, + ), + ) + ) + + def _register_ports( + self, + component: Component, + path: ComponentPath, + ) -> None: + for port_id, port in component.interface.ports.items(): + if not isinstance(port, BondPort): + continue + network_port = NetworkPort(path, component, port_id, port) + self._ports[(path, port_id)] = network_port + self._network.ports.append(network_port) + + def _resolve_port( + self, + graph_path: ComponentPath, + children: dict[ComponentID, Component], + port_id: PortID, + ) -> NetworkPort: + candidate_paths = [graph_path] + candidate_paths.extend( + graph_path + (component_id,) + for component_id in children + ) + matches = [ + self._ports[(path, port_id)] + for path in candidate_paths + if (path, port_id) in self._ports + ] + if len(matches) == 1: + return matches[0] + if not matches: + raise FlattenError( + f"port {port_id!s} is not present in graph at " + f"{_format_path(graph_path)}" + ) + raise FlattenError( + f"port {port_id!s} is ambiguous in graph at " + f"{_format_path(graph_path)}; port IDs must be unique within a graph" + ) + + +def _format_path(path: ComponentPath) -> str: + return "/".join(map(str, path)) if path else "" diff --git a/src/bedit_core/bondgraph/network.py b/src/bedit_core/bondgraph/network.py new file mode 100644 index 0000000..0d5eebf --- /dev/null +++ b/src/bedit_core/bondgraph/network.py @@ -0,0 +1,60 @@ +"""Flat bond-graph network types used by analysis algorithms.""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from bedit_core.models import ( + BondConnection, + BondPort, + Component, + ComponentID, + ConnectionID, + PortID, +) + +ComponentPath = tuple[ComponentID, ...] + + +@dataclass(eq=False) +class NetworkPort: + """A bond port together with its owning component and hierarchical path.""" + + component_path: ComponentPath + component: Component + port_id: PortID + port: BondPort + + +@dataclass +class NetworkBond: + """A bond connection whose endpoints are resolved to flattened ports.""" + + component_path: ComponentPath + connection_id: ConnectionID + connection: BondConnection + source: NetworkPort + target: NetworkPort + + +@dataclass +class BondGraphNetwork: + """All components, bond ports, and bonds in a component hierarchy.""" + + root: Component + components: list[Component] = field(default_factory=list) + ports: list[NetworkPort] = field(default_factory=list) + bonds: list[NetworkBond] = field(default_factory=list) + + def bonds_for(self, port: NetworkPort) -> list[NetworkBond]: + """Return all bonds incident to ``port``.""" + return [ + bond + for bond in self.bonds + if bond.source is port or bond.target is port + ] + + @property + def connections(self) -> list[NetworkBond]: + """Alias for ``bonds`` for callers using connection terminology.""" + return self.bonds diff --git a/src/bedit_core/models.py b/src/bedit_core/models.py index aa169d2..0d197e1 100644 --- a/src/bedit_core/models.py +++ b/src/bedit_core/models.py @@ -46,6 +46,8 @@ class PortCausality(Enum): PREFERRED_FLOW_OUT = "preferred_flow_out" LIKES_EFFORT_OUT = "likes_effort_out" LIKES_FLOW_OUT = "likes_flow_out" + SINGLE_EFFORT_IN = "single_effort_in" + SINGLE_FLOW_IN = "single_flow_in" @dataclass class Document: