Scanner API Integration & Routing in Automated Ingestion Workflows

Scanner API integration and routing is the subsystem that turns a floor full of imaging hardware into a single programmable capture surface, then decides — for every staged object — which device runs the scan and where the resulting master flows next. Within the parent Automated Ingestion & Batch Scanning Workflows pipeline, this is the stage that sits at the boundary between physical hardware and the software pipeline: it exposes each scanner behind a deterministic API contract, dispatches typed job payloads, and hands validated masters onward to Async Task Queuing for Batches for downstream fan-out. Modern cultural heritage institutions no longer treat imaging hardware as isolated peripherals wired to a single operator workstation; they publish device capabilities through RESTful or gRPC endpoints so a scheduler can orchestrate a multi-device capture farm, enforce consistent technical metadata at the point of creation, and eliminate the manual handoffs that historically introduced chain-of-custody gaps. This page specifies the device contract, the routing logic that selects a target scanner, the resilience rules that isolate a failing device, and the compliance mapping that stamps every captured object with preservation-ready provenance before it leaves the capture stage.

# The Scan Job Contract

Before any command reaches a device, the routing layer enforces a strict, typed contract. The scan job is the unit of work a producer submits: it names the batch, the optical target, the color space, the output encoding, and the media class the object belongs to. Modelling this contract with Pydantic gives the ingestion gateway a single validation surface and gives every downstream worker a self-describing, machine-checkable payload. The field specification below is the authoritative contract every dispatched job must honour; a request that fails any constraint is rejected at the gateway and never reaches scanner firmware, which is what prevents malformed DPI values or unsupported color profiles from locking up a device mid-run.

The two constraints that carry the most preservation weight are the resolution floor and the enumerated color space. A target_dpi below 300 silently produces masters that fail later quality review and must be re-captured — an expensive, sometimes impossible outcome for fragile originals. Pinning color_space to a closed enumeration is what lets a captured object be reconciled against its embedded ICC profile downstream; an unconstrained free-text field is the most common source of derivatives that render with shifted color across a collection.

# Routing Architecture and Job Dispatch

Routing logic is the traffic controller for a high-throughput capture farm. When a batch of archival material is staged, the routing engine evaluates media characteristics, the required optical resolution, and available device capacity before dispatching a payload to a specific scanner. This decision layer deliberately decouples job submission from hardware execution, leaning on Async Task Queuing for Batches so that a long-running TIFF or JPEG 2000 generation cycle never blocks the operator interface or starves other devices. A submitted job returns immediately with an accepted status; the actual capture runs asynchronously against whichever device the router selected, and its outcome is reconciled through the results backend.

Device selection is not round-robin. The router maintains a live health view of every scanner in the pool and dispatches only to devices that are both capable of the requested profile and currently healthy. A fragile bound volume routed to a sheet-feed ADF would be physically damaged; an overhead capture rig requested at a resolution it cannot reach would silently down-sample. Encoding device_class in the contract lets the router refuse those pairings before a command is ever issued. Adaptive compression thresholds and prioritised I/O scheduling then prevent storage-array saturation during peak capture windows, when dozens of multi-hundred-megabyte masters land within the same minute.

The health view is maintained by a circuit breaker per device. When a scanner starts failing — a firmware hang, a jammed feeder, a dropped socket — the breaker trips and the router removes that device from the eligible pool until it recovers, rather than repeatedly dispatching work that will fail. The circuit breaker guarding each scanner transitions through three states, isolating a degraded device from the routing pool until a probe confirms it is healthy again.

In the Open state, dispatch is halted until the recovery timeout allows a single probe in HalfOpen; a successful probe closes the breaker and returns the device to the eligible pool.

# Core Implementation: Async Dispatch with Fault Isolation

Implementing scanner coordination in Python requires an asynchronous execution model, strict type enforcement at the gateway, and resilient network communication that fails closed. The pattern below is a production-ready client that validates the payload against the scan job contract, guards each device with a circuit breaker, dispatches the job asynchronously, and writes a deterministic audit hash for chain-of-custody verification. Every failure is classified and logged rather than swallowed, so an interrupted capture surfaces as an actionable event instead of a silent gap.

import asyncio
import hashlib
import json
import logging
from datetime import datetime, timezone
from typing import Dict, Any
import aiohttp
from pydantic import BaseModel, Field, ValidationError

# Audit logger configured for immutable preservation records
audit_logger = logging.getLogger("scanner.ingest.audit")
logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")

class ScanJobPayload(BaseModel):
    """Strict schema validation for ingestion gateway compliance."""
    batch_id: str = Field(pattern=r"^BATCH-\d{4}-[A-Z]{3}$")
    target_dpi: int = Field(ge=300, le=1200)
    color_space: str = Field(pattern=r"^(AdobeRGB|sRGB|ProPhotoRGB)$")
    output_format: str = Field(default="TIFF")
    media_type: str = Field(default="document")

class CircuitBreaker:
    """Stateful circuit breaker for hardware API resilience."""
    def __init__(self, failure_threshold: int = 3, recovery_timeout: float = 30.0):
        self.failure_count = 0
        self.threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure_time = 0.0
        self.state = "CLOSED"

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = asyncio.get_running_loop().time()
        if self.failure_count >= self.threshold:
            self.state = "OPEN"

    def record_success(self):
        self.failure_count = 0
        self.state = "CLOSED"

    def allow_request(self) -> bool:
        if self.state == "CLOSED":
            return True
        if self.state == "OPEN":
            elapsed = asyncio.get_running_loop().time() - self.last_failure_time
            if elapsed > self.recovery_timeout:
                self.state = "HALF-OPEN"
                return True
            return False
        return True  # HALF-OPEN allows one probe

async def dispatch_scan_job(api_endpoint: str, payload: Dict[str, Any], breaker: CircuitBreaker) -> Dict[str, Any]:
    """Asynchronous job dispatch with validation, audit logging, and fault tolerance."""
    if not breaker.allow_request():
        audit_logger.warning("Circuit breaker OPEN. Job dispatch halted for %s", payload.get("batch_id"))
        raise ConnectionError("Scanner API circuit breaker tripped")

    timeout = aiohttp.ClientTimeout(total=15.0)
    try:
        async with aiohttp.ClientSession(timeout=timeout) as session:
            async with session.post(api_endpoint, json=payload) as response:
                if response.status == 200:
                    breaker.record_success()
                    result = await response.json()
                    # Deterministic audit hash for chain-of-custody verification.
                    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
                    payload_hash = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
                    audit_logger.info(
                        "SCAN_DISPATCH_SUCCESS | batch=%s | hash=%s | status=%s | ts=%s",
                        payload["batch_id"], payload_hash, response.status, 
                        datetime.now(timezone.utc).isoformat()
                    )
                    return result
                else:
                    raise aiohttp.ClientResponseError(
                        request_info=response.request_info,
                        history=response.history,
                        status=response.status,
                        message=f"Scanner rejected job: {response.status}"
                    )
    except Exception as e:
        breaker.record_failure()
        audit_logger.error("SCAN_DISPATCH_FAILURE | batch=%s | error=%s", payload.get("batch_id"), str(e))
        raise

# Execution wrapper demonstrating async coordination
async def run_ingest_pipeline():
    breaker = CircuitBreaker(failure_threshold=2, recovery_timeout=10.0)
    job = {
        "batch_id": "BATCH-2024-MSX",
        "target_dpi": 600,
        "color_space": "AdobeRGB",
        "output_format": "TIFF",
        "media_type": "manuscript"
    }
    try:
        validated = ScanJobPayload(**job)
        await dispatch_scan_job("https://scanner-farm.internal/api/v1/jobs", validated.model_dump(), breaker)
    except ValidationError as ve:
        audit_logger.error("SCHEMA_VALIDATION_FAILED | details=%s", ve.json())
    except Exception as e:
        audit_logger.critical("PIPELINE_TERMINATED | reason=%s", str(e))

if __name__ == "__main__":
    asyncio.run(run_ingest_pipeline())

The canonical JSON serialisation before hashing is deliberate: sorting keys and stripping insignificant whitespace makes the SHA-256 digest stable regardless of field ordering, so the same logical job always produces the same audit hash. That determinism is what lets an auditor reconcile a dispatched payload against its recorded fixity months later. The device-level state machine this pattern manages is explored in depth in Automating batch scanner coordination with Python, which covers the ADF acknowledgment loop and buffer-drain edge cases that a naive dispatch loop gets wrong.

# Integration Points

Scanner API integration is a boundary stage, so most of its value comes from how cleanly it hands work to its neighbours. Immediately after a master is captured and its payload validated, the router enqueues it through Async Task Queuing for Batches, which fans the object out to the compute-heavy stages without blocking capture. For text-bearing collections, routing rules redirect derivatives to OCR Processing Pipelines as soon as the master clears validation, so recognition runs in parallel with the next physical scan rather than serially behind it.

Every transient network fault, firmware timeout, or rejected job surfaced here is classified and handled by Error Handling & Retry Logic; the circuit breaker in the dispatch client is one half of that contract, re-dispatching to a healthy device rather than looping on a dead one. Structural correctness of the captured package — expected file counts, naming conventions, per-object checksums — is asserted by Batch Validation Schemas before the object is accepted. Technical characteristics returned by the device are then normalised and enriched by Metadata Extraction Workflows, which reconcile the scanner’s reported bit depth and compression against what the file actually contains.

# Validation and Compliance Mapping

Compliance mapping transforms raw API responses into preservation-ready assets. Scanner APIs typically return technical metadata in proprietary or device-specific shapes, but a trusted repository requires structured alignment with PREMIS, METS, and FADGI. The integration layer parses the device telemetry, normalises bit depth and compression tags, and attaches a provenance event before the object is handed onward. Each captured master is recorded as a PREMIS event of type capture, cross-referenced to the agent (the device and its firmware version) and the deterministic payload hash — the event vocabulary and its serialisation are defined by the preservation-side PREMIS Metadata Mapping. The returned format identifier is confirmed against a signature database through Preservation Format Identification and resolved to a canonical registry entry via Format Registry Integration, so a device that mislabels a JPEG 2000 master as a plain TIFF is caught at the boundary rather than propagated into archival storage.

The compliance targets this stage must satisfy are concrete. A target_dpi of at least 300 for text and 400 for continuous-tone material meets the FADGI 4-star baseline. Every dispatched job and every returned master must carry a SHA-256 digest for chain-of-custody, and every capture event must be written to append-only storage to satisfy the ISO 14721 (OAIS) audit requirements that the parent OAIS-Compliant Digital Preservation Architecture enforces end to end. For authoritative preservation metadata modelling, institutions should align with the PREMIS Data Dictionary maintained by the Library of Congress.

# Security and Auditability

Hardware control surfaces are high-value attack vectors in digitization infrastructure: a compromised scanner endpoint can issue unauthorised capture commands or tamper with technical metadata at the point of creation. Every scanner endpoint must therefore enforce mutual TLS (mTLS) at the transport layer, role-based access control (RBAC) on the dispatch API, and signed payload verification so that a request the gateway cannot cryptographically attribute is refused. Strict egress firewall rules at the management VLAN boundary keep device control traffic off any routable path, and the hardening posture the whole program inherits is specified by Digital Preservation Security Policies.

Beyond the perimeter, auditability requires immutable logging of every API transaction — request payloads, device telemetry, cryptographic checksums, and the operator or service authentication token that authorised the job. Those logs are synchronised into the repository’s preservation store, feeding the same append-only audit trail that Long-term Storage Architecture protects with replication and scheduled fixity re-validation. For engineers designing the concurrent hardware sessions this stage depends on, Python’s native asyncio documentation covers the patterns for managing many device connections without resource contention: Asynchronous I/O in Python.

# Troubleshooting Reference

# Frequently Asked Questions

# Why validate the scan job payload before it reaches the scanner?

Because a malformed request that reaches firmware can lock up a device, waste operator time, or — worse — produce a non-compliant master that only fails review after the fragile original has been re-shelved. Validating target_dpi, color_space, and naming against a typed contract at the ingestion gateway means an invalid job is refused in microseconds and never touches hardware, so the device stays available and the pipeline never has to re-capture an irreplaceable object.

# What does the circuit breaker protect against that retries alone do not?

Retries handle an isolated, transient fault on an otherwise healthy device. The circuit breaker handles a device that has failed systemically — a jammed feeder, a hung firmware process, a severed socket. Once consecutive failures cross the threshold the breaker opens and the router stops dispatching to that scanner for a cooldown window, so the pool stops hammering hardware that cannot respond. After the cooldown a single probe is allowed through; success returns the device to the eligible pool without a human intervening.

# How does routing decide which scanner runs a given job?

The router filters the device pool by two facts: capability and health. Capability comes from the job’s media_type and device_class — a bound volume must go to an overhead rig, not a sheet-feed ADF, and a 600 DPI request must go to a device that can physically reach it. Health comes from each device’s circuit-breaker state; only closed (or probing) breakers are eligible. Among the devices that pass both filters, the router balances by current capacity so no single scanner becomes the bottleneck.

# How is a captured master tied to an auditable chain of custody?

Every dispatched payload is serialised to canonical JSON — sorted keys, no insignificant whitespace — and hashed with SHA-256, so the same logical job always yields the same digest. That hash is recorded on a PREMIS capture event alongside the device agent and timestamp, and the event is written to append-only storage. An auditor can later re-hash the recorded payload and confirm it matches, proving the master that entered the archive is exactly the one the operator dispatched.

# Where do scanner integration faults get retried?

They do not get retried inside the dispatch client itself. The client classifies each fault and surfaces it; the actual backoff, jitter, and re-dispatch policy lives in the shared fault-tolerance layer described in Error Handling & Retry Logic. That separation keeps the retry policy version-pinned and auditable in one place, rather than scattered across every device driver.

# Related

• Async Task Queuing for Batches — the broker layer that fans validated masters out to workers without blocking capture.
• OCR Processing Pipelines — the derivative stage routing redirects text-bearing masters to after validation.
• Error Handling & Retry Logic — the fault-classification and backoff policy the dispatch client surfaces exceptions into.
• Batch Validation Schemas — asserts structural correctness of the captured package before acceptance.
• Automating batch scanner coordination with Python — device-level state machine, ADF acknowledgment loops, and buffer-drain edge cases.
• PREMIS Metadata Mapping — the event vocabulary every capture and dispatch is recorded against.