Runtime#

Composes the pipeline over the data store: the acquisition gateway mints scan identity at the source boundary, the processor drives the module graph per scan, the orchestrator manages run lifecycle in the store registry, and the post-processor runs after-the-fact modules into an existing store.

Multi-threaded pipeline orchestration.

Coordinates downloader and processor threads with queue-based inter-thread communication. Manages lifecycle, monitoring, and graceful shutdown. Owns the single run lifecycle in the store registry and refuses to run against an uninitialized root.

class adapt.runtime.orchestrator.PipelineOrchestrator(config, max_queue_size=None, close_repository_on_stop=True)#

Bases: object

Manages the multi-threaded radar processing pipeline.

This is the main entry point for running adapt. It coordinates two worker threads (downloader, processor) using queues for inter-thread communication. The orchestrator handles startup, monitoring, and graceful shutdown of the processing pipeline.

Pipeline Architecture:

  1. Downloader Thread: Discovers and downloads NEXRAD Level-II files from AWS in realtime or historical mode.

  2. Processor Thread: Processes downloaded files through the full scientific pipeline: - Load and regrid Level-II data to Cartesian grid - Segment cells using configurable threshold method - Compute motion projections (frame 2 and beyond) - Extract cell-level statistics - Persist segmentation to NetCDF and statistics to SQLite - Commit every artifact and table into the store

Modes:

  • Realtime: Polls for latest files within a rolling time window (e.g., “files from last 60 minutes”). Useful for operational monitoring.

  • Historical: Downloads all files within a fixed time range (start_time to end_time). Useful for batch reprocessing and research.

Queue Management:

Inter-thread queues have configurable size limits (default 100 items). Larger queues enable higher throughput but use more memory. Smaller queues provide backpressure (slow down downloader if processor falls behind).

Resumability:

Completed scans are skipped through the collection catalog (scan_is_complete), so a restarted run resumes without reprocessing.

Logging:

All output goes to both console and log file (logs/{radar}_pipeline.log). Log level controlled via config: “DEBUG”, “INFO”, “WARNING”, “ERROR”.

Example usage:

from adapt.runtime.orchestrator import PipelineOrchestrator

config = {
    "mode": "realtime",
    "downloader": {"radar": "KDIX", ...},
    "output_dirs": {...},
    ...
}

orch = PipelineOrchestrator(config)
orch.start(max_runtime=60)  # Run for 60 minutes then stop
Parameters:

  • config (InternalConfig)

  • max_queue_size (int | None)

  • close_repository_on_stop (bool)

close_store()#

Close the store’s collection handles if open.

Return type:

None

request_stop()#

Ask the monitoring loop to exit so the orchestrator’s own thread finalizes.

Called from the CLI on SIGINT/SIGTERM. Distinct from stop() so signalling the loop never trips stop()’s “already finalized” guard — finalize + summary then run on the joined (non-daemon) orchestrator thread, not a daemon the process kills on exit.

Return type:

None

start(max_runtime=None)#

Start the pipeline and run until completion or user interrupt.

This is a blocking call that starts the downloader and processor threads, then enters a monitoring loop. The loop logs status every 30 seconds and handles mode-specific exit conditions.

Realtime Mode: Runs until you press Ctrl+C or max_runtime is exceeded.

Historical Mode: Automatically exits when all files within the start_time/end_time range are queued, processed, and plotted.

All output (console + file) logged to logs/{radar}_pipeline.log at the level specified in config[“logging”][“level”].

Parameters:

max_runtime (int | None) – Maximum runtime in minutes (realtime mode only). If None, runs until KeyboardInterrupt (Ctrl+C). Ignored in historical mode (uses file completion instead).

Raises:

KeyboardInterrupt – User pressed Ctrl+C. Pipeline stops gracefully.

Notes

To stop the pipeline gracefully, press Ctrl+C. Do NOT force-kill. The stop() method is called automatically to close threads and save results.

Examples

Run for 60 minutes in realtime mode:

orch = PipelineOrchestrator(config)
orch.start(max_runtime=60)

Run historical mode until all files processed:

orch = PipelineOrchestrator(config)  # mode="historical" in config
orch.start()  # Runs until all files between start_time/end_time processed
stop()#

Stop the pipeline gracefully and finalize all results.

Called automatically when start() exits (either by user interrupt or mode-specific completion). Safe to call multiple times.

Operations:

  1. Signals all worker threads to stop

  2. Waits for each thread to finish (bounded grace periods)

  3. Finalizes the run record in the store registry

  4. Closes store connections and logs the runtime

Notes

This method should complete within ~20 seconds. If a thread does not respond to the stop signal within its timeout, a warning is logged but the shutdown continues (graceful degradation).

Files processed before the final save() will have their statistics in the SQLite database. The database is safe to query even while the pipeline is running (uses WAL mode).

Radar data processor thread.

Reads acquired-scan messages from the source queue, resolves the raw object through the store, and delegates all scientific processing to GraphExecutors built at startup:

  • _single_executor: ingest + detection (runs every file)

  • _multi_executor: projection + analysis + tracking (runs when 2-frame pair is ready)

Responsibilities of this class (orchestration only): - Queue management: pop acquired-scan messages, mark task done - Completed-scan skip via the collection catalog (scan_is_complete) - Frame pairing: accumulate segmented history, validate time gap - Context assembly: inject dataset_history before calling multi-executor - Persistence handoff: route module-declared outputs through the StoreOutputRouter - Stop/start lifecycle

class adapt.runtime.processor.RadarProcessor(input_queue, config, *, collection, registry, run_id, history, name='RadarProcessor', observability=None, root_trace_id='', reporter=None)#

Bases: Thread

Worker thread that processes NEXRAD files through two execution graphs.

Receives file paths from the downloader queue. For each file:

  1. Runs the single-frame graph (ingest → detection) via _single_executor.

  2. Accumulates segmented datasets in a rolling history.

  3. When a valid 2-frame pair is ready, runs the multi-frame graph (projection → analysis → tracking) via _multi_executor, passing the frame history in context.

Both executors enforce input/output contracts at every DAG edge via GraphExecutor. The processor itself performs no validation.

Example usage (called by PipelineOrchestrator):

processor = RadarProcessor(
    input_queue=downloader_queue,
    config=config,
    collection=collection,
    registry=registry,
    run_id=run_id,
    history=history,
)
processor.start()
...
processor.stop()
Parameters:
current_activity()#

Short ‘what am I doing now’ for the live status line; None when idle.

Return type:

str | None

process_file(filepath)#

Process a NEXRAD file with frame-pairing orchestration.

Phase 1 — pipeline_phase=1 modules (every file):

Per-file executor. Contract-validated by GraphExecutor.

Frame pairing:

Accumulates segmented datasets. Waits until 2 frames are ready.

Phase 2 — pipeline_phase=2 modules (when pair is ready):

Injects dataset_history into context. Per-pair executor. Contract-validated by GraphExecutor.

Phase 3 — pipeline_phase=3 modules (after persistence, when pair ran):

Post-persistence executor. Context: run_id, scan_time, plus the in-memory grid on demand. No-op if no phase-3 modules are registered.

Returns:

True if processed or deferred (waiting for pair), False on error.

Return type:

bool

run()#

Main processor loop (runs in thread).

Binds this thread’s correlation context once — contextvars do not cross threads, so the root trace id is handed in from the orchestrator and re-bound here so every log/span emitted on this thread shares one trace.

stop()#

Signal the processor to stop after the current file finishes.

stopped()#

True if stop() has been called or a ContractViolation forced stop.

Return type:

bool

Acquisition gateway: raw scans become cataloged store objects at the boundary.

The runtime builds one StoreAcquirer per run and injects it into the ingress source — acquisition modules never import persistence. Acquisition is idempotent per source_uri within a collection: bytes already in the store are reused (never rewritten), and the scan is registered for the acquiring run. The returned message is the processor queue contract: {artifact_id, scan_id, scan_time, queued_at}.

class adapt.runtime.acquire.StoreAcquirer(collection, run_id)#

Bases: object

Commit raw scans into one collection’s store, once, at the source boundary.

Parameters:
acquire_existing(source_uri, *, scan_time)#

Register this run against an already-stored source (skip-if-downloaded).

Return type:

dict

Parameters:

  • source_uri (str)

  • scan_time (datetime | None)

acquire_file(path, *, source_uri, scan_time)#

Commit a local raw file (or reuse the cataloged copy) and register its scan.

Return type:

dict

Parameters:

  • path (str | Path)

  • source_uri (str)

  • scan_time (datetime | None)

is_acquired(source_uri)#

True when this source’s bytes are already objects in the store.

Return type:

bool

Parameters:

source_uri (str)

Out-of-band store enrichment by composing the live pipeline’s infra.

PostProcessor enriches an already-processed run with extension tables (e.g. lightning stats), without touching the proven RadarProcessor. It is an after-the-fact module execution: outputs attach to the ORIGINAL run_id, and a re-run replaces the module’s own tables (never anyone else’s).

It reuses — by composition, not inheritance — the same components the pipeline uses: the module registry, GraphBuilder/GraphExecutor for dependency resolution, resolve_module_configs for config, and the StoreOutputRouter for persistence (whose frozen-schema writer enforces module table ownership). Post-process modules declare pipeline_phase = POSTPROCESS_PHASE and are selected exclusively here; the live pipeline never loads them.

class adapt.runtime.postprocessor.PostProcessor(collection, registry, run_id, config, extensions=None)#

Bases: object

Run post-process (phase POSTPROCESS_PHASE) modules over a stored run.

Parameters:
run(modules)#

Resolve, execute, and persist the requested post-process modules.

Parameters:

modules (list[str]) – Names of post-process modules to run (from --module or the postprocess.modules config section).

Return type:

dict