krz/omaha-metro-blotter
Archive of police activity and ALPR surveillance across the Omaha metro.
clone: git clone https://gitbay.org/krz/omaha-metro-blotter.git
3f14d770a2930fd20eddfb4ba6e71133b0df78e6
verified · cmc
author: Christian Cleberg <hello@cleberg.net> · 2026-08-23T06:51:48Z
.github/workflows/daily-pull.yml | 151 +++++++++++++++ .gitignore | 3 + README.nfo | 94 +++++++-- analysis.py | 146 ++++++++++++++ app.py | 317 ++++++++++++++++++++---------- assets/styles.css | 107 ++++++++-- assets/theme.js | 19 ++ ingest.py | 407 +++++++++++++++++++++++++++++++-------- requirements-ingest.txt | 1 + requirements.txt | 5 + schema.sql | 38 ++++ 11 files changed, 1087 insertions(+), 201 deletions(-) new file mode 100644 @@ -0,0 +1,151 @@ +# Sarpy and Council Bluffs serve a rolling 12-month window; records that age out +# of those feeds exist nowhere else. This job is the only thing keeping them, so +# it refuses to publish an archive smaller than the one it started with. +name: daily pull + +on: + schedule: + # 06:00 America/Chicago in summer, 05:00 in winter. Offset from the hour + # because GitHub drops on-the-hour scheduled runs under load. + - cron: "17 11 * * *" + workflow_dispatch: + inputs: + bootstrap: + description: "Start a new archive instead of restoring the published one" + type: boolean + default: false + full: + description: "Pull each feed in full rather than the last 30 days" + type: boolean + default: false + +permissions: + contents: write + +concurrency: + group: archive + cancel-in-progress: false + +env: + TAG: archive + DB: raw_data/metro.db + GH_TOKEN: ${{ github.token }} + +jobs: + pull: + runs-on: ubuntu-latest + timeout-minutes: 30 + + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + cache: pip + cache-dependency-path: requirements-ingest.txt + + - run: pip install -r requirements-ingest.txt + + - name: Restore archive + run: | + if gh release download "$TAG" --pattern metro.db.gz --dir .; then + gunzip -c metro.db.gz > "$DB" + rm metro.db.gz + echo "restored $(du -h "$DB" | cut -f1)" + elif [ "${{ inputs.bootstrap }}" = "true" ]; then + echo "no published archive; starting a new one" + else + echo "::error::No metro.db.gz on release '$TAG'. Anything that has" \ + "already aged out of the Sarpy and Council Bluffs feeds cannot" \ + "be recovered. Re-run with bootstrap only if that is intended." + exit 1 + fi + + - name: Count rows before + run: | + : > before.txt + if [ -f "$DB" ]; then + sqlite3 -noheader -separator ' ' "$DB" \ + "SELECT source, COUNT(*) FROM incidents GROUP BY source ORDER BY source" \ + > before.txt + fi + cat before.txt + + - name: Pull feeds + run: | + if [ "${{ inputs.full }}" = "true" ] || [ "${{ inputs.bootstrap }}" = "true" ]; then + python ingest.py --full + else + python ingest.py + fi + + - name: Check nothing was lost + run: | + sqlite3 -noheader -separator ' ' "$DB" \ + "SELECT source, COUNT(*) FROM incidents GROUP BY source ORDER BY source" \ + > after.txt + cat after.txt + test -s after.txt || { echo "::error::archive is empty"; exit 1; } + # Keyed on FILENAME, not NR == FNR: before.txt is empty on a bootstrap + # run, and awk never resets FNR for a zero-length file. + awk -v first=before.txt ' + FILENAME == first { was[$1] = $2; next } + { now[$1] = $2 } + END { + for (s in was) + if (now[s] + 0 < was[s] + 0) { + printf "::error::%s lost rows: %d -> %d\n", s, was[s], now[s] + bad = 1 + } + exit bad + }' before.txt after.txt + + - name: Publish archive + run: | + sqlite3 "$DB" "VACUUM;" + gzip -c "$DB" > metro.db.gz + gh release view "$TAG" >/dev/null 2>&1 \ + || gh release create "$TAG" --title "Incident archive" --notes "building" + gh release upload "$TAG" metro.db.gz --clobber + { + echo "SQLite archive of Omaha metro police incident feeds, rebuilt daily." + echo "Sarpy County and Council Bluffs publish a rolling 12-month window," + echo "so this holds records their own feeds no longer serve." + echo + echo "Updated $(date -u '+%Y-%m-%d %H:%M UTC'). Schema: schema.sql." + echo + echo '```' + sqlite3 -header -column "$DB" \ + "SELECT agency, COUNT(*) AS rows, SUM(is_stop) AS stops, + MIN(occurred_at) AS earliest, MAX(occurred_at) AS latest + FROM incidents GROUP BY agency ORDER BY rows DESC" + echo '```' + } > notes.md + gh release edit "$TAG" --notes-file notes.md + + # Runs after the upload on purpose: a feed that stopped updating should + # raise the alarm without also blocking the archive from being published. + - name: Check the feeds are still moving + run: | + sqlite3 -noheader "$DB" \ + "SELECT source || ' ' || MAX(occurred_at) FROM incidents + WHERE source <> 'opd_csv' GROUP BY source + HAVING MAX(occurred_at) < datetime('now', '-7 days')" > stale.txt + if [ -s stale.txt ]; then + while read -r line; do echo "::error::feed is stale: $line"; done < stale.txt + exit 1 + fi + echo "all feeds current" + + - name: Summary + if: always() + run: | + { + echo "| source | before | after |" + echo "|---|---|---|" + awk -v first=before.txt ' + FILENAME == first { was[$1] = $2; next } + { printf "| %s | %s | %s |\n", $1, ($1 in was ? was[$1] : 0), $2 }' \ + before.txt after.txt + } >> "$GITHUB_STEP_SUMMARY" @@ -1 +1,4 @@ .DS_Store +*.db +.venv/ +__pycache__/ @@ -3,25 +3,91 @@ └──────────────────────────────────────────────────────────────┘ WHAT - crime data from the omaha police department, for analysis and - visualization. + police activity across the omaha metro, pulled from the agencies' + own feeds and joined against alpr camera locations from + openstreetmap. -API - explore the database over http with sqlite2rest: +COVERAGE + omaha pd dcgis arcgis view, 2022-01-01 onward, + nibrs offence records, updated daily. + no stop or disposition data. + bellevue pd sarpy county publiccrimemap, cad calls + papillion pd for service with stop type, disposition + la vista pd and category. rolling 12-month window: + sarpy county so records age out of the feed, so the local + archive is the only long-term copy. gretna + and springfield appear as fire only; their + police departments do not report to it. + council bluffs pd cbpd public cfs feed, refreshed every ten + minutes, rolling 12-month window. stop + type, disposition, priority, response time. + street addresses withheld; points exact. + ralston pd no machine-readable feed. absent. - pip3 install sqlite2rest - sqlite2rest serve ./raw_data/ingress.db + alpr cameras openstreetmap via overpass, the same data + deflock renders. 168 nodes in the metro + bbox. odbl, attribution required. -TODO - - import script (done) - - drop duplicate header rows (done) - - explore and analyze the data - - plotly dash visualizations - - api to the database +SETUP + uv venv .venv + uv pip install --python .venv/bin/python -r requirements.txt + +USE + .venv/bin/python ingest.py --full # first run, backfill + .venv/bin/python ingest.py # daily, last 30 days + .venv/bin/python ingest.py cbpd sarpy # one source at a time + .venv/bin/python ingest.py opd_csv # 2015-2023 csv archive + .venv/bin/python app.py + +ARCHIVE + .github/workflows/daily-pull.yml runs the pull at 11:17 utc and + keeps the database as metro.db.gz on the "archive" release, so the + archive does not depend on any one machine. each run restores that + asset, pulls, refuses to publish if any source came back with fewer + rows than it started with, then uploads and fails loudly if a feed + has not moved in seven days. + + first run: trigger it manually with bootstrap enabled, which pulls + every feed in full and creates the release. after that the restore + step is mandatory -- a bootstrap over a live archive throws away + whatever has already aged out of the sarpy and council bluffs + feeds. + + github disables scheduled workflows after 60 days without repo + activity, and emails first. that is the most likely way this stops + quietly. + + to run the pull locally instead: + + 0 6 * * * cd /path/to/omaha-incidents && .venv/bin/python ingest.py + +NOTES + all three arcgis services return utc epochs; their where-clause + literals do not agree (opd and council bluffs utc, sarpy central). + ingest.py stores occurred_at in local time. + + each feed has its own taxonomy and none of them are comparable, so + ingest.py derives one cross-agency flag, is_stop, per source. stop + outcomes compare citation and arrest rate by substring, which is + all the two disposition vocabularies support: an agency that + records warnings less thoroughly shows a higher citation rate for + that reason alone. + + colour scheme follows prefers-color-scheme. plotly writes colours + into the figure, so assets/theme.js reports the media query into a + store and app.py builds each figure from it. restyling after the + fact does not work: swapping a maplibre basemap at runtime leaves + it rebuilding with no data layers. + + the camera-proximity panel compares stops against a non-stop + baseline. cameras and stops both concentrate on arterials, so a + gap between the curves is a starting point, not a finding. + + raw_data/ingress.db is the old 2015-2023 sqlite build. nothing + reads it any more. SCREENSHOTS - screenshots/dashboard_01.png - screenshots/dashboard_02.png + screenshots/*.png are from the previous 2015-2023 dashboard. ┌──────────────────────────────────────────────────────────────┐ │ krz.sh │ new file mode 100644 @@ -0,0 +1,146 @@ +"""Queries behind the dashboard: agency activity and distance to the nearest ALPR.""" + +import sqlite3 +from pathlib import Path + +import numpy as np +import pandas as pd + +DB = Path(__file__).parent / "raw_data" / "metro.db" + +# Sarpy and Council Bluffs publish officer-initiated stops; ingest.py flags them +# as is_stop. OPD publishes NIBRS offences only and contributes no stops. +POLICE_AGENCIES = ("Omaha PD", "Council Bluffs PD", "Bellevue PD", "Papillion PD", + "La Vista PD", "Sarpy County SO") + +EARTH_M = 6371000.0 + + +def connect(): + return sqlite3.connect(DB) + + +def load_incidents(conn, agencies=None, start=None, end=None, categories=None): + where, params = ["lat IS NOT NULL", "lon IS NOT NULL"], [] + if agencies: + where.append(f"agency IN ({','.join('?' * len(agencies))})") + params += list(agencies) + if categories: + where.append(f"category IN ({','.join('?' * len(categories))})") + params += list(categories) + if start: + where.append("occurred_at >= ?") + params.append(f"{start}T00:00:00") + if end: + where.append("occurred_at <= ?") + params.append(f"{end}T23:59:59") + df = pd.read_sql_query( + f"""SELECT source, agency, case_id, occurred_at, category, call_type, + disposition, offense_desc, is_stop, address, lat, lon + FROM incidents WHERE {' AND '.join(where)}""", + conn, params=params) + df["occurred_at"] = pd.to_datetime(df["occurred_at"]) + return df + + +def load_cameras(conn): + return pd.read_sql_query( + "SELECT osm_id, lat, lon, manufacturer, operator, direction, first_seen" + " FROM alpr_cameras", conn) + + +def nearest_camera_m(df, cameras): + """Great-circle distance in metres from each incident to the closest camera. + + 164 cameras against ~250k incidents, so the full distance matrix is computed + in chunks rather than all at once.""" + if df.empty or cameras.empty: + return np.full(len(df), np.nan) + + lat1 = np.radians(df["lat"].to_numpy(dtype=float)) + lon1 = np.radians(df["lon"].to_numpy(dtype=float)) + lat2 = np.radians(cameras["lat"].to_numpy(dtype=float)) + lon2 = np.radians(cameras["lon"].to_numpy(dtype=float)) + + out = np.empty(len(df)) + cos2 = np.cos(lat2) + for i in range(0, len(df), 20000): + s = slice(i, i + 20000) + dlat = lat2[None, :] - lat1[s, None] + dlon = lon2[None, :] - lon1[s, None] + a = (np.sin(dlat / 2) ** 2 + + np.cos(lat1[s, None]) * cos2[None, :] * np.sin(dlon / 2) ** 2) + out[s] = (2 * EARTH_M * np.arcsin(np.sqrt(a))).min(axis=1) + return out + + +def daily_counts(df): + if df.empty: + return pd.DataFrame(columns=["date", "agency", "incidents"]) + g = (df.assign(date=df["occurred_at"].dt.floor("D")) + .groupby(["date", "agency"], as_index=False) + .size().rename(columns={"size": "incidents"})) + return g + + +def stop_outcomes(df): + """Citation and arrest rate on vehicle stops, per agency. + + The two CAD systems use different disposition vocabularies -- Sarpy writes + WRITTEN WARNING / CITATION, Council Bluffs writes "3 - Citation" and folds + warnings into "7 - Handled by Officer" -- and both allow several outcomes per + stop. Substring matching on citation and arrest is the only comparison the + two vocabularies actually support, and a department that records warnings + less thoroughly will show a higher citation rate for that reason alone.""" + stops = df[(df["is_stop"] == 1) & df["disposition"].notna()] + if stops.empty: + return pd.DataFrame(columns=["agency", "outcome", "rate", "stops"]) + d = stops["disposition"].str.upper() + stops = stops.assign(cited=d.str.contains("CITATION"), + arrested=d.str.contains("ARREST")) + g = stops.groupby("agency").agg(stops=("cited", "size"), + Cited=("cited", "mean"), + Arrested=("arrested", "mean")).reset_index() + return (g.melt(id_vars=["agency", "stops"], value_vars=["Cited", "Arrested"], + var_name="outcome", value_name="rate") + .sort_values("rate", ascending=False)) + + +def camera_proximity(df, cameras, bin_m=200, max_m=2000): + """Share of stops vs other incidents falling in each distance band. + + Both series are normalised, so a gap between them means stops cluster + differently around cameras than the rest of the call volume does. It is not + evidence of causation: cameras and stops both concentrate on arterials.""" + if df.empty or cameras.empty: + return pd.DataFrame(columns=["distance_m", "kind", "share"]) + d = df.assign(dist=nearest_camera_m(df, cameras)) + d = d[d["dist"] <= max_m] + if d.empty: + return pd.DataFrame(columns=["distance_m", "kind", "share"]) + d["kind"] = np.where(d["is_stop"] == 1, "Vehicle stops", "All other incidents") + d["distance_m"] = (d["dist"] // bin_m * bin_m).astype(int) + g = (d.groupby(["kind", "distance_m"], as_index=False) + .size().rename(columns={"size": "n"})) + g["share"] = g["n"] / g.groupby("kind")["n"].transform("sum") + return g + + +def agency_options(conn): + rows = conn.execute( + "SELECT agency, COUNT(*) FROM incidents GROUP BY agency ORDER BY 2 DESC" + ).fetchall() + return [a for a, _ in rows] + + +def category_options(conn): + rows = conn.execute( + "SELECT category, COUNT(*) FROM incidents WHERE category IS NOT NULL" + " GROUP BY category ORDER BY 2 DESC").fetchall() + return [c for c, _ in rows] + + +def date_bounds(conn): + lo, hi = conn.execute( + "SELECT MIN(occurred_at), MAX(occurred_at) FROM incidents").fetchone() + return lo[:10], hi[:10] @@ -1,116 +1,231 @@ -from dash import Dash, html, dcc, callback, Output, Input, dash_table +"""Omaha metro police activity dashboard. + +Covers Omaha PD, Council Bluffs PD and the Sarpy County agencies (Bellevue, +Papillion, La Vista, Sheriff). Ralston PD publishes no machine-readable feed and +is absent. OPD publishes NIBRS offence records with no stop or disposition data, +so it contributes nothing to the enforcement panels. +""" + +from dash import Dash, Input, Output, callback, dash_table, dcc, html import plotly.express as px -import pandas as pd -import numpy as np -import sqlite3 -from datetime import datetime, date +import plotly.graph_objects as go + +import analysis + +conn = analysis.connect() +INCIDENTS = analysis.load_incidents(conn) +CAMERAS = analysis.load_cameras(conn) +AGENCIES = analysis.agency_options(conn) +CATEGORIES = analysis.category_options(conn) +DATE_LO, DATE_HI = analysis.date_bounds(conn) +conn.close() -# Connect to database and query all incidents -connection = sqlite3.connect("./raw_data/ingress.db") -cursor = connection.cursor() -query = "SELECT * FROM incidents;" -df = pd.read_sql_query(query, connection).sort_values(by="description") +DEFAULT_AGENCIES = [a for a in analysis.POLICE_AGENCIES if a in AGENCIES] +CENTER = {"lat": 41.21, "lon": -95.97} +MAP_SAMPLE = 15000 +# Plotly writes colours into the figure, so the scheme has to be known before a +# figure is built. assets/theme.js reports the media query into the theme store +# and every figure callback reads it; nothing is restyled after the fact. +PALETTES = { + "light": {"fg": "#111", "grid": "#e6e6e6", "legend": "rgba(255,255,255,.85)", + "basemap": "open-street-map"}, + "dark": {"fg": "#e8e8ea", "grid": "#333840", "legend": "rgba(27,30,36,.85)", + "basemap": "carto-darkmatter"}, +} + + +def themed(fig, theme): + p = PALETTES.get(theme, PALETTES["light"]) + fig.update_layout(paper_bgcolor="rgba(0,0,0,0)", plot_bgcolor="rgba(0,0,0,0)", + font_color=p["fg"], legend_bgcolor=p["legend"], + legend_font_color=p["fg"]) + # update_xaxes would bolt empty axis objects onto the map figure, which has + # no cartesian axes at all. + if not any(tr.type == "scattermap" for tr in fig.data): + fig.update_xaxes(gridcolor=p["grid"], zerolinecolor=p["grid"]) + fig.update_yaxes(gridcolor=p["grid"], zerolinecolor=p["grid"]) + return fig -# Replace empty cells with NaN -df = df.replace(r'^\s*$', np.nan, regex=True) -# Convert date column to datetime -# TODO: Create a combined datetime column in the db to load/convert here -df["date"] = pd.to_datetime(df["date"]) +def empty(theme): + return themed(go.Figure().update_layout( + annotations=[{"text": "No incidents match these filters", + "showarrow": False, "font": {"size": 14}}], + xaxis={"visible": False}, yaxis={"visible": False}), theme) -# Configure HTML layout app = Dash(__name__) -app.layout = html.Div(children = [ - html.H1(children="Omaha Crime Mapping & Analysis", style={"textAlign":"center"}), - html.Div([ - html.Div(className="flex", - children = [ - html.P("Crime:"), - dcc.Dropdown(df.sort_values("description").description.unique(), "INJURY", id="dropdown"), - ] - ), - html.Div(className="flex", - children = [ - html.P("Date Range:"), - dcc.DatePickerRange( - id='date-picker-range', - min_date_allowed=date(2015, 1, 1), - max_date_allowed=date(2023, 12, 31), - start_date=date(2015, 1, 1), - end_date=date(2023,12,31) - ), - ] - ), - dcc.Graph(id="map-graph"), - dcc.Graph(id="line-graph"), - dash_table.DataTable(data=df.to_dict("records"), page_size=5, id="table") - ]) +app.title = "Omaha Metro Police Activity" + +app.layout = html.Div([ + dcc.Store(id="theme", data="light"), + html.H1("Omaha Metro Police Activity"), + html.P([ + f"{len(INCIDENTS):,} geolocated incidents, {DATE_LO} to {DATE_HI}. ", + f"{len(CAMERAS)} ALPR cameras from OpenStreetMap. ", + "Ralston PD publishes no feed and is not represented. ", + "Omaha PD reports no stops or dispositions, so it is absent from the ", + "enforcement panels below.", + ], className="subtitle"), + + html.Div(className="controls", children=[ + html.Div([html.Label("Agency"), + dcc.Dropdown(AGENCIES, DEFAULT_AGENCIES, id="agencies", + multi=True)]), + html.Div([html.Label("Category"), + dcc.Dropdown(CATEGORIES, [], id="categories", multi=True, + placeholder="all categories")]), + html.Div([html.Label("Dates"), + dcc.DatePickerRange(id="dates", min_date_allowed=DATE_LO, + max_date_allowed=DATE_HI, + start_date=DATE_LO, end_date=DATE_HI)]), + html.Div([html.Label("Show cameras"), + dcc.Checklist([{"label": " ALPR layer", "value": "on"}], + ["on"], id="show-cameras")]), + ]), + + html.Div(id="kpis", className="kpis"), + + dcc.Graph(id="map"), + dcc.Graph(id="timeline"), + html.Div(className="row", children=[ + dcc.Graph(id="dispositions", className="half"), + dcc.Graph(id="proximity", className="half"), + ]), + html.H2("Incidents"), + dash_table.DataTable(id="table", page_size=10, sort_action="native", + style_table={"overflowX": "auto"}), ]) -# Create map -@callback( - Output("map-graph", "figure"), - Input("dropdown", "value"), - Input('date-picker-range', 'start_date'), - Input('date-picker-range', 'end_date') -) -def update_map(description, start_date, end_date): - dff = df[(df['date'] > start_date) & (df['date'] < end_date)] - dff = dff.reset_index() - dff = dff[dff.description == description] - - fig = px.scatter_mapbox( - dff, - lat="lat", - lon="lon", - hover_name="description", - hover_data=["date", "time"], - title="Incident Count by Coordinates", - center={"lat": 41.257160, "lon": -95.995102}, - zoom=10 - ) - - fig.update_layout(showlegend=False) - fig.update_layout(mapbox_style="open-street-map") - fig.update_layout(margin={"r": 0, "t": 0, "l": 0, "b": 0}) - fig.update_layout(mapbox_bounds={"west": -180, "east": -50, "south": 20, "north": 90}) - return fig +def filtered(agencies, categories, start, end): + df = INCIDENTS + if agencies: + df = df[df["agency"].isin(agencies)] + if categories: + df = df[df["category"].isin(categories)] + if start: + df = df[df["occurred_at"] >= start] + if end: + df = df[df["occurred_at"] <= f"{end[:10]} 23:59:59"] + return df -# Create line graph -@callback( - Output("line-graph", "figure"), - Input("dropdown", "value"), - Input('date-picker-range', 'start_date'), - Input('date-picker-range', 'end_date') -) -def update_line(description, start_date, end_date): - dff = df[(df['date'] > start_date) & (df['date'] < end_date)] - dff = dff.reset_index() - dff = dff[dff.description == description] - dff = dff.groupby(by="date").count() - dff = dff.reset_index() - - fig = px.line( - dff, - x="date", - y="description", - ) - return fig +INPUTS = [Input("agencies", "value"), Input("categories", "value"), + Input("dates", "start_date"), Input("dates", "end_date"), + Input("theme", "data")] + + +@callback(Output("kpis", "children"), *INPUTS) +def update_kpis(agencies, categories, start, end, _theme): + df = filtered(agencies, categories, start, end) + stops = df[df["is_stop"] == 1] + # Rate over the span the stops actually cover, not the full incident range: + # OPD reaches back to 2022 but reports no stops at all. + if len(stops): + span = (stops["occurred_at"].max() - stops["occurred_at"].min()).days + rate = f"{len(stops) / max(span, 1):.1f}" + else: + rate = "0" + cards = [ + ("Incidents", f"{len(df):,}"), + ("Vehicle stops", f"{len(stops):,}"), + ("Stops per day", rate), + ("Agencies", f"{df['agency'].nunique()}"), + ] + return [html.Div(className="kpi", children=[html.Span(v, className="kpi-value"), + html.Span(k, className="kpi-label")]) + for k, v in cards] + + +@callback(Output("map", "figure"), *INPUTS, Input("show-cameras", "value")) +def update_map(agencies, categories, start, end, theme, show_cameras): + df = filtered(agencies, categories, start, end) + # A density layer over a quarter-million points saturates at any radius and + # hides which agency is where, so plot a sample of the points instead. + sampled = len(df) > MAP_SAMPLE + if sampled: + df = df.sample(MAP_SAMPLE, random_state=0) + + fig = go.Figure() + for agency, g in df.groupby("agency", sort=False): + fig.add_trace(go.Scattermap( + lat=g["lat"], lon=g["lon"], mode="markers", name=agency, + marker={"size": 4, "opacity": 0.45}, + text=g["call_type"].fillna(g["category"]).fillna(g["offense_desc"]), + hovertemplate="%{text}<extra>" + agency + "</extra>")) + if show_cameras and len(CAMERAS): + fig.add_trace(go.Scattermap( + lat=CAMERAS["lat"], lon=CAMERAS["lon"], mode="markers", + # Amber reads against both the light and the dark basemap. + marker={"size": 7, "color": "#ffb300"}, + name="ALPR camera", + text=[f"{m or 'unknown make'} — {o or 'operator not tagged'}" + for m, o in zip(CAMERAS["manufacturer"], CAMERAS["operator"])], + hovertemplate="%{text}<extra>ALPR</extra>")) + title = f"{MAP_SAMPLE:,}-incident sample" if sampled else f"{len(df):,} incidents" + fig.update_layout(map={"style": PALETTES[theme]["basemap"], "center": CENTER, + "zoom": 9.6}, + height=560, margin={"r": 0, "t": 30, "l": 0, "b": 0}, + title=title, uirevision="map", + legend={"x": 0.01, "y": 0.99}) + return themed(fig, theme) + + +@callback(Output("timeline", "figure"), *INPUTS) +def update_timeline(agencies, categories, start, end, theme): + df = filtered(agencies, categories, start, end) + counts = analysis.daily_counts(df) + if counts.empty: + return empty(theme) + fig = px.line(counts, x="date", y="incidents", color="agency", + title="Incidents per day by agency") + fig.update_layout(margin={"t": 40}, hovermode="x unified") + return themed(fig, theme) + + +@callback(Output("dispositions", "figure"), *INPUTS) +def update_dispositions(agencies, categories, start, end, theme): + # Category filter is ignored: stops are identified by is_stop, not category. + out = analysis.stop_outcomes(filtered(agencies, None, start, end)) + if out.empty: + return empty(theme) + fig = px.bar(out, x="rate", y="agency", color="outcome", orientation="h", + barmode="group", custom_data=["stops"], + title="Vehicle stop outcomes", + labels={"rate": "share of that agency's stops"}) + fig.update_traces(hovertemplate="%{x:.1%} of %{customdata[0]:,} stops" + "<extra>%{fullData.name}</extra>") + fig.update_layout(margin={"t": 40}, xaxis_tickformat=".0%", + yaxis_title=None, legend_title_text=None) + return themed(fig, theme) + + +@callback(Output("proximity", "figure"), *INPUTS) +def update_proximity(agencies, categories, start, end, theme): + # Category filter is deliberately ignored: the comparison needs both stops + # and the non-stop baseline in the same window. + df = filtered(agencies, None, start, end) + prox = analysis.camera_proximity(df, CAMERAS) + if prox.empty: + return empty(theme) + fig = px.line(prox, x="distance_m", y="share", color="kind", markers=True, + title="Distance to nearest ALPR camera", + labels={"distance_m": "metres to nearest camera", + "share": "share of incidents"}) + fig.update_layout(margin={"t": 40}, yaxis_tickformat=".1%") + return themed(fig, theme) + + +@callback(Output("table", "data"), Output("table", "columns"), *INPUTS) +def update_table(agencies, categories, start, end, _theme): + cols = ["occurred_at", "agency", "category", "call_type", "disposition", + "address"] + df = (filtered(agencies, categories, start, end) + .sort_values("occurred_at", ascending=False) + .head(500)[cols]) + df = df.assign(occurred_at=df["occurred_at"].dt.strftime("%Y-%m-%d %H:%M")) + return df.to_dict("records"), [{"name": c, "id": c} for c in cols] -# Create table -@callback( - Output("table", "data"), - Input("dropdown", "value"), - Input('date-picker-range', 'start_date'), - Input('date-picker-range', 'end_date') -) -def update_table(description, start_date, end_date): - dff = df[(df['date'] > start_date) & (df['date'] < end_date)] - dff = dff.reset_index() - dff = dff[dff.description == description] - return dff.to_dict("records") if __name__ == "__main__": app.run(debug=True) @@ -1,4 +1,37 @@ +:root { + --bg: #fff; + --fg: #111; + --muted: #555; + --border: #ddd; + --panel: #fafafa; +} + +@media (prefers-color-scheme: dark) { + :root { + --bg: #14161a; + --fg: #e8e8ea; + --muted: #9aa0a6; + --border: #333840; + --panel: #1b1e24; + + /* Dash 3 ships light-only values for its component design tokens. */ + --Dash-Text-Primary: #e8e8ea; + --Dash-Text-Strong: #f2f2f4; + --Dash-Text-Weak: #9aa0a6; + --Dash-Text-Disabled: #6b7178; + --Dash-Stroke-Strong: rgba(232, 232, 234, 0.45); + --Dash-Stroke-Weak: rgba(232, 232, 234, 0.15); + --Dash-Fill-Inverse-Strong: #1b1e24; + --Dash-Fill-Interactive-Weak: rgba(255, 255, 255, 0.06); + --Dash-Fill-Primary-Hover: rgba(255, 255, 255, 0.06); + --Dash-Fill-Primary-Active: rgba(255, 255, 255, 0.1); + --Dash-Fill-Disabled: rgba(255, 255, 255, 0.1); + } +} + body { + background: var(--bg); + color: var(--fg); font-family: -apple-system, BlinkMacSystemFont, avenir next, avenir, segoe ui, helvetica neue, helvetica, Cantarell, Ubuntu, roboto, noto, arial, sans-serif; max-width: 80vw; margin: 2rem auto; @@ -14,21 +47,73 @@ body { max-width: 100%; } -.flex { - display: flex; - flex-direction: row; - align-items: center; +.subtitle { + color: var(--muted); + margin-top: -0.5rem; +} + +.controls { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(220px, 1fr)); + gap: 1rem; + margin: 1.5rem 0; +} + +.controls label { + display: block; + font-size: 0.85rem; + font-weight: 600; + margin-bottom: 0.25rem; } -.flex p { - margin-right: 1rem; - flex-shrink: 0; +.kpis { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(150px, 1fr)); + gap: 1rem; } -.flex .dash-dropdown { - width: 100%; +.kpi { + background: var(--panel); + border: 1px solid var(--border); + border-radius: 6px; + padding: 0.75rem 1rem; } -.flex #date-picker-range { - width: 100%; +.kpi-value { + display: block; + font-size: 1.6rem; + font-weight: 600; +} + +.kpi-label { + display: block; + font-size: 0.8rem; + color: var(--muted); +} + +.row { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(420px, 1fr)); + gap: 1rem; +} + +.row .half { + min-width: 0; +} + +/* DataTable paints its own colours instead of reading the tokens above. */ +@media (prefers-color-scheme: dark) { + .dash-spreadsheet-inner td, + .dash-spreadsheet-inner th, + .dash-spreadsheet-inner td.dash-cell, + .dash-spreadsheet-inner th.dash-header { + background-color: var(--panel) !important; + border-color: var(--border) !important; + color: var(--fg) !important; + } + + .dash-spreadsheet-menu, + .dash-spreadsheet-container { + color: var(--fg); + } } new file mode 100644 @@ -0,0 +1,19 @@ +// Plotly bakes colours into the rendered figure, so the server needs to know the +// colour scheme before it builds one. Push the media query result into the +// theme store on load and whenever it changes; app.py styles from there. +(function () { + var query = window.matchMedia('(prefers-color-scheme: dark)'); + + function publish(tries) { + var dc = window.dash_clientside; + try { + // set_props needs the layout rendered, not just the bundle loaded. + dc.set_props('theme', {data: query.matches ? 'dark' : 'light'}); + } catch (e) { + if ((tries || 0) < 100) setTimeout(function () { publish((tries || 0) + 1); }, 50); + } + } + + query.addEventListener('change', function () { publish(0); }); + publish(0); +})(); @@ -1,77 +1,334 @@ -# Import required modules +"""Pull incident feeds and ALPR camera locations into raw_data/metro.db. + +Sources: + opd Omaha Police incident data (DCGIS ArcGIS view), 2022-01-01 onward, daily. + sarpy Sarpy County PublicCrimeMap CAD calls, rolling 12-month window. Covers + Bellevue, Papillion, La Vista, Gretna, Springfield and the Sheriff. + Records age out of the feed, so run this often enough to keep the archive. + cbpd Council Bluffs PD public CFS feed, rolling 12-month window, refreshed + every 10 minutes. Same ageing-out caveat as sarpy. + alpr ALPR cameras from OpenStreetMap via Overpass (the data behind DeFlock). + opd_csv One-time backfill of raw_data/Incidents_*.csv (2015-2023). Statute text + only, no NIBRS category. + +All three ArcGIS services return UTC epochs. Their WHERE literals do not agree: +OPD and Council Bluffs use UTC, Sarpy uses America/Chicago, so each source +carries its own literal timezone and page size. +""" + +import argparse import csv +import json import sqlite3 -import os - -# Create the database file -connection = sqlite3.connect('./raw_data/test.db') - -# Creating a cursor object to execute SQL queries -cursor = connection.cursor() - -# Table Definition -# rb = RB Number -# date = Reported Date -# time = Reported Time -# description = Statute/Ordinance Description -# location = Occurred Location -# district = Occurred District -# lat = Occurred Block LAT -# lon = Occurred Block LON -create_table = '''CREATE TABLE incidents( - id INTEGER PRIMARY KEY AUTOINCREMENT, - rb TEXT NOT NULL, - date TEXT NOT NULL, - time TEXT NOT NULL, - description TEXT NOT NULL, - location TEXT NOT NULL, - district TEXT NOT NULL, - lat REAL NOT NULL, - lon REAL NOT NULL); - ''' - -# Create the table -cursor.execute(create_table) - -# Point to the data directory -directory = os.fsencode("./raw_data/") - -# Loop through all raw data files -for file in os.listdir(directory): - filename = os.fsdecode(file) - if filename.endswith(".csv"): - # Opening the file - file = open("./raw_data/" + filename) - - # Reading the contents of the file - contents = csv.reader(file) - - # SQL query to insert data into the - # table - insert_records = "INSERT INTO incidents (rb, date, time, description, location, district, lat, lon) VALUES(?, ?, ?, ?, ?, ?, ?, ?)" - - # Importing the contents of the file - # into our table - cursor.executemany(insert_records, contents) - print("Inserted data from: ", filename) - continue - else: - continue - -# Delete extra copies of the header row that were inserted -delete_headers = "DELETE FROM incidents WHERE rb = 'RB Number'" -cursor.execute(delete_headers) - -# Test query to see if the data loaded -select_all = "SELECT * FROM incidents" -rows = cursor.execute(select_all).fetchall() - -# Output to the console screen -for r in rows: - print(r) - -# Commit the changes -connection.commit() - -# Close the database connection -connection.close() +import ssl +import sys +import time +import urllib.error +import urllib.parse +import urllib.request + +import certifi +from datetime import datetime, timedelta, timezone +from pathlib import Path +from zoneinfo import ZoneInfo + +# Python does not use the macOS keychain, and Council Bluffs' host is not in the +# bundled trust store on every platform. +SSL_CTX = ssl.create_default_context(cafile=certifi.where()) + +ROOT = Path(__file__).parent +DB = ROOT / "raw_data" / "metro.db" +LOCAL = ZoneInfo("America/Chicago") + +OPD = { + "url": "https://services1.arcgis.com/tIBLyYZX96jUntYm/arcgis/rest/services" + "/Omaha_Police_Incident_Data_(View)/FeatureServer/0", + "date_field": "dteMidpoint", + "literal_tz": timezone.utc, + "oid": "OBJECTID", + "page": 2000, +} + +SARPY = { + "url": "https://geodata.sarpy.gov/arcgis/rest/services/PublicSafety" + "/PublicCrimeMap/FeatureServer/1", + "date_field": "IncidentDate", + "literal_tz": LOCAL, + "oid": "ObjectID", + "page": 2000, +} + +CBPD = { + "url": "https://gispublic.councilbluffs-ia.gov/publicserver/rest/services" + "/Hosted/Public_Facing_CFS_view/FeatureServer/0", + "date_field": "cfs_datetime", + "literal_tz": timezone.utc, + "oid": "objectid", + "page": 1000, +} + +# Council Bluffs files officer-initiated stops under one incident_code. +CBPD_STOP_CODE = "TRAFFIC : TRAFFIC STOP" + +# Sarpy IncidentId prefixes. Fire/EMS agencies share the feed with the police +# agencies; they are kept so the archive stays complete and filtered in the app. +SARPY_AGENCIES = { + "LBP": "Bellevue PD", + "LPP": "Papillion PD", + "LLP": "La Vista PD", + "LSO": "Sarpy County SO", + "LGP": "Gretna PD", + "LSP": "Springfield PD", + "BVF": "Bellevue Fire", + "PAF": "Papillion Fire", + "GRF": "Gretna Fire", + "SPF": "Springfield Fire", + "LVF": "La Vista Fire", +} + +# The feed's other vehicle category, "Traffic", is crashes, parking and DUI +# calls -- reactive, not officer-initiated. +SARPY_STOP_CATEGORY = "Proactive Policing - Vehicle Stop" + +OVERPASS = "https://overpass-api.de/api/interpreter" +# Douglas and Sarpy counties in Nebraska plus Council Bluffs across the river. +BBOX = (40.95, -96.35, 41.45, -95.65) +OVERPASS_QUERY = f""" +[out:json][timeout:120]; +( + node["man_made"="surveillance"]["surveillance:type"="ALPR"]{BBOX}; + node["man_made"="surveillance"]["surveillance:zone"="traffic"]["brand"~"Flock",i]{BBOX}; +); +out body; +""" + + +def get(url, params, retries=4): + body = urllib.parse.urlencode(params).encode() + for attempt in range(retries): + try: + req = urllib.request.Request(url, data=body, + headers={"User-Agent": "omaha-incidents/1.0"}) + with urllib.request.urlopen(req, timeout=120, context=SSL_CTX) as r: + payload = json.load(r) + except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as e: + if attempt == retries - 1: + raise + time.sleep(2 ** attempt) + continue + if "error" in payload: + raise RuntimeError(f"{url}: {payload['error']}") + return payload + raise AssertionError("unreachable") + + +def query_all(service, since): + """Page through an ArcGIS layer, yielding feature attribute dicts.""" + where = "1=1" + if since is not None: + stamp = since.astimezone(service["literal_tz"]).strftime("%Y-%m-%d %H:%M:%S") + where = f"{service['date_field']} >= TIMESTAMP '{stamp}'" + offset = 0 + while True: + page = get(service["url"] + "/query", { + "where": where, + "outFields": "*", + "returnGeometry": "true", + "outSR": "4326", + "orderByFields": f"{service['oid']} ASC", + "resultOffset": offset, + "resultRecordCount": service["page"], + "f": "json", + }) + feats = page.get("features", []) + if not feats: + return + for f in feats: + yield f + offset += len(feats) + print(f" {offset} rows", end="\r", file=sys.stderr, flush=True) + if not page.get("exceededTransferLimit") and len(feats) < service["page"]: + return + + +def local_iso(epoch_ms): + if epoch_ms is None: + return None + dt = datetime.fromtimestamp(epoch_ms / 1000, timezone.utc) + return dt.astimezone(LOCAL).strftime("%Y-%m-%dT%H:%M:%S") + + +def upsert(conn, rows): + conn.executemany( + """INSERT INTO incidents + (source, source_key, agency, case_id, occurred_at, category, + call_type, disposition, offense_desc, is_stop, address, lat, lon) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(source, source_key) DO UPDATE SET + agency=excluded.agency, case_id=excluded.case_id, + occurred_at=excluded.occurred_at, category=excluded.category, + call_type=excluded.call_type, disposition=excluded.disposition, + offense_desc=excluded.offense_desc, is_stop=excluded.is_stop, + address=excluded.address, lat=excluded.lat, lon=excluded.lon""", + rows) + return len(rows) + + +def ingest_opd(conn, since): + rows = [] + for f in query_all(OPD, since): + a = f["attributes"] + occurred = local_iso(a["dteMidpoint"]) + if occurred is None: + continue + rows.append(("opd", str(a["PK"]), "Omaha PD", a.get("RB"), occurred, + a.get("NIBRSCategory"), None, None, None, 0, + a.get("AddressBlock"), a.get("LatBlock"), a.get("LonBlock"))) + return upsert(conn, rows) + + +def ingest_sarpy(conn, since): + rows, unmapped = [], set() + for f in query_all(SARPY, since): + a = f["attributes"] + occurred = local_iso(a["IncidentDate"]) + if occurred is None: + continue + iid = a["IncidentId"] + prefix = iid[:3] + agency = SARPY_AGENCIES.get(prefix) + if agency is None: + unmapped.add(prefix) + agency = f"Unmapped {prefix}" + g = f.get("geometry") or {} + rows.append(("sarpy", iid, agency, iid, occurred, a.get("Category"), + a.get("CadTypeDesc"), a.get("CadDisposition"), + a.get("StatuteDesc"), + int(a.get("Category") == SARPY_STOP_CATEGORY), + a.get("BlkAddress"), g.get("y"), g.get("x"))) + n = upsert(conn, rows) + if unmapped: + print(f" unmapped IncidentId prefixes: {sorted(unmapped)}") + return n + + +def ingest_cbpd(conn, since): + rows = [] + for f in query_all(CBPD, since): + a = f["attributes"] + occurred = local_iso(a["cfs_datetime"]) + if occurred is None: + continue + g = f.get("geometry") or {} + code = a.get("incident_code") + # Council Bluffs withholds the street address; the point is still exact. + rows.append(("cbpd", a["cfs_number"], "Council Bluffs PD", + a.get("case_number") or a.get("cfs_number"), occurred, + a.get("incident_category"), code, a.get("disp_code"), None, + int(code == CBPD_STOP_CODE), + None, g.get("y"), g.get("x"))) + return upsert(conn, rows) + + +def ingest_opd_csv(conn, _since): + """Backfill the 2015-2023 CSV archive. Statute text goes to offense_desc; + category stays NULL because it is not a NIBRS category.""" + rows = [] + for path in sorted((ROOT / "raw_data").glob("Incidents_*.csv")): + with path.open(newline="") as fh: + if fh.readline().startswith("version https://git-lfs"): + print(f" {path.name}: git-lfs pointer, run 'git lfs pull'") + continue + fh.seek(0) + for i, r in enumerate(csv.reader(fh)): + if len(r) < 8 or r[0] == "RB Number": + continue + rb, date, tm, desc, loc, district, lat, lon = r[:8] + try: + when = datetime.strptime(f"{date} {tm}", "%m/%d/%Y %H:%M") + except ValueError: + continue + rows.append(("opd_csv", f"{path.stem}:{i}", "Omaha PD", rb, + when.strftime("%Y-%m-%dT%H:%M:%S"), None, None, None, + desc, 0, loc, + float(lat) if lat else None, + float(lon) if lon else None)) + return upsert(conn, rows) + + +def ingest_alpr(conn, _since): + payload = get(OVERPASS, {"data": OVERPASS_QUERY}) + now = datetime.now(LOCAL).strftime("%Y-%m-%dT%H:%M:%S") + rows = [] + for e in payload["elements"]: + t = e.get("tags", {}) + rows.append((e["id"], e["lat"], e["lon"], + t.get("manufacturer") or t.get("brand"), + t.get("operator"), + t.get("direction") or t.get("camera:direction"), + now, now, json.dumps(t, sort_keys=True))) + conn.executemany( + """INSERT INTO alpr_cameras + (osm_id, lat, lon, manufacturer, operator, direction, + first_seen, last_seen, tags) + VALUES (?,?,?,?,?,?,?,?,?) + ON CONFLICT(osm_id) DO UPDATE SET + lat=excluded.lat, lon=excluded.lon, + manufacturer=excluded.manufacturer, operator=excluded.operator, + direction=excluded.direction, last_seen=excluded.last_seen, + tags=excluded.tags""", + rows) + return len(rows) + + +def migrate(conn): + """Add columns introduced after a database was first built. Runs before + schema.sql so its indexes can reference the new columns.""" + have = {r[1] for r in conn.execute("PRAGMA table_info(incidents)")} + if have and "is_stop" not in have: + conn.execute("ALTER TABLE incidents ADD COLUMN is_stop INTEGER NOT NULL" + " DEFAULT 0") + conn.execute("UPDATE incidents SET is_stop = 1 WHERE category = ?", + (SARPY_STOP_CATEGORY,)) + conn.commit() + + +SOURCES = { + "opd": ingest_opd, + "sarpy": ingest_sarpy, + "cbpd": ingest_cbpd, + "alpr": ingest_alpr, + "opd_csv": ingest_opd_csv, +} + + +def main(): + p = argparse.ArgumentParser(description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("sources", nargs="*", choices=list(SOURCES), + help="default: opd sarpy cbpd alpr") + p.add_argument("--since-days", type=int, default=30, + help="only pull incidents this recent (default 30); " + "ignored by alpr and opd_csv") + p.add_argument("--full", action="store_true", + help="pull the complete feed instead of --since-days") + args = p.parse_args() + sources = args.sources or ["opd", "sarpy", "cbpd", "alpr"] + + since = None if args.full else datetime.now(timezone.utc) - timedelta(days=args.since_days) + + DB.parent.mkdir(exist_ok=True) + conn = sqlite3.connect(DB) + migrate(conn) + conn.executescript((ROOT / "schema.sql").read_text()) + for name in sources: + start = time.monotonic() + print(f" {name}: pulling...") + n = SOURCES[name](conn, since) + conn.commit() + print(f" {name}: {n} rows in {time.monotonic() - start:.1f}s") + conn.close() + + +if __name__ == "__main__": + main() new file mode 100644 @@ -0,0 +1 @@ +certifi new file mode 100644 @@ -0,0 +1,5 @@ +-r requirements-ingest.txt +dash>=3.0 +numpy>=2.0 +pandas>=2.2 +plotly>=6.0 new file mode 100644 @@ -0,0 +1,38 @@ +-- Incidents from every ingested feed. occurred_at is local time (America/Chicago); +-- the upstream services return UTC epochs and ingest.py converts on the way in. +CREATE TABLE IF NOT EXISTS incidents ( + source TEXT NOT NULL, -- opd | sarpy | cbpd | opd_csv + source_key TEXT NOT NULL, -- PK (opd) | IncidentId (sarpy) | cfs_number (cbpd) + agency TEXT NOT NULL, + case_id TEXT, + occurred_at TEXT NOT NULL, -- ISO 8601, no offset + category TEXT, -- each feed's own taxonomy, not comparable + call_type TEXT, -- CadTypeDesc (sarpy) | incident_code (cbpd) + disposition TEXT, -- CAD disposition, sarpy and cbpd + offense_desc TEXT, -- StatuteDesc (sarpy) | statute text (opd_csv) + is_stop INTEGER NOT NULL DEFAULT 0, -- officer-initiated vehicle stop + address TEXT, + lat REAL, + lon REAL, + PRIMARY KEY (source, source_key) +); + +CREATE INDEX IF NOT EXISTS incidents_occurred ON incidents (occurred_at); +CREATE INDEX IF NOT EXISTS incidents_agency ON incidents (agency, occurred_at); +CREATE INDEX IF NOT EXISTS incidents_category ON incidents (category); +CREATE INDEX IF NOT EXISTS incidents_stop ON incidents (is_stop, occurred_at); + +-- ALPR cameras from OpenStreetMap (ODbL). first_seen/last_seen track when a node +-- entered and was last present in the Overpass result, so cameras that appear or +-- are removed are visible over time. +CREATE TABLE IF NOT EXISTS alpr_cameras ( + osm_id INTEGER PRIMARY KEY, + lat REAL NOT NULL, + lon REAL NOT NULL, + manufacturer TEXT, + operator TEXT, + direction TEXT, + first_seen TEXT NOT NULL, + last_seen TEXT NOT NULL, + tags TEXT NOT NULL -- full OSM tag dict as JSON +);