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. timeline must be a single day: the clearance, alarm, preempt, and conflict checks all assume it, and alarm and preempt history is accumulated across runs for you.

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)
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 the three updated_* values and pass them back unchanged next run. 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.