The Daily Processing Pipeline#

The daily processing loop is coordinated by daily.py, a command-line tool that runs four stages: file ingestion, statistics, modal analysis, and plotting. In production the loop is driven by daily2.sh, a cron wrapper that iterates over quantities and window durations.

This page documents the orchestration layer. For the pyOMA-internal per-window workflow (PreProcessSignals → VarSSIRef → StabilCluster), see pyOMA’s continuous monitoring page.

CLI reference#

python daily.py -d 120 -q accel --file_info --stats --modal --plot \
    --tmp_dir=/dev/shm/tower_tmp

Flag	Effect
`-d <minutes>`	Window duration in minutes. Supported values: 10, 30, 60, 120.
`-q <quantity>`	Measurement quantity. One of `accel`, `wind`, `temp`, `strain_rosettes`.
`--file_info`	Scan newly arrived files and update the file-info database.
`--stats`	Compute per-channel statistics for every new window.
`--modal`	Run modal analysis (VarSSIRef + StabilCluster) for every valid window. Only valid with `-q accel` or `-q strain_rosettes`.
`--plot`	Generate daily trend and waterfall plots and write them to `--tmp_dir`.
`--tmp_dir=<path>`	Directory for temporary files and plots. Defaults to `cwd`.
`--dtstart=YYYY-MM-DD HH:MM`	Process only windows at or after this UTC-naive timestamp. When `--file_info` is given, `dtstart` is derived automatically from the first newly arrived file.
`--loglevel=INFO`	Python logging level (`DEBUG`, `INFO`, `WARNING`, `ERROR`).

Stage 1 — File ingestion and quality assessment#

get_file_info(origin, create_new=True) scans new files returned by the active site’s file_list_fn and records per-file metadata plus per-channel statistics:

# Condensed from monitoring.create_file_info
for file in filelist:
    file_time, file_size, headers, units, start_time, sample_rate, measurement \
        = read_file(file)

    if close_to_utc_transition(file_time):
        continue                        # skip files near DST transition

    duration = datetime.timedelta(seconds=measurement.shape[0] / sample_rate)
    sync_time = get_synchronized_time(start_time, file_time, duration)

    dst['start_time'] = (['time'], [TC.to_posix(start_time)])
    dst['file_time']  = (['time'], [TC.to_posix(file_time)])
    dst.coords['time'] = [TC.to_storage_coord(sync_time)]

    ds = ds.combine_first(dst)

Three things happen here simultaneously:

The raw timestamps (start_time, file_time) are stored as POSIX float seconds — timezone-agnostic and arithmetic-friendly.
The synchronised wall-clock time is converted to UTC-naive datetime64 and used as the time coordinate (the database index).
Quality flags — kurtosis, min == max, out-of-range values — are set per channel. Channels with error == True are excluded from downstream OMA.

compute_gap_lengths is appended to every file-info dataset on load. It computes the gap in samples between consecutive files so that get_slice can skip windows where files are non-consecutive.

Stage 2 — Signal preprocessing (statistics)#

create_stats iterates over fixed-duration windows aligned to the local Berlin time grid and extracts one signal slice per window:

# Condensed from monitoring.create_stats
_aware_iter, time_iter_naive = TC.make_index(dtstart, until, minutes)

for time_ in time_iterator:
    start_time = TC.to_local(time_)          # stored UTC-naive → Berlin-aware

    data_slice = get_slice_corrected(start_time, duration, quantity, file_info)
    if data_slice is None:
        continue                              # no coverage or gap in files

    this_ds.coords['time'] = [start_time.to_datetime64()]
    process_ds = process_ds.combine_first(this_ds)

get_slice_corrected manages three sub-steps:

Raw slice extraction (get_slice) — selects files covering the window, truncates samples at both ends to align with the requested boundaries, fills small gaps (≤ 32 samples for strain channels) with np.nan, and stacks the result into a single (T, N) array.
Site transform — the active site’s transforms[quantity] callback is applied. For accel no transform is registered; for wind the callback converts the raw Wg/Wr channels to Cartesian components and applies a circular-mean correction; for strain_rosettes it converts FBG wavelengths to strain with temperature compensation.
Slice caching — the corrected slice is saved as a .npz file under <slice_root>/<dur>-minutes/slices_<quantity>/YYYY/MM/. On subsequent runs get_slice_corrected loads the cached .npz directly, bypassing the reader and transform entirely.

This two-level caching (process-local Dataset and on-disk .npz) means that interrupted runs resume exactly where they left off without reprocessing already-completed windows.

Stage 4 — Concurrent write safety (MultiLock)#

Both create_stats and create_modal_results support parallel execution across num_workers processes. Each worker writes to a private <name>.<pid>.nc file during the long computation phase, then merges into the shared master file at the end of a chunk:

# Condensed from monitoring.save_ds
with MultiLock(savepath):
    if reload_current:
        current_ds = xr.open_dataset(savepath)
        current_ds.load(); current_ds.close()

    dupes, _, _ = np.intersect1d(new_ds.time, current_ds.time, ...)
    current_ds = current_ds.drop_sel(time=dupes)
    current_ds = current_ds.combine_first(new_ds)

    current_ds.to_netcdf(tempfile)
    os.rename(tempfile, savepath)   # atomic on POSIX filesystems

MultiLock layers a per-process .pid.lock file on top of simpleflock to handle a race condition where simpleflock can briefly grant the lock to two processes simultaneously.

The cron wrapper (daily2.sh)#

In production, daily2.sh is called once per day by cron. It iterates over quantities and durations, passing the --dtstart flag derived from the most recently arrived file to each subsequent duration run:

TMPDIR=/dev/shm/tower_tmp
mkdir ${TMPDIR}

# 120-minute file ingestion sets DTSTART for shorter-duration runs
python daily.py -d 120 -q accel --file_info --stats --modal \
    --tmp_dir=${TMPDIR} > ${TMPDIR}/tower_out.txt

if test -f "${TMPDIR}/dtstart.tmp"; then
    DTSTART=$(cat ${TMPDIR}/dtstart.tmp)
    python daily.py -d  60 -q accel --stats --modal \
        --dtstart=${DTSTART} >> ${TMPDIR}/tower_out.txt
    python daily.py -d  30 -q accel --stats --modal \
        --dtstart=${DTSTART} >> ${TMPDIR}/tower_out.txt
    python daily.py -d  10 -q accel --stats --modal --plot \
        --dtstart=${DTSTART} >> ${TMPDIR}/tower_out.txt
fi
# results and plots are emailed to the monitoring team
cat ${TMPDIR}/tower_out.txt | mail -s "..." \
    -a ${TMPDIR}/modal_accel_10.png your@email.de

The --file_info flag at 120 minutes writes the earliest new-file timestamp to dtstart.tmp. All subsequent shorter-duration runs use that timestamp so they process exactly the same new data range, rather than re-scanning the full history.

Quantities without modal analysis (wind, temp) run through the same loop structure but omit the --modal flag.