Geographic Zone Routing for Maintenance Dispatch

Geographic zone routing is the spatial-assignment stage of the Technician Routing & SLA Enforcement pipeline — the step that decides where a job lives and whose zone owns it before any technician is named.

On a campus or a multi-site estate, the difference between a well-run maintenance day and a backlog is drive time. A technician who crosses three buildings to reach a fault burns wrench-time that never appears on a timesheet but shows up as missed SLAs and lower first-visit completion. This component treats the estate as a set of partitioned routing zones anchored to the asset tree, resolves every work order to the zone that physically contains its asset, and assigns the nearest eligible technician inside that zone — spilling over to an adjacent zone only when the home zone has no one available. For facilities managers it converts an unstructured travel free-for-all into predictable, ownable territory; for Python automation and CMMS integration teams it demands clean location data, a deterministic distance function, and an assignment rule that degrades gracefully under load. This guide builds that stage end to end: prerequisites, the data contract, a step-by-step implementation, a configuration reference, validation, and the failure modes you will actually hit.

Prerequisites

Zone routing runs as a synchronous resolver invoked by the dispatch worker once a work order has cleared ingestion and validation. Before you deploy it, confirm the following are in place.

• Python 3.11+ with haversine>=2.8 for great-circle drive-time estimation between coordinates, and optionally shapely>=2.0 when your zones are true polygons rather than point-radius or building-code lookups. pydantic>=2.6 models the zone and technician records so a malformed roster is rejected at load time rather than mid-dispatch. No external routing service is required — the estimator is offline and deterministic.
• Asset location data. Every asset must carry a resolvable (latitude, longitude) and a site_id/building_id sourced from the equipment tree in asset hierarchy design. Zone routing is only as accurate as the coordinates hanging off each asset node; an asset with no location cannot be resolved to a zone.
• A zone registry. A version-controlled table mapping each zone to its owning site, its centroid (or polygon), its adjacency list, and the technician roster permitted to work it. Adjacency is what makes overflow possible without sending someone across the estate.
• Environment variables: CMMS_BASE_URL, CMMS_API_TOKEN (carrying the routing:read and technician:read scopes), ZONE_REGISTRY_PATH pointing at the registry, and DRIVE_SPEED_KMH (default 30) used to convert distance into an estimated drive-time budget. A token missing technician:read must fail closed at the roster step rather than assigning against an empty candidate list.
• Permissions. The resolver reads technician records but never mutates a shift or a certification. Assignment writes are the caller’s responsibility; keep this component read-only so a routing bug can never corrupt the roster.

Architecture and Data Contract

The resolver sits between a validated work order and the technician-assignment writer. It never edits the asset tree and it never books a shift — it reads an asset’s location, maps it to the owning zone, filters the zone’s roster to eligible technicians, ranks them by estimated drive time, and emits an assignment directive. Four boundaries keep the decision honest and stop spatial logic from leaking into scheduling or execution.

• Location boundary: the work order’s location_id is resolved to concrete coordinates from the asset tree. A location that cannot be resolved is rejected here, before it consumes roster quota.
• Zone boundary: coordinates map to exactly one owning zone via polygon containment or nearest-centroid fallback. Zone ownership is deterministic — the same asset always resolves to the same home zone.
• Eligibility boundary: the zone’s roster is filtered to technicians who are both permitted in the zone and available, so combining spatial fit with skills and shift state happens in one place.
• Assignment boundary: candidates are ranked by haversine drive-time estimate; the nearest is chosen, or, when the home zone is empty, the search widens to adjacent zones. Downstream consumers act on the directive — they do not re-run the geometry.

Assignment is not a single yes/no. The resolver returns one of ASSIGN_IN_ZONE, ASSIGN_OVERFLOW, or DEFER_NO_CANDIDATE, together with the chosen technician, the resolved zone, and the estimated drive-time minutes so the caller can enforce SLA budgets without re-querying.

The contract across the assignment boundary is explicit: the input is a RoutingRequest derived from the canonical work order plus the resolved coordinates, and the output is an AssignmentDecision. Encoding both with pydantic means a roster with a missing home coordinate or a zone that references a non-existent adjacency is rejected at load time instead of producing a plausible-but-wrong assignment.

from enum import Enum
from typing import List, Optional, Tuple
from pydantic import BaseModel, Field


class Priority(str, Enum):
    CRITICAL = "critical"
    HIGH = "high"
    STANDARD = "standard"
    PLANNED = "planned"


class Directive(str, Enum):
    ASSIGN_IN_ZONE = "ASSIGN_IN_ZONE"
    ASSIGN_OVERFLOW = "ASSIGN_OVERFLOW"
    DEFER_NO_CANDIDATE = "DEFER_NO_CANDIDATE"


class Zone(BaseModel):
    """A routing territory anchored to one site."""
    zone_id: str = Field(..., min_length=2)
    site_id: str
    centroid: Tuple[float, float]        # (lat, lon)
    adjacent_zone_ids: List[str] = Field(default_factory=list)
    radius_km: float = Field(2.0, gt=0)  # point-radius containment fallback


class Technician(BaseModel):
    """A dispatchable resource with a home base and zone permissions."""
    technician_id: str = Field(..., min_length=3)
    home_base: Tuple[float, float]       # (lat, lon) technician starts from
    permitted_zone_ids: List[str] = Field(..., min_length=1)
    available: bool = True
    skills: List[str] = Field(default_factory=list)


class RoutingRequest(BaseModel):
    """Input contract: one spatial assignment for one work order."""
    work_order_id: str = Field(..., min_length=8, max_length=64)
    asset_id: str = Field(..., min_length=6)
    location_id: str
    asset_coords: Tuple[float, float]    # resolved from the asset tree
    priority: Priority = Priority.STANDARD
    escalation_tier: int = Field(0, ge=0, le=3)


class AssignmentDecision(BaseModel):
    """Output contract: the directive plus the evidence behind it."""
    directive: Directive
    technician_id: Optional[str] = None
    zone_id: Optional[str] = None
    drive_minutes: Optional[float] = None

Step-by-Step Implementation

1. Define the canonical work order payload

The resolver is invoked against a work order, so it imports the same WorkOrderPayload used everywhere on this site rather than redefining its shape. The location_id is the field that anchors a job in space; the SLA fields — priority, requested_completion, and escalation_tier — are what the routing engine reads when it decides whether an overflow drive is acceptable or whether the job should defer.

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Dict, List, Optional


@dataclass
class WorkOrderPayload:
    """Canonical CMMS work order — SLA fields are mandatory site-wide."""
    work_order_id: str
    asset_id: str
    part_skus: List[str]
    required_quantities: Dict[str, int]
    location_id: str
    priority: Priority = Priority.STANDARD
    requested_completion: Optional[datetime] = None
    escalation_tier: int = 0
    status: str = "open"
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

2. Resolve the asset location to coordinates and its owning zone

Look up the asset’s coordinates from the location service backed by the asset tree, then map those coordinates to exactly one owning zone. Prefer polygon containment when shapely is available and your zones are surveyed; fall back to nearest-centroid within radius_km otherwise. Resolving to a single home zone is what makes ownership deterministic — the same asset always lands in the same territory.

from haversine import haversine, Unit
from typing import Dict, List, Optional


def resolve_zone(coords: tuple, zones: List[Zone]) -> Optional[Zone]:
    """Map coordinates to one owning zone by nearest centroid within radius."""
    candidates = []
    for zone in zones:
        distance_km = haversine(coords, zone.centroid, unit=Unit.KILOMETERS)
        if distance_km <= zone.radius_km:
            candidates.append((distance_km, zone))
    if not candidates:
        return None
    # Deterministic: nearest centroid wins; ties break on zone_id for stability.
    candidates.sort(key=lambda pair: (pair[0], pair[1].zone_id))
    return candidates[0][1]


def build_request(
    wo: WorkOrderPayload, location_index: Dict[str, tuple]
) -> RoutingRequest:
    """Narrow the work order to the fields the resolver needs."""
    coords = location_index.get(wo.location_id)
    if coords is None:
        raise ValueError(f"location {wo.location_id!r} has no resolvable coordinates")
    return RoutingRequest(
        work_order_id=wo.work_order_id,
        asset_id=wo.asset_id,
        location_id=wo.location_id,
        asset_coords=coords,
        priority=wo.priority,
        escalation_tier=wo.escalation_tier,
    )

3. Estimate drive time with a haversine budget

Convert great-circle distance between a technician’s home base and the asset into an estimated drive-time budget using a configurable average speed. The estimate is deliberately offline and deterministic — it never calls a live routing API — because dispatch decisions must be reproducible and must not block on a third-party service. A road-factor multiplier compensates for the fact that straight-line distance always understates real driving.

from haversine import haversine, Unit


def estimate_drive_minutes(
    origin: tuple,
    destination: tuple,
    speed_kmh: float = 30.0,
    road_factor: float = 1.3,
) -> float:
    """Great-circle distance → estimated drive time in minutes.

    road_factor inflates straight-line distance to approximate road network
    travel; speed_kmh is the campus/estate average, not a highway speed.
    """
    straight_km = haversine(origin, destination, unit=Unit.KILOMETERS)
    road_km = straight_km * road_factor
    hours = road_km / speed_kmh
    return round(hours * 60.0, 1)

4. Filter the roster to eligible in-zone technicians

Reduce the zone’s roster to technicians who are permitted in the zone and available, then rank the survivors by estimated drive time. This is the single place where spatial fit is combined with the other gates: skill matching from skill-based dispatch and shift state from shift calendar integration both plug in through the available flag and an optional required-skill check, so the resolver never assigns someone who is off-shift or uncertified.

from typing import List, Optional, Tuple


def eligible_in_zone(
    zone: Zone,
    request: RoutingRequest,
    technicians: List[Technician],
    required_skill: Optional[str] = None,
    speed_kmh: float = 30.0,
) -> List[Tuple[float, Technician]]:
    """Return (drive_minutes, technician) ranked nearest-first for one zone."""
    ranked: List[Tuple[float, Technician]] = []
    for tech in technicians:
        if zone.zone_id not in tech.permitted_zone_ids:
            continue
        if not tech.available:
            continue
        if required_skill and required_skill not in tech.skills:
            continue
        minutes = estimate_drive_minutes(
            tech.home_base, request.asset_coords, speed_kmh=speed_kmh
        )
        ranked.append((minutes, tech))
    ranked.sort(key=lambda pair: (pair[0], pair[1].technician_id))
    return ranked

5. Assign in-zone, then overflow to an adjacent zone

Compose the pieces into the entry point the dispatch worker calls. Try the home zone first; if it yields a candidate, emit ASSIGN_IN_ZONE. When the home zone is empty — every permitted technician is busy or off-shift — widen the search to the zone’s adjacency list and emit ASSIGN_OVERFLOW for the nearest candidate found there. If neither yields anyone, emit DEFER_NO_CANDIDATE so the caller can escalate through the SLA timer and escalation engine rather than silently dropping the job. Every decision is logged with a structured audit line for root-cause analysis.

import logging
from typing import Dict, List, Optional

logger = logging.getLogger(__name__)


def assign_technician(
    request: RoutingRequest,
    zones: List[Zone],
    technicians: List[Technician],
    required_skill: Optional[str] = None,
    speed_kmh: float = 30.0,
) -> AssignmentDecision:
    """Nearest eligible in-zone technician, overflowing to adjacent zones."""
    zones_by_id: Dict[str, Zone] = {z.zone_id: z for z in zones}
    home = resolve_zone(request.asset_coords, zones)
    if home is None:
        logger.warning("no zone for wo:%s loc:%s", request.work_order_id, request.location_id)
        return AssignmentDecision(directive=Directive.DEFER_NO_CANDIDATE)

    in_zone = eligible_in_zone(home, request, technicians, required_skill, speed_kmh)
    if in_zone:
        minutes, tech = in_zone[0]
        decision = AssignmentDecision(
            directive=Directive.ASSIGN_IN_ZONE,
            technician_id=tech.technician_id,
            zone_id=home.zone_id,
            drive_minutes=minutes,
        )
    else:
        overflow: list = []
        for adj_id in home.adjacent_zone_ids:
            adj = zones_by_id.get(adj_id)
            if adj is not None:
                overflow.extend(
                    eligible_in_zone(adj, request, technicians, required_skill, speed_kmh)
                )
        if overflow:
            overflow.sort(key=lambda pair: (pair[0], pair[1].technician_id))
            minutes, tech = overflow[0]
            decision = AssignmentDecision(
                directive=Directive.ASSIGN_OVERFLOW,
                technician_id=tech.technician_id,
                zone_id=home.zone_id,      # home zone still owns the job
                drive_minutes=minutes,
            )
        else:
            decision = AssignmentDecision(directive=Directive.DEFER_NO_CANDIDATE)

    logger.info(
        "routed wo:%s directive:%s tech:%s zone:%s drive:%s",
        request.work_order_id, decision.directive.value,
        decision.technician_id, decision.zone_id, decision.drive_minutes,
    )
    return decision

Configuration Reference

Keep every tunable in a version-controlled configuration registry, not in the worker source. The defaults below are conservative starting points for a mixed campus and multi-site estate.

Validation and Testing

Zone resolution and drive-time ranking are pure functions, so the highest-value test asserts that a fixed roster and asset location always yield the same technician. A deterministic assertion catches accidental nondeterminism — unstable tie-breaks, dict ordering, or a road factor drifting between calls — before it reaches production.

def test_empty_home_zone_overflows_to_nearest_adjacent():
    zones = [
        Zone(zone_id="Z-NORTH", site_id="SITE-1", centroid=(40.7130, -74.0060),
             adjacent_zone_ids=["Z-SOUTH"], radius_km=1.5),
        Zone(zone_id="Z-SOUTH", site_id="SITE-1", centroid=(40.7000, -74.0060),
             adjacent_zone_ids=["Z-NORTH"], radius_km=1.5),
    ]
    technicians = [
        # Only permitted in the home zone but unavailable → forces overflow.
        Technician(technician_id="TECH-N1", home_base=(40.7128, -74.0059),
                   permitted_zone_ids=["Z-NORTH"], available=False),
        # Permitted in the adjacent zone and available → the overflow pick.
        Technician(technician_id="TECH-S1", home_base=(40.7010, -74.0060),
                   permitted_zone_ids=["Z-SOUTH"], available=True),
    ]
    request = RoutingRequest(
        work_order_id="WO-2026-0715",
        asset_id="AHU-021",
        location_id="SITE-1-BLDG-A",
        asset_coords=(40.7129, -74.0060),   # inside Z-NORTH
    )
    decision = assign_technician(request, zones, technicians)
    assert decision.directive is Directive.ASSIGN_OVERFLOW
    assert decision.technician_id == "TECH-S1"
    assert decision.zone_id == "Z-NORTH"     # home zone still owns the job
    assert decision.drive_minutes is not None

On a successful assignment the resolver emits a single structured line per work order — routed wo:WO-2026-0715 directive:ASSIGN_OVERFLOW tech:TECH-S1 zone:Z-NORTH drive:2.9 — which is the canonical signal that a spatial decision was made. A deferral instead emits no zone for wo:... or a routed ... directive:DEFER_NO_CANDIDATE ... line with a null technician; seeing the latter under a shift shortage confirms the roster is empty rather than the geometry being broken. Assert against both lines in integration tests to verify the full location-to-assignment path.

Failure Modes and Troubleshooting

Expand each scenario for the root cause, the diagnostic log excerpt, and the fix. The checklist items render as interactive checkboxes — work through them in order.

An asset resolves to no zone at all

• Look for a no zone for wo:<id> loc:<location_id> line — it means resolve_zone returned None for the asset’s coordinates.
• Confirm the asset carries a (latitude, longitude) in the equipment tree; an asset added after a site merge often inherits a null location from asset hierarchy design and cannot be placed.
• Check the nearest zone’s radius_km — coordinates just outside every containment radius fall through to None; widen the radius or add the missing zone rather than routing to a guessed neighbour.

Every in-zone technician is busy, so the job overflows unexpectedly

• Confirm the routed ... directive:ASSIGN_OVERFLOW line names a technician from an adjacent zone, not a distant one — an over-wide max_overflow_hops reaches too far.
• Verify the home-zone roster is genuinely empty by checking each technician’s available flag against shift calendar integration; a stale availability flag makes a present technician look busy.
• If overflow drive time blows the SLA budget, set overflow_enabled to false for that priority so the job defers and escalates instead of dispatching a long drive.

Stale GPS or location data sends technicians to the wrong place

• Compare the asset’s stored coordinates against a recent physical scan; a drifted or copy-pasted coordinate silently resolves the asset into the wrong home zone.
• Check whether the location_index was rebuilt after the last asset-tree change — a cached index resolves moved assets to their old zone until it is refreshed.
• Confirm road_factor and DRIVE_SPEED_KMH match reality; a drive-time estimate that is wildly off usually points at an estate-wide speed misconfiguration, not a single bad coordinate.

Frequently Asked Questions

How is a work order’s home zone decided?

The resolver reads the asset’s coordinates from the equipment tree and maps them to exactly one owning zone — by polygon containment when surveyed zones and shapely are available, or by nearest centroid within each zone’s radius_km otherwise. Ownership is deterministic: the same asset always resolves to the same home zone, and ties break on zone_id for stability.

What happens when every technician in the home zone is busy?

The search widens to the zone’s adjacency list and the nearest available technician there is assigned with an ASSIGN_OVERFLOW directive, while the home zone still owns the job for reporting. If overflow is disabled or no adjacent zone has anyone, the resolver emits DEFER_NO_CANDIDATE so the caller can escalate rather than silently dropping the work order.

Does this component pick technicians by skill as well as location?

It combines both in one place. Location decides the zone and the drive-time ranking; the required_skill and available checks fold in trade matching from skill-based dispatch and shift state from shift calendar integration. The resolver never assigns an off-shift or uncertified technician even if they are the closest.

Why estimate drive time offline instead of calling a routing API?

Dispatch decisions must be reproducible and must not block on a third party. A haversine estimate inflated by a road factor is deterministic, instant, and good enough to rank candidates within a zone; a live routing service adds latency and a failure mode without changing which technician is nearest in the common case.

Related

Anchor every zone to accurate coordinates from asset hierarchy design, layer trade matching on top with skill-based dispatch, gate candidates on shift state from shift calendar integration, escalate deferrals through SLA timer and escalation, and see the drive-time payoff quantified in reducing drive time with zone-partitioned dispatch.

Part of: Technician Routing & SLA Enforcement.