Action Flow Port Connectivity

This example checks whether every port (in item / out item) on each sub-action inside an action definition is covered by at least one item flow. Unconnected ports are reported so that incomplete wiring can be caught early.

This example uses the same DataPipeline model as Action Flow Type Compatibility and Action Flow Latency Analysis Labs. In this model, the Publish action has an in item tag : Tag port that is intentionally left without a flow.

Concepts Used

API	Purpose
FlowUsage	Port endpoints of an item flow (source_output_feature / target_input_feature)
ConnectorAsUsage	Owning action usages at each end of the flow (source_feature / target_features)
Type.inputs / Type.outputs	Item ports declared on an action usage
FeatureDirectionKind	Direction of a port feature (In, Out, Inout)

Example Model

package DataPipeline {
    private import ISQBase::DurationValue;
    private import SI::min;

    item def RawData;
    item def CleanData;
    item def AnalysisResult;
    item def Report;
    item def Tag;

    action def Ingest {
        doc /* Read raw data from a source. */
        out item raw : RawData;
        attribute latency : DurationValue = 2 [min];
    }

    action def Clean {
        doc /* Remove noise and normalize the raw data. */
        in item raw : RawData;
        out item clean : CleanData;
        attribute latency : DurationValue = 5 [min];
    }

    action def Analyze {
        doc /* Run statistical analysis on clean data. */
        in item clean : CleanData;
        out item result : AnalysisResult;
        attribute latency : DurationValue = 10 [min];
    }

    action def Publish {
        doc /* Format and publish the result.
             Expects a Report but receives an AnalysisResult — a type mismatch. */
        in item report : Report;
        in item tag : Tag;  // intentionally left unconnected
        attribute latency : DurationValue = 1 [min];
    }

    action def Pipeline {
        first start;
        then action ingest : Ingest;
        then action clean : Clean;
        then action analyze : Analyze;
        then action publish : Publish;

        flow from ingest.raw to clean.raw;
        flow from clean.clean to analyze.clean;
        // Type mismatch: AnalysisResult is not a subtype of Report
        flow from analyze.result to publish.report;
        // publish.tag is intentionally left without a flow
    }
}

Example Script

import pathlib
import syside

EXAMPLE_DIR = pathlib.Path(__file__).parent
MODEL_FILE_PATH = EXAMPLE_DIR / "example_model.sysml"

type ConnectedPort = tuple[str, str]


def check_port_connectivity(model: syside.Model, action_def_name: str) -> None:
    """Check whether every port on each sub-action inside a named action
    definition is covered by at least one item flow. Report any port that
    is not connected."""

    # Build a set of (action qualified name, port name) pairs that appear
    # as endpoints in at least one FlowUsage.
    connected: set[ConnectedPort] = set()

    for flow in model.nodes(syside.FlowUsage):
        owning_def = flow.owning_definition
        if owning_def is None or owning_def.name != action_def_name:
            continue

        # Record source port
        src_action = flow.source_feature
        src_port = flow.source_output_feature
        if (
            src_action is not None
            and src_port is not None
            and src_port.name is not None
        ):
            connected.add((str(src_action.qualified_name), src_port.name))

        # Record target port
        for tgt_action in flow.target_features.collect():
            tgt_port = flow.target_input_feature
            if tgt_port is not None and tgt_port.name is not None:
                connected.add((str(tgt_action.qualified_name), tgt_port.name))

    # Compare each sub-action's ports against the connected set.
    unconnected_found = False
    for action in model.nodes(syside.ActionUsage):
        owning_def = action.owning_definition
        if owning_def is None or owning_def.name != action_def_name:
            continue

        ports = {*action.inputs.collect(), *action.outputs.collect()}
        for port in ports:
            if port.name is None:
                continue
            action_qname = str(action.qualified_name)
            key: ConnectedPort = (action_qname, port.name)
            if key not in connected:
                direction = "in"
                if port.direction == syside.FeatureDirectionKind.Out:
                    direction = "out"
                elif port.direction == syside.FeatureDirectionKind.Inout:
                    direction = "inout"
                print(
                    f"  [UNCONNECTED] {action.name}.{port.name} ({direction})"
                )
                unconnected_found = True

    if not unconnected_found:
        print("  All ports are connected.")


def main() -> None:
    (model, diagnostics) = syside.load_model([MODEL_FILE_PATH])
    assert not diagnostics.contains_errors(warnings_as_errors=True)

    action_def = "Pipeline"
    print(f"Port connectivity report for {action_def}:")
    check_port_connectivity(model, action_def)


if __name__ == "__main__":
    main()

Output

Port connectivity report for Pipeline:
  [UNCONNECTED] publish.tag (in)

Download

Download this example here.