Skip to content

dataflow_view

This module defines the collectors for the DataFlowDiagram.

collector 🔗

collector(diagram: context.ContextDiagram, params: dict[str, t.Any], exchange_filter: cabc.Callable[[cabc.Iterable[fa.FunctionalExchange], cabc.Iterable[fa.FunctionalExchange], tuple[str, str]], cabc.Iterable[fa.FunctionalExchange]] = only_involved) -> _elkjs.ELKInputData

Main collector that calls either default or portless collectors.

Source code in capellambse_context_diagrams/collectors/dataflow_view.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def collector(
    diagram: context.ContextDiagram,
    params: dict[str, t.Any],
    exchange_filter: cabc.Callable[
        [
            cabc.Iterable[fa.FunctionalExchange],
            cabc.Iterable[fa.FunctionalExchange],
            tuple[str, str],
        ],
        cabc.Iterable[fa.FunctionalExchange],
    ] = only_involved,
) -> _elkjs.ELKInputData:
    """Main collector that calls either default or portless collectors."""
    return _collect_data(
        diagram, params, exchange_filter, **COLLECTOR_PARAMS[diagram.type]
    )

only_involved 🔗

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.

Source code in capellambse_context_diagrams/collectors/dataflow_view.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
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
    ]