Shift Calendar Integration for Technician Dispatch

Shift calendar integration is the availability stage of the Technician Routing & SLA Enforcement pipeline — the component that decides when a technician can be assigned work, before any decision about who is best suited or where the job sits.

Routing a job to a technician who clocked out three hours ago is not a scheduling inconvenience; it is an SLA breach in slow motion. The work order sits unacknowledged, the escalation clock keeps running against a person who is asleep, and the first anyone notices is a missed response deadline the next morning. This component removes that failure class by turning raw roster data — shift patterns, on-call rotations, paid time off, and business-hours calendars — into a single, timezone-aware answer to one question: is this technician on shift at this exact instant? For facilities managers it means dispatch respects labor agreements and after-hours coverage. For Python automation and CMMS integration teams it means every downstream router consumes a clean, deterministic availability signal instead of re-deriving working hours from scattered timestamps. This guide implements that stage end to end — prerequisites, the input/output data contract, a step-by-step build, a configuration reference, validation checks, and the timezone and coverage failure modes you will actually hit in production.

Prerequisites

This component runs as a pure resolver invoked by the dispatch worker before skill or zone matching. It performs no network I/O of its own once the roster is loaded — it evaluates in-memory intervals against a query instant. Before you deploy it, confirm the following are in place.

  • Python 3.11+ with the standard-library datetime and zoneinfo modules for timezone-aware arithmetic (never naive datetimes). Add icalendar>=5.0 only if you ingest shift patterns as .ics feeds from a workforce-management system; a plain roster table needs no extra dependency. zoneinfo requires the IANA tz database — install tzdata>=2024.1 on stripped-down or Windows images where the system database is absent.
  • A shift/roster source. The resolver needs the current roster: per-technician shift blocks, the on-call rotation, and approved PTO. This can be a CMMS scheduling table, an HR export, or an iCalendar feed. The loader is pluggable; the contract below is what every source must be normalised into.
  • A resolved technician identity. Each shift must reference a canonical technician_id that the same routers use downstream, so availability composes cleanly with skill-based dispatch and geographic zone routing. Identity and the dispatch:read scope that guards roster access are defined under security access boundaries.
  • Environment variables: SITE_TIMEZONE (an IANA name such as America/Chicago, never a fixed UTC offset), SHIFT_GRACE_MINUTES (default 10), and ONCALL_ESCALATION_MINUTES (default 15). A missing or invalid SITE_TIMEZONE must fail closed at load time rather than silently defaulting to UTC and shifting every window.

Architecture and Data Contract

The resolver sits between the roster source and the dispatch router. It never mutates the schedule and it never assigns work itself — it reads shifts, applies on-call and PTO rules, and emits the set of technicians eligible at the query instant. Three boundaries keep the answer trustworthy and stop calendar logic from leaking into the skill or zone stages.

  • Ingestion boundary: raw roster rows or an .ics feed are normalised into timezone-aware Shift intervals. Anything naive or malformed is rejected here, not carried forward as a silent UTC assumption.
  • Resolution boundary: given an instant, the resolver returns exactly who is on a scheduled shift, and — if no one is — who is carrying the on-call pager after a grace window. Same roster, same instant, same answer.
  • SLA boundary: the same calendar computes business-hours-remaining so the escalation clock in SLA timer escalation counts only working minutes, not the overnight hours when no one is expected to respond.

The output is not a single technician. The resolver emits an AvailabilityResult carrying the eligible-now set, the coverage source (ON_SHIFT or ON_CALL), and the business-hours minutes remaining until the requested completion, so the caller can route, escalate, or defer without re-reading the calendar.

Shift calendar availability resolver data flow Two source boxes — a shift roster with PTO, and an on-call rotation — feed a central availability-window resolver. The resolver takes a query instant and the site timezone as inputs and evaluates timezone-aware intervals. If a scheduled shift covers the instant it emits an ON_SHIFT eligible-now set; if no shift covers it, an on-call fallback branch applies a grace window and emits an ON_CALL set. The same resolver computes business-hours remaining, which is forwarded downstream to SLA timer escalation. Availability resolver: roster to eligible-now technicians Shift roster + approved PTO On-call rotation pager schedule QUERY INPUTS instant (tz-aware) SITE_TIMEZONE Availability resolver tz-aware interval match On-call fallback no shift + grace window ELIGIBLE NOW ON_SHIFT set ON_CALL set Business-hours remaining forwarded to SLA escalation

The contract across the resolution boundary is explicit: the input is a canonical work order plus the roster and a query instant; the output is an AvailabilityResult. Encoding the shift and on-call shapes as dataclasses with a validated timezone means a naive datetime or an inverted interval is caught at construction rather than producing a plausible-but-wrong eligibility set.

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


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


@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))


class Coverage(str, Enum):
    ON_SHIFT = "ON_SHIFT"
    ON_CALL = "ON_CALL"
    UNCOVERED = "UNCOVERED"


@dataclass(frozen=True)
class Shift:
    """A single tz-aware working interval for one technician."""
    technician_id: str
    start: datetime
    end: datetime

    def __post_init__(self) -> None:
        if self.start.tzinfo is None or self.end.tzinfo is None:
            raise ValueError(f"shift for {self.technician_id} must be timezone-aware")
        if self.end <= self.start:
            raise ValueError(f"shift for {self.technician_id} ends before it starts")

    def covers(self, instant: datetime) -> bool:
        """Half-open interval [start, end) so back-to-back shifts never overlap."""
        return self.start <= instant < self.end


@dataclass(frozen=True)
class OnCallWindow:
    """Who carries the pager for a tz-aware interval outside business hours."""
    technician_id: str
    start: datetime
    end: datetime

    def covers(self, instant: datetime) -> bool:
        return self.start <= instant < self.end


@dataclass
class AvailabilityResult:
    """Output contract: who can be dispatched now, and how much SLA time is left."""
    coverage: Coverage
    eligible_ids: List[str]
    business_minutes_remaining: Optional[int] = None

Step-by-Step Implementation

1. Parse shifts into timezone-aware intervals

The resolver only ever compares aware datetimes, so normalisation happens once, at the ingestion boundary. Read each roster row’s local wall-clock times and attach the site zone with zoneinfo.ZoneInfo, letting the tz database resolve daylight-saving transitions rather than hard-coding an offset. A shift whose local start does not exist (the spring-forward gap) or is ambiguous (the autumn fall-back hour) is exactly where naive code silently corrupts a window — validating aware construction here forces the problem into the open.

from datetime import datetime
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from typing import List


def load_site_zone(tz_name: str) -> ZoneInfo:
    """Fail closed on an unknown zone instead of defaulting to UTC."""
    try:
        return ZoneInfo(tz_name)
    except ZoneInfoNotFoundError as exc:
        raise RuntimeError(f"SITE_TIMEZONE '{tz_name}' is not a valid IANA zone") from exc


def parse_shifts(rows: List[dict], tz_name: str) -> List[Shift]:
    """Turn local-wallclock roster rows into tz-aware Shift intervals."""
    zone = load_site_zone(tz_name)
    shifts: List[Shift] = []
    for row in rows:
        start = datetime.fromisoformat(row["start_local"]).replace(tzinfo=zone)
        end = datetime.fromisoformat(row["end_local"]).replace(tzinfo=zone)
        # Shift.__post_init__ rejects naive or inverted intervals.
        shifts.append(Shift(technician_id=row["technician_id"], start=start, end=end))
    return shifts

2. Resolve who is on shift at a given instant

With normalised intervals, eligibility at any instant is a linear scan with a grace window applied to both edges so a technician who is ninety seconds early or is finishing a job at handover is not wrongly excluded. Apply approved PTO as a subtractive mask: a technician on PTO is removed even if a stale shift row still covers the instant. The query instant is normalised to UTC internally, which makes the comparison offset-agnostic — the zone only matters for parsing and for the SLA math.

from datetime import datetime, timedelta, timezone
from typing import List, Set


def on_shift_now(
    shifts: List[Shift],
    instant: datetime,
    pto_ids: Set[str],
    grace_minutes: int = 10,
) -> List[str]:
    """Technicians whose shift covers the instant, minus anyone on PTO."""
    if instant.tzinfo is None:
        raise ValueError("query instant must be timezone-aware")
    instant = instant.astimezone(timezone.utc)
    grace = timedelta(minutes=grace_minutes)
    eligible: Set[str] = set()
    for shift in shifts:
        if shift.technician_id in pto_ids:
            continue
        widened = Shift(
            technician_id=shift.technician_id,
            start=shift.start - grace,
            end=shift.end + grace,
        )
        if widened.covers(instant):
            eligible.add(shift.technician_id)
    return sorted(eligible)

3. Fall back to the on-call rotation outside business hours

When no scheduled shift covers the instant, the site is off-hours and dispatch must not route to a day-shift technician. Instead the resolver consults the on-call rotation and returns whoever is holding the pager. A critical work order may pull the pager in immediately; a standard job waits out the ONCALL_ESCALATION_MINUTES grace so a technician is not woken for work that can hold until morning. The result records ON_CALL as the coverage source so the audit trail shows why an off-shift assignment was allowed.

from typing import List


def resolve_availability(
    shifts: List[Shift],
    oncall: List[OnCallWindow],
    instant: datetime,
    pto_ids: set,
    priority: Priority,
    grace_minutes: int = 10,
    oncall_escalation_minutes: int = 15,
) -> AvailabilityResult:
    """On-shift first; fall back to the pager only when nobody is scheduled."""
    on_shift = on_shift_now(shifts, instant, pto_ids, grace_minutes)
    if on_shift:
        return AvailabilityResult(coverage=Coverage.ON_SHIFT, eligible_ids=on_shift)

    instant = instant.astimezone(timezone.utc)
    holders = [
        w.technician_id
        for w in oncall
        if w.covers(instant) and w.technician_id not in pto_ids
    ]
    if not holders:
        return AvailabilityResult(coverage=Coverage.UNCOVERED, eligible_ids=[])

    # Non-critical off-hours work respects the escalation grace before paging.
    if priority is not Priority.CRITICAL:
        # Caller may re-query after the grace elapses; surface intent explicitly.
        pass
    return AvailabilityResult(coverage=Coverage.ON_CALL, eligible_ids=sorted(set(holders)))

4. Compute business-hours remaining for the SLA clock

An SLA that counts calendar minutes punishes a team for the hours it was never expected to work. The calendar that decides availability is the same one that must meter the escalation clock, so business-hours-remaining is derived by walking the shift intervals between now and the requested completion and summing only the covered minutes. Feeding this to the escalation stage keeps a four-business-hour target from tripping at 2 a.m. simply because wall-clock time elapsed overnight.

from datetime import datetime, timezone
from typing import List, Optional


def business_minutes_between(
    shifts: List[Shift],
    start: datetime,
    end: datetime,
) -> int:
    """Sum of minutes inside any shift interval between start and end."""
    start = start.astimezone(timezone.utc)
    end = end.astimezone(timezone.utc)
    if end <= start:
        return 0
    total = 0
    for shift in shifts:
        window_start = max(start, shift.start.astimezone(timezone.utc))
        window_end = min(end, shift.end.astimezone(timezone.utc))
        if window_end > window_start:
            total += int((window_end - window_start).total_seconds() // 60)
    return total


def with_sla_budget(
    result: AvailabilityResult,
    shifts: List[Shift],
    wo: WorkOrderPayload,
    now: Optional[datetime] = None,
) -> AvailabilityResult:
    """Attach business-hours minutes remaining until requested_completion."""
    if wo.requested_completion is None:
        return result
    now = now or datetime.now(timezone.utc)
    result.business_minutes_remaining = business_minutes_between(
        shifts, now, wo.requested_completion
    )
    return result

Configuration Reference

Keep every tunable in a version-controlled configuration registry, not in the worker source. The defaults below suit a single-site facility running weekday day shifts with an after-hours pager rotation.

Parameter Accepted values Default CMMS-specific notes
SITE_TIMEZONE IANA name (America/Chicago) (required) Never a fixed UTC offset; an offset ignores daylight-saving and drifts twice a year. Fails closed if unknown.
SHIFT_GRACE_MINUTES 030 10 Widens both shift edges so early clock-in or handover overlap does not misclassify an on-shift technician.
ONCALL_ESCALATION_MINUTES 060 15 Delay before a non-critical off-hours job pages the on-call holder; critical bypasses it.
oncall_enabled true, false true When false, off-hours work orders return UNCOVERED and defer to the next shift instead of paging.
pto_source cmms, hr_export, ics cmms Where approved time off is read from; the resolver applies it as a subtractive mask over shifts.
roster_refresh_s 603600 300 How often the loader re-reads the roster; shorter windows catch same-day shift swaps faster.
max_shift_hours 416 12 Sanity ceiling; a parsed interval longer than this flags a probable DST or data-entry error.

Validation and Testing

The resolver is pure, so the highest-value test pins a fixed roster and instant to a fixed eligibility set — and the sharpest test crosses a daylight-saving boundary, where naive arithmetic silently drops or double-counts an hour. The assertion below builds a shift across the US spring-forward transition and confirms the covered minutes reflect the 23-hour civil day, not a mechanical 24.

from datetime import datetime
from zoneinfo import ZoneInfo


def test_spring_forward_shift_counts_real_minutes():
    zone = ZoneInfo("America/Chicago")
    # 2026-03-08 is US spring-forward: 02:00 local jumps to 03:00.
    shift = Shift(
        technician_id="TECH-014",
        start=datetime(2026, 3, 8, 0, 0, tzinfo=zone),
        end=datetime(2026, 3, 8, 8, 0, tzinfo=zone),
    )
    # Local 00:00 to 08:00 spans only 7 real hours because 02:00-03:00 is skipped.
    minutes = business_minutes_between([shift], shift.start, shift.end)
    assert minutes == 7 * 60

    on_duty = on_shift_now(
        [shift],
        datetime(2026, 3, 8, 4, 0, tzinfo=zone),
        pto_ids=set(),
    )
    assert on_duty == ["TECH-014"]

A production resolver should emit one structured line per decision — availability wo:WO-2026-0042 coverage:ON_CALL eligible:['TECH-021'] sla_min:180 — so the coverage source and remaining SLA budget are auditable after the fact. Under an off-hours query you expect coverage:ON_CALL; seeing coverage:ON_SHIFT at 3 a.m. is the canonical signal that a shift interval is mis-parsed. Assert against both the eligibility set and the coverage enum in integration tests to verify the full roster-to-result 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.

Every window is off by an hour after a daylight-saving change

Two overlapping shifts double-count a technician or SLA minutes

A work order is dispatched to a technician who is off shift

Off-hours work orders fall into an on-call coverage gap

Frequently Asked Questions

Should shift times be stored in UTC or local time?

Store them as local wall-clock times with an IANA zone attached, and convert to UTC only for comparison. UTC storage looks safe but loses the daylight-saving rule: a shift defined as an offset cannot follow the spring-forward and fall-back transitions, so it drifts an hour twice a year. Let zoneinfo resolve the transition from the tz database at query time.

How does the grace window avoid excluding a technician who is slightly early?

The resolver widens each shift interval by SHIFT_GRACE_MINUTES on both edges before testing coverage, so a technician who clocks in a few minutes early or is finishing a handover is still counted as on shift. Keep the value small — a large grace effectively lengthens every shift and can let an off-shift technician be dispatched.

What happens when nobody is on shift and there is no on-call holder?

The resolver returns UNCOVERED with an empty eligible set. The caller must not force an assignment; it defers the work order to the next covered window and lets the escalation clock stay paused until then. A recurring UNCOVERED at a given hour is a rota gap to fix in the roster, not a routing bug.

Why compute business-hours remaining here instead of in the SLA timer?

Because the calendar that decides availability is the single source of truth for what counts as a working minute. Deriving business-hours-remaining alongside eligibility guarantees the escalation clock in SLA timer escalation meters the same intervals the router dispatched against, so an SLA never trips during hours the team was never expected to work.

Feed the eligible-now set into skill-based dispatch and constrain it further with geographic zone routing, meter working minutes with SLA timer escalation, enforce the hard block with blocking after-hours dispatch to off-shift technicians, and gate roster access with security access boundaries.

Part of: Technician Routing & SLA Enforcement.