Near-Real-Time Processing

Poll a directory spool, process newly indexed patches, and stop after two empty polls. Choose the polling interval for the data cadence and storage latency.

In a large archive where new files land in one folder, root.update(paths="incoming") walks only that folder; run a plain update() now and then to catch removals and moves elsewhere.

import time
from pathlib import Path

import dascore as dc
from dascore.units import s

input_dir = Path("/path/to/incoming/data")
output_dir = Path("/path/to/results")
output_dir.mkdir(parents=True, exist_ok=True)

root = dc.spool(input_dir)
window = step = 10 * s
poll_seconds = 30
processed = set()
empty_polls = 0

while empty_polls < 2:
    root = root.update()
    spool = root.sort("time")
    contents = spool.get_contents()
    identities = list(zip(contents["source_path"], contents["source_patch_key"]))
    new_indices = [i for i, identity in enumerate(identities) if identity not in processed]

    if not new_indices:
        empty_polls += 1
        time.sleep(poll_seconds)
        continue

    empty_polls = 0
    for index in new_indices:
        identity = identities[index]
        patch = spool[index]
        result = patch.rolling(time=window, step=step, engine="numpy").mean()
        name = f"processed_{len(processed):06d}.h5"
        result.io.write(output_dir / name, "dasdae")
        processed.add(identity)

    time.sleep(poll_seconds)

Tracking source identities instead of row positions handles late files that sort before earlier arrivals. For a long-running service, replace the two-poll exit with monitoring and retries, and persist processed so restarts do not repeat work.