Slotting Architecture · 9 min read

How to Detect Aisle Congestion from Pick-Density Heatmaps

You have a stream of confirmed pick events, each stamped with an aisle and a timestamp, and you need to know where and when your floor bottlenecks — not as a vague sense that “the middle aisles get busy at lunch,” but as a concrete list of aisle-timeslot cells that crossed a congestion threshold. A raw pick log cannot tell you that directly: picks are scattered across seconds and aisles, and the signal only appears once you bin them into intervals and count. This page builds the focused function that turns that event stream into a density heatmap and flags the congested cells, guarding the flag with a concurrent-picker floor so a single fast picker’s burst is never mistaken for a crowd. It is the detection how-to behind the Congestion Modeling & Zone Routing cluster.

Binning a pick-event stream into an aisle-by-interval density heatmap On the left, a stream of individual pick events, each carrying an aisle, a timestamp, and a picker id. An arrow labelled bin by aisle and 30-minute interval, count picks, points to a three-by-three heatmap on the right with aisle rows A-01, A-02, A-03 and time columns 10:30, 11:00, 11:30. Each cell shows its pick count and is shaded by density. The cell at aisle A-02, time 11:00, holds 48 picks and is outlined in coral because it crosses the congested threshold of 45 picks per interval with at least two pickers present. From event stream to flagged density cells PickEvent stream A-02 · 11:04 · P1 A-02 · 11:07 · P4 A-01 · 11:12 · P2 A-02 · 11:18 · P1 A-03 · 11:26 · P5 A-02 · 11:29 · P4 bin by aisle × 30-min interval 10:30 11:00 11:30 A-01 A-02 A-03 81410 224826 6129 congested ≥ 45 picks / interval (guarded by ≥ 2 concurrent pickers)

Prerequisites

Confirm each of these before running the detector against a live feed:

  • Python 3.10+ — the implementation uses list[...] generics, X | None unions, and frozen dataclass records.
  • A confirmed pick-event feed — one row per pick line with an aisle id, a timezone-normalized timestamp, and a picker id, sourced from WMS pick confirmations.
  • Consistent aisle keys — the aisle_id values must match those used by the travel-distance and wave-sequencing layers, or the flagged cells will not join back.
  • A congestion threshold from your own history — a picks-per-interval cut derived from a diagnostic baseline of a busy week, not a guessed constant.
  • Timezone-normalized timestamps — all events in facility-local time so interval boundaries land where the shift boundaries actually are.

Configuration Block

Every tunable lives in one externalized profile. The lever that decides everything is congested_picks_per_interval — the pick count above which a cell is a bottleneck — paired with min_pickers, the concurrency floor that stops a single picker’s burst from tripping the flag.

# congestion_heatmap.yaml — one profile per facility
congestion_heatmap:
  interval_minutes: 30              # bucket width; align to the wave cadence
  congested_picks_per_interval: 45  # cell picks at/above this = congested
  warn_picks_per_interval: 30       # cell picks at/above this = warning
  min_pickers: 2                    # concurrency floor; below this, never "congested"
# Equivalent Python config dict consumed by the detector
CONGESTION_HEATMAP = {
    "interval_minutes": 30,
    "congested_picks_per_interval": 45,
    "warn_picks_per_interval": 30,
    "min_pickers": 2,
}

Implementation

The detector is one focused function. It floors each event’s timestamp to an interval bucket, folds the stream into per-cell pick counts and distinct-picker sets, then classifies each cell against the two thresholds — with the congested level gated behind the concurrent-picker floor. It returns a sorted list of typed HeatCell records so the caller can render the heatmap, alert on the congested tail, or hand the cells to the wave scheduler.

from __future__ import annotations

import logging
from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime

logger = logging.getLogger("space.congestion.detect")


@dataclass(frozen=True)
class PickEvent:
    aisle_id: str
    picker_id: str
    picked_at: datetime


@dataclass(frozen=True)
class HeatCell:
    aisle_id: str
    time_slot: str
    picks: int
    pickers: int
    level: str  # ok | warn | congested


def detect_congestion(
    events: list[PickEvent],
    interval_minutes: int = 30,
    congested_picks_per_interval: int = 45,
    warn_picks_per_interval: int = 30,
    min_pickers: int = 2,
) -> list[HeatCell]:
    """Bin pick events into an aisle x interval heatmap and flag congested cells."""
    pick_counts: dict[tuple[str, str], int] = defaultdict(int)
    picker_sets: dict[tuple[str, str], set[str]] = defaultdict(set)

    for ev in events:
        bucket_min = (ev.picked_at.minute // interval_minutes) * interval_minutes
        slot = ev.picked_at.replace(minute=bucket_min, second=0, microsecond=0).strftime("%H:%M")
        key = (ev.aisle_id, slot)
        pick_counts[key] += 1
        picker_sets[key].add(ev.picker_id)

    cells: list[HeatCell] = []
    for key, picks in pick_counts.items():
        pickers = len(picker_sets[key])
        if picks >= congested_picks_per_interval and pickers >= min_pickers:
            level = "congested"
        elif picks >= warn_picks_per_interval:
            level = "warn"
        else:
            level = "ok"
        cells.append(HeatCell(key[0], key[1], picks, pickers, level))

    cells.sort(key=lambda c: c.picks, reverse=True)
    n_congested = sum(1 for c in cells if c.level == "congested")
    logger.info("Heatmap: %d cells from %d events, %d congested",
                len(cells), len(events), n_congested)
    return cells

Step-by-Step Walkthrough

  1. Floor each timestamp to an interval. (minute // interval_minutes) * interval_minutes snaps every event to the start of its bucket, and strftime("%H:%M") turns it into a stable slot label. Reading interval_minutes from config — and aligning it to the wave cadence — is what lets the flagged cells join back to the sequencer’s slots.
  2. Fold into two accumulators. pick_counts measures throughput per cell; picker_sets measures concurrency. Both are keyed on (aisle_id, slot). Keeping distinct pickers as a set means duplicate events from the same picker never inflate the concurrency count.
  3. Classify against thresholds, congested first. A cell is congested only if it clears congested_picks_per_interval and carries at least min_pickers distinct pickers; otherwise it falls to warn on pick count alone, or ok. Ordering the check this way is what stops a lone picker who smashed out 50 picks in an empty aisle from being reported as a crowd — high throughput, but no interference.
  4. Sort by pick density. Returning cells sorted by picks descending puts the worst bottlenecks at the head of the list, so an alerting job or a wave scheduler can act on the top-N without re-scanning.
  5. Log the congested count. The single logger.info line gives the batch job a one-line summary — total cells, events consumed, and how many crossed the line — which is the signal you trend to see whether a re-slot or a smoothing change actually reduced hotspots.

Verification

Assert the detector’s invariants directly — a genuine crowd flags, a solo burst does not, and the picker floor is enforced. These checks run without a live feed.

import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO)


def _ev(aisle: str, minute: int, picker: str) -> PickEvent:
    return PickEvent(aisle, picker, datetime(2026, 7, 16, 11, minute))


# A crowded cell: 46 picks in 11:00-11:30 from three pickers -> congested.
crowd = [_ev("A-02", 5 + i % 25, f"P{i % 3}") for i in range(46)]
# A solo burst: 50 picks from ONE picker -> high throughput, but not a crowd.
solo = [_ev("A-09", 5 + i % 25, "P7") for i in range(50)]

cells = detect_congestion(crowd + solo)
by_aisle = {c.aisle_id: c for c in cells}

assert by_aisle["A-02"].level == "congested", by_aisle["A-02"]
assert by_aisle["A-09"].level == "warn", by_aisle["A-09"]   # gated by min_pickers
assert cells[0].picks >= cells[-1].picks                     # sorted densest-first
print(f"OK — A-02={by_aisle['A-02'].level} ({by_aisle['A-02'].pickers} pickers), "
      f"A-09={by_aisle['A-09'].level} ({by_aisle['A-09'].pickers} picker)")

Sample expected output:

INFO:space.congestion.detect:Heatmap: 2 cells from 96 events, 1 congested
OK — A-02=congested (3 pickers), A-09=warn (1 picker)

Common Pitfalls

  • Flagging throughput without concurrency. A single picker clearing a big replenishment in an empty aisle produces a high pick count but zero interference. Without the min_pickers floor you would dispatch a supervisor to a “hotspot” that is one person working fast. Always gate the congested level on distinct pickers, not picks alone.
  • An interval that fights the wave cadence. Binning at 30 minutes against 60-minute waves scatters one wave’s crowding across two cells and can hide a genuine peak. Set interval_minutes to the wave interval so a flagged cell maps to exactly one scheduling slot.
  • Un-normalized timestamps. Mixing UTC and local time, or spanning a daylight-saving change mid-feed, shifts events into the wrong buckets and smears the peak. Normalize every timestamp to facility-local time before binning.
  • Reusing an old threshold after a re-slot. A congestion cut calibrated before a layout change no longer reflects the floor. Re-derive congested_picks_per_interval from a fresh baseline whenever slotting or staffing shifts, and trend the congested-cell count to confirm the threshold still discriminates.