fix: Handle Operational Capabilities

capellambse_context_diagrams/collectors/dataflow_view.py

@@ -5,35 +5,124 @@
 from __future__ import annotations
 
 import collections.abc as cabc
+import functools
+import operator
 import typing as t
 
 from capellambse import helpers
-from capellambse.model import common
+from capellambse.model import common, modeltypes
 from capellambse.model.crosslayer import fa
 from capellambse.model.layers import ctx, oa
 
 from .. import _elkjs, context
 from . import default, generic, makers, portless
 
 
+def only_involved(
+    exchanges: cabc.Iterable[fa.FunctionalExchange],
+    functions: cabc.Iterable[fa.FunctionalExchange],
+    attributes: tuple[str, str],
+) -> cabc.Iterable[fa.FunctionalExchange]:
+    src_attr, trg_attr = attributes
+    src_getter = operator.attrgetter(src_attr)
+    trg_getter = operator.attrgetter(trg_attr)
+    return [
+        ex
+        for ex in exchanges
+        if src_getter(ex) in functions and trg_getter(ex) in functions
+    ]
+
+
 def collector(
     diagram: context.ContextDiagram,
     params: dict[str, t.Any],
-    attribute: str = "involved_functions",
     exchange_filter: cabc.Callable[
-        [cabc.Iterable[fa.FunctionalExchange]],
+        [
+            cabc.Iterable[fa.FunctionalExchange],
+            cabc.Iterable[fa.FunctionalExchange],
+            tuple[str, str],
+        ],
         cabc.Iterable[fa.FunctionalExchange],
-    ] = lambda exs: exs,
+    ] = only_involved,
+) -> _elkjs.ELKInputData:
+    return COLLECTORS[diagram.type](diagram, params, exchange_filter)
+
+
+def collector_portless(
+    diagram: context.ContextDiagram,
+    params: dict[str, t.Any],
+    exchange_filter: cabc.Callable[
+        [
+            cabc.Iterable[fa.FunctionalExchange],
+            cabc.Iterable[fa.FunctionalExchange],
+            tuple[str, str],
+        ],
+        cabc.Iterable[fa.FunctionalExchange],
+    ],
+    attribute: str = "involved_activities",
+) -> _elkjs.ELKInputData:
+    data = makers.make_diagram(diagram)
+    activities = getattr(diagram.target, attribute)
+    filter = functools.partial(
+        exchange_filter,
+        functions=activities,
+        attributes=("source", "target"),
+    )
+    made_edges: set[str] = set()
+    for act in activities:
+        data["children"].append(act_box := makers.make_box(act))
+        connections = list(portless.get_exchanges(act, filter=filter))
+
+        in_act: dict[str, oa.OperationalActivity] = {}
+        out_act: dict[str, oa.OperationalActivity] = {}
+        for edge in connections:
+            if edge.source == act:
+                out_act.setdefault(edge.source.uuid, edge.source)
+            else:
+                in_act.setdefault(edge.target.uuid, edge.target)
+
+        act_box["height"] += (
+            makers.PORT_SIZE + 2 * makers.PORT_PADDING
+        ) * max(len(in_act), len(out_act))
+
+        ex_datas: list[generic.ExchangeData] = []
+        for ex in connections:
+            if ex.uuid in made_edges:
+                continue
+
+            ex_data = generic.ExchangeData(ex, data, diagram.filters, params)
+            generic.exchange_data_collector(ex_data)
+            made_edges.add(ex.uuid)
+            ex_datas.append(ex_data)
+
+    return data
+
+
+def collector_default(
+    diagram: context.ContextDiagram,
+    params: dict[str, t.Any],
+    exchange_filter: cabc.Callable[
+        [
+            cabc.Iterable[fa.FunctionalExchange],
+            cabc.Iterable[fa.FunctionalExchange],
+            tuple[str, str],
+        ],
+        cabc.Iterable[fa.FunctionalExchange],
+    ],
+    attribute: str = "involved_functions",
 ) -> _elkjs.ELKInputData:
     data = makers.make_diagram(diagram)
     functions = getattr(diagram.target, attribute)
+    filter = functools.partial(
+        exchange_filter,
+        functions=functions,
+        attributes=("source.owner", "target.owner"),
+    )
     made_edges: set[str] = set()
     for fnc in functions:
         data["children"].append(fnc_box := makers.make_box(fnc))
         _ports = default.port_collector(fnc, diagram.type)
-        connections = default.port_exchange_collector(
-            _ports, filter=exchange_filter
-        )
+        connections = default.port_exchange_collector(_ports, filter=filter)
         in_ports: dict[str, fa.FunctionPort] = {}
         out_ports: dict[str, fa.FunctionPort] = {}
         for edge in connections:
@@ -61,12 +150,7 @@ def collector(
     return data
 
 
-def only_involved(
-    exchanges: cabc.Iterable[fa.FunctionalExchange],
-    functions: cabc.Iterable[fa.FunctionalExchange],
-) -> cabc.Iterable[fa.FunctionalExchange]:
-    return [
-        ex
-        for ex in exchanges
-        if ex.source.owner in functions and ex.target.owner in functions
-    ]
+COLLECTORS: dict[modeltypes.DiagramType, cabc.Callable] = {
+    modeltypes.DiagramType.OAIB: collector_portless,
+    modeltypes.DiagramType.SDFB: collector_default,
+}

capellambse_context_diagrams/collectors/portless.py

@@ -7,6 +7,7 @@
 """
 from __future__ import annotations
 
+import collections.abc as cabc
 import typing as t
 from itertools import chain
 
@@ -145,6 +146,10 @@ def context_collector(
 
 def get_exchanges(
     obj: common.GenericElement,
+    filter: cabc.Callable[
+        [cabc.Iterable[common.GenericElement]],
+        cabc.Iterable[common.GenericElement],
+    ] = lambda i: i,
 ) -> t.Iterator[common.GenericElement]:
     """Yield exchanges safely.
 
@@ -185,4 +190,5 @@ def get_exchanges(
     elif is_capability:
         exchanges += [obj.component_involvements, obj.incoming_exploitations]
 
-    yield from {i.uuid: i for i in chain.from_iterable(exchanges)}.values()
+    filtered = filter(chain.from_iterable(exchanges))
+    yield from {i.uuid: i for i in filtered}.values()

capellambse_context_diagrams/context.py

@@ -525,13 +525,7 @@ def name(self) -> str:  # type: ignore
         return f"DatFlow view of {self.target.name}"
 
     def _create_diagram(self, params: dict[str, t.Any]) -> cdiagram.Diagram:
-        filter = functools.partial(
-            dataflow_view.only_involved,
-            functions=self.target.involved_functions,
-        )
-        params["elkdata"] = dataflow_view.collector(
-            self, params, exchange_filter=filter
-        )
+        params["elkdata"] = dataflow_view.collector(self, params)
         return super()._create_diagram(params)