Skip to content

dataflow_view

Collector for the DataFlowDiagram.

collector 🔗

collector(
    diagram: ContextDiagram,
    exchange_filter: Callable[
        [
            Iterable[FunctionalExchange],
            Iterable[FunctionalExchange],
            tuple[str, str],
        ],
        Iterable[FunctionalExchange],
    ] = only_involved,
) -> cabc.Iterator[m.ModelElement]

Collect model elements through default or portless collectors.

Source code in src/capellambse_context_diagrams/collectors/dataflow_view.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def collector(
    diagram: context.ContextDiagram,
    exchange_filter: cabc.Callable[
        [
            cabc.Iterable[fa.FunctionalExchange],
            cabc.Iterable[fa.FunctionalExchange],
            tuple[str, str],
        ],
        cabc.Iterable[fa.FunctionalExchange],
    ] = only_involved,
) -> cabc.Iterator[m.ModelElement]:
    """Collect model elements through default or portless collectors."""
    yield from _collect_data(
        diagram, exchange_filter, **COLLECTOR_PARAMS[diagram.type]
    )

only_involved 🔗

only_involved(
    exchanges: Iterable[FunctionalExchange],
    functions: Iterable[FunctionalExchange],
    attributes: tuple[str, str],
) -> cabc.Iterable[fa.FunctionalExchange]

Exchange filter function for collecting edges.

Source code in src/capellambse_context_diagrams/collectors/dataflow_view.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def only_involved(
    exchanges: cabc.Iterable[fa.FunctionalExchange],
    functions: cabc.Iterable[fa.FunctionalExchange],
    attributes: tuple[str, str],
) -> cabc.Iterable[fa.FunctionalExchange]:
    """Exchange filter function for collecting edges."""
    src_attr, trg_attr = attributes
    src_getter = operator.attrgetter(src_attr)
    trg_getter = operator.attrgetter(trg_attr)
    return [
        ex
        for ex in exchanges
        if src_getter(ex) in functions and trg_getter(ex) in functions
    ]