Skip to content

portless

Collector for portless ContextDiagrams.

This collector is used to collect ELKInputData on diagrams that don't involve ports or any connectors.

collect_exchange_endpoints 🔗

collect_exchange_endpoints(
    e: ModelElement,
) -> tuple[m.ModelElement, m.ModelElement]

Safely collect exchange endpoints from e.

Source code in src/capellambse_context_diagrams/collectors/portless.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def collect_exchange_endpoints(
    e: m.ModelElement,
) -> tuple[m.ModelElement, m.ModelElement]:
    """Safely collect exchange endpoints from `e`."""

    def _get(e: m.ModelElement, attrs: frozenset[str]) -> m.ModelElement:
        for attr in attrs:
            try:
                obj = getattr(e, attr)
                assert isinstance(obj, m.ModelElement)
                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 🔗

collector(diagram: ContextDiagram) -> cabc.Iterator[m.ModelElement]

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 src/capellambse_context_diagrams/collectors/portless.py 
26
27
28
29
30
31
32
33
34
35
def collector(
    diagram: context.ContextDiagram,
) -> cabc.Iterator[m.ModelElement]:
    """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).
    """
    yield from get_exchanges(diagram.target)

get_exchanges 🔗

get_exchanges(
    obj: ModelElement,
    filter: Callable[
        [Iterable[ModelElement]], Iterable[ModelElement]
    ] = lambda i: i,
) -> t.Iterator[t.Any]

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 src/capellambse_context_diagrams/collectors/portless.py 
 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
def get_exchanges(
    obj: m.ModelElement,
    filter: cabc.Callable[
        [cabc.Iterable[m.ModelElement]],
        cabc.Iterable[m.ModelElement],
    ] = lambda i: i,
) -> t.Iterator[t.Any]:
    """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, oa.OperationalCapability)
    is_capability = isinstance(obj, sa.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, sa.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()