The 100 problems, solved / Booking, allocation and simulators / 063 · Design a train platform management system

Design a train platform management system

Assign platforms to arriving trains, delays and conflicts over time.

Concepts this design leans on

01The design, in the six-step framework

1. Requirements and use cases

A station has a few platforms and many trains; each train occupies a platform from arrival to departure, delays move those windows, and a train that cannot get a platform waits outside. It is interval scheduling with resources, plus a simulation clock.

  • Operations loads a timetable: trains with scheduled arrival and departure times and required platform features (length, electrified).
  • The system assigns a platform to each train so no two trains overlap on one platform, respecting features and a minimum turnaround gap.
  • Operations reports a delay; the train's window moves and conflicts are re-resolved for it and anything it now blocks.
  • A train with no feasible platform is held and assigned as soon as a platform frees.
  • Operations queries which train is on which platform at a time, and the hold queue.

Out of scope: track routing between platforms, passenger information, multi-station coordination. Constraints: assignment for a day of a few hundred trains runs in well under a second; the preference rule (nearest free platform, or the one that frees latest) is swappable; the simulation is deterministic given the event list.

2. Core entities and relationships

  • Train: id, scheduled window, length and needs; an immutable record.
  • Platform: id, length, features, and a sorted list of Occupations.
  • Occupation: a train on a platform over a window, including turnaround.
  • Window: half-open arrival and departure with overlaps.
  • PlatformPolicy: which feasible platform to prefer; a strategy.
  • Station: owns platforms, the hold queue (a priority queue by arrival), and the event handlers for arrivals, departures and delays.
StationResourcesValues1..*0..*Stationplatformstrainsheld: heap[(arrival, id)]policyturnaroundload(trains)assign(train) -> Platform | Nonedelay(train_id, minutes)depart(train_id)at(time) -> dictPlatformPolicychoose(train, feasible) -> PlatformPlatformidlengthfeaturesoccupations: list[Occupation]fits(train)free_for(window)book(train, window)release(train_id)Occupationtrain_idwindow: WindowTrainidwindow: Windowlengthneeds: frozensetWindowarrivedepartoverlaps(other)shifted(minutes)padded(gap)

Platforms compose their occupations: an occupation is meaningless without its platform. Occupations aggregate trains because a delayed train is the same train on possibly another platform.

3. Interfaces and key methods

from dataclasses import dataclass, field
from typing import Optional, Protocol

@dataclass(frozen=True)
class Window:
    arrive: int      # minutes from midnight
    depart: int
    def overlaps(self, other: "Window") -> bool:
        return self.arrive < other.depart and other.arrive < self.depart
    def shifted(self, minutes: int) -> "Window":
        return Window(self.arrive + minutes, self.depart + minutes)
    def padded(self, gap: int) -> "Window":
        """Add the turnaround gap after departure."""
        return Window(self.arrive, self.depart + gap)

@dataclass(frozen=True)
class Train:
    id: str
    window: Window
    length: int
    needs: frozenset = frozenset()

class Platform:
    def fits(self, train: Train) -> bool:
        """Length and every needed feature."""
    def free_for(self, window: Window) -> bool:
        """No occupation overlaps; binary search on arrival times."""
    def book(self, train: Train, window: Window) -> None: ...
    def release(self, train_id: str) -> None: ...

class PlatformPolicy(Protocol):
    def choose(self, train: Train, feasible: list) -> "Platform":
        """Pick among platforms that fit and are free; e.g. the one that frees latest before arrival."""

class Station:
    def assign(self, train: Train) -> Optional["Platform"]:
        """Book a feasible platform or push the train onto the hold heap; returns the platform or None."""
    def delay(self, train_id: str, minutes: int) -> None:
        """Release, shift the window, re-assign; re-run assignment for held trains."""
    def depart(self, train_id: str) -> None:
        """Release the platform and try to place held trains in arrival order."""

Invariants: on each platform, occupations are sorted by arrival and pairwise non-overlapping once padded by the turnaround gap; a train is on at most one platform; a train is in the hold heap if and only if it has no occupation; the policy is only ever asked to choose among already-feasible platforms.

4. Design patterns and the choices behind them

Per-platform sorted intervals

The force: "is platform 3 free from 10:05 to 10:20" asked hundreds of times. Each platform keeps its occupations sorted by arrival; free_for is a bisect plus a check of the two neighbours, the standard move from interval scheduling. Padding the window by the turnaround gap at booking time means the overlap test is one rule everywhere, instead of a special case in every comparison.

A heap of held trains

The force: when a platform frees, the earliest-arriving held train should get it, not the one that happened to be checked first. A priority queue keyed by arrival time gives that ordering in O(log n) and keeps the station's departure handler to a loop of pop, try, push back.

Preference as a strategy, train life cycle as state

Which platform to prefer varies by station: the one nearest the concourse, the one that minimises fragmentation (frees latest before this arrival), the one that balances usage. PlatformPolicy.choose is a strategy fed only feasible options, so correctness never depends on it. A train's status (held, on platform, departed) is a small state machine the station keeps through its three event methods: a train is in held, in where, or gone from trains.

5. Core flow and code

  1. load sorts trains by arrival and calls assign for each, which is the batch plan for the day.
  2. assign pads the train's window by the turnaround gap, filters platforms by fits and free_for, asks the policy to choose, and books; with no feasible platform the train is pushed onto the hold heap.
  3. delay releases the train's occupation, shifts its window, and re-assigns it; then it re-runs assign for every train that was booked after it on the same platform, because the shifted window may now overlap them.
  4. depart releases the platform, forgets the train, and tries every held train in arrival order; one that still fits nowhere stays held and does not block a later one that fits.
import bisect, heapq
from dataclasses import dataclass, field
from typing import Optional

@dataclass(frozen=True)
class Occupation:
    train_id: str
    window: Window

class Platform:
    def __init__(self, id: str, length: int, features: frozenset = frozenset()):
        self.id, self.length, self.features = id, length, features
        self.occupations: list = []            # sorted by window.arrive

    def fits(self, train: Train) -> bool:
        return train.length <= self.length and train.needs <= self.features

    def _idx(self, arrive: int) -> int:
        return bisect.bisect_left(self.occupations, arrive, key=lambda o: o.window.arrive)

    def free_for(self, window: Window) -> bool:
        i = self._idx(window.arrive)
        for j in (i - 1, i):
            if 0 <= j < len(self.occupations) and self.occupations[j].window.overlaps(window):
                return False
        return True

    def book(self, train: Train, window: Window) -> None:
        self.occupations.insert(self._idx(window.arrive), Occupation(train.id, window))

    def release(self, train_id: str) -> Optional[Occupation]:
        for k, o in enumerate(self.occupations):
            if o.train_id == train_id:
                return self.occupations.pop(k)
        return None

    def frees_before(self, t: int) -> int:
        """Departure (padded) of the last occupation ending at or before t; -1 if none."""
        ends = [o.window.depart for o in self.occupations if o.window.depart <= t]
        return max(ends) if ends else -1

class LatestFreePolicy:
    """Prefer the platform that became free most recently: keeps long gaps intact for long trains."""
    def choose(self, train: Train, feasible: list) -> Platform:
        return min(feasible, key=lambda p: (-p.frees_before(train.window.arrive), p.id))

class Station:
    def __init__(self, platforms: list, policy: PlatformPolicy = None, turnaround: int = 5):
        self.platforms = {p.id: p for p in platforms}
        self.policy = policy or LatestFreePolicy()
        self.turnaround = turnaround
        self.trains: dict = {}
        self.where: dict = {}                  # train id -> platform id
        self.held: list = []                   # heap of (arrive, train id)

    def assign(self, train: Train) -> Optional[Platform]:
        self.trains[train.id] = train
        window = train.window.padded(self.turnaround)
        feasible = [p for p in self.platforms.values() if p.fits(train) and p.free_for(window)]
        if not feasible:
            heapq.heappush(self.held, (train.window.arrive, train.id))
            return None
        platform = self.policy.choose(train, feasible)
        platform.book(train, window)
        self.where[train.id] = platform.id
        return platform

    def load(self, trains: list) -> None:
        for t in sorted(trains, key=lambda t: (t.window.arrive, t.id)):
            self.assign(t)

    def _unbook(self, train_id: str) -> None:
        pid = self.where.pop(train_id, None)
        if pid is not None:
            self.platforms[pid].release(train_id)
        self.held = [(a, i) for a, i in self.held if i != train_id]
        heapq.heapify(self.held)

    def delay(self, train_id: str, minutes: int) -> None:
        old = self.trains[train_id]
        pid = self.where.get(train_id)
        later = []
        if pid is not None:   # trains booked after it on the same platform may now be in conflict
            later = [o.train_id for o in self.platforms[pid].occupations if o.window.arrive > old.window.arrive]
        self._unbook(train_id)
        moved = Train(old.id, old.window.shifted(minutes), old.length, old.needs)
        self.assign(moved)
        for tid in later:
            if self.platforms[self.where[tid]].release(tid) is not None:
                del self.where[tid]
                self.assign(self.trains[tid])
        self._place_held()

    def depart(self, train_id: str) -> None:
        self._unbook(train_id)
        self.trains.pop(train_id, None)        # a departed train cannot be delayed back onto a platform
        self._place_held()

    def _place_held(self) -> None:
        waiting, self.held = self.held, []
        for arrive, tid in sorted(waiting):
            self.assign(self.trains[tid])

    def at(self, t: int) -> dict:
        def on_platform(o: Occupation) -> bool:   # the turnaround gap blocks the platform, but the train has left
            return o.window.arrive <= t < self.trains[o.train_id].window.depart
        return {p.id: next((o.train_id for o in p.occupations if on_platform(o)), None)
                for p in self.platforms.values()}

Delays are handled by unbooking and re-running the same assign, not by a separate repair algorithm; that keeps one code path for feasibility and makes the simulation deterministic because the re-assignment order is fixed (the delayed train first, then later trains on its old platform by arrival, then the held heap).

6. Extensions, concurrency and what interviewers probe

  • "Minimise the number of platforms needed." It is the maximum number of padded windows overlapping at any instant; run a sweep line over the loaded timetable and say the matching argument.
  • "A platform closes for maintenance." Remove it, re-assign its occupations by arrival through assign; the heap catches what no longer fits.
  • "Prefer short walks for passengers." Another PlatformPolicy; feasibility is untouched.
  • Concurrency: the station is a single-threaded simulation driven by an event list; if signals and dispatchers push events concurrently, serialise them through one queue rather than locking per platform, because delay touches several platforms.
  • Test first: two trains separated by less than the turnaround gap do not share a platform; a delayed train that no longer fits is held and placed when the next departs; the feature filter excludes an electric train from a non-electrified platform.
  • Common mistake: checking only the next occupation on a platform and not the previous one, and forgetting the turnaround gap so a train is assigned to a platform the moment the previous one leaves.

02Where this question was reported

  • LLD Mastery · Google-tagged (Medium); platform assignment, schedules
  • PracHub · Train-schedule coding question alongside a solutions-design round