atspm-report

Turns the output tables of the atspm package into per-region PDF reports of new traffic signal issues. Repeat alerts are suppressed against history you store between runs, so each report shows what changed, not what is already known.

pip install atspm-report

What it does

ReportGenerator(config).generate(...) takes DataFrames (pandas or Ibis), runs the detectors, suppresses repeats, and returns PDFs as BytesIO plus the alert tables behind them.

It does not fetch data, read or write files, send email, or schedule anything. You supply the inputs, save the PDFs, and persist the returned history for the next run.

Alert types

Pass only the inputs you have; every argument except signals is optional and a missing input just leaves its section out.

Inputs

All inputs except signals come straight from atspm's output tables. DeviceId may be int or string in any table; it is cast to string internally.

signals (required, you provide this) one row per signal:

atspm tables and how much history to pass each run:

The CUSUM detectors compute each entity's baseline from everything you pass, so keep the window consistent.

The latest day in timeline is the report day, and the clearance and conflict checks only ever look at that day. The controller alarm and preempt checks need six weeks of daily counts, which you can supply either way:

• One day of timeline. Persist the returned updated_alarm_history and updated_preempt_history and pass them back as alarm_history and preempt_history; the counts build up over runs. Days present in the timeline replace the matching history rows, so re-running a day does not double count it.
• Up to 42 days of timeline. The counts are rebuilt from it every run and there is nothing to persist. Older days can be trimmed to the ALARM_EVENT_CLASSES and Preempt rows to keep it small. Add device_days (columns DeviceId, Date: the days each signal reported data) so the preempt check counts a reported day with no calls as zero; without it, reported days are inferred from each day's timeline span, which a trimmed day no longer has.

detector_health needs the prediction and anomaly columns, which come from atspm's detector_health aggregation, not the plain actuations one.

Usage

from pathlib import Path
import pandas as pd
from atspm_report import ReportGenerator

STATE = Path('state')
STATE.mkdir(exist_ok=True)

def load(name):
    path = STATE / f'{name}.parquet'
    return pd.read_parquet(path) if path.exists() else pd.DataFrame()

config = {'verbosity': 1}   # every key has a default, see below

result = ReportGenerator(config).generate(
    signals=signals,                 # your own table
    terminations=terminations,       # atspm outputs, pandas or Ibis
    detector_health=detector_health,
    has_data=has_data,
    pedestrian=full_ped,
    phase_wait=phase_wait,
    coordination_agg=coordination_agg,
    timeline=timeline,
    # history from the previous run (empty on the first run)
    past_alerts={k: load(f'past_{k}') for k in ReportGenerator.ALERT_TYPES},
    alarm_history=load('alarm_history'),
    preempt_history=load('preempt_history'),
)

# 1. Persist history for the next run. Store it verbatim.
for alert_type, df in result['updated_past_alerts'].items():
    df.to_parquet(STATE / f'past_{alert_type}.parquet', index=False)
# Only needed when timeline is a single day (see Inputs).
result['updated_alarm_history'].to_parquet(STATE / 'alarm_history.parquet', index=False)
result['updated_preempt_history'].to_parquet(STATE / 'preempt_history.parquet', index=False)

# 2. Do something with the PDFs. Empty dict means nothing new today.
for region, pdf in result['reports'].items():
    Path(f'report_{region}.pdf').write_bytes(pdf.getvalue())

Ibis tables work anywhere a DataFrame does, so a DuckDB, Polars, or Spark backend can do the heavy lifting:

import ibis
con = ibis.duckdb.connect('atspm.duckdb')
result = ReportGenerator(config).generate(
    signals=con.table('signals'),
    terminations=con.table('terminations').filter(ibis._.TimeStamp >= start),
    # ...
)

Returns

State between runs

generate() never writes anything. Save updated_past_alerts and pass it back unchanged next run; do the same with updated_alarm_history and updated_preempt_history unless you pass a multi-day timeline, in which case those two are rebuilt every run and need not be stored. Every retention and suppression rule is already applied inside, so do not filter, dedupe, or reshape them. Dropping them does not error; it silently resets suppression, alarm totals, or the preempt baseline.

Configuration

Pass a plain dict; every key has a default.

The full list with clearance noise filters and chart options is in the ReportGenerator docstring.

License

MIT, see LICENSE. Contributions welcome; open an issue for problems or help.