diff --git a/capellambse_context_diagrams/__init__.py b/capellambse_context_diagrams/__init__.py
index 0c53da73..da41a9b8 100644
--- a/capellambse_context_diagrams/__init__.py
+++ b/capellambse_context_diagrams/__init__.py
@@ -60,7 +60,11 @@ def register_classes() -> None:
"""Add the `context_diagram` property to the relevant model objects."""
supported_classes: list[SupportedClass] = [
(oa.Entity, DiagramType.OAB, {}),
- (oa.OperationalActivity, DiagramType.OAB, {}),
+ (
+ oa.OperationalActivity,
+ DiagramType.OAB,
+ {"display_parent_relation": True},
+ ),
(oa.OperationalCapability, DiagramType.OCB, {}),
(ctx.Mission, DiagramType.MCB, {}),
(ctx.Capability, DiagramType.MCB, {"display_symbols_as_boxes": False}),
@@ -69,32 +73,41 @@ def register_classes() -> None:
DiagramType.SAB,
{
"display_symbols_as_boxes": True,
+ "display_parent_relation": True,
+ "include_inner_objects": True,
"render_styles": styling.BLUE_ACTOR_FNCS,
},
),
(
ctx.SystemFunction,
- DiagramType.SDFB,
+ DiagramType.SAB,
{"render_styles": styling.BLUE_ACTOR_FNCS},
),
(
la.LogicalComponent,
DiagramType.LAB,
- {"render_styles": styling.BLUE_ACTOR_FNCS},
+ {
+ "display_parent_relation": True,
+ "include_inner_objects": True,
+ "render_styles": styling.BLUE_ACTOR_FNCS,
+ },
),
(
la.LogicalFunction,
- DiagramType.LDFB,
+ DiagramType.LAB,
{"render_styles": styling.BLUE_ACTOR_FNCS},
),
(
pa.PhysicalComponent,
DiagramType.PAB,
- {"render_styles": styling.BLUE_ACTOR_FNCS},
+ {
+ "display_parent_relation": True,
+ "render_styles": styling.BLUE_ACTOR_FNCS,
+ },
),
(
pa.PhysicalFunction,
- DiagramType.PDFB,
+ DiagramType.PAB,
{"render_styles": styling.BLUE_ACTOR_FNCS},
),
]
diff --git a/capellambse_context_diagrams/collectors/dataflow_view.py b/capellambse_context_diagrams/collectors/dataflow_view.py
index e706f2cb..906b217f 100644
--- a/capellambse_context_diagrams/collectors/dataflow_view.py
+++ b/capellambse_context_diagrams/collectors/dataflow_view.py
@@ -8,6 +8,7 @@
import functools
import operator
import typing as t
+from itertools import chain
from capellambse.model import modeltypes
from capellambse.model.crosslayer import fa
@@ -128,7 +129,7 @@ def collector_default(
connections = default.port_exchange_collector(_ports, filter=filter)
in_ports: dict[str, fa.FunctionPort] = {}
out_ports: dict[str, fa.FunctionPort] = {}
- for edge in connections:
+ for edge in (edges := list(chain.from_iterable(connections.values()))):
if edge.source.owner == fnc:
out_ports.setdefault(edge.source.uuid, edge.source)
else:
@@ -142,7 +143,7 @@ def collector_default(
) * max(len(in_ports), len(out_ports))
ex_datas: list[generic.ExchangeData] = []
- for ex in connections:
+ for ex in edges:
if ex.uuid in made_edges:
continue
diff --git a/capellambse_context_diagrams/collectors/default.py b/capellambse_context_diagrams/collectors/default.py
index fbe4de4b..7c1f28a9 100644
--- a/capellambse_context_diagrams/collectors/default.py
+++ b/capellambse_context_diagrams/collectors/default.py
@@ -8,6 +8,7 @@
import collections.abc as cabc
import typing as t
+from itertools import chain
from capellambse import helpers
from capellambse.model import common
@@ -22,13 +23,22 @@ def collector(
diagram: context.ContextDiagram, params: dict[str, t.Any] | None = None
) -> _elkjs.ELKInputData:
"""Collect context data from ports of centric box."""
+ diagram.display_parent_relation = (params or {}).pop(
+ "display_parent_relation", diagram.display_parent_relation
+ )
+ diagram.include_inner_objects = (params or {}).pop(
+ "include_inner_objects", diagram.include_inner_objects
+ )
data = generic.collector(diagram, no_symbol=True)
ports = port_collector(diagram.target, diagram.type)
centerbox = data["children"][0]
- centerbox["ports"] = [makers.make_port(i.uuid) for i in ports]
connections = port_exchange_collector(ports)
+ centerbox["ports"] = [
+ makers.make_port(uuid) for uuid, edges in connections.items() if edges
+ ]
ex_datas: list[generic.ExchangeData] = []
- for ex in connections:
+ edges: common.ElementList[fa.AbstractExchange]
+ for ex in (edges := list(chain.from_iterable(connections.values()))):
if is_hierarchical := exchanges.is_hierarchical(ex, centerbox):
if not diagram.include_inner_objects:
continue
@@ -46,21 +56,22 @@ def collector(
continue
global_boxes = {centerbox["id"]: centerbox}
- if diagram.display_parent_relation:
+ made_boxes = {centerbox["id"]: centerbox}
+ if diagram.display_parent_relation and diagram.target.owner:
box = makers.make_box(
- diagram.target.parent,
+ diagram.target.owner,
no_symbol=diagram.display_symbols_as_boxes,
layout_options=makers.DEFAULT_LABEL_LAYOUT_OPTIONS,
)
box["children"] = [centerbox]
del data["children"][0]
- global_boxes[diagram.target.parent.uuid] = box
+ global_boxes[diagram.target.owner.uuid] = box
+ made_boxes[diagram.target.owner.uuid] = box
stack_heights: dict[str, float | int] = {
"input": -makers.NEIGHBOR_VMARGIN,
"output": -makers.NEIGHBOR_VMARGIN,
}
- child_boxes: list[_elkjs.ELKInputChild] = []
for child, local_ports, side in port_context_collector(ex_datas, ports):
_, label_height = helpers.get_text_extent(child.name)
height = max(
@@ -82,71 +93,47 @@ def collector(
no_symbol=diagram.display_symbols_as_boxes,
)
box["ports"] = [makers.make_port(j.uuid) for j in local_ports]
- if child.parent.uuid == centerbox["id"]:
- child_boxes.append(box)
- else:
- global_boxes[child.uuid] = box
-
- if diagram.display_parent_relation:
- if child == diagram.target.parent:
- _move_edge_to_local_edges(
- box, connections, local_ports, diagram, data
- )
- elif child.parent == diagram.target.parent:
- parent_box = global_boxes[child.parent.uuid]
- parent_box.setdefault("children", []).append(
- global_boxes.pop(child.uuid)
- )
+ global_boxes[child.uuid] = box
+ made_boxes[child.uuid] = box
+
+ if diagram.display_parent_relation and child.owner is not None:
+ child_box = global_boxes.pop(child.uuid)
+ for uuid in generic.get_all_owners(child):
+ owner = diagram.target._model.by_uuid(uuid)
+ assert owner is not None
+ if not (parent_box := global_boxes.get(uuid)):
+ parent_box = makers.make_box(
+ owner,
+ no_symbol=diagram.display_symbols_as_boxes,
+ )
+ global_boxes[uuid] = parent_box
+ made_boxes[uuid] = parent_box
+
+ parent_box.setdefault("children", []).append(child_box)
for label in parent_box["labels"]:
label["layoutOptions"] = (
- makers.CENTRIC_LABEL_LAYOUT_OPTIONS
+ makers.DEFAULT_LABEL_LAYOUT_OPTIONS
)
- _move_edge_to_local_edges(
- parent_box, connections, local_ports, diagram, data
- )
+ child_box = parent_box
stack_heights[side] += makers.NEIGHBOR_VMARGIN + height
del global_boxes[centerbox["id"]]
data["children"].extend(global_boxes.values())
- if child_boxes:
- centerbox["children"] = child_boxes
- centerbox["width"] = makers.EOI_WIDTH
- for label in centerbox.get("labels", []):
- label.setdefault("layoutOptions", {}).update(
- makers.DEFAULT_LABEL_LAYOUT_OPTIONS
- )
+ if diagram.display_parent_relation:
+ owner_boxes: dict[str, _elkjs.ELKInputChild] = {
+ uuid: box
+ for uuid, box in made_boxes.items()
+ if box.get("children")
+ }
+ generic.move_parent_boxes_to_owner(owner_boxes, diagram.target, data)
+ generic.move_edges(owner_boxes, edges, data)
centerbox["height"] = max(centerbox["height"], *stack_heights.values())
return data
-def _move_edge_to_local_edges(
- box: _elkjs.ELKInputChild,
- connections: list[common.GenericElement],
- local_ports: list[common.GenericElement],
- diagram: context.ContextDiagram,
- data: _elkjs.ELKInputData,
-) -> None:
- edges_to_remove: list[str] = []
- for c in connections:
- if (
- c.target in local_ports
- and c.source in diagram.target.ports
- or c.source in local_ports
- and c.target in diagram.target.ports
- ):
- for edge in data["edges"]:
- if edge["id"] == c.uuid:
- box.setdefault("edges", []).append(edge)
- edges_to_remove.append(edge["id"])
-
- data["edges"] = [
- e for e in data["edges"] if e["id"] not in edges_to_remove
- ]
-
-
def port_collector(
target: common.GenericElement | common.ElementList, diagram_type: DT
) -> list[common.GenericElement]:
@@ -182,13 +169,12 @@ def port_exchange_collector(
[cabc.Iterable[common.GenericElement]],
cabc.Iterable[common.GenericElement],
] = lambda i: i,
-) -> list[common.GenericElement]:
+) -> dict[str, common.ElementList[fa.AbstractExchange]]:
"""Collect exchanges from `ports` savely."""
- edges: list[common.GenericElement] = []
+ edges: dict[str, common.ElementList[fa.AbstractExchange]] = {}
for i in ports:
try:
- filtered = filter(getattr(i, "exchanges"))
- edges.extend(filtered)
+ edges[i.uuid] = filter(getattr(i, "exchanges"))
except AttributeError:
pass
return edges
diff --git a/capellambse_context_diagrams/collectors/generic.py b/capellambse_context_diagrams/collectors/generic.py
index 9f350be2..15952df6 100644
--- a/capellambse_context_diagrams/collectors/generic.py
+++ b/capellambse_context_diagrams/collectors/generic.py
@@ -11,7 +11,7 @@
import logging
import typing as t
-from capellambse.model import common
+from capellambse.model import common, layers
from capellambse.model.crosslayer import interaction
from capellambse.model.modeltypes import DiagramType as DT
@@ -43,6 +43,12 @@
"""Default size of marker-ends in pixels."""
MARKER_PADDING = makers.PORT_PADDING
"""Default padding of markers in pixels."""
+PackageTypes: tuple[type[common.GenericElement], ...] = (
+ layers.oa.EntityPkg,
+ layers.la.LogicalComponentPkg,
+ layers.ctx.SystemComponentPkg,
+ layers.pa.PhysicalComponentPkg,
+)
def collector(
@@ -185,3 +191,76 @@ def collect_label(obj: common.GenericElement) -> str | None:
elif isinstance(obj, interaction.AbstractCapabilityInclude):
return "« i »"
return "" if obj.name.startswith("(Unnamed") else obj.name
+
+
+def move_parent_boxes_to_owner(
+ boxes: dict[str, _elkjs.ELKInputChild],
+ obj: common.GenericElement,
+ data: _elkjs.ELKInputData,
+ filter_types: tuple[type, ...] = PackageTypes,
+) -> None:
+ """Move boxes to their owner box."""
+ boxes_to_remove: list[str] = []
+ for child in data["children"]:
+ if not child.get("children"):
+ continue
+
+ owner = obj._model.by_uuid(child["id"])
+ if (
+ isinstance(owner, filter_types)
+ or not (oowner := owner.owner)
+ or isinstance(oowner, filter_types)
+ or not (oowner_box := boxes.get(oowner.uuid))
+ ):
+ continue
+
+ oowner_box.setdefault("children", []).append(child)
+ boxes_to_remove.append(child["id"])
+
+ data["children"] = [
+ b for b in data["children"] if b["id"] not in boxes_to_remove
+ ]
+
+
+def move_edges(
+ boxes: dict[str, _elkjs.ELKInputChild],
+ connections: list[common.GenericElement],
+ data: _elkjs.ELKInputData,
+) -> None:
+ """Move edges to boxes."""
+ edges_to_remove: list[str] = []
+ for c in connections:
+ source_owner_uuids = get_all_owners(c.source)
+ target_owner_uuids = get_all_owners(c.target)
+ common_owner_uuid = None
+ for owner in source_owner_uuids:
+ if owner in set(target_owner_uuids):
+ common_owner_uuid = owner
+ break
+
+ if not common_owner_uuid or not (
+ owner_box := boxes.get(common_owner_uuid)
+ ):
+ continue
+
+ for edge in data["edges"]:
+ if edge["id"] == c.uuid:
+ owner_box.setdefault("edges", []).append(edge)
+ edges_to_remove.append(edge["id"])
+
+ data["edges"] = [
+ e for e in data["edges"] if e["id"] not in edges_to_remove
+ ]
+
+
+def get_all_owners(obj: common.GenericElement) -> list[str]:
+ """Return the UUIDs from all owners of ``obj``."""
+ owners: list[str] = []
+ current = obj
+ while current is not None and not isinstance(current, PackageTypes):
+ owners.append(current.uuid)
+ try:
+ current = current.owner
+ except AttributeError:
+ break
+ return owners
diff --git a/capellambse_context_diagrams/collectors/portless.py b/capellambse_context_diagrams/collectors/portless.py
index 151b4af4..3d3b1c45 100644
--- a/capellambse_context_diagrams/collectors/portless.py
+++ b/capellambse_context_diagrams/collectors/portless.py
@@ -41,11 +41,8 @@ def collector(
except AttributeError:
continue
- stack_heights: dict[str, float | int] = {
- "input": -makers.NEIGHBOR_VMARGIN,
- "output": -makers.NEIGHBOR_VMARGIN,
- }
contexts = context_collector(connections, diagram.target)
+ global_boxes = {centerbox["id"]: centerbox}
made_boxes = {centerbox["id"]: centerbox}
if diagram.display_parent_relation and diagram.target.owner is not None:
box = makers.make_box(
@@ -55,8 +52,13 @@ def collector(
)
box["children"] = [centerbox]
del data["children"][0]
+ global_boxes[diagram.target.owner.uuid] = box
made_boxes[diagram.target.owner.uuid] = box
+ stack_heights: dict[str, float | int] = {
+ "input": -makers.NEIGHBOR_VMARGIN,
+ "output": -makers.NEIGHBOR_VMARGIN,
+ }
for i, exchanges, side in contexts:
var_height = generic.MARKER_PADDING + (
generic.MARKER_SIZE + generic.MARKER_PADDING
@@ -68,7 +70,7 @@ def collector(
else:
height = var_height
- if box := made_boxes.get(i.uuid): # type: ignore[assignment]
+ if box := global_boxes.get(i.uuid): # type: ignore[assignment]
if box is centerbox:
continue
box["height"] = height
@@ -78,33 +80,37 @@ def collector(
height=height,
no_symbol=diagram.display_symbols_as_boxes,
)
+ global_boxes[i.uuid] = box
made_boxes[i.uuid] = box
- if diagram.display_parent_relation:
- if i.owner is not None:
- if not (parent_box := made_boxes.get(i.owner.uuid)):
- parent_box = makers.make_box(
- i.owner,
- no_symbol=diagram.display_symbols_as_boxes,
- )
- made_boxes[i.owner.uuid] = parent_box
-
- parent_box.setdefault("children", []).append(
- made_boxes.pop(i.uuid)
+ if diagram.display_parent_relation and i.owner is not None:
+ if not (parent_box := global_boxes.get(i.owner.uuid)):
+ parent_box = makers.make_box(
+ i.owner,
+ no_symbol=diagram.display_symbols_as_boxes,
)
- for label in parent_box["labels"]:
- label["layoutOptions"] = (
- makers.DEFAULT_LABEL_LAYOUT_OPTIONS
- )
+ global_boxes[i.owner.uuid] = parent_box
+ made_boxes[i.owner.uuid] = parent_box
+
+ parent_box.setdefault("children", []).append(
+ global_boxes.pop(i.uuid)
+ )
+ for label in parent_box["labels"]:
+ label["layoutOptions"] = makers.DEFAULT_LABEL_LAYOUT_OPTIONS
stack_heights[side] += makers.NEIGHBOR_VMARGIN + height
- del made_boxes[centerbox["id"]]
- data["children"].extend(made_boxes.values())
+ del global_boxes[centerbox["id"]]
+ data["children"].extend(global_boxes.values())
if diagram.display_parent_relation:
- _move_parent_boxes(diagram.target, data)
- _move_edges(made_boxes, connections, data)
+ owner_boxes: dict[str, _elkjs.ELKInputChild] = {
+ uuid: box
+ for uuid, box in made_boxes.items()
+ if box.get("children")
+ }
+ generic.move_parent_boxes_to_owner(owner_boxes, diagram.target, data)
+ generic.move_edges(owner_boxes, connections, data)
centerbox["height"] = max(centerbox["height"], *stack_heights.values())
if not diagram.display_symbols_as_boxes and makers.is_symbol(
@@ -223,79 +229,3 @@ def get_exchanges(
filtered = filter(chain.from_iterable(exchanges))
yield from {i.uuid: i for i in filtered}.values()
-
-
-def _move_parent_boxes(
- obj: common.GenericElement,
- data: _elkjs.ELKInputData,
-) -> None:
- owner_boxes: dict[str, _elkjs.ELKInputChild] = {
- child["id"]: child
- for child in data["children"]
- if child.get("children")
- }
- boxes_to_remove: list[str] = []
- for child in data["children"]:
- if not child.get("children"):
- continue
-
- owner = obj._model.by_uuid(child["id"])
- if (
- not (oowner := owner.owner)
- or isinstance(oowner, layers.oa.EntityPkg)
- or not (oowner_box := owner_boxes.get(oowner.uuid))
- ):
- continue
-
- oowner_box.setdefault("children", []).append(child)
- boxes_to_remove.append(child["id"])
-
- data["children"] = [
- b for b in data["children"] if b["id"] not in boxes_to_remove
- ]
-
-
-def _move_edges(
- boxes: dict[str, _elkjs.ELKInputChild],
- connections: list[common.GenericElement],
- data: _elkjs.ELKInputData,
-) -> None:
- owner_boxes: dict[str, _elkjs.ELKInputChild] = {
- uuid: box for uuid, box in boxes.items() if box.get("children")
- }
- edges_to_remove: list[str] = []
-
- for c in connections:
- source_owner_uuids = _get_all_owners(c.source)
- target_owner_uuids = _get_all_owners(c.target)
- common_owner_uuid = None
- for owner in source_owner_uuids:
- if owner in set(target_owner_uuids):
- common_owner_uuid = owner
- break
-
- if not common_owner_uuid or not (
- owner_box := owner_boxes.get(common_owner_uuid)
- ):
- continue
-
- for edge in data["edges"]:
- if edge["id"] == c.uuid:
- owner_box.setdefault("edges", []).append(edge)
- edges_to_remove.append(edge["id"])
-
- data["edges"] = [
- e for e in data["edges"] if e["id"] not in edges_to_remove
- ]
-
-
-def _get_all_owners(obj: common.GenericElement) -> list[str]:
- owners = []
- current = obj
- while current is not None and not isinstance(current, layers.oa.EntityPkg):
- owners.append(current.uuid)
- try:
- current = current.owner
- except AttributeError:
- break
- return owners
diff --git a/tests/data/ContextDiagram.aird b/tests/data/ContextDiagram.aird
index 9cc0da80..1437a998 100644
--- a/tests/data/ContextDiagram.aird
+++ b/tests/data/ContextDiagram.aird
@@ -86,7 +86,7 @@
-
+