Acquisition

Downloads raw NEXRAD Level-II files from AWS S3. Runs as a background thread, independent of all processing. Delivers local file paths via a thread-safe queue.

AWS S3 NEXRAD Level-II file discovery and download.

Monitors AWS S3 bucket for new NEXRAD radar files and acquires them into the store in realtime or historical batches. Deduplicates via the catalog (by source URI) to avoid re-downloading.

class adapt.modules.acquisition.module.AwsNexradDownloader(config, result_queue=None, acquire=None, conn=None, clock=None, sleeper=None)

Bases: Thread

Downloads NEXRAD Level-II files from AWS S3 in realtime or historical mode.

Realtime Mode: Continuously monitors S3 for new files within a rolling time window (e.g., “last 60 minutes”). Useful for operational nowcasting. The latest N files are retained; older files are not re-downloaded.

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

AWS S3 Bucket: Files stored at s3://unidata-nexrad-level2/{YYYY}/{MM}/{DD}/{radar}/ Example: s3://unidata-nexrad-level2/2025/03/05/KDIX/KDIX20250305_000310_V06

Deduplication: Maintains set of known files to avoid re-downloading. Safe to restart mid-execution.

Queue Communication: Puts one acquired-scan message per new file on result_queue — {artifact_id, scan_id, scan_time, queued_at} — built by the injected acquisition gateway, which commits the raw bytes into the store at the source boundary. Downstream processor can begin work immediately (streaming architecture).

File Size Filtering: Ignores files < 1 KB (corrupted downloads or metadata-only files).

Thread Safety: Safe to call status methods while run() is executing. Uses locks for shared state (_known_files, historical tracking).

Example usage (typically called by orchestrator):

downloader = AwsNexradDownloader(
    config={
        "radar": "KDIX",
        "output_dir": "/data/nexrad",
        "latest_files": 5,
        "latest_minutes": 60,
        "poll_interval_sec": 30
    },
    result_queue=processor_queue
)
downloader.start()
...
downloader.stop()
downloader.join(timeout=10)

get_historical_progress()

Get progress of historical download as (processed, expected) counts.

Returns:

First element: number of scans successfully processed/queued. Second element: total number of scans expected in time range.

Return type:

tuple

Notes

Only meaningful in historical mode. In realtime mode, expected is always 0. Progress is (processed, expected) where processed <= expected. This can be used to display download progress to the user.

Examples

>>> processed, expected = downloader.get_historical_progress()
>>> print(f"Downloaded {processed}/{expected} scans")

>>> while not downloader.is_historical_complete():
...     processed, expected = downloader.get_historical_progress()
...     print(f"Progress: {processed}/{expected}")
...     time.sleep(10)

is_historical_complete()

Check if historical download has finished.

In historical mode, this indicates whether all scans in the time range have been downloaded and queued. In realtime mode, this always returns False.

Returns:

True if all available scans in the start_time to end_time range have been processed.

Return type:

bool

Notes

“Complete” means scans have been QUEUED, not necessarily processed by the pipeline. The main loop will exit automatically when this is True.

Examples

>>> while not downloader.is_historical_complete():
...     downloader.run()  # Wait for completion

queue_backpressure_stats()

How often, and for how long, the downloader waited on the processor.

Return type:

dict

run()

Main thread loop - automatically invoked when thread starts.

This is the primary execution method called by threading.Thread.start(). Do NOT call directly; use thread.start() instead.

Behavior:

1. Logs startup with mode (realtime or historical)

2. Repeatedly calls download_task() at intervals

3. Handles exceptions in download task (logs and continues)

4. In historical mode: exits automatically when complete

5. In realtime mode: runs until stop() is called

6. Performs interruptible sleep between iterations (can exit quickly)

Notes

• Running in a background daemon thread (set at __init__)

• All exceptions are logged; failures don’t crash the thread

• Sleep is interruptible: responds to stop() and historical completion within 2 seconds (default is 2-second sleep chunks)

• Thread-safe: can call stop() from another thread

Examples

>>> downloader = AwsNexradDownloader(config, queue)
>>> downloader.start()  # Calls run() in background thread
>>> downloader.join()   # Wait for thread to finish

stop()

Signal the downloader thread to stop gracefully.

Calling this method sets the internal stop event, which causes the run() main loop to exit. The thread will finish its current download task before stopping (not immediate).

Safe to call from any thread. Can be called multiple times.

Examples

>>> downloader = AwsNexradDownloader(config, queue)
>>> downloader.start()
>>> time.sleep(60)
>>> downloader.stop()  # Signal thread to exit
>>> downloader.join()  # Wait for thread termination

stopped()

Check if a stop request has been issued.

Returns:

True if stop() has been called, False otherwise.

Return type:

bool

Notes

This is non-blocking and reflects whether the stop event has been set. The thread may still be running even if this returns True (it finishes its current task before exiting).

Examples

>>> if downloader.stopped():
...     print("Stop was requested")