exchanges

ExchangeCollector

ExchangeCollector(diagram: context.InterfaceContextDiagram | context.FunctionalContextDiagram, data: _elkjs.ELKInputData, params: dict[str, t.Any])

Base class for context collection on Exchanges.

Source code in capellambse_context_diagrams/collectors/exchanges.py

def __init__(
    self,
    diagram: (
        context.InterfaceContextDiagram | context.FunctionalContextDiagram
    ),
    data: _elkjs.ELKInputData,
    params: dict[str, t.Any],
) -> None:
    self.diagram = diagram
    self.data: _elkjs.ELKInputData = data
    self.obj = self.diagram.target
    self.params = params

    src, trg, alloc_fex, fncs = self.intermap[diagram.type]
    self.get_source = operator.attrgetter(src)
    self.get_target = operator.attrgetter(trg)
    self.get_alloc_fex = operator.attrgetter(alloc_fex)
    self.get_alloc_functions = operator.attrgetter(fncs)

collect abstractmethod

collect() -> None

Populate the elkdata container.

Source code in capellambse_context_diagrams/collectors/exchanges.py

@abc.abstractmethod
def collect(self) -> None:
    """Populate the elkdata container."""
    raise NotImplementedError

update_children_size

update_children_size(data: _elkjs.ELKInputChild, exchanges: t.Sequence[_elkjs.ELKInputEdge]) -> None

Adjust size of functions.

Source code in capellambse_context_diagrams/collectors/exchanges.py

def update_children_size(
    self,
    data: _elkjs.ELKInputChild,
    exchanges: t.Sequence[_elkjs.ELKInputEdge],
) -> None:
    """Adjust size of functions."""
    stack_height: int | float = -makers.NEIGHBOR_VMARGIN
    for child in data.children:
        inputs, outputs = [], []
        obj = self.obj._model.by_uuid(child.id)
        if isinstance(obj, cs.Component):
            self.update_children_size(child, exchanges)
            return

        port_ids = {p.id for p in child.ports}
        for ex in exchanges:
            source, target = ex.sources[0], ex.targets[0]
            if source in port_ids:
                outputs.append(source)
            elif target in port_ids:
                inputs.append(target)

        childnum = max(len(inputs), len(outputs))
        height = max(
            child.height + 2 * makers.LABEL_VPAD,
            makers.PORT_PADDING
            + (makers.PORT_SIZE + makers.PORT_PADDING) * childnum,
        )
        child.height = height
        stack_height += makers.NEIGHBOR_VMARGIN + height

    if stack_height > 0:
        data.height = stack_height

InterfaceContextCollector

InterfaceContextCollector(diagram: context.InterfaceContextDiagram, data: _elkjs.ELKInputData, params: dict[str, t.Any])

Bases: ExchangeCollector

Collect necessary _elkjs.ELKInputData for building the interface context.

Source code in capellambse_context_diagrams/collectors/exchanges.py

def __init__(
    self,
    diagram: context.InterfaceContextDiagram,
    data: _elkjs.ELKInputData,
    params: dict[str, t.Any],
) -> None:
    self.left: _elkjs.ELKInputChild | None = None
    self.right: _elkjs.ELKInputChild | None = None
    self.incoming_edges = {}
    self.outgoing_edges = {}

    super().__init__(diagram, data, params)

left instance-attribute

left: ELKInputChild | None = None

Left (source) Component Box of the interface.

right instance-attribute

right: ELKInputChild | None = None

Right (target) Component Box of the interface.

add_interface

add_interface() -> None

Add the ComponentExchange (interface) to the collected data.

Source code in capellambse_context_diagrams/collectors/exchanges.py

def add_interface(self) -> None:
    """Add the ComponentExchange (interface) to the collected data."""
    ex_data = generic.ExchangeData(
        self.obj,
        self.data,
        self.diagram.filters,
        self.params,
        is_hierarchical=False,
    )
    src, tgt = generic.exchange_data_collector(ex_data)
    self.data.edges[-1].layoutOptions = copy.deepcopy(
        _elkjs.EDGE_STRAIGHTENING_LAYOUT_OPTIONS
    )
    assert self.right is not None
    assert self.left is not None
    self.left.ports.append(makers.make_port(src.uuid))
    self.right.ports.append(makers.make_port(tgt.uuid))

collect

collect() -> None

Collect all allocated FunctionalExchanges in the context.

Source code in capellambse_context_diagrams/collectors/exchanges.py

def collect(self) -> None:
    """Collect all allocated `FunctionalExchange`s in the context."""
    self.get_left_and_right()
    if self.diagram._hide_functions:
        assert self.left is not None
        self.left.children = []
        assert self.right is not None
        self.right.children = []
        self.incoming_edges = {}
        self.outgoing_edges = {}

    if self.diagram._include_interface or self.diagram._hide_functions:
        self.add_interface()

    try:
        for ex in (self.incoming_edges | self.outgoing_edges).values():
            ex_data = generic.ExchangeData(
                ex,
                self.data,
                self.diagram.filters,
                self.params,
                is_hierarchical=False,
            )
            src, tgt = generic.exchange_data_collector(ex_data)
            if ex in self.incoming_edges.values():
                self.data.edges[-1].sources = [tgt.uuid]
                self.data.edges[-1].targets = [src.uuid]

        if not self.data.edges:
            logger.warning(
                "There are no FunctionalExchanges allocated to '%s'.",
                self.obj.name,
            )
    except AttributeError:
        pass

PhysicalLinkContextCollector

PhysicalLinkContextCollector(diagram: context.InterfaceContextDiagram, data: _elkjs.ELKInputData, params: dict[str, t.Any])

Bases: ExchangeCollector

Collect necessary _elkjs.ELKInputData for building the PhysicalLink context.

Source code in capellambse_context_diagrams/collectors/exchanges.py

def __init__(
    self,
    diagram: context.InterfaceContextDiagram,
    data: _elkjs.ELKInputData,
    params: dict[str, t.Any],
) -> None:
    self.left: _elkjs.ELKInputChild | None = None
    self.right: _elkjs.ELKInputChild | None = None

    super().__init__(diagram, data, params)

left instance-attribute

left: ELKInputChild | None = None

Left partner of the interface.

right instance-attribute

right: ELKInputChild | None = None

Right partner of the interface.

add_interface

add_interface() -> None

Add the ComponentExchange (interface) to the collected data.

Source code in capellambse_context_diagrams/collectors/exchanges.py

def add_interface(self) -> None:
    """Add the ComponentExchange (interface) to the collected data."""
    ex_data = generic.ExchangeData(
        self.obj,
        self.data,
        self.diagram.filters,
        self.params,
        is_hierarchical=False,
    )
    src, tgt = generic.exchange_data_collector(ex_data)
    self.data.edges[-1].layoutOptions = copy.deepcopy(
        _elkjs.EDGE_STRAIGHTENING_LAYOUT_OPTIONS
    )
    assert self.right is not None
    assert self.left is not None
    left_port, right_port = self.get_source_and_target_ports(src, tgt)
    self.left.ports.append(left_port)
    self.right.ports.append(right_port)

collect

collect() -> None

Collect all allocated PhysicalLinks in the context.

Source code in capellambse_context_diagrams/collectors/exchanges.py

def collect(self) -> None:
    """Collect all allocated `PhysicalLink`s in the context."""
    self.get_left_and_right()
    if self.diagram._include_interface:
        self.add_interface()

get_source_and_target_ports

get_source_and_target_ports(src: m.ModelElement, tgt: m.ModelElement) -> tuple[_elkjs.ELKInputPort, _elkjs.ELKInputPort]

Return the source and target ports of the interface.

Source code in capellambse_context_diagrams/collectors/exchanges.py

def get_source_and_target_ports(
    self, src: m.ModelElement, tgt: m.ModelElement
) -> tuple[_elkjs.ELKInputPort, _elkjs.ELKInputPort]:
    """Return the source and target ports of the interface."""
    left_port = makers.make_port(src.uuid)
    right_port = makers.make_port(tgt.uuid)
    if self.diagram._display_port_labels:
        left_port.labels = makers.make_label(src.name)
        right_port.labels = makers.make_label(tgt.name)

        _plp = self.diagram._port_label_position
        if not (plp := getattr(_elkjs.PORT_LABEL_POSITION, _plp, None)):
            raise ValueError(f"Invalid port label position '{_plp}'.")

        assert isinstance(plp, _elkjs.PORT_LABEL_POSITION)
        port_label_position = plp.name

        assert self.left is not None
        self.left.layoutOptions["portLabels.placement"] = (
            port_label_position
        )
        assert self.right is not None
        self.right.layoutOptions["portLabels.placement"] = (
            port_label_position
        )
    return left_port, right_port

get_elkdata_for_exchanges

get_elkdata_for_exchanges(diagram: context.InterfaceContextDiagram | context.FunctionalContextDiagram, collector_type: type[ExchangeCollector], params: dict[str, t.Any]) -> _elkjs.ELKInputData

Return exchange data for ELK.

Source code in capellambse_context_diagrams/collectors/exchanges.py

def get_elkdata_for_exchanges(
    diagram: (
        context.InterfaceContextDiagram | context.FunctionalContextDiagram
    ),
    collector_type: type[ExchangeCollector],
    params: dict[str, t.Any],
) -> _elkjs.ELKInputData:
    """Return exchange data for ELK."""
    data = makers.make_diagram(diagram)
    data.layoutOptions["layered.nodePlacement.strategy"] = "NETWORK_SIMPLEX"
    collector = collector_type(diagram, data, params)
    collector.collect()
    for comp in data.children:
        collector.update_children_size(comp, data.edges)
    return data

is_hierarchical

is_hierarchical(ex: m.ModelElement, box: _elkjs.ELKInputChild, key: t.Literal['ports'] | t.Literal['children'] = 'ports') -> bool

Check if the exchange is hierarchical (nested) inside box.

Source code in capellambse_context_diagrams/collectors/exchanges.py

def is_hierarchical(
    ex: m.ModelElement,
    box: _elkjs.ELKInputChild,
    key: t.Literal["ports"] | t.Literal["children"] = "ports",
) -> bool:
    """Check if the exchange is hierarchical (nested) inside ``box``."""
    src, trg = generic.collect_exchange_endpoints(ex)
    objs = {o.id for o in getattr(box, key)}
    attr_map = {"children": "parent.uuid", "ports": "parent.parent.uuid"}
    attr_getter = operator.attrgetter(attr_map[key])
    source_contained = src.uuid in objs or attr_getter(src) == box.id
    target_contained = trg.uuid in objs or attr_getter(trg) == box.id
    return source_contained and target_contained