Skip to content

gds_business.vsm.checks

VSM verification checks (VSM-001..VSM-004).

VSM-001: Each step has at most 1 incoming and 1 outgoing material flow.

Source code in packages/gds-business/gds_business/vsm/checks.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def check_vsm001_linear_process_flow(model: ValueStreamModel) -> list[Finding]:
    """VSM-001: Each step has at most 1 incoming and 1 outgoing material flow."""
    findings: list[Finding] = []
    for step in model.steps:
        incoming = sum(1 for f in model.material_flows if f.target == step.name)
        outgoing = sum(1 for f in model.material_flows if f.source == step.name)
        linear = incoming <= 1 and outgoing <= 1
        findings.append(
            Finding(
                check_id="VSM-001",
                severity=Severity.WARNING,
                message=(
                    f"Step {step.name!r}: {incoming} incoming, {outgoing} outgoing "
                    f"material flow(s) — "
                    f"{'linear' if linear else 'non-linear (branching detected)'}"
                ),
                source_elements=[step.name],
                passed=linear,
            )
        )
    return findings

VSM-002: Identifies where flow_type transitions from push to pull.

Source code in packages/gds-business/gds_business/vsm/checks.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def check_vsm002_push_pull_boundary(model: ValueStreamModel) -> list[Finding]:
    """VSM-002: Identifies where flow_type transitions from push to pull."""
    findings: list[Finding] = []
    transitions: list[tuple[str, str]] = []

    # Look for adjacent flows where type changes
    for i, flow in enumerate(model.material_flows):
        for other in model.material_flows[i + 1 :]:
            # Adjacent if one's target is the other's source
            if flow.target == other.source and flow.flow_type != other.flow_type:
                transitions.append(
                    (flow.target, f"{flow.flow_type}->{other.flow_type}")
                )
            elif other.target == flow.source and flow.flow_type != other.flow_type:
                transitions.append(
                    (flow.source, f"{other.flow_type}->{flow.flow_type}")
                )

    if transitions:
        for boundary_element, transition in transitions:
            findings.append(
                Finding(
                    check_id="VSM-002",
                    severity=Severity.INFO,
                    message=(
                        f"Push/pull boundary at {boundary_element!r}: {transition}"
                    ),
                    source_elements=[boundary_element],
                    passed=True,
                )
            )
    else:
        # Check if all flows are same type
        flow_types = {f.flow_type for f in model.material_flows}
        if len(flow_types) <= 1:
            ftype = next(iter(flow_types), "push")
            findings.append(
                Finding(
                    check_id="VSM-002",
                    severity=Severity.INFO,
                    message=f"All material flows are {ftype} — no push/pull boundary",
                    source_elements=[],
                    passed=True,
                )
            )
        else:
            findings.append(
                Finding(
                    check_id="VSM-002",
                    severity=Severity.INFO,
                    message="Mixed push/pull flows but no clear boundary detected",
                    source_elements=[],
                    passed=True,
                )
            )

    return findings

VSM-003: All flow source/target are declared elements.

Source code in packages/gds-business/gds_business/vsm/checks.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def check_vsm003_flow_reference_validity(model: ValueStreamModel) -> list[Finding]:
    """VSM-003: All flow source/target are declared elements."""
    findings: list[Finding] = []
    all_names = model.element_names

    for flow in model.material_flows:
        src_valid = flow.source in all_names
        findings.append(
            Finding(
                check_id="VSM-003",
                severity=Severity.ERROR,
                message=(
                    f"MaterialFlow source {flow.source!r} "
                    f"{'is' if src_valid else 'is NOT'} a declared element"
                ),
                source_elements=[flow.source],
                passed=src_valid,
            )
        )
        tgt_valid = flow.target in all_names
        findings.append(
            Finding(
                check_id="VSM-003",
                severity=Severity.ERROR,
                message=(
                    f"MaterialFlow target {flow.target!r} "
                    f"{'is' if tgt_valid else 'is NOT'} a declared element"
                ),
                source_elements=[flow.target],
                passed=tgt_valid,
            )
        )

    for flow in model.information_flows:
        src_valid = flow.source in all_names
        findings.append(
            Finding(
                check_id="VSM-003",
                severity=Severity.ERROR,
                message=(
                    f"InformationFlow source {flow.source!r} "
                    f"{'is' if src_valid else 'is NOT'} a declared element"
                ),
                source_elements=[flow.source],
                passed=src_valid,
            )
        )
        tgt_valid = flow.target in all_names
        findings.append(
            Finding(
                check_id="VSM-003",
                severity=Severity.ERROR,
                message=(
                    f"InformationFlow target {flow.target!r} "
                    f"{'is' if tgt_valid else 'is NOT'} a declared element"
                ),
                source_elements=[flow.target],
                passed=tgt_valid,
            )
        )

    return findings

VSM-004: Max cycle_time should not exceed customer takt_time.

Source code in packages/gds-business/gds_business/vsm/checks.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def check_vsm004_bottleneck_vs_takt(model: ValueStreamModel) -> list[Finding]:
    """VSM-004: Max cycle_time should not exceed customer takt_time."""
    findings: list[Finding] = []

    if not model.steps or not model.customers:
        findings.append(
            Finding(
                check_id="VSM-004",
                severity=Severity.WARNING,
                message="No steps or customers to check bottleneck vs takt",
                source_elements=[],
                passed=True,
            )
        )
        return findings

    max_cycle = max(s.cycle_time for s in model.steps)
    bottleneck = next(s for s in model.steps if s.cycle_time == max_cycle)

    for customer in model.customers:
        within_takt = max_cycle <= customer.takt_time
        findings.append(
            Finding(
                check_id="VSM-004",
                severity=Severity.WARNING,
                message=(
                    f"Bottleneck {bottleneck.name!r} (cycle_time={max_cycle}) "
                    f"{'<=' if within_takt else '>'} "
                    f"customer {customer.name!r} takt_time={customer.takt_time}"
                ),
                source_elements=[bottleneck.name, customer.name],
                passed=within_takt,
            )
        )

    return findings