Design a traffic signal controller

01The design, in the six-step framework

1. Requirements and use cases

A traffic signal controller is a small real-time state machine. Safety is the invariant: two conflicting directions must never be green together, and every design choice serves that.

• The controller cycles through phases at a four-way junction: north-south green, yellow, all-red, east-west green, yellow, all-red.
• A traffic engineer sets timings per plan, such as a longer green at rush hour.
• A pedestrian presses a button, and a walk signal is served at the next safe gap.
• An emergency vehicle requests priority for a direction; the controller moves safely to that green and holds, then resumes.
• Displays and logs are told of every change.
• On a fault, the junction falls back to flashing red.

Out of scope: coordination between junctions, vehicle sensors and adaptive timing beyond a plan switch. Non-functional constraints: every green-to-red goes through yellow and all-red; the logic is a pure function of time and inputs, so it is testable without sleeping; a new plan or phase is data.

2. Core entities and relationships

• Direction: an enum of NS and EW approaches.
• Light: an enum of GREEN, YELLOW, RED.
• Phase: an immutable step: which direction is green or yellow, and for how long.
• TimingPlan: an ordered, cyclic list of phases with a name.
• Controller: current phase, deadline, pending requests; advanced by tick.
• ControllerMode: NORMAL, PREEMPTED, FLASHING.
• SignalListener: observers such as lamp drivers and logs.
• Clock: injected so tests advance time by hand.

A plan composes its phases because they are defined together and ordered. The controller aggregates a plan so plans can be swapped, and a plan can be shared across junctions.

3. Interfaces and key methods

class Clock(Protocol):
    def now(self) -> float: ...

class SignalListener(Protocol):
    def on_change(self, phase: Phase) -> None: ...

class TimingPlan:
    def validate(self) -> None:
        """Reject any phase with two conflicting greens, or a green not followed by yellow."""

class Controller:
    def tick(self) -> None:
        """Advance if the deadline has passed; idempotent between deadlines."""
    def request_walk(self) -> None:
        """Serve the walk at the next all-red phase."""
    def preempt(self, direction: Direction) -> None:
        """Move safely to green for direction and hold."""
    def clear_preempt(self) -> None: ...
    def change_plan(self, plan: TimingPlan) -> None:
        """Takes effect at the next cycle start."""
    def fail(self) -> None:
        """Enter flashing red; only an operator reset leaves it."""

Invariants: at most one direction is non-red in any phase; a change away from green always passes through yellow and all-red; listeners see every phase once, in order.

4. Design patterns and the choices behind them

A state machine with phases as data

The core is a state machine, but its states are data, not classes: a phase has no behaviour of its own, only lights and a duration. Encoding phases in a plan means rush hour is a different list. State classes would suit only the controller's modes, which do behave differently, and a small enum plus one branch is enough at this size.

Time is an input

tick reads an injected clock and acts when a deadline has passed. A sleeping thread per light is the rejected form: it cannot be tested without waiting, and threads drift. This is the inject the clock rule, and it makes every behaviour below a short unit test.

Observers for lamps and logs

The controller does not know about bulbs. Listeners receive the phase; a lamp driver, an audit log and a city dashboard are all equal subscribers.

Validate plans at load

A plan with two greens is dangerous, so validate refuses it before it ever runs. Keeping the safety check at the boundary means the run-time loop stays simple and cannot reach an unsafe state through configuration.

5. Core flow and code

1. A driver loop calls tick several times a second.
2. If the deadline has passed, the controller picks the next phase, sets a new deadline, and notifies listeners.
3. At the all-red phase following a walk request, a walk phase is inserted.
4. preempt(d) jumps to the yellow of the current green unless already green for d, then all-red, then holds green for d until clear_preempt.
5. fail switches to flashing red at once.

from __future__ import annotations
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Protocol


class UnsafePlan(Exception): ...


class Direction(Enum):
    NS = "ns"
    EW = "ew"


class Light(Enum):
    GREEN = "green"; YELLOW = "yellow"; RED = "red"; FLASH_RED = "flash_red"


class ControllerMode(Enum):
    NORMAL = "normal"; PREEMPTED = "preempted"; FLASHING = "flashing"


@dataclass(frozen=True)
class Phase:
    name: str
    ns: Light
    ew: Light
    seconds: float
    walk: bool = False

    def is_green(self, d: Direction) -> bool:
        return (self.ns if d is Direction.NS else self.ew) is Light.GREEN

    def clearing(self) -> bool:
        return self.ns is Light.RED and self.ew is Light.RED


ALL_RED = Phase("all_red", Light.RED, Light.RED, 2)
WALK = Phase("walk", Light.RED, Light.RED, 10, walk=True)
FLASHING = Phase("flashing", Light.FLASH_RED, Light.FLASH_RED, 3600)


def green(d: Direction, seconds: float) -> Phase:
    if d is Direction.NS:
        return Phase("ns_green", Light.GREEN, Light.RED, seconds)
    return Phase("ew_green", Light.RED, Light.GREEN, seconds)


def yellow(d: Direction) -> Phase:
    if d is Direction.NS:
        return Phase("ns_yellow", Light.YELLOW, Light.RED, 3)
    return Phase("ew_yellow", Light.RED, Light.YELLOW, 3)


@dataclass(frozen=True)
class TimingPlan:
    name: str
    phases: tuple[Phase, ...]

    def validate(self) -> None:
        for i, p in enumerate(self.phases):
            if p.ns is not Light.RED and p.ew is not Light.RED:
                raise UnsafePlan(f"{p.name}: two directions active")
            nxt = self.phases[(i + 1) % len(self.phases)]
            if Light.GREEN in (p.ns, p.ew) and Light.YELLOW not in (nxt.ns, nxt.ew):
                raise UnsafePlan(f"{p.name}: green must be followed by yellow")


def standard_plan(ns: float = 30, ew: float = 30) -> TimingPlan:
    return TimingPlan("standard", (green(Direction.NS, ns), yellow(Direction.NS), ALL_RED,
                                   green(Direction.EW, ew), yellow(Direction.EW), ALL_RED))


class Clock(Protocol):
    def now(self) -> float: ...


class SystemClock:
    def now(self) -> float: return time.monotonic()


class SignalListener(Protocol):
    def on_change(self, phase: Phase) -> None: ...


class Controller:
    def __init__(self, plan: TimingPlan, clock: Clock) -> None:
        plan.validate()
        self.plan, self.clock = plan, clock
        self.index = 0
        self.mode = ControllerMode.NORMAL
        self.listeners: list[SignalListener] = []
        self.walk_requested = False
        self.pending_plan: Optional[TimingPlan] = None
        self.script: list[Phase] = []        # a safe sequence that overrides the plan
        self.held: Optional[Phase] = None
        self.current = plan.phases[0]
        self.deadline = clock.now() + self.current.seconds

    def _show(self, phase: Phase) -> None:
        self.current = phase
        self.deadline = self.clock.now() + phase.seconds
        for l in self.listeners:
            l.on_change(phase)

    def _next_from_plan(self) -> Phase:
        if self.current.clearing() and not self.current.walk and self.walk_requested:
            self.walk_requested = False
            return WALK
        self.index = (self.index + 1) % len(self.plan.phases)
        if self.index == 0 and self.pending_plan:
            self.plan, self.pending_plan = self.pending_plan, None
        return self.plan.phases[self.index]

    def tick(self) -> None:
        if self.mode is ControllerMode.FLASHING or self.clock.now() < self.deadline:
            return
        if self.script:
            self._show(self.script.pop(0))
        elif self.held is not None:
            self.deadline = self.clock.now() + self.held.seconds     # keep holding
        else:
            self._show(self._next_from_plan())

    def request_walk(self) -> None:
        self.walk_requested = True

    def change_plan(self, plan: TimingPlan) -> None:
        plan.validate()
        self.pending_plan = plan

    def preempt(self, direction: Direction) -> None:
        self.mode = ControllerMode.PREEMPTED
        self.held = green(direction, 5)
        if self.current.is_green(direction):
            return
        steps: list[Phase] = []
        for d in Direction:
            if self.current.is_green(d):
                steps.append(yellow(d))
        self.script = steps + [ALL_RED, self.held]
        self.deadline = self.clock.now()     # the next tick starts clearing now

    def clear_preempt(self) -> None:
        self.mode = ControllerMode.NORMAL
        direction = Direction.NS if self.current.is_green(Direction.NS) else Direction.EW
        self.held = None
        self.script = [yellow(direction), ALL_RED]
        self.index = len(self.plan.phases) - 1     # resume at the start of the cycle

    def fail(self) -> None:
        self.mode = ControllerMode.FLASHING
        self._show(FLASHING)

6. Extensions, concurrency and what interviewers probe

• Emergency during yellow: preempting mid-yellow must not cut straight to green. The script starts from the current phase; for a yellow phase it adds no second yellow and continues to all-red, so the invariant holds.
• Minimum green: a preempt may cut a green short only after a minimum time so cross traffic can clear; a real controller enforces this in preempt, and pedestrians mid-crossing need their clearance too.
• Rush hour is change_plan, applied at the cycle boundary so timings never jump mid-phase.
• Left-turn arrows and more approaches: replace the two light fields with a map from movement to light and a conflict table, and validate against it. The controller loop does not change.
• Concurrency: buttons and emergency signals arrive from other threads. Put a lock around the controller's methods, or better, queue inputs and let tick drain them, so there is one writer.
• Test first: with a fake clock, 30 seconds gives NS yellow and 33 gives all-red; no phase ever has both directions non-red; a walk request is served after the next all-red, once; a preempt from NS green reaches EW green through yellow and all-red; fail stops the cycle.
• Common mistake: giving each direction its own timer. Two independent timers drift, and that is how both end up green. One controller owns the junction.

02Where this question was reported

• interviewkickstart.com · Interview Kickstart Google list: 'How do you design a traffic control system?'
• Codemia · Codemia Easy
• GitHub · awesome-low-level-design, Easy
• javarevisited.blogspot.com · javarevisited: four directions, pedestrians, rush hour
• irc.pcper.com · 'Google SWE Teaches Low Level Design' YouTube series episodes: Tic-Tac-Toe, Cache, Elevator, Parking Garage, Limit Order Book, Air Traffic Controller (only scra
• interviewkickstart.com · Weak: listicle
• YouTube · Episode 6 of 'Google SWE Teaches Low Level Design' per search listings; watch URL not resolved