Orchestrating Photogrammetry Jobs with Python Schedulers

A single survey is a script. Forty surveys a week, on shared hardware, with reruns and partial failures and a deadline attached to each one, is a scheduling problem — and the scripts that worked beautifully for the single case fail in a specific, predictable way at fleet scale: two jobs start dense matching at the same moment, the host swaps, both are killed, and the queue helpfully restarts them together. This page builds the orchestration layer that prevents that. It covers modelling a reconstruction as a dependency graph, admitting jobs by memory rather than by core count, making every stage resumable so a restart costs minutes instead of hours, and keeping the run state somewhere that survives the orchestrator itself.

Everything here sits above the reconstruction engine rather than inside it. The engine invocation is the one described in setting up OpenDroneMap with Python, and the driver choice — a long-running service or a subprocess per job — is the one weighed in PyODM vs direct ODM CLI for automation. What changes at fleet scale is not how a job runs but what decides when it runs and what happens when it does not finish.

Audience and prerequisites. Python 3.10+, comfort with subprocesses and a task queue of some kind, and a working single-survey pipeline. No specific scheduler is assumed: the patterns here are expressed in the standard library and map onto Celery, RQ, Prefect, Airflow, or a systemd timer and a database table, because the hard parts are the same in all of them.

# Prerequisites

The dependency worth arguing about is the last one. Run state held in a Python dictionary disappears with the process, and an orchestrator that cannot answer “what was running when I died?” cannot safely restart anything — it either re-runs finished work or abandons unfinished work, and both are worse than the crash.

# Conceptual architecture

A reconstruction is not one task; it is a chain of stages with different resource profiles. Ingestion is I/O-bound and cheap. Feature extraction is CPU-bound and parallel. Dense matching is memory-bound and hostile to concurrency. Export is I/O-bound again. Scheduling them as a single opaque unit means the whole job is sized for its worst stage, so the cheap stages hold a dense-matching-sized reservation while doing almost nothing.

Modelling the pipeline as a graph of stages, each declaring its own resource class, lets the scheduler interleave: several jobs can extract features concurrently while exactly one is in dense matching, and the export of a finished job can proceed alongside both.

Figure 1 — Scheduling by stage rather than by job. The memory-bound stage gets its own narrow gate; the rest of the pipeline is not throttled to match it.

# Step 1: Describe a job as validated data, not arguments

The first thing a scheduler needs is a job specification it can store, hash, compare and re-submit. Passing a bag of keyword arguments through a queue works until the day two versions of the caller disagree about a default, at which point two jobs that look identical produce different outputs.

from pathlib import Path
from typing import Literal

from pydantic import BaseModel, Field, field_validator


class ReconstructionJob(BaseModel):
    """Everything needed to reproduce one run, and nothing else."""

    project: str = Field(min_length=1)
    images_dir: Path
    gcp_file: Path | None = None
    target_epsg: int
    ortho_resolution_cm: float = Field(gt=0, le=100)
    split: int = Field(default=500, ge=50)
    engine_digest: str            # pinned image, never a floating tag
    priority: Literal["normal", "urgent"] = "normal"

    @field_validator("images_dir")
    @classmethod
    def _must_exist(cls, v: Path) -> Path:
        if not v.is_dir():
            raise ValueError(f"images_dir does not exist: {v}")
        if not any(v.iterdir()):
            raise ValueError(f"images_dir is empty: {v}")
        return v

    def fingerprint(self) -> str:
        """Stable identity for deduplication and cache lookup."""
        import hashlib
        payload = self.model_dump_json(exclude={"priority"}).encode()
        return hashlib.sha256(payload).hexdigest()[:16]

Two details earn their keep. Validation happens at submission, so a job with a missing image directory is rejected in milliseconds by the process that has the context to explain it, rather than an hour later by a worker that does not. And fingerprint() deliberately excludes priority, so re-submitting the same work at a higher priority is recognised as the same work rather than queued twice.

# Step 2: Admit jobs by memory, not by worker slot

Most task queues size their pool in workers. That is the wrong unit for photogrammetry, because a worker’s memory footprint varies by two orders of magnitude across stages and by an order of magnitude across datasets. A pool of four workers is safe for four small jobs and fatal for four large ones.

The fix is admission control against a measured budget: before a memory-bound stage starts, ask whether its estimated footprint fits in what is actually free, minus a reserve the operating system keeps for itself.

import psutil
from filelock import FileLock, Timeout


class MemoryGate:
    """A host-wide gate for memory-bound stages.

    The lock file makes this work across processes — a per-process semaphore
    would let two independent workers each believe they were alone.
    """

    def __init__(self, lock_path: str, reserve_gb: float = 4.0):
        self.lock = FileLock(lock_path)
        self.reserve_gb = reserve_gb

    def acquire(self, need_gb: float, timeout_s: float = 3600) -> bool:
        try:
            self.lock.acquire(timeout=timeout_s)
        except Timeout:
            return False
        free_gb = psutil.virtual_memory().available / (1024 ** 3)
        if free_gb - self.reserve_gb < need_gb:
            self.lock.release()
            return False
        return True

    def release(self) -> None:
        if self.lock.is_locked:
            self.lock.release()

The estimate for need_gb should be measured rather than guessed, and it scales with the dataset. A serviceable model is linear in image count at a fixed resolution — measure two points on your own hardware and interpolate — refreshed whenever the engine version or the default quality settings change.

# Step 3: Make every stage resumable

A job that must restart from zero after any failure is a job whose expected cost rises sharply with its length, because the probability of some interruption over eight hours is not small. Resumability turns that into a probability of losing one stage.

The reconstruction engine already leaves stage markers on disk, as described in the OpenDroneMap setup guide. The orchestrator’s job is to read them rather than to keep its own parallel notion of progress, because only the filesystem survives the orchestrator crashing.

Figure 2 — The same interruption, twice. Reading stage markers before starting turns a total loss into the cost of one stage, and the check itself is a filesystem call.

from pathlib import Path

# Ordered stages and the directory each one leaves behind on success.
STAGE_MARKERS = [
    ("dataset", "images.json"),
    ("opensfm", "opensfm/reconstruction.json"),
    ("openmvs", "opensfm/undistorted/openmvs/scene_dense.ply"),
    ("odm_filterpoints", "odm_filterpoints/point_cloud.ply"),
    ("odm_meshing", "odm_meshing/odm_mesh.ply"),
    ("mvs_texturing", "odm_texturing/odm_textured_model_geo.obj"),
    ("odm_georeferencing", "odm_georeferencing/odm_georeferenced_model.laz"),
    ("odm_orthophoto", "odm_orthophoto/odm_orthophoto.tif"),
]


def resume_stage(project_dir: Path) -> str | None:
    """Name the earliest stage whose output is missing, or None if complete."""
    for stage, marker in STAGE_MARKERS:
        if not (project_dir / marker).exists():
            return stage
    return None

One caveat: a marker proves a stage wrote something, not that it wrote something valid. A run killed mid-write can leave a truncated file that satisfies exists(). For the markers that matter — the georeferenced model, the orthophoto — verify openability rather than existence, which for a raster is one rasterio.open and for a point cloud one header read.

# Step 4: Keep run state where a restart can find it

The last piece is durable state. A table with one row per job, updated at each transition, answers the three questions a restarting orchestrator needs: what was running, what finished, and what has been retried too many times already.

import sqlite3
from contextlib import closing

SCHEMA = """
CREATE TABLE IF NOT EXISTS runs (
  fingerprint   TEXT PRIMARY KEY,
  project       TEXT NOT NULL,
  state         TEXT NOT NULL,          -- queued | running | done | failed
  stage         TEXT,                   -- last stage entered
  attempts      INTEGER NOT NULL DEFAULT 0,
  worker        TEXT,                   -- host:pid that claimed it
  heartbeat_ts  REAL,                   -- updated while running
  last_error    TEXT
);
"""


def claim_next(db: str, worker: str, stale_after_s: float = 900.0) -> str | None:
    """Atomically claim a queued job, or one whose worker stopped reporting."""
    with closing(sqlite3.connect(db, isolation_level="IMMEDIATE")) as con:
        con.executescript(SCHEMA)
        row = con.execute(
            """
            SELECT fingerprint FROM runs
             WHERE state = 'queued'
                OR (state = 'running'
                    AND heartbeat_ts < strftime('%s','now') - ?)
             ORDER BY attempts ASC, rowid ASC LIMIT 1
            """,
            (stale_after_s,),
        ).fetchone()
        if row is None:
            return None
        con.execute(
            """UPDATE runs
                  SET state='running', worker=?, attempts=attempts+1,
                      heartbeat_ts=strftime('%s','now')
                WHERE fingerprint=?""",
            (worker, row[0]),
        )
        con.commit()
        return row[0]

The heartbeat is what makes this safe. A worker that dies without updating its row leaves the job in running forever unless something can decide the claim has gone stale; the timestamp lets another worker reclaim it after a grace period, and the attempts counter stops that from becoming an infinite loop. The transition rules that decide whether a reclaim is safe are the subject of retrying failed photogrammetry jobs idempotently.

Figure 3 — Four states and two ways back to the queue. The heartbeat path is what distinguishes a fleet that heals itself from one that quietly stops making progress.

# Parameter deep-dive

The relationship between heartbeat_interval_s and stale_after_s is the one to get right. A ratio below about five invites a healthy but briefly-stalled worker to have its job stolen and run twice; a ratio above about thirty means a dead host holds a job for a quarter of an hour before anyone notices.

# Verification and output inspection

An orchestrator is verified by killing it. The properties worth asserting are that no job is lost, none is run twice concurrently, and the queue drains.

import sqlite3
from contextlib import closing


def assert_queue_invariants(db: str) -> None:
    """Properties that must hold at any moment, checkable in one pass."""
    with closing(sqlite3.connect(db)) as con:
        # 1. No two workers hold the same job.
        dup = con.execute(
            "SELECT fingerprint, COUNT(DISTINCT worker) c FROM runs "
            "WHERE state='running' GROUP BY fingerprint HAVING c > 1"
        ).fetchall()
        assert not dup, f"job claimed by multiple workers: {dup}"

        # 2. Nothing is running without a recent heartbeat.
        stale = con.execute(
            "SELECT fingerprint FROM runs WHERE state='running' "
            "AND heartbeat_ts < strftime('%s','now') - 1800"
        ).fetchall()
        assert not stale, f"running with a stale heartbeat: {stale}"

        # 3. Attempts never exceed the configured ceiling.
        over = con.execute("SELECT fingerprint FROM runs WHERE attempts > 3").fetchall()
        assert not over, f"retried past max_attempts: {over}"

        # 4. Every done job has an orthophoto that opens.
        import rasterio
        for (proj,) in con.execute("SELECT project FROM runs WHERE state='done'"):
            with rasterio.open(f"{proj}/odm_orthophoto/odm_orthophoto.tif") as ds:
                assert ds.count >= 1 and ds.crs is not None

Run that after a deliberate SIGKILL of a worker mid-job, and again after the fleet has drained the resulting backlog. An orchestrator that passes both is one that can be left alone over a weekend.

# Troubleshooting

Two workers processed the same project and the outputs are interleaved. The claim was not atomic. sqlite3.connect(..., isolation_level="IMMEDIATE") takes a write lock at the start of the transaction, which is what makes the select-then-update pair safe; the default deferred mode allows two readers to see the same queued row. On a client-server database, the equivalent is SELECT ... FOR UPDATE SKIP LOCKED.

Jobs sit in running forever after a host reboot. Nothing reclaims stale claims. Either the heartbeat is not being written during long stages, or stale_after_s is longer than anyone waits before intervening manually. Write the heartbeat from a background thread rather than between stages — a dense-matching stage can run for hours without returning to the orchestrator’s control flow.

The memory gate is held by a job that finished. The lock was not released on the exception path. Wrap the gate in a context manager, or use try/finally; a worker killed by the OOM killer releases the file lock automatically only because the process died, and a worker that raises and continues does not.

Everything queues behind one enormous survey. Ordering by attempts ASC, rowid ASC is fair but not priority-aware, and a job that needs 38 GB will block the memory gate whenever it reaches the front. Give the memory gate a size-aware policy: allow a small job to pass a waiting large one when the large one could not start anyway. Without that, a single oversized dataset starves the fleet.

Retries make the failure worse. The failure was not transient. Classify before retrying, as described in retrying failed photogrammetry jobs idempotently; an out-of-memory kill retried unchanged consumes hours to fail identically.

The fingerprint changed and everything re-ran. A field was added to the job model, or a default moved. Fingerprints are stable only across a stable schema, so version the model and treat a schema change as a deliberate cache invalidation rather than an accident.

# Related

• Setting up OpenDroneMap with Python
• PyODM vs direct ODM CLI for automation
• Structuring drone imagery for batch processing
• Retrying failed photogrammetry jobs idempotently
• Writing a manifest-driven batch runner for ODM

← Core Photogrammetry Fundamentals