Bondgraph causality inference

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"
+]

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

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 "<root>"

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

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: