app.workers

ARQ entrypoint — thin shim over app.workers.dispatcher.

The actual lease+dispatch+heartbeat lives in dispatcher.execute_task so any queue backend (Celery, SQS, in-memory) can drive the same code path. This module is the ARQ-specific wrapper: WorkerSettings for the arq CLI and a run_task(ctx, task_id) function with ARQ’s expected signature.

class app.workers.runner.WorkerSettings

Bases: object

functions: ClassVar[list[Callable[[...], Any]]] = [<function run_task>]

static on_startup(ctx)

static get_redis_url()

Return type:

str

async app.workers.runner.run_task(ctx, task_id)

ARQ-shaped wrapper. ctx is unused — ARQ passes its job context dict here, but execute_task is queue-agnostic.

Parameters:

• ctx (dict)

• task_id (str)

Return type:

dict

ProgressEvent emitter.

The worker writes events to two sinks:

1. events.jsonl — durable, used for SSE replay via Last-Event-ID.

2. JobEvent table — durable index for the API to count/serve events.

Both are appended in order. The event_id in the DB doubles as the SSE event id; clients reconnect with Last-Event-ID: <int> to resume from that point.

class app.workers.events.JsonlEventSink(path)

Bases: object

Append-only JSONL sink, thread-safe under a process. Workers run in subprocess so cross-process locking isn’t required for v0.

Parameters:

path (Path)

append(payload)

Parameters:

payload (dict[str, Any])

Return type:

None

read_from(after_seq=0)

Parameters:

after_seq (int)

Return type:

list[dict]

app.workers.events.now_iso()

Return type:

str

Worker-side ProgressEvent persistence.

class app.workers.progress.WorkerProgressReporter(*, job_id, task_id, loop, event_path=None)

Bases: object

Synchronous reporter used by worker task handlers.

Handlers run in a worker thread. Event persistence is scheduled back onto the dispatcher’s asyncio loop, then mirrored to events.jsonl after the DB row is committed.

Parameters:

• job_id (str)

• task_id (str)

• loop (AbstractEventLoop)

• event_path (Path | None)

phase_started(phase)

Parameters:

phase (Literal['feature_extraction', 'matching', 'geometric_verification', 'incremental_init', 'incremental_register', 'incremental_ba', 'global_rotation_avg', 'global_positioning', 'global_ba', 'hierarchical_cluster', 'hierarchical_merge', 'panorama', 'spherical', 'bundle_adjust', 'triangulate', 'relocalize', 'pose_graph_optimize', 'segment', 'export', 'vlad_index', 'backend_action', 'artifact_conversion'])

Return type:

None

phase_progress(phase, *, current, total=None, rate=None)

Parameters:

• phase (Literal['feature_extraction', 'matching', 'geometric_verification', 'incremental_init', 'incremental_register', 'incremental_ba', 'global_rotation_avg', 'global_positioning', 'global_ba', 'hierarchical_cluster', 'hierarchical_merge', 'panorama', 'spherical', 'bundle_adjust', 'triangulate', 'relocalize', 'pose_graph_optimize', 'segment', 'export', 'vlad_index', 'backend_action', 'artifact_conversion'])

• current (int)

• total (int | None)

• rate (float | None)

Return type:

None

phase_completed(phase)

Parameters:

phase (Literal['feature_extraction', 'matching', 'geometric_verification', 'incremental_init', 'incremental_register', 'incremental_ba', 'global_rotation_avg', 'global_positioning', 'global_ba', 'hierarchical_cluster', 'hierarchical_merge', 'panorama', 'spherical', 'bundle_adjust', 'triangulate', 'relocalize', 'pose_graph_optimize', 'segment', 'export', 'vlad_index', 'backend_action', 'artifact_conversion'])

Return type:

None

metric(key, value)

Parameters:

• key (str)

• value (float)

Return type:

None

snapshot_available(*, snapshot_seq, summary)

Parameters:

• snapshot_seq (int)

• summary (dict[str, Any])

Return type:

None

log_line(level, message)

Parameters:

• level (Literal['DEBUG', 'INFO', 'WARNING', 'ERROR'])

• message (str)

Return type:

None

warning(message)

Parameters:

message (str)

Return type:

None

error(*, error_class, message, detail=None)

Parameters:

• error_class (str)

• message (str)

• detail (dict[str, Any] | None)

Return type:

None

async app.workers.progress.append_job_event(job_id, payload)

Persist one validated ProgressEvent and return its JSON payload.

JobEvent.event_id is supplied explicitly so the same code works on SQLite, where BigInteger primary keys do not autoincrement as rowids. A short retry loop handles rare concurrent inserts.

Parameters:

• job_id (str)

• payload (dict[str, Any])

Return type:

dict[str, Any]

app.workers.progress.get_progress_reporter()

Return the reporter bound to the currently running worker task.

Return type:

ProgressReporter | None

app.workers.progress.reset_progress_reporter(token)

Restore the previous bound reporter.

Parameters:

token (Token)

Return type:

None

app.workers.progress.set_progress_reporter(reporter)

Bind reporter for the next asyncio.to_thread handler call.

Parameters:

reporter (ProgressReporter | None)

Return type:

Token

Feature extraction task.

Materializes the dataset’s images, ensures a database.db, calls pycolmap.extract_features, returns a result reference. Sealed snapshot emission is handled by app.storage.snapshots.SnapshotStore and is optional for this stage (DB-only mutation).

The materialization step is what lets the API stay clean: the HTTP caller only has to know the dataset_id; this task reads the materialization blob the orchestrator put together (kind + image_list + blob_shas / image_root / s3 coords) and produces a real local directory pycolmap can read.

app.workers.tasks.extract.run(task)

Parameters:

task (Task)

Return type:

dict[str, Any]

Matching task — pair selection (PairsSpec) + per-pair matcher (MatcherSpec).

The spec half of the task state carries {pairs: {...}, matcher: {...}} (the AIP-202 split shape). pairs.strategy selects which image pairs to consider; matcher.type selects the per-pair algorithm.

app.workers.tasks.match.run(task)

Parameters:

task (Task)

Return type:

dict[str, Any]

Geometric verification task.

app.workers.tasks.verify.run(task)

Parameters:

task (Task)

Return type:

dict[str, Any]

Mapping task — incremental | global | hierarchical | spherical.

Dispatches to backend.run_mapping(kind=...) and then runs the sfmapi-side post-processing: per-submodel snapshot emit, primary- submodel emit at the flat sparse/ root for the legacy snapshot read endpoint, and a sealed snapshot.

Resume support is internal to the backend — the colmap_mod backend writes MappingInput checkpoints into jobs/{job_id}/checkpoints/ and resumes from the latest one when the same task re-runs. Other backends may use their own checkpoint format; the interface here just threads job_dir through.

app.workers.tasks.map.run(task)

Parameters:

task (Task)

Return type:

dict[str, Any]

Standalone bundle adjustment — produces a new SubModel revision.

Supports two algorithms via spec.mode (see app.schemas.pipeline_spec.BundleAdjustmentSpec):

• standard (default): single solve over all parameters.

• two_stage: a poses-only pass, then a full unlock pass — requires the backend to expose two_stage_bundle_adjustment (capability ba.two_stage).

app.workers.tasks.ba.run(task)

Parameters:

task (Task)

Return type:

dict

Re-triangulation against an existing database.

app.workers.tasks.triangulate.run(task)

Parameters:

task (Task)

Return type:

dict

Relocalize images into an existing reconstruction.

app.workers.tasks.relocalize.run(task)

Parameters:

task (Task)

Return type:

dict

Pose graph optimization.

app.workers.tasks.pgo.run(task)

Parameters:

task (Task)

Return type:

dict

Export a reconstruction to PLY / NVM / COLMAP text/binary.

app.workers.tasks.export.run(task)

Parameters:

task (Task)

Return type:

dict