Action Flow Type Compatibility

This example checks whether the item type flowing out of each source port is compatible with the type expected by the destination port. A flow is compatible when the source type is the same as, or a specialization of, the destination type.

This example uses the same DataPipeline model as Action Flow Port Connectivity and Action Flow Latency Analysis Labs. In this model, the flow from Analyze to Publish passes an AnalysisResult where a Report is expected — a type mismatch, since AnalysisResult does not specialize Report.

Concepts Used

API	Purpose
FlowUsage	Port features at each end of a flow (source_output_feature / target_input_feature)
Feature.types	Type chain for a feature; filter by DocumentTier.Project to isolate the user-defined type
Type.specializes()	Returns whether one type is the same as, or a specialization of, another
DocumentTier	Source tier of an element (StandardLibrary, External, Project)

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"


def get_project_type(port: syside.Feature) -> syside.Type | None:
    """Return the most-specific project-defined type of a port, or None if
    the port carries only standard-library types."""
    return next(
        (
            t
            for t in port.types.collect()
            if t.document.document_tier == syside.DocumentTier.Project
        ),
        None,
    )


def check_flow_types(model: syside.Model, action_def_name: str) -> None:
    """Check whether the item type at each flow source conforms to the
    type expected at the destination. Report any type mismatch."""

    incompatible_found = False

    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

        src_port = flow.source_output_feature
        tgt_port = flow.target_input_feature
        if src_port is None or tgt_port is None:
            continue

        src_type = get_project_type(src_port)
        tgt_type = get_project_type(tgt_port)
        if src_type is None or tgt_type is None:
            continue

        # Identify the owning actions for labeling
        src_action = flow.source_feature
        tgt_actions = flow.target_features.collect()
        tgt_action = tgt_actions[0] if tgt_actions else None

        src_label = (
            f"{src_action.name}.{src_port.name}"
            if src_action
            else src_port.name
        )
        tgt_label = (
            f"{tgt_action.name}.{tgt_port.name}"
            if tgt_action
            else tgt_port.name
        )

        # The source type must be the same as, or a specialization of,
        # the destination type.
        compatible = src_type.specializes(tgt_type)
        status = "OK" if compatible else "TYPE MISMATCH"
        print(
            f"  [{status}] {src_label} ({src_type.name})"
            f" -> {tgt_label} ({tgt_type.name})"
        )
        if not compatible:
            incompatible_found = True

    if not incompatible_found:
        print("  All flows are type-compatible.")


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"Flow type-compatibility report for {action_def}:")
    check_flow_types(model, action_def)


if __name__ == "__main__":
    main()

Output

Flow type-compatibility report for Pipeline:
  [OK] ingest.raw (RawData) -> clean.raw (RawData)
  [OK] clean.clean (CleanData) -> analyze.clean (CleanData)
  [TYPE MISMATCH] analyze.result (AnalysisResult) -> publish.report (Report)

Download

Download this example here.