Skip to content

portless

Collection of ELKInputData on diagrams that don't involve ports or any connectors.

ContextInfo 🔗

Bases: NamedTuple

ContextInfo data.

Source code in capellambse_context_diagrams/collectors/portless.py
147
148
149
150
151
152
153
154
155
class ContextInfo(t.NamedTuple):
    """ContextInfo data."""

    element: common.GenericElement
    """An element of context."""
    connections: list[common.GenericElement]
    """The context element's relevant exchanges."""
    side: t.Literal["input", "output"]
    """Whether this is an input or output to the element of interest."""

connections: list[common.GenericElement] instance-attribute 🔗

The context element's relevant exchanges.

element: common.GenericElement instance-attribute 🔗

An element of context.

side: t.Literal['input', 'output'] instance-attribute 🔗

Whether this is an input or output to the element of interest.

collect_exchange_endpoints(e) 🔗

Safely collect exchange endpoints from e.

Source code in capellambse_context_diagrams/collectors/portless.py
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def collect_exchange_endpoints(
    e: common.GenericElement,
) -> tuple[common.GenericElement, common.GenericElement]:
    """Safely collect exchange endpoints from `e`."""

    def _get(
        e: common.GenericElement, attrs: t.FrozenSet[str]
    ) -> common.GenericElement:
        for attr in attrs:
            try:
                obj = getattr(e, attr)
                assert isinstance(obj, common.GenericElement)
                return obj
            except AttributeError:
                continue
        raise AttributeError()

    try:
        return _get(e, SOURCE_ATTR_NAMES), _get(e, TARGET_ATTR_NAMES)
    except AttributeError:
        pass
    return generic.collect_exchange_endpoints(e)

collector(diagram, params=None) 🔗

Collect context data from exchanges of centric box.

This is the special context collector for the operational architecture layer diagrams (diagrams where elements don't exchange via ports/connectors).

Source code in capellambse_context_diagrams/collectors/portless.py
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def collector(
    diagram: context.ContextDiagram, params: dict[str, t.Any] | None = None
) -> _elkjs.ELKInputData:
    """Collect context data from exchanges of centric box.

    This is the special context collector for the operational
    architecture layer diagrams (diagrams where elements don't exchange
    via ports/connectors).
    """
    data = generic.collector(diagram, no_symbol=True)
    centerbox = data["children"][0]
    connections = list(get_exchanges(diagram.target))
    for ex in connections:
        try:
            generic.exchange_data_collector(
                generic.ExchangeData(ex, data, diagram.filters, params),
                collect_exchange_endpoints,
            )
        except AttributeError:
            continue

    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(
            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.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
        ) * len(exchanges)
        if not diagram.display_symbols_as_boxes and makers.is_symbol(
            diagram.target
        ):
            height = makers.MIN_SYMBOL_HEIGHT + var_height
        else:
            height = var_height

        if box := global_boxes.get(i.uuid):  # type: ignore[assignment]
            if box is centerbox:
                continue
            box["height"] = height
        else:
            box = makers.make_box(
                i,
                height=height,
                no_symbol=diagram.display_symbols_as_boxes,
            )
            global_boxes[i.uuid] = box
            made_boxes[i.uuid] = box

        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,
                )
                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 global_boxes[centerbox["id"]]
    data["children"].extend(global_boxes.values())

    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, connections, data)

    centerbox["height"] = max(centerbox["height"], *stack_heights.values())
    if not diagram.display_symbols_as_boxes and makers.is_symbol(
        diagram.target
    ):
        data["layoutOptions"]["spacing.labelNode"] = 5.0
    return data

get_exchanges(obj, filter=lambda i: i) 🔗

Yield exchanges safely.

Yields exchanges from .related_exchanges or exclusively by obj classtype: * Capabilities: * .extends, * .includes, * .generalizes and their reverse * .included_by, * .extended_by, * .generalized_by and optionally * .entity_involvements (Operational) * .component_involvements and .incoming_exploitations (System) * Mission: * .involvements and * .exploitations.

Source code in capellambse_context_diagrams/collectors/portless.py
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
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.

    Yields exchanges from ``.related_exchanges`` or exclusively by
    ``obj`` classtype:
        * Capabilities:
            * ``.extends``,
            * ``.includes``,
            * ``.generalizes`` and their reverse
            * ``.included_by``,
            * ``.extended_by``,
            * ``.generalized_by`` and optionally
            * ``.entity_involvements`` (Operational)
            * ``.component_involvements`` and ``.incoming_exploitations``
              (System)
        * Mission:
            * ``.involvements`` and
            * ``.exploitations``.
    """
    is_op_capability = isinstance(obj, layers.oa.OperationalCapability)
    is_capability = isinstance(obj, layers.ctx.Capability)
    if is_op_capability or is_capability:
        exchanges = [
            obj.includes,
            obj.extends,
            obj.generalizes,
            obj.included_by,
            obj.extended_by,
            obj.generalized_by,
        ]
    elif isinstance(obj, layers.ctx.Mission):
        exchanges = [obj.involvements, obj.exploitations]
    else:
        exchanges = [obj.related_exchanges]

    if is_op_capability:
        exchanges += [obj.entity_involvements]
    elif is_capability:
        exchanges += [obj.component_involvements, obj.incoming_exploitations]

    filtered = filter(chain.from_iterable(exchanges))
    yield from {i.uuid: i for i in filtered}.values()