default

Collection of ELKInputData on diagrams that involve ports.

DERIVATORS module-attribute

DERIVATORS: dict[type[ModelElement], DerivatorFunction] = {LogicalComponent: derive_from_functions, SystemComponent: derive_from_functions}

Supported objects to build derived contexts for.

ContextInfo

Bases: NamedTuple

ContextInfo data.

element instance-attribute

element: ModelElement

An element of context.

ports instance-attribute

ports: list[ModelElement]

The context element's relevant ports.

This list only contains ports that at least one of the exchanges passed into collect_exchanges sees.

side instance-attribute

side: Literal['input', 'output']

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

collector

collector(diagram: context.ContextDiagram, params: dict[str, t.Any] | None = None) -> _elkjs.ELKInputData

Collect context data from ports of centric box.

Source code in capellambse_context_diagrams/collectors/default.py

def collector(
    diagram: context.ContextDiagram, params: dict[str, t.Any] | None = None
) -> _elkjs.ELKInputData:
    """Collect context data from ports of centric box."""
    data = generic.collector(diagram, no_symbol=True)
    processor = ContextProcessor(diagram, data, params=params)
    processor.process_context()
    derivator = DERIVATORS.get(type(diagram.target))
    if diagram._display_derived_interfaces and derivator is not None:
        derivator(
            diagram,
            data,
            processor.made_boxes[diagram.target.uuid],
        )
    return data

derive_from_functions

derive_from_functions(diagram: context.ContextDiagram, data: _elkjs.ELKInputData, centerbox: _elkjs.ELKInputChild) -> None

Derive Components from allocated functions of the context target.

A Component, a ComponentExchange and two ComponentPorts are added to data. These elements are prefixed with Derived- to receive special styling in the serialization step.

Source code in capellambse_context_diagrams/collectors/default.py

def derive_from_functions(
    diagram: context.ContextDiagram,
    data: _elkjs.ELKInputData,
    centerbox: _elkjs.ELKInputChild,
) -> None:
    """Derive Components from allocated functions of the context target.

    A Component, a ComponentExchange and two ComponentPorts are added
    to ``data``. These elements are prefixed with ``Derived-`` to
    receive special styling in the serialization step.
    """
    assert isinstance(diagram.target, cs.Component)
    ports = []
    for fnc in diagram.target.allocated_functions:
        inc, out = port_collector(fnc, diagram.type)
        ports.extend(inc + out)

    context_box_ids = {child.id for child in data.children}
    components: dict[str, cs.Component] = {}
    for port in ports:
        for fex in port.exchanges:
            if isinstance(port, fa.FunctionOutputPort):
                attr = "target"
            else:
                attr = "source"

            try:
                derived_comp = getattr(fex, attr).owner.owner
                if (
                    derived_comp == diagram.target
                    or derived_comp.uuid in context_box_ids
                ):
                    continue

                if derived_comp.uuid not in components:
                    components[derived_comp.uuid] = derived_comp
            except AttributeError:  # No owner of owner.
                pass

    # Idea: Include flow direction of derived interfaces from all functional
    # exchanges. Mixed means bidirectional. Just even out bidirectional
    # interfaces and keep flow direction of others.

    for i, (uuid, derived_component) in enumerate(components.items(), 1):
        box = makers.make_box(
            derived_component,
            no_symbol=diagram._display_symbols_as_boxes,
        )
        class_ = type(derived_comp).__name__
        box.id = f"{makers.STYLECLASS_PREFIX}-{class_}:{uuid}"
        data.children.append(box)
        source_id = f"{makers.STYLECLASS_PREFIX}-CP_INOUT:{i}"
        target_id = f"{makers.STYLECLASS_PREFIX}-CP_INOUT:{-i}"
        box.ports.append(makers.make_port(source_id))
        centerbox.ports.append(makers.make_port(target_id))
        if i % 2 == 0:
            source_id, target_id = target_id, source_id

        data.edges.append(
            _elkjs.ELKInputEdge(
                id=f"{makers.STYLECLASS_PREFIX}-ComponentExchange:{i}",
                sources=[source_id],
                targets=[target_id],
            )
        )

    data.children[0].height += (
        makers.PORT_PADDING
        + (makers.PORT_SIZE + makers.PORT_PADDING) * len(components) // 2
    )

port_collector

port_collector(target: m.ModelElement | m.ElementList, diagram_type: DT) -> tuple[list[m.ModelElement], list[m.ModelElement]]

Savely collect ports from target.

Source code in capellambse_context_diagrams/collectors/default.py

def port_collector(
    target: m.ModelElement | m.ElementList, diagram_type: DT
) -> tuple[list[m.ModelElement], list[m.ModelElement]]:
    """Savely collect ports from `target`."""

    def __collect(target):
        incoming_ports: list[m.ModelElement] = []
        outgoing_ports: list[m.ModelElement] = []
        for attr in generic.DIAGRAM_TYPE_TO_CONNECTOR_NAMES[diagram_type]:
            try:
                ports = getattr(target, attr)
                if not ports or not isinstance(
                    ports[0],
                    (fa.FunctionPort, fa.ComponentPort, cs.PhysicalPort),
                ):
                    continue
                if attr == "inputs":
                    incoming_ports.extend(ports)
                elif attr == "ports":
                    for port in ports:
                        if port.direction == "IN":
                            incoming_ports.append(port)
                        else:
                            outgoing_ports.append(port)
                else:
                    outgoing_ports.extend(ports)
            except AttributeError:
                pass
        return incoming_ports, outgoing_ports

    if isinstance(target, cabc.Iterable):
        assert not isinstance(target, m.ModelElement)
        incoming_ports: list[m.ModelElement] = []
        outgoing_ports: list[m.ModelElement] = []
        for obj in target:
            inc, out = __collect(obj)
            incoming_ports.extend(inc)
            outgoing_ports.extend(out)
    else:
        incoming_ports, outgoing_ports = __collect(target)
    return incoming_ports, outgoing_ports

port_context_collector

port_context_collector(exchange_datas: t.Iterable[generic.ExchangeData], local_ports: t.Container[m.ModelElement]) -> t.Iterator[ContextInfo]

Collect the context objects.

PARAMETER	DESCRIPTION
exchange_datas	The ExchangeDatas to look at to find new elements. TYPE: Iterable[ExchangeData]
local_ports	Connectors/Ports lookup where exchange_datas is checked against. If an exchange connects via a port from local_ports it is collected. TYPE: Container[ModelElement]

RETURNS	DESCRIPTION
contexts	An iterator over ContextDiagram.ContextInfos.

Source code in capellambse_context_diagrams/collectors/default.py

def port_context_collector(
    exchange_datas: t.Iterable[generic.ExchangeData],
    local_ports: t.Container[m.ModelElement],
) -> t.Iterator[ContextInfo]:
    """Collect the context objects.

    Parameters
    ----------
    exchange_datas
        The ``ExchangeData``s to look at to find new elements.
    local_ports
        Connectors/Ports lookup where ``exchange_datas`` is checked
        against. If an exchange connects via a port from ``local_ports``
        it is collected.

    Returns
    -------
    contexts
        An iterator over
        [`ContextDiagram.ContextInfo`s][capellambse_context_diagrams.context.ContextDiagram].
    """

    ctx: dict[str, ContextInfo] = {}
    side: t.Literal["input", "output"]
    for exd in exchange_datas:
        try:
            source, target = generic.collect_exchange_endpoints(exd)
        except AttributeError:
            continue

        if source in local_ports:
            port = target
            side = "output"
        elif target in local_ports:
            port = source
            side = "input"
        else:
            continue

        try:
            owner = port.owner  # type: ignore[attr-defined]
        except AttributeError:
            continue

        info = ContextInfo(owner, [], side)
        info = ctx.setdefault(owner.uuid, info)
        if port not in info.ports:
            info.ports.append(port)

    return iter(ctx.values())

port_exchange_collector

port_exchange_collector(ports: t.Iterable[m.ModelElement], filter: Filter = lambda i: i) -> dict[str, list[fa.AbstractExchange]]

Collect exchanges from ports savely.

Source code in capellambse_context_diagrams/collectors/default.py

def port_exchange_collector(
    ports: t.Iterable[m.ModelElement],
    filter: Filter = lambda i: i,
) -> dict[str, list[fa.AbstractExchange]]:
    """Collect exchanges from `ports` savely."""
    edges: dict[str, list[fa.AbstractExchange]] = {}

    for port in ports:
        if exs := _extract_edges(port, "exchanges", filter):
            edges[port.uuid] = t.cast(list[fa.AbstractExchange], exs)
            continue

        if links := _extract_edges(port, "links", filter):
            edges[port.uuid] = t.cast(list[fa.AbstractExchange], links)

    return edges