Mechanic Assignment & Workload Balancing

Once a defect has been routed to a repair lane, a legal clock starts: under 49 CFR § 396.11©(2) a driver-reported condition affecting safe operation must be corrected — and certified corrected by a qualified person — before the asset returns to service, and under 49 CFR § 396.25 the individual performing that brake work must be a documented, qualified brake inspector. Assigning the wrong technician, or letting a critical repair order sit in an overloaded queue, is therefore not a scheduling inconvenience — it is a Vehicle Maintenance BASIC exposure and, for brake and steering work, a direct § 396.25 qualification violation. This page specifies the deterministic assignment layer inside the Defect Classification & Repair Order Routing pipeline: it consumes a routed repair order, matches it against a certification-aware technician registry, balances load so no bay saturates while critical work waits, and emits an auditable assignment record that maps a named, qualified mechanic to a specific DVIR defect code.

The assignment service is a deterministic consumer that sits downstream of routing and upstream of the maintenance-execution (CMMS) layer. It does not decide severity and it does not decide the critical/non-critical branch — it receives a repair order that has already passed through the Critical vs Non-Critical Routing Logic layer, carrying the 0–100 score produced by the Severity Scoring Algorithms for DVIR Defects and the fleet-aware priority weighting from Dynamic Threshold Tuning for Fleet Types. Target Python 3.10+ (the code below uses match statements, structural typing, and the | union syntax) with:

• Pydantic 2.x — schema validation and immutable (frozen) assignment payloads.
• enum / datetime / dataclasses (stdlib) — certification enums, UTC timestamps, and the technician registry record.
• NumPy 1.26+ — vectorised candidate scoring across the registry when the technician pool is large.
• Celery 5.x or Prefect 2.x — asynchronous emission of the assignment event and the escalation fan-out.
• A message broker (Kafka, RabbitMQ, or AWS SNS/SQS) — the transport the assignment decision and audit entry are published to.

Every inbound repair order is assumed to already conform to the canonical contract defined in the Standardized DVIR JSON Schema Design; reject any payload that fails that contract before it reaches this layer rather than assigning it to a default technician. Access to the technician registry and to assignment writes must be gated by role-based access for DVIR data, so that only a maintenance supervisor role can override a computed assignment and every override is attributed.

The assignment layer operates on three records: the routed repair order it consumes, the technician registry it queries, and the assignment decision it emits. Field names and the 0–100 severity scale are held identical to the scoring, threshold, and routing pages so a single defect can be reconstructed end-to-end during a DOT audit.

Repair-order input (consumed):

Technician registry (queried):

Assignment decision (emitted):

The engine runs a fixed, auditable sequence: a certification hard filter, a capacity gate, a weighted scoring and selection, and a terminal escalation path when no candidate survives. Because every stage is deterministic and its inputs are logged, replaying the same registry snapshot and repair order always reproduces the same assignment — a property a DOT auditor can verify.

The records are modelled as frozen Pydantic types so a decision, once emitted, cannot be silently mutated before it reaches the audit store:

from __future__ import annotations

from datetime import datetime, timezone
from enum import StrEnum

from pydantic import BaseModel, Field


class Cert(StrEnum):
    ASE_BRAKES = "ase_brakes"        # required for brake work under 49 CFR § 396.25
    ASE_STEERING = "ase_steering"
    ASE_ELECTRICAL = "ase_electrical"
    OEM_POWERTRAIN = "oem_powertrain"
    GENERAL = "general"


class Tier(StrEnum):
    MINOR = "minor"
    MAJOR = "major"
    CRITICAL = "critical"


class RepairOrder(BaseModel, frozen=True):
    defect_id: str
    asset_id: str
    component_code: str
    severity_score: int = Field(ge=0, le=100)
    severity_tier: Tier
    compliance_hold: bool
    required_cert: Cert
    facility_code: str
    est_labor_hours: float = Field(gt=0)
    deadline_utc: datetime


class Technician(BaseModel, frozen=True):
    technician_id: str
    facility_code: str
    certifications: frozenset[Cert]
    status: str                       # "available" | "on_leave" | "oos_hold"
    active_labor_hours: float = Field(ge=0)
    shift_capacity_hours: float = Field(gt=0)
    open_oos_count: int = Field(ge=0)
    historical_efficiency: float = Field(ge=0.0, le=1.5)

    @property
    def utilisation(self) -> float:
        return self.active_labor_hours / self.shift_capacity_hours

The first stage removes every technician who is not legally allowed to perform the repair. For brake components this is not a preference but a § 396.25 requirement, so a missing ase_brakes credential drops the candidate unconditionally. The filter also excludes staff at the wrong facility, on leave, or already parked behind an OOS hold.

UTILISATION_CEILING = 0.85          # deprioritise above 85% shift utilisation
MAX_OOS_PER_SHIFT = 2               # hard business-logic ceiling per technician per shift


def eligible(order: RepairOrder, techs: list[Technician]) -> list[Technician]:
    """Hard filter: legal qualification + facility + availability. No scoring here."""
    return [
        t for t in techs
        if t.facility_code == order.facility_code
        and t.status == "available"
        and order.required_cert in t.certifications      # § 396.25 qualification gate
    ]

A qualified technician who is already saturated must not receive a routine order, and no technician may exceed the OOS ceiling. A critical / compliance_hold order is allowed to breach the soft utilisation ceiling when it is the only way to place safety-critical work — but it can never breach the hard OOS count, which caps concurrent immobilised assets per person.

def within_capacity(order: RepairOrder, t: Technician) -> bool:
    if t.open_oos_count >= MAX_OOS_PER_SHIFT:
        return False                                      # hard ceiling, never overridden
    if order.severity_tier is Tier.CRITICAL or order.compliance_hold:
        return True                                       # critical work may breach the soft ceiling
    return t.utilisation < UTILISATION_CEILING            # routine work respects the soft ceiling

Survivors are ranked by a normalised weighted score that rewards spare capacity and proven efficiency while giving deadline pressure and severity a direct pull. The weights are configuration, not code, so tuning them is an auditable config change rather than a silent behavioural drift.

WEIGHTS = {                          # must sum to 1.0; validated at load time
    "spare_capacity": 0.40,          # 1 - utilisation → prefer the least-loaded qualified tech
    "efficiency": 0.25,              # historical actual/estimated labour ratio
    "deadline_pull": 0.20,          # closeness of deadline_utc → urgency
    "severity_pull": 0.15,          # severity_score / 100 → risk weight
}


def _deadline_pull(order: RepairOrder, now: datetime) -> float:
    remaining_h = (order.deadline_utc - now).total_seconds() / 3600
    # Nearer deadlines pull harder; clamp to [0, 1] over a 48-hour horizon.
    return min(1.0, max(0.0, 1.0 - remaining_h / 48.0))


def score(order: RepairOrder, t: Technician, now: datetime) -> float:
    components = {
        "spare_capacity": 1.0 - min(t.utilisation, 1.0),
        "efficiency": min(t.historical_efficiency, 1.0),
        "deadline_pull": _deadline_pull(order, now),
        "severity_pull": order.severity_score / 100.0,
    }
    return round(sum(WEIGHTS[k] * v for k, v in components.items()), 4)


def assign(order: RepairOrder, techs: list[Technician], now: datetime | None = None):
    now = now or datetime.now(timezone.utc)
    pool = [t for t in eligible(order, techs) if within_capacity(order, t)]
    if not pool:
        return escalate(order, reason=_diagnose_empty_pool(order, techs))
    winner = max(pool, key=lambda t: score(order, t, now))
    return {
        "defect_id": order.defect_id,
        "technician_id": winner.technician_id,
        "assignment_score": score(order, winner, now),
        "outcome": "assigned",
        "decided_at_utc": now.isoformat(),
    }

When no qualified, in-capacity technician exists — a skill gap, a closed facility, or extreme saturation — the order must not be dropped or force-fitted. Escalate it to a mobile service unit or regional partner and record which constraint could not be satisfied, because a critical defect with an unmet deadline is itself a compliance event.

def _diagnose_empty_pool(order: RepairOrder, techs: list[Technician]) -> str:
    at_facility = [t for t in techs if t.facility_code == order.facility_code]
    match at_facility:
        case []:
            return f"no technicians at facility {order.facility_code}"
        case _ if not any(order.required_cert in t.certifications for t in at_facility):
            return f"no {order.required_cert} qualification at facility (49 CFR § 396.25)"
        case _:
            return "all qualified technicians at or above capacity / OOS ceiling"


def escalate(order: RepairOrder, reason: str) -> dict:
    return {
        "defect_id": order.defect_id,
        "technician_id": None,
        "assignment_score": 0.0,
        "outcome": "escalated",
        "escalation_reason": reason,
        "decided_at_utc": datetime.now(timezone.utc).isoformat(),
    }

For multi-defect vehicles, group compatible repairs under one technician with a bin-packing pass before scoring, so that tooling, bay, and parts co-location are exploited only when the same qualification covers every grouped defect; never group a brake defect with a non-brake defect onto a technician who lacks ase_brakes.

The assignment layer converts computed values into named FMCSA obligations. The mapping below is deterministic and imperative — there is no hedged middle ground.

The assignment record never certifies the repair and never clears an OOS hold — that is the assigned technician’s § 396.11©(2) sign-off, captured downstream. Keeping those responsibilities separate is what lets an auditor distinguish “work was dispatched to a qualified person” from “work was certified complete.”

The assignment decision is emitted as an idempotent event so retries, broker redeliveries, and replays resolve to a single work-order assignment rather than a duplicate ticket. Key the CMMS write on a deterministic identifier derived from the defect and the target technician, and hash-link consecutive decisions per asset so the assignment history is tamper-evident.

import hashlib
import json


def assignment_key(defect_id: str, technician_id: str | None) -> str:
    """Deterministic idempotency key: same defect + same technician → same key."""
    basis = f"{defect_id}|{technician_id or 'ESCALATED'}"
    return hashlib.sha256(basis.encode("utf-8")).hexdigest()


def chain(decision: dict, prev_hash: str) -> dict:
    """Append prev_hash and self-hash so consecutive decisions for an asset are linked."""
    body = json.dumps(decision, sort_keys=True, separators=(",", ":"))
    decision["prev_hash"] = prev_hash
    decision["record_hash"] = hashlib.sha256((prev_hash + body).encode()).hexdigest()
    return decision

Downstream, publish assigned and escalated events to the CMMS (to open or route the work order), to the telematics/ELD platform (so a compliance_hold asset stays flagged until sign-off), and to the fleet dashboard (for capacity and SLA monitoring). Consumers must upsert on assignment_key, never insert blindly, so a redelivered event is a no-op. Expose utilisation and assignment-latency metrics over a read-only endpoint for predictive capacity planning; do not let a metrics reader mutate the registry.

• Schema validation — reject any repair order or registry record that fails the Pydantic contract; never default a missing required_cert to general.
• Deterministic execution — the same registry snapshot and order always produce the same assignment; keep scoring weights in validated config that sums to 1.0.
• Qualification gate is non-negotiable — a missing § 396.25 credential drops the candidate; it can never be scored around.
• Hard vs soft ceilings — the OOS-per-shift count is a hard ceiling; the 85% utilisation ceiling is soft and may be breached only by critical / compliance_hold work.
• Idempotency — key CMMS writes on the SHA-256 assignment_key; redeliveries upsert, never duplicate.
• Audit logging — persist the named technician_id, all score components, the timestamp, and the escalation reason to an append-only, hash-chained store.
• Separation of duties — assignment dispatches work; it never certifies the repair or clears an OOS hold.
• Configuration management — weights, ceilings, and the certification map are versioned config changes, each attributable to a supervisor role.

• Critical vs Non-Critical Routing Logic — produces the routed repair order this layer consumes.
• Severity Scoring Algorithms for DVIR Defects — the 0–100 score that feeds the severity pull in the assignment weighting.
• Dynamic Threshold Tuning for Fleet Types — supplies the fleet-aware priority context that reaches assignment.
• Standardized DVIR JSON Schema Design — the canonical contract every inbound repair order must satisfy.
• Implementing Role-Based Access for DVIR Data — gates who may override an assignment or read the registry.

Back to Defect Classification & Repair Order Routing.