Capturing Field Inspection Results on Mobile Devices

18 min read 1 sub-topic

This guide is part of the Inspection Scheduling & Field Operations track, which covers how a permit leaves the office and gets adjudicated in the field. Once an inspector arrives on site, the platform’s job is to capture a pass, fail, or correction result — with checklists, photos, and geotags — reliably enough to hold up in a hearing, even when the cell signal drops to one bar in a basement mechanical room.

Problem Statement and Scope

Field capture is where a permitting system meets physical reality, and physical reality is hostile to clean data. An inspector stands in a half-built structure with no reliable connectivity, needs to record a code-referenced result against a checklist, attach photographic evidence, and stamp it with a location — then move to the next stop before the first record has finished uploading. If the capture layer assumes a live connection, one of three failures follows almost immediately:

  • Lost results. The app POSTs a result, the request times out on flaky cellular, and the inspector believes the visit was recorded when it was not. The permit never advances, or worse, advances on a stale assumption.
  • Unverifiable evidence. A contractor disputes a failed inspection months later. The photo exists, but nothing binds it to the moment of capture — no hash, no geotag, no tamper-evident timestamp — so it is worthless as a record.
  • Ambiguous results. “Failed” is recorded with no structured reason, no code citation, and no correction list, so the office cannot generate a notice, the contractor cannot fix the right thing, and the re-inspection re-litigates the whole visit.

The people who feel this are concrete: field inspectors who cannot trust the app and fall back to paper; municipal clerks who must reconcile paper against the system days later; compliance officers who need a defensible chain of custody for every piece of evidence; and Python automation builders who have to make all of it deterministic across dozens of inspection types.

The design principle that resolves this is offline-first: the device is the source of truth for the duration of a visit, and the network is an eventually-consistent replication channel, not a precondition for work. The inputs to this subsystem are a scheduled inspection assignment, a structured checklist template, and whatever the inspector observes. The outputs are an immutable, hashed result record; one or more evidence artifacts each bound to that record by content hash and geotag; and a lifecycle transition written back to the permit once the record reaches the server. Everything the inspector does must succeed with the radio off.

Offline-first capture and replay pipeline for field inspection results On the device side of a trust and connectivity boundary, an inspector completes a structured checklist that produces a pass, fail, or correction result. The result is canonicalized and content-hashed, then written to a local durable queue together with photo and geotag evidence, each artifact hashed for chain of custody. When connectivity returns the sync engine replays queued records with idempotency keys to a FastAPI ingestion endpoint, which re-verifies every evidence hash, appends the result to an immutable store, and applies a permit lifecycle transition: a pass advances the permit toward issuance while a fail opens a code violation. connectivity boundary · sync ON DEVICE (offline-first) SERVER (eventually consistent) Checklist capture pass / fail / correction code-referenced items Canonicalize SHA-256 result hash idempotency key Evidence photo hash · geotag chain of custody Durable queue local SQLite (WAL) survives app kill replay + retry FastAPI ingest verify hashes dedup on idem-key append-only store Pass advance Fail violation

Prerequisites

This implementation targets Python 3.10+ on the server (the examples use match/case, StrEnum, and modern type syntax). The device-side queue is illustrated with SQLite because it ships on every mobile platform and gives you crash-durable writes; the same contract works with any embedded key-value store the mobile runtime exposes.

pip install "fastapi>=0.110" "uvicorn>=0.29" "pydantic>=2.6" \
            "python-multipart>=0.0.9" "structlog>=24.1"

Environment assumptions:

  • Each inspector authenticates through your municipal identity provider and carries a stable subject and a field_inspector role. Result capture must be gated by the same permission model used elsewhere in the platform — see implementing role-based access for clerk portals, which defines the inspector tier and the transitions it is allowed to drive.
  • The device holds a synced copy of each assigned inspection and its checklist template so capture never requires a round trip.
  • The server exposes an append-only store (a WORM bucket or an immutable table) for results and evidence, and a permit datastore whose lifecycle state this pipeline mutates.
  • Object storage (S3-compatible) for evidence blobs, addressed by content hash so an identical photo is stored once.

Step 1 — Define a Structured Result Schema

“Failed” with no structure is unusable downstream. Model the result as a typed record where every checklist item carries an outcome and, on failure, a code citation and a correction instruction. Encoding this as a schema — rather than free text — is what lets the office generate a notice automatically and lets a re-inspection target the exact deficiency.

from __future__ import annotations
from datetime import datetime, timezone
from enum import Enum
from pydantic import BaseModel, ConfigDict, Field


class Outcome(str, Enum):
    PASS = "pass"
    FAIL = "fail"
    CORRECTION = "correction"  # passes conditionally; re-check specific items
    NOT_APPLICABLE = "n/a"


class ChecklistItem(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")
    item_code: str = Field(min_length=1, max_length=32)   # e.g. "IBC-1704.2"
    outcome: Outcome
    note: str = Field(default="", max_length=1000)
    # Populated only when outcome is FAIL/CORRECTION; drives the notice.
    correction: str = Field(default="", max_length=1000)


class InspectionResult(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")
    permit_id: str
    inspection_id: str
    inspector_sub: str                     # actor identity from the IdP
    inspection_type: str                   # "rough_electrical", "final_framing", ...
    captured_at: datetime                  # tz-aware, set on device at capture time
    items: list[ChecklistItem] = Field(min_length=1, max_length=300)
    evidence: list[EvidenceRef] = Field(default_factory=list, max_length=50)

    @property
    def overall(self) -> Outcome:
        outcomes = {i.outcome for i in self.items}
        if Outcome.FAIL in outcomes:
            return Outcome.FAIL
        if Outcome.CORRECTION in outcomes:
            return Outcome.CORRECTION
        return Outcome.PASS

The overall outcome is derived, never entered by hand: a single failed item fails the visit, so the inspector cannot accidentally tap “pass” over a recorded deficiency. captured_at is stamped on the device at the moment of capture, not on arrival at the server, because the two can be hours apart on a bad-signal route.

Step 2 — Bind Evidence with Content Hashes for Chain of Custody

A photo is only evidence if you can prove it is the same photo that was taken, at that place, at that time. Hash each artifact with SHA-256 on the device the instant it is captured, and treat that hash as its permanent identifier. The record references evidence by hash; the blob is uploaded separately and verified against the hash on arrival. Any later edit to the pixels changes the hash and breaks the reference — that is the chain of custody.

import hashlib
from pydantic import BaseModel, ConfigDict, Field


def sha256_file(data: bytes) -> str:
    """Content address for an evidence blob. Computed on-device at capture."""
    return hashlib.sha256(data).hexdigest()


class GeoTag(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")
    lat: float = Field(ge=-90, le=90)
    lon: float = Field(ge=-180, le=180)
    accuracy_m: float = Field(ge=0)        # GPS horizontal accuracy in meters
    fixed_at: datetime                     # when the location fix was taken


class EvidenceRef(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")
    sha256: str = Field(pattern=r"^[0-9a-f]{64}$")
    media_type: str = Field(pattern=r"^(image/jpeg|image/png|image/heic)$")
    byte_len: int = Field(gt=0, le=25 * 1024 * 1024)   # cap per-photo size
    geo: GeoTag | None = None              # None only if the device had no fix
    item_code: str = Field(default="")     # ties a photo to the item it documents


InspectionResult.model_rebuild()  # resolve the forward reference to EvidenceRef

Because the geotag carries its own accuracy and fix time, the office can distinguish a precise on-site fix from a stale or coarse one and weigh the evidence accordingly. Store the blob in object storage keyed by its hash so a re-uploaded duplicate deduplicates for free and never overwrites the original.

Step 3 — Canonicalize and Sign the Result on Device

Before the record enters the outbound queue, freeze it. Serialize it with sorted keys and compute a result hash over the canonical bytes; that hash is the record’s identity and, combined with the inspection ID, its idempotency key. If the same result is replayed three times over a flaky link, the server sees one identity and commits once.

import json
import hashlib


def canonical_bytes(result: InspectionResult) -> bytes:
    """Deterministic serialization: identical results hash identically."""
    payload = result.model_dump(mode="json")
    return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()


def result_hash(result: InspectionResult) -> str:
    return hashlib.sha256(canonical_bytes(result)).hexdigest()


def idempotency_key(result: InspectionResult) -> str:
    # One logical result per (inspection, content). A correction re-visit is a new
    # inspection_id, so it is a distinct record — not a duplicate to be dropped.
    return f"{result.inspection_id}:{result_hash(result)[:16]}"

The idempotency key is what makes the whole offline model safe: capture, queue, and replay can all fail and retry at any point, and the server still ends up with exactly one committed result.

Step 4 — Persist to a Durable On-Device Queue

The device is the source of truth until sync succeeds, so its queue must survive an app crash, a force-quit, or a dead battery mid-visit. A local SQLite database in write-ahead-logging mode gives you atomic, crash-durable writes without a server. Each queued row holds the canonical payload, the idempotency key, and a delivery state.

import sqlite3
from contextlib import closing

SCHEMA = """
CREATE TABLE IF NOT EXISTS outbox (
    idem_key    TEXT PRIMARY KEY,          -- dedup within the device
    permit_id   TEXT NOT NULL,
    payload     BLOB NOT NULL,             -- canonical_bytes(result)
    state       TEXT NOT NULL DEFAULT 'pending',   -- pending|sent|acked
    attempts    INTEGER NOT NULL DEFAULT 0,
    created_at  TEXT NOT NULL
);
"""


def open_outbox(path: str) -> sqlite3.Connection:
    conn = sqlite3.connect(path, isolation_level=None)
    conn.execute("PRAGMA journal_mode=WAL;")   # durable across app kill
    conn.execute("PRAGMA synchronous=NORMAL;")
    conn.executescript(SCHEMA)
    return conn


def enqueue(conn: sqlite3.Connection, result: InspectionResult) -> str:
    key = idempotency_key(result)
    with closing(conn.cursor()) as cur:
        # INSERT OR IGNORE makes local enqueue idempotent too: tapping "save"
        # twice does not create two rows.
        cur.execute(
            "INSERT OR IGNORE INTO outbox(idem_key, permit_id, payload, created_at) "
            "VALUES (?,?,?,?)",
            (key, result.permit_id, canonical_bytes(result),
             datetime.now(timezone.utc).isoformat()),
        )
    return key

The inspector’s “save” action is complete the moment this row commits — no network involved. The queue is drained later by the sync engine, whose replay, ordering, and conflict handling are involved enough to warrant their own walkthrough: see syncing offline inspection data with conflict resolution.

Step 5 — Ingest on the Server and Write Back to the Permit Lifecycle

When a queued record reaches the FastAPI endpoint, the server re-verifies every evidence hash, deduplicates on the idempotency key, appends the result to the immutable store, and only then drives the permit’s lifecycle transition. Ordering matters: the result is committed as a fact before the permit state is allowed to change, so a crash between the two leaves a recorded result to replay against, never a phantom transition.

from fastapi import FastAPI, HTTPException, Header, status
from fastapi.responses import JSONResponse

app = FastAPI()


@app.post("/inspections/{inspection_id}/result")
async def ingest_result(
    inspection_id: str,
    result: InspectionResult,
    idempotency_key_hdr: str = Header(alias="Idempotency-Key"),
) -> JSONResponse:
    # 1. Reject a replay cheaply, before doing any work.
    if await store.already_committed(idempotency_key_hdr):
        return JSONResponse({"status": "duplicate"}, status_code=status.HTTP_200_OK)

    # 2. Re-derive the key server-side; a client cannot smuggle a mismatched body.
    if idempotency_key_hdr != idempotency_key(result):
        raise HTTPException(status.HTTP_400_BAD_REQUEST, "idempotency key mismatch")

    # 3. Verify each blob's bytes match the hash the device claimed.
    for ref in result.evidence:
        blob = await blobs.fetch(ref.sha256)
        if blob is None or sha256_file(blob) != ref.sha256:
            raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY,
                                f"evidence hash unverified: {ref.sha256[:12]}")

    # 4. Commit the result as an immutable fact FIRST.
    await store.append(idempotency_key_hdr, result)

    # 5. Then transition the permit — a fail branches to enforcement.
    match result.overall:
        case Outcome.PASS:
            await permits.transition(result.permit_id, to="inspection_passed")
        case Outcome.FAIL:
            await permits.open_violation(result.permit_id, result)
        case Outcome.CORRECTION:
            await permits.transition(result.permit_id, to="correction_required")
    return JSONResponse({"status": "committed"}, status_code=status.HTTP_201_CREATED)

A failed inspection is not a dead end — it is the trigger for enforcement. Routing Outcome.FAIL into a violation is where this track hands off to compliance work; the record shape and the state it opens are defined in modeling code violation lifecycles in Python, so a field failure becomes a tracked, escalating case rather than a note nobody actions.

Configuration Reference

Parameter Type Default Municipal-context notes
captured_at_source str "device" Always stamp on-device; server arrival time can trail by hours on bad-signal routes.
max_photo_bytes int 26214400 25 MiB per artifact; HEIC/JPEG originals from modern phones fit, TIFF scans do not.
evidence_per_result int 50 Bounds a single visit’s upload burst so one stop cannot exhaust the queue.
geo_accuracy_floor_m float 50.0 Flag fixes coarser than this; a basement often yields no usable GPS at all.
outbox_journal_mode str "WAL" Write-ahead logging keeps the queue durable across an app force-quit.
idem_key_ttl_days int 90 How long the server remembers a committed key to reject late replays.
hash_algorithm str "sha256" Content address for results and evidence; do not downgrade to MD5/SHA-1.
result_store str "append_only" WORM bucket or immutable table; never a mutable row a re-visit overwrites.

Error Handling and Edge Cases

Field capture breaks in specific, physical ways. Handle each explicitly rather than letting it surface as a lost record.

  • No GPS fix underground. A mechanical or electrical inspection often happens where the device gets no satellite lock. Make geo optional but record why it is absent (last known fix plus accuracy), and let the office treat a fix-less photo as weaker evidence rather than rejecting the whole result.
  • App killed mid-capture. The OS reclaims memory and terminates the app before the inspector taps save. Because capture writes to the WAL-mode outbox incrementally, a partially entered checklist is recoverable on relaunch; never hold the in-progress result only in memory.
  • Clock skew on the device. A phone with a wrong clock stamps captured_at in the future or past. Record the device clock but also capture the server receive time on ingest, and reconcile: a large gap is a data-quality flag, not a silent overwrite.
  • Evidence hash mismatch on upload. The blob was truncated by a dropped connection, so its bytes no longer match the claimed hash. The server rejects it with 422 and the sync engine re-queues the blob; the result row stays pending until every referenced artifact verifies.
  • Duplicate submission after a timeout. The device never saw the 201 and replays. The idempotency key makes the second POST a no-op that returns duplicate — the permit transitions exactly once. This is the single most common field failure and the whole design exists to make it harmless.
  • Unauthorized capture. A device whose inspector token has expired or lacks the field_inspector scope must not commit a result. Enforce the role at ingest using the same evaluator as the rest of the platform, and reject with 403 so a stale device cannot mutate a permit.

Testing and Verification

The properties that matter are determinism (identical results hash identically), idempotency (replays commit once), and correct outcome derivation. Test them as data.

import pytest


def _result(outcomes: list[Outcome]) -> InspectionResult:
    return InspectionResult(
        permit_id="P-1", inspection_id="I-9", inspector_sub="[email protected]",
        inspection_type="rough_electrical",
        captured_at=datetime(2026, 7, 15, 14, 0, tzinfo=timezone.utc),
        items=[ChecklistItem(item_code=f"IBC-{i}", outcome=o)
               for i, o in enumerate(outcomes)],
    )


def test_overall_outcome_is_derived_not_entered():
    assert _result([Outcome.PASS, Outcome.FAIL]).overall is Outcome.FAIL
    assert _result([Outcome.PASS, Outcome.CORRECTION]).overall is Outcome.CORRECTION
    assert _result([Outcome.PASS, Outcome.PASS]).overall is Outcome.PASS


def test_canonical_hash_is_stable_across_key_order():
    r = _result([Outcome.PASS])
    assert result_hash(r) == result_hash(r.model_copy(deep=True))


def test_idempotency_key_survives_a_replay():
    r = _result([Outcome.FAIL])
    assert idempotency_key(r) == idempotency_key(r)  # same content → same key

A passing derivation test plus a stable-hash test gives you a regression net: change the schema or the serialization and the suite tells you immediately, before a device in the field starts producing records the server rejects.

Integration Notes

This capture layer is the field-facing edge of the inspection track and it touches subsystems on both sides. Upstream, every result is scoped by the permission model that governs clerk portals and inspector roles — an inspector may only commit results for visits assigned to them, and the ingest endpoint enforces that scope before it appends anything. The structured result record it produces is the input to the sync engine detailed in syncing offline inspection data with conflict resolution, which owns queue replay, ordering, and the merge rules for concurrent edits. Downstream, a failed result opens a case in the code violation lifecycle, so field enforcement and office enforcement share one record rather than diverging. When bulk result uploads arrive after a full day offline, route them through the platform’s shared error handling and retry logic for ingestion pipelines so a rejected artifact is queued and traceable rather than dropped.

Frequently Asked Questions

Why hash evidence on the device instead of on the server?

Because chain of custody depends on binding the artifact to the moment and place of capture, and the server never saw that moment. Hashing on the device the instant a photo is taken creates a fingerprint that travels with the record; the server later re-computes the hash over the uploaded bytes and confirms they match. If the two agree, the photo is provably the one captured on site. If someone edits the pixels afterward, the hash changes and the reference breaks — which is exactly the tamper-evidence a hearing requires.

What makes a submission idempotent when the network keeps dropping?

Each result carries an idempotency key derived from its inspection ID and a hash of its canonical content. The server records committed keys and treats any repeat as a no-op that returns a duplicate acknowledgment. So the device can POST the same result any number of times — after a timeout, after an app restart, after a day offline — and the permit transitions exactly once. The design assumes replays will happen and makes them harmless rather than trying to prevent them.

How should the app behave when there is no GPS fix at all?

Record the result anyway and mark the geotag as absent, ideally with the last known fix and its age. An inspection in a basement or a steel-framed structure frequently yields no satellite lock, and refusing to save the result would push the inspector back to paper. The office can weigh a fix-less photo as weaker evidence, but a captured result with honest metadata always beats a lost one.

Where does a failed inspection go after capture?

Straight into enforcement. When the overall outcome is a fail, the ingest endpoint opens a case in the violation lifecycle rather than merely marking the permit. That means the code citation and correction list the inspector entered on site become the seed of the notice the office sends, so field observation and office action share a single structured record instead of being re-keyed.

Explore deeper

Sub-topic Syncing Offline Inspection Data with Conflict Resolution