UML state machine diagrams

vending/check_vending.py

"""One test for every row of the vending machine's transition table, then every pair
that has no row, then the timeout, driven by a clock the test controls."""
from vending import EVENTS, STATES, TABLE, IllegalTransition, VendingMachine


class FakeClock:
    def __init__(self):
        self.now = 0

    def __call__(self):
        return self.now


def machine_in(state, credit=0, stock=None):
    m = VendingMachine(stock or {"tea": 2, "coffee": 1}, FakeClock())
    m.state, m.credit = state, credit
    return m


# (row, state, credit, stock, event, args, expected state, expected output)
CASES = [
    ((("Idle", "coin"), 0), "Idle", 0, None, "coin", (10,), "Collecting", []),
    ((("Collecting", "coin"), 0), "Collecting", 10, None, "coin", (20,), "Collecting", []),
    ((("Collecting", "select"), 0), "Collecting", 30, None, "select", ("tea",), "Dispensing", ["tea"]),
    ((("Collecting", "select"), 1), "Collecting", 30, {"coffee": 0}, "select", ("coffee",), "Collecting", ["display: coffee costs 30"]),
    ((("Collecting", "cancel"), 0), "Collecting", 20, None, "cancel", (), "Idle", ["coins: 20"]),
    ((("Collecting", "timeout"), 0), "Collecting", 20, None, "timeout", (), "Idle", ["coins: 20"]),
    ((("Dispensing", "take"), 0), "Dispensing", 10, None, "take", (), "Idle", ["coins: 10"]),
    ((("Maintenance", "refill"), 0), "Maintenance", 0, None, "refill", ("tea", 5), "Maintenance", []),
    ((("Maintenance", "door_close"), 0), "Maintenance", 0, None, "door_close", (), "Idle", ["coin slot unlocked"]),
    ((("Idle", "door_open"), 0), "Idle", 0, None, "door_open", (), "Maintenance", ["coin slot locked"]),
    ((("Collecting", "door_open"), 0), "Collecting", 20, None, "door_open", (), "Maintenance", ["coins: 20", "coin slot locked"]),
    ((("Dispensing", "door_open"), 0), "Dispensing", 0, None, "door_open", (), "Maintenance", ["coin slot locked"]),
]
ARGS = {"coin": (10,), "select": ("tea",), "refill": ("tea", 5)}

rows = {(key, i) for key, alternatives in TABLE.items() for i in range(len(alternatives))}
tested = {case[0] for case in CASES}
assert rows == tested, f"rows without a test: {sorted(rows - tested)}"

passed = 0
for n, (row, state, credit, stock, event, args, want_state, want_out) in enumerate(CASES, start=1):
    m = machine_in(state, credit, stock)
    m.fire(event, *args)
    ok = m.state == want_state and m.out == want_out
    passed += ok
    call = f"{event}({', '.join(map(str, args))})"
    print(f"row {n:>2}  {state:<11} {call:<17} -> {m.state:<11} {'ok' if ok else 'FAILED: ' + str(m.out)}")
print(f"{passed} of {len(CASES)} rows pass")

refused = 0
for state in STATES:
    for event in EVENTS:
        if (state, event) in TABLE:
            continue
        m = machine_in(state, 10)
        try:
            m.fire(event, *ARGS.get(event, ()))
        except IllegalTransition:
            refused += (m.state, m.credit, m.out) == (state, 10, [])
print(f"{refused} of {len(STATES) * len(EVENTS) - len(TABLE)} unlisted pairs refused, with the state unchanged")

clock = FakeClock()
m = VendingMachine({"tea": 1}, clock)
m.fire("coin", 10)
clock.now = 29
before = m.tick()
clock.now = 30
print(f"timeout: {before} at 29 s, {m.tick()} at 30 s, handed back {m.out}")

row  1  Idle        coin(10)          -> Collecting  ok
row  2  Collecting  coin(20)          -> Collecting  ok
row  3  Collecting  select(tea)       -> Dispensing  ok
row  4  Collecting  select(coffee)    -> Collecting  ok
row  5  Collecting  cancel()          -> Idle        ok
row  6  Collecting  timeout()         -> Idle        ok
row  7  Dispensing  take()            -> Idle        ok
row  8  Maintenance refill(tea, 5)    -> Maintenance ok
row  9  Maintenance door_close()      -> Idle        ok
row 10  Idle        door_open()       -> Maintenance ok
row 11  Collecting  door_open()       -> Maintenance ok
row 12  Dispensing  door_open()       -> Maintenance ok
12 of 12 rows pass
21 of 21 unlisted pairs refused, with the state unchanged
timeout: Collecting at 29 s, Idle at 30 s, handed back ['coins: 10']

vending/vending.py

"""A vending machine's lifecycle, kept as a transition table.

TABLE maps (state, event) to a list of (guard, effect, target) rows, tried in order;
the first row whose guard holds is taken. A pair with no row is an illegal transition
and is refused. A target of None is an internal transition: the state does not change
and its entry and exit behaviour do not run. The clock is passed in, so the timeout
is an ordinary event in tests.
"""

PRICES = {"tea": 20, "coffee": 30}
TIMEOUT = 30  # seconds without a selection before the credit is handed back
IN_SERVICE = ("Idle", "Collecting", "Dispensing")  # the substates of the composite state InService
STATES = (*IN_SERVICE, "Maintenance")
EVENTS = ("coin", "select", "cancel", "timeout", "take", "door_open", "door_close", "refill")


class IllegalTransition(Exception):
    pass


def always(m, *args):
    return True


def can_buy(m, item):
    return m.credit >= PRICES[item] and m.stock.get(item, 0) > 0


def add_credit(m, amount):
    m.credit += amount
    m.since = m.clock()


def vend(m, item):
    m.stock[item] -= 1
    m.credit -= PRICES[item]
    m.out.append(item)


def ask_for_more(m, item):
    m.out.append(f"display: {item} costs {PRICES[item]}")


def hand_back_credit(m, *args):
    if m.credit:
        m.out.append(f"coins: {m.credit}")
        m.credit = 0


def add_stock(m, item, count):
    m.stock[item] = m.stock.get(item, 0) + count


def nothing(m, *args):
    pass


TABLE = {
    ("Idle", "coin"): [(always, add_credit, "Collecting")],
    ("Collecting", "coin"): [(always, add_credit, "Collecting")],
    ("Collecting", "select"): [(can_buy, vend, "Dispensing"), (always, ask_for_more, "Collecting")],
    ("Collecting", "cancel"): [(always, hand_back_credit, "Idle")],
    ("Collecting", "timeout"): [(always, hand_back_credit, "Idle")],
    ("Dispensing", "take"): [(always, hand_back_credit, "Idle")],
    ("Maintenance", "refill"): [(always, add_stock, None)],
    ("Maintenance", "door_close"): [(always, nothing, "Idle")],
}
# One arrow on the diagram, from the composite state InService, is one row per substate here.
for _state in IN_SERVICE:
    TABLE[(_state, "door_open")] = [(always, hand_back_credit, "Maintenance")]

ENTRY = {"Maintenance": lambda m: m.out.append("coin slot locked")}
EXIT = {"Maintenance": lambda m: m.out.append("coin slot unlocked")}


class VendingMachine:
    def __init__(self, stock, clock):
        self.state = "Idle"
        self.credit = 0
        self.stock = dict(stock)
        self.clock = clock
        self.since = clock()
        self.out = []

    def fire(self, event, *args):
        rows = TABLE.get((self.state, event))
        if rows is None:
            raise IllegalTransition(f"{event} is not allowed in {self.state}")
        for guard, effect, target in rows:
            if guard(self, *args):
                if target is not None:
                    EXIT.get(self.state, nothing)(self)
                effect(self, *args)
                if target is not None:
                    self.state = target
                    ENTRY.get(target, nothing)(self)
                return self.state
        raise IllegalTransition(f"no guard of {event} holds in {self.state}")

    def tick(self):
        """Turns the passing of time into the timeout event, when one is due."""
        if self.state == "Collecting" and self.clock() - self.since >= TIMEOUT:
            return self.fire("timeout")
        return self.state

session · compiled/java_session.py

from session import run, versions

# VendingFlow.java runs as a single source file; MissingCase.java is expected not to compile.
versions("java -version")
run(
    """
java VendingFlow.java
javac -d classes MissingCase.java
""",
    may_fail=("javac -d classes MissingCase.java",),
)

# openjdk version "25.0.4.1" LTS
$ java VendingFlow.java
Coin[amount=10] -> Collecting[credit=10]
Select[item=tea] -> Collecting[credit=10]
Coin[amount=20] -> Collecting[credit=30]
Select[item=tea] -> Dispensing[item=tea, change=10]
Take[] -> Idle[]
refused: Take[] is not allowed in Idle[]
$ javac -d classes MissingCase.java
MissingCase.java:9: error: the switch expression does not cover all possible input values
        return switch (state) {
               ^
1 error
[exit status 1]

compiled/VendingFlow.java

// The vending machine's states and events as sealed interfaces of records. A switch over
// a sealed type must cover every permitted subtype, so the compiler lists the states.
import java.util.Map;

public class VendingFlow {
    static final Map<String, Integer> PRICES = Map.of("tea", 20, "coffee", 30);

    static VendingState next(VendingState state, Event event) {
        return switch (state) {
            case Idle _ -> switch (event) {
                case Coin(int amount) -> new Collecting(amount);
                case DoorOpen _ -> new Maintenance();
                default -> throw refused(state, event);
            };
            case Collecting(int credit) -> switch (event) {
                case Coin(int amount) -> new Collecting(credit + amount);
                case Select(String item) when credit >= PRICES.get(item) -> new Dispensing(item, credit - PRICES.get(item));
                case Select _ -> state;
                case Cancel _ -> new Idle();
                case DoorOpen _ -> new Maintenance();
                default -> throw refused(state, event);
            };
            case Dispensing(String item, int change) -> switch (event) {
                case Take _ -> new Idle();
                case DoorOpen _ -> new Maintenance();
                default -> throw refused(state, event);
            };
            case Maintenance _ -> switch (event) {
                case DoorClose _ -> new Idle();
                default -> throw refused(state, event);
            };
        };
    }

    static IllegalStateException refused(VendingState state, Event event) {
        return new IllegalStateException(event + " is not allowed in " + state);
    }

    public static void main(String[] args) {
        VendingState state = new Idle();
        for (Event event : new Event[] {new Coin(10), new Select("tea"), new Coin(20), new Select("tea"), new Take()}) {
            state = next(state, event);
            System.out.println(event + " -> " + state);
        }
        try {
            next(state, new Take());
        } catch (IllegalStateException e) {
            System.out.println("refused: " + e.getMessage());
        }
    }
}

sealed interface VendingState permits Idle, Collecting, Dispensing, Maintenance {}
record Idle() implements VendingState {}
record Collecting(int credit) implements VendingState {}
record Dispensing(String item, int change) implements VendingState {}
record Maintenance() implements VendingState {}

sealed interface Event permits Coin, Select, Cancel, Take, DoorOpen, DoorClose {}
record Coin(int amount) implements Event {}
record Select(String item) implements Event {}
record Cancel() implements Event {}
record Take() implements Event {}
record DoorOpen() implements Event {}
record DoorClose() implements Event {}

compiled/MissingCase.java

// VendingFlow.java with one state forgotten: the switch over VendingState has no case for
// Maintenance, and javac refuses to compile it.
import java.util.Map;

public class MissingCase {
    static final Map<String, Integer> PRICES = Map.of("tea", 20, "coffee", 30);

    static VendingState next(VendingState state, Event event) {
        return switch (state) {
            case Idle _ -> switch (event) {
                case Coin(int amount) -> new Collecting(amount);
                case DoorOpen _ -> new Maintenance();
                default -> throw refused(state, event);
            };
            case Collecting(int credit) -> switch (event) {
                case Coin(int amount) -> new Collecting(credit + amount);
                case Select(String item) when credit >= PRICES.get(item) -> new Dispensing(item, credit - PRICES.get(item));
                case Select _ -> state;
                case Cancel _ -> new Idle();
                case DoorOpen _ -> new Maintenance();
                default -> throw refused(state, event);
            };
            case Dispensing(String item, int change) -> switch (event) {
                case Take _ -> new Idle();
                case DoorOpen _ -> new Maintenance();
                default -> throw refused(state, event);
            };
        };
    }

    static IllegalStateException refused(VendingState state, Event event) {
        return new IllegalStateException(event + " is not allowed in " + state);
    }

    public static void main(String[] args) {
        VendingState state = new Idle();
        for (Event event : new Event[] {new Coin(10), new Select("tea"), new Coin(20), new Select("tea"), new Take()}) {
            state = next(state, event);
            System.out.println(event + " -> " + state);
        }
        try {
            next(state, new Take());
        } catch (IllegalStateException e) {
            System.out.println("refused: " + e.getMessage());
        }
    }
}

sealed interface VendingState permits Idle, Collecting, Dispensing, Maintenance {}
record Idle() implements VendingState {}
record Collecting(int credit) implements VendingState {}
record Dispensing(String item, int change) implements VendingState {}
record Maintenance() implements VendingState {}

sealed interface Event permits Coin, Select, Cancel, Take, DoorOpen, DoorClose {}
record Coin(int amount) implements Event {}
record Select(String item) implements Event {}
record Cancel() implements Event {}
record Take() implements Event {}
record DoorOpen() implements Event {}
record DoorClose() implements Event {}

compiled/session.py

"""Replays commands the way a terminal shows them: "$ " and the command, then all it printed.

Standard output and standard error are merged in the order they were written. A command
that exits with a status other than 0 is followed by "[exit status N]"; unless the script
lists it in may_fail, the session then exits with status 1, so a run that could not build
a program is never recorded as if it had worked.

Java is the pinned Java 25 (Eclipse Temurin 25.0.4) from the mise installation next to
the Python that runs this file. It lives in the home folder, which learn:verify's sandbox
does not let a run read, so the sessions of this folder are recorded with
LEARN_VERIFY_NETWORK=allow. They read only this folder and the JDK, write only inside the
run's own folder and temporary home, and use no network.
"""

import os
import re
import subprocess
import sys
from pathlib import Path

# <mise>/installs/python/3.14.8/bin/python3 -> <mise>/installs
INSTALLS = Path(sys.executable).resolve().parents[3]
JDK = INSTALLS / "java" / "temurin-25.0.4+101.0.LTS" / "bin"
# java -version prints a release date; pages show no dates.
RELEASE_DATE = re.compile(r"\s+\d{4}-\d{2}-\d{2}")


def _shell(command):
    env = dict(os.environ, PATH=os.pathsep.join([str(JDK), os.environ.get("PATH", "")]))
    return subprocess.run(command, shell=True, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)


def versions(*commands):
    """Prints the first line of each version command, as a comment."""
    for command in commands:
        done = _shell(command)
        lines = done.stdout.splitlines()
        if done.returncode != 0 or not lines:
            sys.exit(f"{command} did not run: the pinned JDK is missing on this computer")
        print("#", RELEASE_DATE.sub("", lines[0]))


def run(script, may_fail=()):
    """Runs each line of the script as one shell command."""
    unexpected = 0
    for command in (line.strip() for line in script.strip().splitlines()):
        print(f"$ {command}")
        done = _shell(command)
        print(done.stdout, end="")
        if done.returncode != 0:
            print(f"[exit status {done.returncode}]")
            if command not in may_fail:
                unexpected += 1
    if unexpected:
        sys.exit(1)