Design a traffic signal controller

TL;DR

• You build an intersection controller that walks a ring of phases, green to yellow to all-red to the next green, and publishes each change to signal heads that colour themselves.
• Three decisions carry the interview: the safety invariant is enforced by a constructor, not by scheduling logic; presses are queued commands applied at safe points, so nothing can shorten a yellow; timing is a Strategy whose answer is clamped by the phase.
• Patterns that earn their place: State, Strategy, Command, Observer, Mediator. Singleton is discussed and deliberately not used.

Problem statement

"Design the controller for a signalised intersection. It has several approaches; each approach has a signal head that shows red, yellow or green. Only non-conflicting movements may be green together, and there is a clearance interval between phases. Pedestrians press a button to be let across, emergency vehicles can demand a green, and an engineer can put the intersection into flashing-red maintenance. Timing should be able to adapt to vehicle detectors. Show me how you would prove it is safe."

Requirements

Functional

• An intersection with N approaches grouped into phases; only non-conflicting movements share a phase.
• Each phase runs green, then yellow, then an all-red clearance before the next phase gets green.
• Pedestrian call buttons: the phase that carries that crossing must not be skipped, and its green is at least the walk interval.
• Emergency override: hold a green for one approach until it is cleared, reached through a safe transition.
• Maintenance mode: every head flashes red, and the intersection resumes through a clearance.
• Adaptive timing from vehicle detectors, with fixed timing as the fallback.
• A tick-based simulation and an event log of every stage change and every press.

Non-functional and constraints

• The safety invariant: two conflicting approaches are never green at the same instant, and every change of the green set is preceded by an all-red tick. This is the requirement; everything else is throughput.
• Deterministic: durations are counted in ticks, never measured against a wall clock.
• Thread-safe: presses arrive from buttons, radios and a maintenance laptop while the cycle runs.

Out of scope: the cabinet's hardware conflict monitor (the electromechanical backstop that exists precisely because software fails), signal head optics, and coordination between intersections, which is the follow-up.

Clarifying questions and assumptions

Question to ask	Assumption taken here
Which movements conflict?	Opposing through movements are compatible; anything crossing is not. Direction.conflicts_with is one line and the whole model rests on it.
What is a tick worth?	One second of signal time. Durations are tick counts, so a slow process runs the intersection slowly rather than dangerously.
Can an emergency vehicle get an instant green?	No. It gets the next green: the current green ends early, but the yellow and the all-red run in full.
May a phase be skipped?	Yes, when the detectors report nobody waiting and no pedestrian has called. A pedestrian call makes a phase unskippable.
Who enforces the minimum green?	The phase. The strategy proposes a number and Phase.clamp bounds it, so a bad strategy costs throughput, never safety.
One controller per process, so a Singleton?	No. A green-wave corridor is a dozen controllers in one process, and every test builds its own.

Core entities and relationships

• Phase — a set of movements that may run together plus the bounds on its green. Its constructor rejects conflicting movements, so an unsafe phase cannot be built.
• PhaseCycle — the ring: the fixed order phases are offered in, with lookup by name and by approach.
• IntersectionController — the mediator: it owns the ring position, the stage, the elapsed ticks, the command queue, the pedestrian calls and the event log, all under one lock.
• ControllerState — GREEN, YELLOW, ALL_RED, EMERGENCY, MAINTENANCE: where the intersection is, which is not the same as what any one head is showing.
• TrafficLight and PedestrianSignal — observers. Each receives a SignalUpdate and works out its own colour; neither is ever polled.
• PedestrianCall, EmergencyOverride, ClearOverride — presses as command objects, queued and executed at a safe point.
• TimingStrategy (FixedTiming, AdaptiveTiming) and Sensor (LoopSensor) — the policy and the detectors. The strategy sees only an immutable PhaseDemand.

Multiplicities: intersection 1 -> 1 cycle, cycle 1 -> * phases, phase 1 -> * movements, intersection 1 -> * heads, intersection 1 -> 1 timing strategy.

Class diagram

The ring, the controller and the timing policy.

classDiagram
    class IntersectionController {
        -_stage: ControllerState
        -_index: int
        -_elapsed: int
        -_pending: List~SignalCommand~
        -_pedestrian: Set~str~
        -_override: Direction
        -_lock: Lock
        +tick() int
        +pedestrian_call(direction: Direction) None
        +request_emergency(direction: Direction) None
        +clear_emergency() None
        +enter_maintenance() None
        +green_directions() Set~Direction~
        +event_log() List~str~
    }
    class PhaseCycle {
        +phases: Tuple~Phase~
        +phase(index: int) Phase
        +index_of(name: str) int
        +serving(direction: Direction) Phase
        +directions() Tuple~Direction~
    }
    class Phase {
        +name: str
        +movements: Set~Direction~
        +min_green: int
        +max_green: int
        +serves(direction: Direction) bool
        +clamp(ticks: int) int
    }
    class PhaseDemand {
        +waiting_vehicles: int
        +pedestrian_waiting: bool
    }
    class TimingStrategy {
        <<interface>>
        +name: str
        +green_ticks(demand: PhaseDemand) int
    }
    class FixedTiming
    class AdaptiveTiming
    class Sensor {
        <<interface>>
        +waiting(direction: Direction) int
    }
    class LoopSensor {
        +arrive(direction: Direction, count: int) None
        +depart(direction: Direction, count: int) None
    }
    class Direction {
        <<enumeration>>
        NORTH
        SOUTH
        EAST
        WEST
    }
    class ControllerState {
        <<enumeration>>
        GREEN
        YELLOW
        ALL_RED
        EMERGENCY
        MAINTENANCE
    }
    IntersectionController "1" *-- "1" PhaseCycle
    PhaseCycle "1" *-- "many" Phase
    IntersectionController --> TimingStrategy
    IntersectionController --> Sensor
    IntersectionController ..> PhaseDemand : builds
    TimingStrategy <|.. FixedTiming
    TimingStrategy <|.. AdaptiveTiming
    Sensor <|.. LoopSensor
    Phase --> "many" Direction
    IntersectionController --> ControllerState
    TimingStrategy ..> PhaseDemand

Presses as commands, heads as observers.

classDiagram
    class SignalCommand {
        <<interface>>
        +requested_at: float
        +apply(sink: CommandSink) str
        +label() str
    }
    class CommandSink {
        <<interface>>
        +apply_pedestrian_call(direction: Direction) str
        +apply_emergency(direction: Direction) str
        +apply_clear_override() str
    }
    class PedestrianCall {
        +direction: Direction
    }
    class EmergencyOverride {
        +direction: Direction
    }
    class ClearOverride
    class PhaseListener {
        <<interface>>
        +on_phase_changed(update: SignalUpdate) None
    }
    class SignalUpdate {
        +tick: int
        +phase: str
        +movements: Set~Direction~
        +stage: ControllerState
    }
    class TrafficLight {
        +direction: Direction
        +status() SignalState
    }
    class PedestrianSignal {
        +direction: Direction
        +status() PedestrianState
    }
    class SignalState {
        <<enumeration>>
        RED
        GREEN
        YELLOW
        FLASHING_RED
    }
    class SignalEvent {
        +tick: int
        +at: float
        +message: str
    }
    class IntersectionController
    SignalCommand <|.. PedestrianCall
    SignalCommand <|.. EmergencyOverride
    SignalCommand <|.. ClearOverride
    CommandSink <|.. IntersectionController
    PedestrianCall ..> CommandSink : apply
    PhaseListener <|.. TrafficLight
    PhaseListener <|.. PedestrianSignal
    IntersectionController --> "many" PhaseListener : publishes
    IntersectionController ..> SignalUpdate : builds
    IntersectionController o-- "many" SignalEvent
    TrafficLight --> SignalState

Design patterns applied

Pattern	Where	Why it earns its place
State	ControllerState and the branches in tick	Five stages, and a tick means something different in each: run the current interval down, hold indefinitely, or do nothing at all. The enum-and-table form fits because each transition is a line; classes per state would hide the one thing a reviewer must be able to see at a glance, which is that YELLOW always goes to ALL_RED.
Strategy	TimingStrategy with fixed and adaptive implementations	"Now make it adapt to traffic" is the follow-up. The strategy sees an immutable PhaseDemand and returns a number, so it is a pure function and cannot reach the lights.
Command	PedestrianCall, EmergencyOverride, ClearOverride	Presses arrive at arbitrary moments and must be applied at safe ones. As objects they queue, log and replay; as direct method calls they would mutate the cycle mid-transition.
Observer	PhaseListener, implemented by every head and pedestrian signal	The controller publishes one SignalUpdate; each head decides its own colour from it. Adding telemetry is one more subscriber, and the publish happens outside the lock.
Mediator	IntersectionController	Detectors, heads, buttons and the timing policy never reference each other. The rules of the conversation live in one object you can read top to bottom.
Dependency injection	Sensor, TimingStrategy, Clock	A LoopSensor in tests, a hardware bus in the field; the controller cannot tell.

What was deliberately not used: Singleton. One intersection is one controller, which is why the pattern is tempting, and it is still wrong — the green-wave follow-up is a corridor of controllers in one process, and the safety test builds a fresh one per case. Say that, then say the real backstop: a production cabinet has a hardware conflict monitor that drops the intersection to flashing red when the software commands conflicting greens. Software invariants are the first line, not the only one.

Key flows

A pedestrian press, one tick later, and the heads colouring themselves.

sequenceDiagram
    autonumber
    actor P as Pedestrian
    participant C as IntersectionController
    participant D as LoopSensor
    participant T as AdaptiveTiming
    participant NL as TrafficLight north
    participant EL as TrafficLight east
    P->>C: pedestrian_call(WEST)
    Note over C: queued, not applied yet
    C->>C: tick, drain the queue, mark phase EW unskippable
    C->>C: green expires, yellow, then all-red
    C->>D: waiting(EAST), waiting(WEST)
    D-->>C: 0 and 0
    C->>T: green_ticks(EW, pedestrian waiting)
    T-->>C: 3 ticks
    C->>C: clamp to the phase, stretch to the 8-tick walk
    C-)NL: on_phase_changed(EW is green)
    C-)EL: on_phase_changed(EW is green)
    NL->>NL: not my movement, stay red
    EL->>EL: my movement, stage green, show green

The cycle as a state machine. Note that nothing reaches GREEN or EMERGENCY except through ALL_RED.

stateDiagram-v2
    [*] --> ALL_RED
    ALL_RED --> GREEN : clearance done, next phase chosen
    GREEN --> YELLOW : green ticks used up
    YELLOW --> ALL_RED : yellow ticks used up
    GREEN --> EMERGENCY : override names the phase already green
    ALL_RED --> EMERGENCY : override, clearance done
    EMERGENCY --> YELLOW : override cleared
    GREEN --> MAINTENANCE : cabinet opened
    ALL_RED --> MAINTENANCE : cabinet opened
    MAINTENANCE --> ALL_RED : back in service

The one arrow worth defending is GREEN --> EMERGENCY. When the ambulance is coming from an approach that is already green, dropping to yellow and back would be worse than useless, so the controller converts the running green into a held one with no colour change. When the ambulance needs a different phase, there is no shortcut: the green ends early, and the yellow and all-red run at their full length.

Implementation

Write the vocabulary first, and put the safety invariant in it. Everything after that is scheduling, and scheduling bugs then cost throughput instead of lives.

conflicts_with is the whole safety model: opposing through movements are compatible, anything crossing is not.

code/lld/traffic_signal/models.py — approaches and stages

class Direction(StrEnum):
    """One approach to the intersection, named by where the traffic comes from."""

    NORTH = "north"
    SOUTH = "south"
    EAST = "east"
    WEST = "west"

    def opposite(self) -> Direction:
        pairs = {
            Direction.NORTH: Direction.SOUTH,
            Direction.SOUTH: Direction.NORTH,
            Direction.EAST: Direction.WEST,
            Direction.WEST: Direction.EAST,
        }
        return pairs[self]

    def conflicts_with(self, other: Direction) -> bool:
        """Opposing through movements are compatible; anything crossing is not."""
        return other is not self and other is not self.opposite()


class SignalState(StrEnum):
    """What one signal head is showing."""

    RED = "red"
    GREEN = "green"
    YELLOW = "yellow"
    FLASHING_RED = "flashing_red"


class ControllerState(StrEnum):
    """Where the intersection is in its cycle. Only GREEN and EMERGENCY show green."""

    GREEN = "green"
    YELLOW = "yellow"
    ALL_RED = "all_red"
    EMERGENCY = "emergency"
    MAINTENANCE = "maintenance"


class PedestrianState(StrEnum):
    WALK = "walk"
    DONT_WALK = "dont_walk"

code/lld/traffic_signal/models.py — errors

class ConflictingMovementError(ValidationError):
    """Two movements in the same phase would cross each other."""


class UnknownApproachError(NotFoundError):
    """No phase in this cycle ever gives that approach a green."""


class SignalStateError(InvalidStateError):
    """The operation is not legal for the controller's current state."""

Phase.__post_init__ is the line to point at when you are asked to prove the intersection is safe: a phase whose movements conflict raises at construction, so no scheduler downstream can produce one.

code/lld/traffic_signal/models.py — phases and the ring

@dataclass(frozen=True, slots=True)
class Phase:
    """A set of movements that may run together, and the bounds on its green.

    The constructor is the safety check: `Phase("bad", {NORTH, EAST})` raises,
    so an unsafe phase never exists to be scheduled.
    """

    name: str
    movements: frozenset[Direction]
    min_green: int = 6
    max_green: int = 20

    def __post_init__(self) -> None:
        if not self.movements:
            raise ValidationError(f"phase {self.name!r} has no movements")
        for one in self.movements:
            for other in self.movements:
                if one.conflicts_with(other):
                    raise ConflictingMovementError(
                        f"phase {self.name!r}: {one} and {other} cannot be green together"
                    )
        if not 1 <= self.min_green <= self.max_green:
            raise ValidationError(f"phase {self.name!r} has an impossible green range")

    def serves(self, direction: Direction) -> bool:
        return direction in self.movements

    def clamp(self, ticks: int) -> int:
        return max(self.min_green, min(self.max_green, ticks))


@dataclass(frozen=True, slots=True)
class PhaseCycle:
    """The ring: the fixed order phases are offered in."""

    phases: tuple[Phase, ...]

    def __post_init__(self) -> None:
        if not self.phases:
            raise ValidationError("a cycle needs at least one phase")
        names = [phase.name for phase in self.phases]
        if len(set(names)) != len(names):
            raise ValidationError("phase names must be unique within a cycle")

    def __len__(self) -> int:
        return len(self.phases)

    def phase(self, index: int) -> Phase:
        return self.phases[index % len(self.phases)]

    def index_of(self, name: str) -> int:
        for index, phase in enumerate(self.phases):
            if phase.name == name:
                return index
        raise UnknownApproachError(f"no phase named {name!r}")

    def serving(self, direction: Direction) -> Phase:
        for phase in self.phases:
            if phase.serves(direction):
                return phase
        raise UnknownApproachError(f"no phase serves {direction}")

    def directions(self) -> tuple[Direction, ...]:
        seen: list[Direction] = []
        for phase in self.phases:
            seen.extend(d for d in sorted(phase.movements) if d not in seen)
        return tuple(seen)


@dataclass(frozen=True, slots=True)
class PhaseDemand:
    """What the timing strategy is allowed to know: the phase and who is waiting."""

    phase: Phase
    waiting_vehicles: int
    pedestrian_waiting: bool

Presses become objects because they arrive at the wrong moment. The controller queues them and drains the queue at the top of a tick, before any timing decision.

code/lld/traffic_signal/models.py — presses as commands

class CommandSink(Protocol):
    """The three controller methods a queued command needs."""

    def apply_pedestrian_call(self, direction: Direction) -> str: ...

    def apply_emergency(self, direction: Direction) -> str: ...

    def apply_clear_override(self) -> str: ...


class SignalCommand(Protocol):
    """Command: a button press as an object, executed at the next safe point."""

    requested_at: float

    def apply(self, sink: CommandSink) -> str: ...

    def label(self) -> str: ...


@dataclass(frozen=True, slots=True)
class PedestrianCall:
    """Someone wants to cross alongside `direction`; that phase must not be skipped."""

    direction: Direction
    requested_at: float

    def apply(self, sink: CommandSink) -> str:
        return sink.apply_pedestrian_call(self.direction)

    def label(self) -> str:
        return f"pedestrian call on {self.direction}"


@dataclass(frozen=True, slots=True)
class EmergencyOverride:
    """Hold a green for `direction` until it is cleared."""

    direction: Direction
    requested_at: float

    def apply(self, sink: CommandSink) -> str:
        return sink.apply_emergency(self.direction)

    def label(self) -> str:
        return f"emergency override for {self.direction}"


@dataclass(frozen=True, slots=True)
class ClearOverride:
    """End the override and let the ring resume."""

    requested_at: float

    def apply(self, sink: CommandSink) -> str:
        return sink.apply_clear_override()

    def label(self) -> str:
        return "clear override"

The update that goes out to observers carries the stage and the movements, and nothing else. A head that receives it knows enough to colour itself.

code/lld/traffic_signal/models.py — the published update

@dataclass(frozen=True, slots=True)
class SignalUpdate:
    """What the controller publishes after every tick. Heads decide their own colour."""

    tick: int
    phase: str
    movements: frozenset[Direction]
    stage: ControllerState


@dataclass(frozen=True, slots=True)
class SignalEvent:
    tick: int
    at: float
    message: str

    def render(self) -> str:
        return f"t{self.tick} {self.message}"


class PhaseListener(Protocol):
    """Observer: signal heads, pedestrian signals, telemetry."""

    def on_phase_changed(self, update: SignalUpdate) -> None: ...


class Sensor(Protocol):
    """An induction loop. It reports; it never acts."""

    def waiting(self, direction: Direction) -> int: ...

Timing is a pure function of demand, and whatever it returns is clamped by the phase.

code/lld/traffic_signal/strategies.py — fixed and adaptive timing

class TimingStrategy(Protocol):
    name: str

    def green_ticks(self, demand: PhaseDemand) -> int: ...


class FixedTiming:
    """The same green for every phase, every cycle. What most intersections still run."""

    name = "fixed"

    def __init__(self, green: int = 10) -> None:
        if green < 1:
            raise ValueError("green must be at least one tick")
        self._green = green

    def green_ticks(self, demand: PhaseDemand) -> int:
        return self._green


class AdaptiveTiming:
    """A base green plus one tick for every `per_tick` vehicles the loops report.

    The clamp lives in the controller, so this can be tuned aggressively: the
    worst it can do is ask for a green the phase will refuse to give.
    """

    name = "adaptive"

    def __init__(self, base: int = 6, per_tick: int = 2) -> None:
        if base < 1 or per_tick < 1:
            raise ValueError("base and per_tick must be positive")
        self._base = base
        self._per_tick = per_tick

    def green_ticks(self, demand: PhaseDemand) -> int:
        return self._base + demand.waiting_vehicles // self._per_tick

The heads are observers with a three-line rule; the pedestrian signal walks with the traffic beside it and never during an emergency hold.

code/lld/traffic_signal/services.py — heads, walk signals and loops

class TrafficLight:
    """One signal head. It is an observer: it works out its own colour from the update."""

    def __init__(self, direction: Direction) -> None:
        self.direction = direction
        self._state = SignalState.RED
        self._lock = threading.Lock()

    def on_phase_changed(self, update: SignalUpdate) -> None:
        colour = self._colour_for(update)
        with self._lock:
            self._state = colour

    def status(self) -> SignalState:
        with self._lock:
            return self._state

    def _colour_for(self, update: SignalUpdate) -> SignalState:
        if update.stage is ControllerState.MAINTENANCE:
            return SignalState.FLASHING_RED
        if self.direction not in update.movements:
            return SignalState.RED
        if update.stage in (ControllerState.GREEN, ControllerState.EMERGENCY):
            return SignalState.GREEN
        if update.stage is ControllerState.YELLOW:
            return SignalState.YELLOW
        return SignalState.RED


class PedestrianSignal:
    """Walks with the traffic beside it, and never during an emergency hold."""

    def __init__(self, direction: Direction) -> None:
        self.direction = direction
        self._state = PedestrianState.DONT_WALK
        self._lock = threading.Lock()

    def on_phase_changed(self, update: SignalUpdate) -> None:
        walking = update.stage is ControllerState.GREEN and self.direction in update.movements
        with self._lock:
            self._state = PedestrianState.WALK if walking else PedestrianState.DONT_WALK

    def status(self) -> PedestrianState:
        with self._lock:
            return self._state


class LoopSensor:
    """A stand-in for induction loops: the environment drives it, the controller reads it."""

    def __init__(self, queues: Mapping[Direction, int] | None = None) -> None:
        self._queues: dict[Direction, int] = dict(queues or {})
        self._lock = threading.Lock()

    def waiting(self, direction: Direction) -> int:
        with self._lock:
            return self._queues.get(direction, 0)

    def arrive(self, direction: Direction, count: int = 1) -> None:
        with self._lock:
            self._queues[direction] = self._queues.get(direction, 0) + count

    def depart(self, direction: Direction, count: int = 1) -> None:
        with self._lock:
            self._queues[direction] = max(0, self._queues.get(direction, 0) - count)

    def snapshot(self) -> dict[Direction, int]:
        with self._lock:
            return dict(self._queues)

Then the controller. Read tick first: drain the queue, spend a tick, and advance the stage when the current one is used up. _start_green is where the override, the skipping rule, the strategy and the clamp all meet.

code/lld/traffic_signal/services.py — the controller

class IntersectionController:
    """Mediator: loops, heads, buttons and the timing policy only ever talk to this.

    The cycle is a tick counter, not a wall clock: `tick` is called once per
    second of signal time, and every duration is a number of ticks. A slow
    process therefore runs the intersection slowly; it can never shorten a
    yellow, which is the failure that kills people.
    """

    def __init__(
        self,
        cycle: PhaseCycle,
        sensor: Sensor | None = None,
        timing: TimingStrategy | None = None,
        clock: Clock | None = None,
        yellow_ticks: int = 3,
        all_red_ticks: int = 2,
        walk_ticks: int = 8,
    ) -> None:
        if yellow_ticks < 1 or all_red_ticks < 1:
            raise ValueError("yellow and all-red clearance must be at least one tick each")
        shortest_max = min(phase.max_green for phase in cycle.phases)
        if walk_ticks > shortest_max:
            raise ValueError(f"walk interval {walk_ticks} exceeds the shortest max green")
        self._cycle = cycle
        self._sensor = sensor
        self._timing = timing or FixedTiming()
        self._clock = clock or SystemClock()
        self._yellow = yellow_ticks
        self._all_red = all_red_ticks
        self._walk = walk_ticks
        self._lights = {d: TrafficLight(d) for d in cycle.directions()}
        self._listeners: list[PhaseListener] = list(self._lights.values())
        self._index = 0
        self._stage = ControllerState.ALL_RED  # start dark-safe, first green after a clearance
        self._elapsed = 0
        self._duration = all_red_ticks
        self._ticks = 0
        self._pending: list[SignalCommand] = []
        self._pedestrian: set[str] = set()
        self._override: Direction | None = None
        self._log: list[SignalEvent] = []
        self._lock = threading.Lock()

    # -- the invoker side of Command ---------------------------------------------------
    def submit(self, command: SignalCommand) -> None:
        """Queue a press. It is executed on the next tick, never mid-transition."""
        with self._lock:
            self._pending.append(command)
            self._record(f"queued: {command.label()}")

    def pedestrian_call(self, direction: Direction) -> None:
        self._cycle.serving(direction)  # fail fast: this intersection has no such approach
        self.submit(PedestrianCall(direction, self._clock.now()))

    def request_emergency(self, direction: Direction) -> None:
        self._cycle.serving(direction)
        self.submit(EmergencyOverride(direction, self._clock.now()))

    def clear_emergency(self) -> None:
        self.submit(ClearOverride(self._clock.now()))

    # -- the tick ----------------------------------------------------------------------
    def tick(self) -> int:
        with self._lock:
            self._ticks += 1
            if self._stage is not ControllerState.MAINTENANCE:
                self._drain_commands()
                self._elapsed += 1
                if self._stage is ControllerState.EMERGENCY:
                    if self._override is None:
                        self._begin(ControllerState.YELLOW, self._yellow)
                elif self._elapsed >= self._duration:
                    self._advance()
            update = self._update()
            listeners = list(self._listeners)
        for listener in listeners:  # outside the lock
            listener.on_phase_changed(update)
        return self._ticks

    def run(self, ticks: int) -> None:
        for _ in range(ticks):
            self.tick()

    # -- operator ----------------------------------------------------------------------
    def enter_maintenance(self) -> None:
        with self._lock:
            self._override = None
            self._pending.clear()
            self._begin(ControllerState.MAINTENANCE, 0)
            update, listeners = self._update(), list(self._listeners)
        for listener in listeners:
            listener.on_phase_changed(update)

    def leave_maintenance(self) -> None:
        with self._lock:
            if self._stage is not ControllerState.MAINTENANCE:
                raise SignalStateError("the intersection is not in maintenance")
            self._begin(ControllerState.ALL_RED, self._all_red)  # resume through a clearance
            update, listeners = self._update(), list(self._listeners)
        for listener in listeners:
            listener.on_phase_changed(update)

    # -- reads --------------------------------------------------------------------------
    def subscribe(self, listener: PhaseListener) -> None:
        with self._lock:
            self._listeners.append(listener)
            update = self._update()
        listener.on_phase_changed(update)

    def light(self, direction: Direction) -> TrafficLight:
        return self._lights[direction]

    def stage(self) -> ControllerState:
        with self._lock:
            return self._stage

    def current_phase(self) -> str:
        with self._lock:
            return self._cycle.phase(self._index).name

    def green_directions(self) -> frozenset[Direction]:
        """The safety invariant is asserted against this on every tick of every test."""
        with self._lock:
            if self._stage in (ControllerState.GREEN, ControllerState.EMERGENCY):
                return frozenset(self._cycle.phase(self._index).movements)
            return frozenset()

    def pending_pedestrians(self) -> set[str]:
        with self._lock:
            return set(self._pedestrian)

    def event_log(self) -> list[str]:
        with self._lock:
            return [event.render() for event in self._log]

    # -- the command sink: called with `_lock` already held --------------------------------
    def apply_pedestrian_call(self, direction: Direction) -> str:
        phase = self._cycle.serving(direction)
        self._pedestrian.add(phase.name)
        self._record(f"pedestrian waiting on {direction}, phase {phase.name} cannot be skipped")
        return phase.name

    def apply_emergency(self, direction: Direction) -> str:
        self._cycle.serving(direction)  # UnknownApproachError before anything changes
        self._override = direction
        if self._stage is ControllerState.GREEN:
            if self._cycle.phase(self._index).serves(direction):
                self._begin(ControllerState.EMERGENCY, 0)  # already green: just hold it
            else:
                self._duration = self._elapsed  # end this green now, yellow on the next tick
                self._record(f"emergency for {direction}: cutting the green short")
        return direction.value

    def apply_clear_override(self) -> str:
        self._override = None
        self._record("override cleared, the ring resumes")
        return ""

    # -- the state machine: called with `_lock` already held --------------------------------
    def _drain_commands(self) -> None:
        """Apply the queue. A bad command is logged and dropped: `tick` must never raise."""
        pending, self._pending = self._pending, []
        for command in pending:
            try:
                command.apply(self)
            except HandbookError as exc:
                self._record(f"dropped {command.label()}: {exc}")

    def _advance(self) -> None:
        if self._stage in (ControllerState.GREEN, ControllerState.EMERGENCY):
            self._begin(ControllerState.YELLOW, self._yellow)
        elif self._stage is ControllerState.YELLOW:
            self._begin(ControllerState.ALL_RED, self._all_red)
        else:
            self._start_green()

    def _start_green(self) -> None:
        if self._override is not None:
            phase = self._cycle.serving(self._override)
            self._index = self._cycle.index_of(phase.name)
            self._pedestrian.discard(phase.name)
            self._begin(ControllerState.EMERGENCY, 0)
            return
        self._index = self._next_index()
        phase = self._cycle.phase(self._index)
        demand = PhaseDemand(phase, self._waiting_for(phase), phase.name in self._pedestrian)
        green = phase.clamp(self._timing.green_ticks(demand))
        if demand.pedestrian_waiting:
            green = max(green, self._walk)
            self._pedestrian.discard(phase.name)
        self._begin(ControllerState.GREEN, green)

    def _next_index(self) -> int:
        """Skip phases nobody is waiting for; never skip one with a pedestrian call."""
        count = len(self._cycle)
        for step in range(1, count + 1):
            candidate = (self._index + step) % count
            phase = self._cycle.phase(candidate)
            if self._waiting_for(phase) or phase.name in self._pedestrian:
                return candidate
        return (self._index + 1) % count

    def _waiting_for(self, phase: Phase) -> int:
        if self._sensor is None:
            return 0
        return sum(self._sensor.waiting(direction) for direction in phase.movements)

    def _begin(self, stage: ControllerState, duration: int) -> None:
        self._stage = stage
        self._elapsed = 0
        self._duration = duration
        phase = self._cycle.phase(self._index).name
        detail = f" for {duration} ticks" if duration else ""
        self._record(f"{stage} on {phase}{detail}")

    def _update(self) -> SignalUpdate:
        phase = self._cycle.phase(self._index)
        return SignalUpdate(self._ticks, phase.name, frozenset(phase.movements), self._stage)

    def _record(self, message: str) -> None:
        self._log.append(SignalEvent(self._ticks, self._clock.now(), message))

Running python -m lld.traffic_signal.demo prints one line per stage change, with the four heads and the detector queues:

--- two phases, adaptive green, queues shown as N/S/E/W ---
t  2 green     EW  N:R S:R E:G W:G  west ped walk queues 7/4/6/0
ambulance from the north: the east-west green is cut short, then held for it
t  9 yellow    EW  N:R S:R E:Y W:Y  west ped wait queues 11/7/0/0
t 12 all_red   EW  N:R S:R E:R W:R  west ped wait queues 13/8/0/0
t 14 emergency NS  N:G S:G E:R W:R  west ped wait queues 11/6/0/0
ambulance clear, the ring resumes through yellow and all-red
t 23 yellow    NS  N:Y S:Y E:R W:R  west ped wait queues 1/0/0/0
t 26 all_red   NS  N:R S:R E:R W:R  west ped wait queues 3/1/0/0
t 28 green     NS  N:G S:G E:R W:R  west ped wait queues 1/0/0/0
west button pressed, and no car has waited there for twenty ticks
t 36 yellow    NS  N:Y S:Y E:R W:R  west ped wait queues 0/1/0/0
t 39 all_red   NS  N:R S:R E:R W:R  west ped wait queues 2/2/0/0
t 41 green     EW  N:R S:R E:G W:G  west ped walk queues 3/3/0/0
maintenance N:F S:F E:F W:F  stage maintenance
t39 all_red on NS for 2 ticks
t41 green on EW for 8 ticks
t46 maintenance on EW

Three things in that trace are worth narrating. The ambulance request arrives while east-west is green: the green ends at t9 instead of running its full length, but the yellow still gets its three ticks and the clearance its two, and only then does north-south come up as a held EMERGENCY. After the override is cleared, north-south drains to zero and the east-west phase is skipped entirely, because the loops report nobody there. Then the west button is pressed, and the log line green on EW for 8 ticks shows the walk interval winning over the adaptive strategy's proposal of three.

Concurrency and edge cases

Which lock protects what. One lock, IntersectionController._lock, guards the stage, the elapsed count, the ring index, the command queue, the pedestrian calls, the override and the log. That is deliberately coarse: they are one invariant, not six, and a tick is a handful of integer operations. An uncontended lock costs about 17 ns (see the latency numbers), so even at ten presses a second the locking is noise next to the once-a-second tick. Splitting it would buy nothing and would let a press land halfway through a transition.

The heads have their own small lock because they are read from other threads, and the controller publishes to them outside its own lock. A telemetry subscriber that blocks on a socket therefore delays its own update, not the yellow.

Presses can never break a transition. submit only appends; _drain_commands runs at the top of the next tick. An override that arrives during a yellow sets a flag and nothing else, so the yellow and the all-red are untouchable by design rather than by review. The test for this asserts the exact stage sequence after an override arrives mid-yellow.

tick must never raise. A command that fails validation is logged and dropped rather than propagating out of the tick loop, and the public request_emergency validates the approach eagerly so a caller's mistake is a caller's exception. An intersection that stops ticking because of a bad button press is a worse failure than any it could prevent.

Tick drift. Durations are tick counts, so the controller has no way to shorten an interval when the process is slow. If you drove it from Clock.now() deltas instead, a garbage-collection pause would extend a green harmlessly but a clock adjustment could truncate a yellow. The injected clock only stamps events for correlation with city telemetry.

Other edges handled: an emergency for an approach this intersection does not serve; leaving maintenance twice; a walk interval longer than the shortest maximum green, refused at construction; a strategy that returns an absurd number, clamped by the phase; a phase with no demand and no pedestrian, skipped; every phase quiet at once, in which case the ring simply advances.

Common mistake

Modelling one light per approach with its own timer. It looks natural and it is unsafe: two timers can both be in green, and no amount of testing proves they cannot. Model the intersection as one state machine whose green set is derived from the phase, and the conflicting-green bug becomes unrepresentable rather than unlikely.

Extensibility and follow-ups

• Green-wave corridors: give the controller an offset and a cycle length so its phases start at a fixed point in a shared cycle; a coordinator holds a dozen controllers and sets their offsets. Nothing inside the ring changes, which is the payoff for not making it a Singleton.
• Transit priority: an EmergencyOverride with a lower rank. Extend the queue to a priority queue and let a bus request an extension of a running green rather than a new phase.
• Fault detection: the heads already receive every update; a ConflictMonitor listener that also reads the lamp feedback and calls enter_maintenance on a mismatch is one more subscriber.
• Left-turn phases: a phase per protected movement, added to the ring. The constructor's conflict check does the reasoning for you.
• Actuated timing with gap-out: AdaptiveTiming extends the green only while vehicles keep crossing the loop. The strategy needs one more field on PhaseDemand and no change to the controller.
• City-wide telemetry: the event log becomes a stream, and the question turns into a system-design one about ingest, retention and per-corridor dashboards.

Interview tip

When you are asked to prove a safety property, show where it is impossible to violate, not where it is checked. "A Phase with conflicting movements raises in its constructor, and the green set is always exactly one phase's movements" is a two-sentence proof. Then offer the test that asserts it on every tick of a sixty-tick run with overrides and pedestrian calls injected, and mention the hardware conflict monitor that exists because software is not enough.

Tests

tests/test_traffic_signal.py has 13 cases. The two worth walking through are the ring test, which pins the exact stage sequence, and the safety test, which asserts the invariant on every tick of a run that includes a pedestrian call and an override:

code/lld/traffic_signal/tests/test_traffic_signal.py — the ring

def test_the_ring_runs_green_yellow_all_red_then_the_next_phase() -> None:
    controller = build()
    assert stages(controller, 12) == [
        ControllerState.ALL_RED,  # the intersection starts dark-safe
        *[ControllerState.GREEN] * 5,  # five ticks of green, as FixedTiming asked
        *[ControllerState.YELLOW] * 2,
        *[ControllerState.ALL_RED] * 2,  # mandatory clearance between two greens
        *[ControllerState.GREEN] * 2,
    ]
    assert controller.current_phase() == "NS"  # the ring moved on

code/lld/traffic_signal/tests/test_traffic_signal.py — the safety invariant

def test_conflicting_approaches_are_never_green_and_every_change_clears_first() -> None:
    controller = build(sensor=LoopSensor({N: 4, S: 2, E: 6, W: 3}), timing=AdaptiveTiming())
    seen: list[frozenset[Direction]] = []
    for tick in range(1, 61):
        if tick == 12:
            controller.pedestrian_call(W)
        if tick == 20:
            controller.request_emergency(E)
        if tick == 30:
            controller.clear_emergency()
        controller.tick()
        green = controller.green_directions()
        assert not any(one.conflicts_with(other) for one in green for other in green)
        seen.append(green)

    changes = [g for i, g in enumerate(seen) if i == 0 or g != seen[i - 1]]
    for index, green in enumerate(changes):
        if green and index:
            assert not changes[index - 1], "a green set must be preceded by an all-red"

The concurrency test runs one thread ticking forty times while twenty-four threads press buttons, and asserts that no tick ever saw conflicting greens and that all twenty-four presses reached the queue. Be honest about what it proves: the invariant survives because it is structural — a Phase cannot hold conflicting movements, and _index only moves while the stage is ALL_RED — so this test would still pass with the lock removed. It is a regression net over the whole machine, not evidence for the lock. What the lock actually buys is that a reader never sees a half-applied transition and that enter_maintenance cannot be overwritten by a tick already in flight; say that rather than claiming the mutex is what keeps the intersection safe.

code/lld/traffic_signal/tests/test_traffic_signal.py — presses under load

def test_presses_from_many_threads_never_break_the_safety_invariant() -> None:
    controller = build(sensor=LoopSensor({N: 5, S: 5, E: 5, W: 5}), timing=AdaptiveTiming())
    violations: list[frozenset[Direction]] = []
    presses = 24

    def press(index: int) -> None:
        if index % 3 == 0:
            controller.pedestrian_call(W)
        elif index % 3 == 1:
            controller.request_emergency(N if index % 2 else E)
        else:
            controller.clear_emergency()

    def drive(_: int) -> None:
        for _ in range(40):
            controller.tick()
            green = controller.green_directions()
            if any(one.conflicts_with(other) for one in green for other in green):
                violations.append(green)

    with ThreadPoolExecutor(max_workers=8) as pool:
        futures = [pool.submit(drive, 0)]
        futures += [pool.submit(press, index) for index in range(presses)]
        for future in futures:
            future.result()

    assert violations == []
    log = controller.event_log()
    assert sum(1 for line in log if "queued:" in line) == presses  # no press was lost

The rest cover: a conflicting phase refused at construction; an override arriving mid-yellow; an override for an approach that is already green; a quiet phase skipped until a pedestrian calls, then stretched to the walk interval; adaptive green growing with the queue and stopping at the cap; maintenance flashing and resuming through a clearance; an unknown approach; and the heads updating without being polled. Run them with uv run pytest code/lld/traffic_signal -q.

45-minute pacing

Minutes	What to do	What to say or write
0-5	Clarify	Which movements conflict? Pedestrians, emergency vehicles, maintenance? Adaptive timing? Out of scope: the hardware conflict monitor, inter-intersection coordination.
5-11	Safety first	Write conflicts_with and Phase with its constructor check. State the invariant out loud before any scheduling code.
11-18	The ring	Draw the five-stage diagram, then PhaseCycle. Mark that nothing reaches green except through all-red.
18-33	Code	tick, _advance, _start_green. Add the command queue and say why a press cannot be applied where it arrives.
33-39	Timing and observers	TimingStrategy with the clamp, then the heads colouring themselves from one update.
39-45	Extensions	Green waves, transit priority, fault detection, and the hand-off to a city-scale system design.

Related

• State — the five stages as an enum with guarded transitions
• Strategy — fixed and adaptive timing behind one method
• Command — presses queued and applied at safe points
• Mediator — the controller as the only object that knows everyone
• Design an elevator system — the other tick-driven controller, with dispatch instead of phases
• Concurrency for LLD in Python — why one coarse lock is right here