Routing Work Orders and Preventive Maintenance via Barcode & QR Integration

Barcode and QR integration is the scan-driven intake component of the Asset Lookup & Inventory Synchronization pipeline: it converts a physical scan at the asset into a validated, SLA-tagged work order without manual triage. When a technician scans an asset tag, the integration must resolve the identifier against ground-truth records, evaluate maintenance history and stock reality, and emit a routing directive with deterministic latency. This page covers the scan-to-dispatch path specifically — capturing the scan, resolving asset context, deciding corrective versus preventive routing, and committing an idempotent work order — and assumes the canonical inventory ledger and taxonomy described in the parent pipeline already exist.

Where This Component Sits

The scan path is a synchronous front end bolted onto an otherwise asynchronous routing pipeline. A scan event must feel instant to the technician holding the device, so the ingress and resolution stages run inline and return an acknowledgment within a tight budget, while any heavy follow-up work (replenishment, escalation notification, audit fan-out) is handed to background workers. The hard requirements are strict idempotency — a repeated scan or a network retry must never spawn a second work order — and graceful degradation when the CMMS registry or inventory API is slow. Every directive this component emits conforms to the same work order contract used everywhere else on the site, so the SLA fields priority, requested_completion, and escalation_tier are assigned here and never recomputed informally downstream.

Prerequisites

Before wiring the scan path, confirm the following are in place:

• Python 3.11+ with pydantic>=2.5 for payload validation, aiohttp>=3.9 for asynchronous CMMS calls, and redis>=5.0 for the asset-context cache and the idempotency key store.
• CMMS REST API v2 reachable at a stable base URL, exposing GET /api/v2/assets/{asset_id} and POST /api/v2/workorders, with the latter honoring an Idempotency-Key request header.
• Environment variables: CMMS_BASE_URL, CMMS_API_TOKEN, REDIS_URL, and SCAN_INGRESS_HMAC_SECRET for verifying device payload signatures.
• Permissions: the service account needs assets:read and workorders:write scopes. Mutating reservations against the ledger additionally requires the role grants defined under security access boundaries — a scan must not be able to commit a reservation the operator is not authorized to make.
• Provisioned hardware. Scanners must already map their decoded symbology to the registry’s canonical asset_id format. The handshake, firmware alignment, and tag-format normalization are covered in syncing barcode scanners with CMMS asset registries; without it, every scan fails resolution.

Architecture and Data Contract

The component is decomposed into four stateless stages that communicate only through typed objects. Each stage accepts a validated input, applies a deterministic transformation, and emits a structured output for the next stage — no shared mutable state — which is what makes the path testable and safe to scale.

The output of rule evaluation is the canonical domain model shared site-wide. Defining it once here means the rest of the path — and any code a reader copies — references the same fields, with the same SLA semantics, that the work order schema standards document governs.

from datetime import datetime
from enum import Enum
from typing import Literal

from pydantic import BaseModel, Field


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


class WorkOrderPayload(BaseModel):
    """Canonical work order contract shared across the pipeline."""

    work_order_id: str = Field(..., min_length=8)
    asset_id: str = Field(..., pattern=r"^SYS-\d{4}-COMP-\d{3}$")
    priority: Priority
    requested_completion: datetime
    escalation_tier: int = Field(0, ge=0, le=3)
    directive: Literal[
        "ROUTE_IMMEDIATE", "ROUTE_SUBSTITUTE", "DEFER_PROCUREMENT", "RESERVE_PM"
    ]
    assigned_crew: str
    required_parts: list[str] = Field(default_factory=list)
    trigger_event: Literal["barcode_scan"] = "barcode_scan"

Step-by-Step Implementation

1. Validate the Scan Payload at Ingress

Edge scanners and mobile apps transmit structured payloads; the ingress gateway enforces strict typing so malformed requests are rejected before they consume registry quota or trigger side effects. A client-generated idempotency_key rides on every scan and is the anchor for deduplication later.

from datetime import datetime
from typing import Literal

from pydantic import BaseModel, Field, ValidationError


class ScanPayload(BaseModel):
    asset_id: str = Field(..., pattern=r"^SYS-\d{4}-COMP-\d{3}$")
    scan_timestamp: datetime
    operator_id: str = Field(..., min_length=1)
    geolocation: tuple[float, float]
    scan_type: Literal["corrective", "preventive", "inspection"]
    idempotency_key: str = Field(..., description="UUIDv4 generated client-side")


def validate_scan_payload(raw_json: dict) -> ScanPayload:
    try:
        return ScanPayload(**raw_json)
    except ValidationError as exc:
        # Reject at ingress; never propagate to the routing engine.
        raise ValueError(f"Malformed scan payload: {exc.errors()}")

2. Resolve Asset Context

Once validated, the path retrieves maintenance schedules, failure modes, criticality, routing tags, and current spare-part position for the asset. High-frequency profiles are cached in Redis to keep resolution off the CMMS critical path during scanning surges; the cache is rebuilt from the registry, never written to independently. The routing tags read here (primary_crew, escalation_tier, dispatch_zone) are inherited through the tree defined under asset hierarchy design.

import json
from typing import Any

import aiohttp
import redis.asyncio as redis

CACHE_TTL_SECONDS = 300


async def resolve_asset_context(
    asset_id: str,
    cmms_base_url: str,
    session: aiohttp.ClientSession,
    cache: redis.Redis,
) -> dict[str, Any]:
    cached = await cache.get(f"asset:{asset_id}")
    if cached is not None:
        return json.loads(cached)

    async with session.get(
        f"{cmms_base_url}/api/v2/assets/{asset_id}",
        timeout=aiohttp.ClientTimeout(total=2.0),
    ) as resp:
        resp.raise_for_status()
        context = await resp.json()

    await cache.set(f"asset:{asset_id}", json.dumps(context), ex=CACHE_TTL_SECONDS)
    return context

3. Evaluate Routing Rules

Business logic decides whether the scan produces a corrective work order, a preventive task, an escalation, or a deferral pending parts. The matrix weighs asset criticality, the scan type, the next PM window, and live stock. When a required spare falls below safety stock the directive becomes DEFER_PROCUREMENT, which is handed to automated reorder triggers rather than blocking the queue. The fine-grained bill-of-materials evaluation against multi-bin stock belongs to parts availability checks; this stage consumes that verdict and assigns SLA metadata around it.

import uuid
from datetime import timedelta


SLA_WINDOWS = {
    Priority.CRITICAL: timedelta(hours=4),
    Priority.HIGH: timedelta(hours=24),
    Priority.STANDARD: timedelta(days=3),
    Priority.PM: timedelta(days=7),
}


def evaluate_routing_rules(
    asset_context: dict, scan: ScanPayload
) -> WorkOrderPayload:
    criticality = asset_context.get("criticality_score", 3)
    crew = asset_context.get("primary_crew", "shift_1_mech")
    parts_stock = asset_context.get("spare_parts_inventory", {})
    required = asset_context.get("required_parts", [])

    def build(priority: Priority, directive: str, parts: list[str],
              crew_id: str, tier: int) -> WorkOrderPayload:
        return WorkOrderPayload(
            work_order_id=f"WO-{uuid.uuid4().hex[:12]}",
            asset_id=scan.asset_id,
            priority=priority,
            requested_completion=scan.scan_timestamp + SLA_WINDOWS[priority],
            escalation_tier=tier,
            directive=directive,
            assigned_crew=crew_id,
            required_parts=parts,
        )

    # Planned preventive demand reserves rather than dispatches immediately.
    if scan.scan_type == "preventive" and asset_context.get("next_pm_date"):
        return build(Priority.PM, "RESERVE_PM", [], "pm_crew_a", tier=0)

    # Critical assets escalate to tier-2 with the SLA clock already running.
    if criticality >= 4:
        return build(Priority.CRITICAL, "ROUTE_IMMEDIATE", required, "tier_2", tier=2)

    # Block dispatch only if a required part is genuinely out of stock.
    missing = [p for p in required if parts_stock.get(p, 0) < 1]
    if missing:
        return build(Priority.HIGH, "DEFER_PROCUREMENT", missing, "unassigned", tier=1)

    return build(Priority.STANDARD, "ROUTE_IMMEDIATE", [], crew, tier=0)

4. Dispatch the Work Order Idempotently

The final stage pushes the payload to the CMMS work order queue. The Idempotency-Key header is mandatory: a network retry or a double-scan replays the same key, and the CMMS returns the existing work order instead of creating a duplicate. The device receives a structured confirmation carrying the work order ID, assigned crew, and the requested_completion window so the technician sees the SLA they are committing to.

async def dispatch_work_order(
    work_order: WorkOrderPayload,
    idempotency_key: str,
    cmms_base_url: str,
    session: aiohttp.ClientSession,
) -> dict:
    headers = {
        "Idempotency-Key": idempotency_key,
        "Content-Type": "application/json",
    }

    async with session.post(
        f"{cmms_base_url}/api/v2/workorders",
        data=work_order.model_dump_json(),
        headers=headers,
        timeout=aiohttp.ClientTimeout(total=5.0),
    ) as resp:
        if resp.status == 409:
            # The CMMS already processed this key — return the prior result.
            return {
                "status": "duplicate_acknowledged",
                "wo_id": work_order.work_order_id,
            }
        resp.raise_for_status()
        return await resp.json()

The end-to-end scan handler chains the four stages. Bound the number of concurrent outbound calls with a semaphore so a shift-change scanning surge cannot exhaust the connection pool, and lean on a durable broker via async batch processing for the non-blocking follow-up work the synchronous path defers.

import asyncio

_dispatch_limit = asyncio.Semaphore(20)


async def handle_scan(
    raw_json: dict,
    cmms_base_url: str,
    session: aiohttp.ClientSession,
    cache: redis.Redis,
) -> dict:
    scan = validate_scan_payload(raw_json)
    context = await resolve_asset_context(scan.asset_id, cmms_base_url, session, cache)
    work_order = evaluate_routing_rules(context, scan)
    async with _dispatch_limit:
        return await dispatch_work_order(
            work_order, scan.idempotency_key, cmms_base_url, session
        )

Configuration Reference

Validation and Testing

Verify the path without a live CMMS by asserting on the typed objects each stage emits. A preventive scan against an asset with a scheduled PM must reserve rather than dispatch, and the SLA window must match the matrix:

from datetime import datetime


def test_preventive_scan_reserves_pm():
    context = {
        "criticality_score": 2,
        "next_pm_date": "2026-07-15T00:00:00Z",
        "primary_crew": "pm_crew_a",
        "spare_parts_inventory": {},
        "required_parts": [],
    }
    scan = ScanPayload(
        asset_id="SYS-0421-COMP-009",
        scan_timestamp=datetime(2026, 6, 28, 9, 0, 0),
        operator_id="tech-114",
        geolocation=(40.71, -74.00),
        scan_type="preventive",
        idempotency_key="11111111-1111-4111-8111-111111111111",
    )
    work_order = evaluate_routing_rules(context, scan)
    assert work_order.directive == "RESERVE_PM"
    assert work_order.priority is Priority.PM
    assert work_order.requested_completion == datetime(2026, 7, 5, 9, 0, 0)

To confirm idempotency end to end, replay the same idempotency_key and assert the second call is acknowledged as a duplicate. A correct run logs a single creation followed by a duplicate acknowledgment:

INFO dispatch wo_id=WO-9f2a1c7b4e08 status=created key=2c1d… asset=SYS-0421-COMP-009
INFO dispatch wo_id=WO-9f2a1c7b4e08 status=duplicate_acknowledged key=2c1d… asset=SYS-0421-COMP-009

The field-by-field enforcement of the payload itself — types, ranges, and required keys — is worked through in the JSON Schema validation guide; run those checks in CI so a contract change cannot ship a payload the CMMS will reject.

Failure Modes and Troubleshooting

Duplicate work orders from double-scans. A technician scans twice, or a mobile client retries on a flaky link, and two work orders appear for one fault. Root cause: the idempotency_key is regenerated per HTTP attempt instead of per scan, so the CMMS sees two distinct keys. Diagnostic — two creations with different keys for one asset within seconds:

INFO dispatch wo_id=WO-aa11 status=created key=7e90… asset=SYS-0188-COMP-002
INFO dispatch wo_id=WO-bb22 status=created key=4c31… asset=SYS-0188-COMP-002

Fix: generate the UUID once when the scan is captured on the device and persist it across retries, so every retransmission carries the same key and the second hits the 409 path.

False DEFER_PROCUREMENT on in-stock parts. Work orders defer for parts that are physically on the shelf. Root cause: a stale cached asset context reflects a stock level from before a recent receipt. Diagnostic — a deferral whose part shows positive on-hand in the ledger. Fix: shorten CACHE_TTL_SECONDS for fast-moving assets and invalidate the asset cache key on any inventory mutation event, so a receipt evicts the stale snapshot immediately. Tuning those reorder points themselves is covered under inventory threshold optimization.

Resolution timeouts under load. During shift change, scans fail with asyncio.TimeoutError on the asset GET. Root cause: the registry read latency exceeds the 2-second budget when the cache is cold. Diagnostic:

ERROR resolve_asset_context asset=SYS-0312-COMP-017 error=TimeoutError elapsed=2.01s

Fix: pre-warm the cache for assets in the active scanning zone ahead of the shift, raise the connection-pool ceiling, and fall back to the last cached snapshot with a lowered confidence flag rather than failing the scan outright.

Unresolved scans on newly tagged assets. A freshly tagged asset returns 404 from the registry and the scan dead-ends. Root cause: the tag was printed before the asset record and its routing tags were provisioned. Fix: gate tag printing on registry creation and reconcile orphaned tags on a fixed cadence, following the provisioning sequence in syncing barcode scanners with CMMS asset registries.

Frequently Asked Questions

How does the scan path prevent duplicate work orders?

Deduplication is anchored on a single client-generated idempotency_key created when the scan is captured, not per HTTP attempt. Every retransmission carries the same key, the key rides in the Idempotency-Key header on dispatch, and the CMMS returns the existing work order with a 409 instead of creating a second one.

Should scan handling be synchronous or asynchronous?

The resolve-and-dispatch path runs inline so the technician gets an instant acknowledgment, but it is bounded by a semaphore and short timeouts. Heavier follow-up — replenishment, escalation notifications, audit fan-out — is handed to background workers so a slow downstream call never blocks the next scan.

Where are the SLA fields set during a scan?

They are assigned in rule evaluation. The scan type and asset criticality select a Priority, the priority maps to a fixed SLA window that yields requested_completion, and the escalation path sets escalation_tier. Those three fields then travel unchanged with the work order so every downstream consumer reads the same contract.

What happens when a required part is out of stock?

The directive becomes DEFER_PROCUREMENT and the work order is handed to automated reorder triggers, which convert the shortfall into a purchase requisition. The dispatch queue is never blocked waiting on parts; the deferred work order is reactivated when stock is received.

How does the path behave when the CMMS registry is slow or down?

Asset resolution enforces a short timeout and falls back to a time-bound cached snapshot rather than failing the scan, recording the lower confidence that decision carries. Dispatch retries transient 5xx responses with backoff, and a commit that exhausts its retries lands in the dead-letter queue for review.

Related

Continue with the sibling components of this domain: parts availability checks for the bill-of-materials evaluation gate, automated reorder triggers for the procurement deferral path, and inventory threshold optimization for the dynamic reorder points that shape stock reality. For device-level setup, see syncing barcode scanners with CMMS asset registries, and for the shared payload contract, work order schema standards.

Part of: Asset Lookup & Inventory Synchronization — production engineering resources for Python and CMMS integration teams.