Start with causality inference

BEdit/src/bedit/core/bond_graph.py

@@ -0,0 +1,134 @@
+"""Bond-graph analysis hooks.
+
+This module intentionally has no GUI or simulation-engine dependencies.  The
+causality inference algorithm can grow here without coupling the document model
+to OpenModelica or Qt.
+"""
+
+from __future__ import annotations
+from typing import Any
+
+from bedit.core.application_log import get_logger
+
+log = get_logger(__name__)
+
+def infer_causality(component: dict[str, Any], toplevel: bool = True) -> dict[str, Any]:
+    """Infer power-connection causality in a serialized component tree.
+
+    This is currently a traversal stub: it ensures every power connection has a
+    causality value, while preserving causality already supplied by callers.
+    Future inference rules should assign ``source``, ``target``,
+    ``warn_source``, or ``warn_target`` here and return the same tree.
+
+    The input is mutated and returned so the composer receives the inferred
+    representation without needing a second document conversion.
+    """
+
+    if toplevel:
+        log.info("Causality inference")
+
+        # Reset causalities
+        reset_causality(component)
+
+    implementation = component.get("implementation", {})
+    graph = implementation.get("graph") if isinstance(implementation, dict) else None
+    if not isinstance(graph, dict):
+        return component
+    
+    id_list = build_id_list(component)
+
+    # Check fixed causality
+    for connection in graph.get("connections", []):
+        if isinstance(connection, dict) and connection.get("type") == "power":
+            source, target = get_ports_from_bond(connection, id_list)
+            log.info(source)
+            log.info(target)
+            if source.get('causality', 'indifferent') == 'fixed effort out':
+                connection['causality'] = 'target'
+            elif source.get('causality', 'indifferent') == 'fixed flow out':
+                connection['causality'] = 'source'
+
+    for block in graph.get("blocks", []):
+        if isinstance(block, dict):
+            infer_causality(block, False)
+
+    return component
+
+def reset_causality(component: dict[str, Any]) -> None:
+    implementation = component.get("implementation", {})
+    graph = implementation.get("graph") if isinstance(implementation, dict) else None
+    if not isinstance(graph, dict):
+        return
+    for connection in graph.get("connections", []):
+        if isinstance(connection, dict) and connection.get("type") == "power":
+            connection["causality"] = "none"
+    for block in graph.get("blocks", []):
+        if isinstance(block, dict):
+            reset_causality(block)
+
+def build_id_list(graph: dict[str, Any]) -> dict[str, Any]:
+    """Index all addressable objects in a component tree by their stable ID."""
+
+    id_list: dict[str, Any] = {}
+    id_kinds: dict[str, str] = {}
+
+    def add(item: dict[str, Any], description: str) -> None:
+        item_id = item.get("id")
+        if not item_id:
+            raise ValueError(f"{description} has no ID")
+        # Port IDs identify a port on a component definition and may therefore
+        # recur in cloned component instances. Component and junction IDs are
+        # document objects and must remain globally unique.
+        if item_id in id_list and not (
+            description == "port" and id_kinds[item_id] == "port"
+        ):
+            raise ValueError(f"Duplicate simulation object ID: {item_id}")
+        id_list[item_id] = item
+        id_kinds[item_id] = description
+
+    def visit(component: dict[str, Any]) -> None:
+        add(component, "component")
+        interface = component.get("interface", {})
+        for port in interface.get("ports", []):
+            add(port, "port")
+
+        implementation = component.get("implementation", {})
+        if implementation.get("kind") != "graph":
+            return
+        nested_graph = implementation.get("graph", {})
+        for junction in nested_graph.get("junctions", []):
+            add(junction, "junction")
+        for block in nested_graph.get("blocks", []):
+            visit(block)
+
+    visit(graph)
+    return id_list
+
+def get_ports_from_bond(bond: dict[str, Any], id_list: dict[str, Any]) -> tuple[dict[str,Any], dict[str,Any]]:
+    source = bond.get('source', None)
+    target = bond.get('target', None)
+    if source is None or target is None:
+        raise ValueError("Source and Target of a power bond cannot be None")
+    
+    source_component = id_list.get(source.get('block'), None)
+    target_component = id_list.get(target.get('block'), None)
+    if source_component is None or target_component is None:
+        raise ValueError("Source or Target blocks not found")
+    
+    def _find_port(ports: list[dict[str, Any]], port: str) -> dict[str, Any] | None:
+        for pi in ports:
+            if pi.get('id', '') == port:
+                return pi
+        return None
+
+    source_port = _find_port(source_component.get('interface', {}).get('ports', []), source.get('port'))
+    target_port = _find_port(target_component.get('interface', {}).get('ports', []), target.get('port'))
+    if source_port is None or target_port is None:
+        raise ValueError("Source or Target port not found")
+    
+    if source_port.get('type') != 'power':
+        raise ValueError("Source port is not a power port")
+    if target_port.get('type') != 'power':
+        raise ValueError("Target port is not a power port")
+    
+    return source_port, target_port