Basic implied paths

src/faebryk/core/cpp/__init__.py

@@ -3,6 +3,7 @@
 
 import json
 import logging
+import subprocess
 from importlib.metadata import Distribution
 
 from faebryk.libs.util import ConfigFlag, at_exit, global_lock
@@ -115,6 +116,7 @@ def compile_and_load():
                 is_pyi=True,
             )
         )
+        subprocess.check_output(["ruff", "check", "--fix", pyi_out])
 
 
 # Re-export c++ with type hints provided by __init__.pyi

src/faebryk/core/cpp/__init__.pyi

@@ -121,8 +121,8 @@ class LinkDirectConditionalFilterResult(enum.Enum):
 
     FILTER_FAIL_UNRECOVERABLE = 2
 
-class LinkDirectDerived(LinkDirect):
-    pass
+class LinkDirectDerived(LinkDirectConditional):
+    def __init__(self, arg: Sequence[GraphInterface], /) -> None: ...
 
 class LinkFilteredException(Exception):
     pass

src/faebryk/core/cpp/src/main.cpp

@@ -145,7 +145,8 @@ PYMOD(m) {
     nb::class_<LinkDirectConditional, LinkDirect>(m, "LinkDirectConditional")
         .def(nb::init<LinkDirectConditional::FilterF, bool>(), "filter"_a,
              "needs_only_first_in_path"_a = false);
-    nb::class_<LinkDirectDerived, LinkDirect>(m, "LinkDirectDerived");
+    nb::class_<LinkDirectDerived, LinkDirectConditional>(m, "LinkDirectDerived")
+        .def(nb::init<Path>());
 
     nb::exception<LinkDirectConditional::LinkFilteredException>(m,
                                                                 "LinkFilteredException");

src/faebryk/core/moduleinterface.py

@@ -1,6 +1,7 @@
 # This file is part of the faebryk project
 # SPDX-License-Identifier: MIT
 import logging
+from itertools import pairwise
 from typing import (
     Sequence,
     cast,
@@ -13,19 +14,24 @@
 )
 from faebryk.core.graphinterface import GraphInterface
 from faebryk.core.link import (
+    Link,
     LinkDirect,
     LinkDirectConditional,
     LinkDirectConditionalFilterResult,
+    LinkDirectDerived,
 )
 from faebryk.core.node import CNode, Node, NodeException
 from faebryk.core.pathfinder import Path, find_paths
 from faebryk.core.trait import Trait
 from faebryk.library.can_specialize import can_specialize
-from faebryk.libs.util import cast_assert, once
+from faebryk.libs.util import ConfigFlag, cast_assert, groupby, once, times
 
 logger = logging.getLogger(__name__)
 
 
+IMPLIED_PATHS = ConfigFlag("IMPLIED_PATHS", default=False, descr="Use implied paths")
+
+
 class ModuleInterface(Node):
     class TraitT(Trait): ...
 
@@ -68,7 +74,9 @@ def __init__(self):
 
     def __preinit__(self) -> None: ...
 
-    def connect(self: Self, *other: Self, linkcls=None) -> Self:
+    def connect(
+        self: Self, *other: Self, linkcls: type[Link] | Link | None = None
+    ) -> Self:
         if not {type(o) for o in other}.issubset({type(self)}):
             raise NodeException(
                 self,
@@ -85,9 +93,15 @@ def connect(self: Self, *other: Self, linkcls=None) -> Self:
             self.connected.connect([o.connected for o in other])
             return ret
 
-        # TODO: give link a proper copy constructor
-        for o in other:
-            self.connected.connect(o.connected, link=linkcls())
+        # TODO: give link a proper copy constructor instead
+        if isinstance(linkcls, Link):
+            assert len(other) <= 1
+            links = [linkcls]
+        else:
+            links = times(len(other), linkcls)
+
+        for o, link in zip(other, links):
+            self.connected.connect(o.connected, link=link)
 
         return ret
 
@@ -108,7 +122,17 @@ def connect_shallow(self, *other: Self) -> Self:
 
     def get_connected(self) -> dict[Self, Path]:
         paths = find_paths(self, [])
-        return {cast_assert(type(self), p[-1].node): p for p in paths}
+        # TODO theoretically we could get multiple paths for the same MIF
+        # practically this won't happen in the current implementation
+        paths_per_mif = groupby(paths, lambda p: cast_assert(type(self), p[-1].node))
+
+        def choose_path(_paths: list[Path]) -> Path:
+            return self._path_with_least_conditionals(_paths)
+
+        path_per_mif = {mif: choose_path(paths) for mif, paths in paths_per_mif.items()}
+        for mif, paths in paths_per_mif.items():
+            self._connect_via_implied_paths(mif, paths)
+        return path_per_mif
 
     def is_connected_to(self, other: "ModuleInterface") -> list[Path]:
         return [
@@ -146,3 +170,35 @@ def __init_subclass__(cls, *, init: bool = True) -> None:
             raise TypeError("Overriding _on_connect is deprecated")
 
         return super().__init_subclass__(init=init)
+
+    @staticmethod
+    def _path_with_least_conditionals(paths: list["Path"]) -> "Path":
+        if len(paths) == 1:
+            return paths[0]
+
+        paths_links = [
+            (path, [e1.is_connected_to(e2) for e1, e2 in pairwise(path)])
+            for path in paths
+        ]
+        paths_conditionals = [
+            (
+                path,
+                [link for link in links if isinstance(link, LinkDirectConditional)],
+            )
+            for path, links in paths_links
+        ]
+        path = min(paths_conditionals, key=lambda x: len(x[1]))[0]
+        return path
+
+    def _connect_via_implied_paths(self, other: Self, paths: list["Path"]):
+        if not IMPLIED_PATHS:
+            return
+
+        if self.connected.is_connected_to(other.connected):
+            # TODO link resolution
+            return
+
+        # heuristic: choose path with fewest conditionals
+        path = self._path_with_least_conditionals(paths)
+
+        self.connect(other, linkcls=LinkDirectDerived(path))

test/core/test_hierarchy_connect.py

@@ -13,7 +13,7 @@
     LinkDirectDerived,
 )
 from faebryk.core.module import Module
-from faebryk.core.moduleinterface import ModuleInterface
+from faebryk.core.moduleinterface import IMPLIED_PATHS, ModuleInterface
 from faebryk.libs.app.erc import ERCPowerSourcesShortedError, simple_erc
 from faebryk.libs.library import L
 from faebryk.libs.util import times
@@ -336,6 +336,7 @@ def test_isolated_connect_erc():
     assert not a1.sda.reference.is_connected_to(b1.sda.reference)
 
 
+@pytest.mark.skipif(not IMPLIED_PATHS, reason="IMPLIED_PATHS is not set")
 def test_direct_implied_paths():
     powers = times(2, F.ElectricPower)
 
@@ -351,6 +352,7 @@ def test_direct_implied_paths():
     assert isinstance(path[1].is_connected_to(path[2]), LinkDirectDerived)
 
 
+@pytest.mark.skipif(not IMPLIED_PATHS, reason="IMPLIED_PATHS is not set")
 def test_children_implied_paths():
     powers = times(3, F.ElectricPower)
 
@@ -367,6 +369,7 @@ def test_children_implied_paths():
     assert isinstance(paths[0][1].is_connected_to(paths[0][2]), LinkDirectDerived)
 
 
+@pytest.mark.skipif(not IMPLIED_PATHS, reason="IMPLIED_PATHS is not set")
 def test_shallow_implied_paths():
     powers = times(4, F.ElectricPower)