krz/omaha-metro-blotter

  1"""Queries behind the dashboard: agency activity and distance to the nearest ALPR.
  2
  3Reads incidents_current, the newest known version of each record. The originals
  4stay in incidents and every superseded version in incident_amendments, so a
  5disposition an agency changed after the fact is still recoverable."""
  6
  7import sqlite3
  8from pathlib import Path
  9
 10import numpy as np
 11import pandas as pd
 12
 13DB = Path(__file__).parent / "raw_data" / "metro.db"
 14
 15# Sarpy and Council Bluffs publish officer-initiated stops; ingest.py flags them
 16# as is_stop. OPD publishes NIBRS offences only and contributes no stops.
 17POLICE_AGENCIES = ("Omaha PD", "Council Bluffs PD", "Bellevue PD", "Papillion PD",
 18                   "La Vista PD", "Sarpy County SO")
 19
 20EARTH_M = 6371000.0
 21
 22
 23def connect():
 24    return sqlite3.connect(DB)
 25
 26
 27def load_incidents(conn, agencies=None, start=None, end=None, categories=None):
 28    where, params = ["lat IS NOT NULL", "lon IS NOT NULL"], []
 29    if agencies:
 30        where.append(f"agency IN ({','.join('?' * len(agencies))})")
 31        params += list(agencies)
 32    if categories:
 33        where.append(f"category IN ({','.join('?' * len(categories))})")
 34        params += list(categories)
 35    if start:
 36        where.append("occurred_at >= ?")
 37        params.append(f"{start}T00:00:00")
 38    if end:
 39        where.append("occurred_at <= ?")
 40        params.append(f"{end}T23:59:59")
 41    df = pd.read_sql_query(
 42        f"""SELECT source, agency, case_id, occurred_at, category, call_type,
 43                   disposition, offense_desc, is_stop, address, lat, lon, amended
 44            FROM incidents_current WHERE {' AND '.join(where)}""",
 45        conn, params=params)
 46    df["occurred_at"] = pd.to_datetime(df["occurred_at"])
 47    return df
 48
 49
 50def load_cameras(conn):
 51    return pd.read_sql_query(
 52        "SELECT osm_id, lat, lon, manufacturer, operator, direction, first_seen"
 53        " FROM alpr_cameras", conn)
 54
 55
 56def nearest_camera_m(df, cameras):
 57    """Great-circle distance in metres from each incident to the closest camera.
 58
 59    164 cameras against ~250k incidents, so the full distance matrix is computed
 60    in chunks rather than all at once."""
 61    if df.empty or cameras.empty:
 62        return np.full(len(df), np.nan)
 63
 64    lat1 = np.radians(df["lat"].to_numpy(dtype=float))
 65    lon1 = np.radians(df["lon"].to_numpy(dtype=float))
 66    lat2 = np.radians(cameras["lat"].to_numpy(dtype=float))
 67    lon2 = np.radians(cameras["lon"].to_numpy(dtype=float))
 68
 69    out = np.empty(len(df))
 70    cos2 = np.cos(lat2)
 71    for i in range(0, len(df), 20000):
 72        s = slice(i, i + 20000)
 73        dlat = lat2[None, :] - lat1[s, None]
 74        dlon = lon2[None, :] - lon1[s, None]
 75        a = (np.sin(dlat / 2) ** 2
 76             + np.cos(lat1[s, None]) * cos2[None, :] * np.sin(dlon / 2) ** 2)
 77        out[s] = (2 * EARTH_M * np.arcsin(np.sqrt(a))).min(axis=1)
 78    return out
 79
 80
 81def daily_counts(df):
 82    if df.empty:
 83        return pd.DataFrame(columns=["date", "agency", "incidents"])
 84    g = (df.assign(date=df["occurred_at"].dt.floor("D"))
 85           .groupby(["date", "agency"], as_index=False)
 86           .size().rename(columns={"size": "incidents"}))
 87    return g
 88
 89
 90def stop_outcomes(df):
 91    """Citation and arrest rate on vehicle stops, per agency.
 92
 93    The two CAD systems use different disposition vocabularies -- Sarpy writes
 94    WRITTEN WARNING / CITATION, Council Bluffs writes "3 - Citation" and folds
 95    warnings into "7 - Handled by Officer" -- and both allow several outcomes per
 96    stop. Substring matching on citation and arrest is the only comparison the
 97    two vocabularies actually support, and a department that records warnings
 98    less thoroughly will show a higher citation rate for that reason alone."""
 99    stops = df[(df["is_stop"] == 1) & df["disposition"].notna()]
100    if stops.empty:
101        return pd.DataFrame(columns=["agency", "outcome", "rate", "stops"])
102    d = stops["disposition"].str.upper()
103    stops = stops.assign(cited=d.str.contains("CITATION"),
104                         arrested=d.str.contains("ARREST"))
105    g = stops.groupby("agency").agg(stops=("cited", "size"),
106                                    Cited=("cited", "mean"),
107                                    Arrested=("arrested", "mean")).reset_index()
108    return (g.melt(id_vars=["agency", "stops"], value_vars=["Cited", "Arrested"],
109                   var_name="outcome", value_name="rate")
110             .sort_values("rate", ascending=False))
111
112
113def camera_proximity(df, cameras, bin_m=200, max_m=2000):
114    """Share of stops vs other incidents falling in each distance band.
115
116    Both series are normalised, so a gap between them means stops cluster
117    differently around cameras than the rest of the call volume does. It is not
118    evidence of causation: cameras and stops both concentrate on arterials."""
119    if df.empty or cameras.empty:
120        return pd.DataFrame(columns=["distance_m", "kind", "share"])
121    d = df.assign(dist=nearest_camera_m(df, cameras))
122    d = d[d["dist"] <= max_m]
123    if d.empty:
124        return pd.DataFrame(columns=["distance_m", "kind", "share"])
125    d["kind"] = np.where(d["is_stop"] == 1, "Vehicle stops", "All other incidents")
126    d["distance_m"] = (d["dist"] // bin_m * bin_m).astype(int)
127    g = (d.groupby(["kind", "distance_m"], as_index=False)
128           .size().rename(columns={"size": "n"}))
129    g["share"] = g["n"] / g.groupby("kind")["n"].transform("sum")
130    return g
131
132
133def amendment_history(conn, source, source_key):
134    """Every version of one record, oldest first."""
135    return pd.read_sql_query(
136        """SELECT 'original' AS version, occurred_at, category, call_type,
137                  disposition, is_stop, address, first_seen AS seen_at
138             FROM incidents WHERE source = ? AND source_key = ?
139           UNION ALL
140           SELECT 'amended', occurred_at, category, call_type,
141                  disposition, is_stop, address, seen_at
142             FROM incident_amendments WHERE source = ? AND source_key = ?
143           ORDER BY seen_at""",
144        conn, params=[source, source_key, source, source_key])
145
146
147def changed_stop_outcomes(conn):
148    """Stops whose disposition the agency changed after first publishing it."""
149    return pd.read_sql_query(
150        """SELECT o.agency, o.case_id, o.occurred_at,
151                  o.disposition AS first_published,
152                  a.disposition AS later_published, a.seen_at
153             FROM incident_amendments a
154             JOIN incidents o
155               ON o.source = a.source AND o.source_key = a.source_key
156            WHERE o.is_stop = 1
157              AND IFNULL(a.disposition, '') <> IFNULL(o.disposition, '')
158            ORDER BY a.seen_at DESC""", conn)
159
160
161def search_audit(conn):
162    """Per-agency summary of ALPR searches from the Flock transparency portals.
163
164    reason is free text the searching officer typed, and most searches have
165    none despite the portals stating that access requires one. user_id is
166    redacted upstream, so nothing here attributes a search to a person."""
167    return pd.read_sql_query(
168        """SELECT agency,
169                  COUNT(*) AS searches,
170                  SUM(reason IS NOT NULL) AS with_reason,
171                  1.0 * SUM(reason IS NOT NULL) / COUNT(*) AS reason_rate,
172                  CAST(AVG(network_count) AS INT) AS avg_networks,
173                  MAX(network_count) AS max_networks,
174                  MIN(searched_at) AS earliest, MAX(searched_at) AS latest
175             FROM alpr_searches GROUP BY agency ORDER BY searches DESC""", conn)
176
177
178def agency_options(conn):
179    rows = conn.execute(
180        "SELECT agency, COUNT(*) FROM incidents_current GROUP BY agency"
181        " ORDER BY 2 DESC").fetchall()
182    return [a for a, _ in rows]
183
184
185def category_options(conn):
186    rows = conn.execute(
187        "SELECT category, COUNT(*) FROM incidents_current"
188        " WHERE category IS NOT NULL GROUP BY category ORDER BY 2 DESC").fetchall()
189    return [c for c, _ in rows]
190
191
192def date_bounds(conn):
193    lo, hi = conn.execute(
194        "SELECT MIN(occurred_at), MAX(occurred_at) FROM incidents_current").fetchone()
195    return lo[:10], hi[:10]