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
archive: ingest.py · raw
1"""Pull incident feeds and ALPR camera locations into raw_data/metro.db.
2
3Sources:
4 opd Omaha Police incident data (DCGIS ArcGIS view), 2022-01-01 onward, daily.
5 sarpy Sarpy County PublicCrimeMap CAD calls, rolling 12-month window. Covers
6 Bellevue, Papillion, La Vista, Gretna, Springfield and the Sheriff.
7 Records age out of the feed, so run this often enough to keep the archive.
8 cbpd Council Bluffs PD public CFS feed, rolling 12-month window, refreshed
9 every 10 minutes. Same ageing-out caveat as sarpy.
10 alpr ALPR cameras from OpenStreetMap via Overpass (the data behind DeFlock).
11 opd_csv One-time backfill of raw_data/Incidents_*.csv (2015-2023). Statute text
12 only, no NIBRS category.
13
14All three ArcGIS services return UTC epochs. Their WHERE literals do not agree:
15OPD and Council Bluffs use UTC, Sarpy uses America/Chicago, so each source
16carries its own literal timezone and page size.
17"""
18
19import argparse
20import csv
21import json
22import sqlite3
23import ssl
24import sys
25import time
26import urllib.error
27import urllib.parse
28import urllib.request
29
30import certifi
31from datetime import datetime, timedelta, timezone
32from pathlib import Path
33from zoneinfo import ZoneInfo
34
35# Python does not use the macOS keychain, and Council Bluffs' host is not in the
36# bundled trust store on every platform.
37SSL_CTX = ssl.create_default_context(cafile=certifi.where())
38
39ROOT = Path(__file__).parent
40DB = ROOT / "raw_data" / "metro.db"
41LOCAL = ZoneInfo("America/Chicago")
42
43OPD = {
44 "url": "https://services1.arcgis.com/tIBLyYZX96jUntYm/arcgis/rest/services"
45 "/Omaha_Police_Incident_Data_(View)/FeatureServer/0",
46 "date_field": "dteMidpoint",
47 "literal_tz": timezone.utc,
48 "oid": "OBJECTID",
49 "page": 2000,
50}
51
52SARPY = {
53 "url": "https://geodata.sarpy.gov/arcgis/rest/services/PublicSafety"
54 "/PublicCrimeMap/FeatureServer/1",
55 "date_field": "IncidentDate",
56 "literal_tz": LOCAL,
57 "oid": "ObjectID",
58 "page": 2000,
59}
60
61CBPD = {
62 "url": "https://gispublic.councilbluffs-ia.gov/publicserver/rest/services"
63 "/Hosted/Public_Facing_CFS_view/FeatureServer/0",
64 "date_field": "cfs_datetime",
65 "literal_tz": timezone.utc,
66 "oid": "objectid",
67 "page": 1000,
68}
69
70# Council Bluffs files officer-initiated stops under one incident_code.
71CBPD_STOP_CODE = "TRAFFIC : TRAFFIC STOP"
72
73# Sarpy IncidentId prefixes. Fire/EMS agencies share the feed with the police
74# agencies; they are kept so the archive stays complete and filtered in the app.
75SARPY_AGENCIES = {
76 "LBP": "Bellevue PD",
77 "LPP": "Papillion PD",
78 "LLP": "La Vista PD",
79 "LSO": "Sarpy County SO",
80 "LGP": "Gretna PD",
81 "LSP": "Springfield PD",
82 "BVF": "Bellevue Fire",
83 "PAF": "Papillion Fire",
84 "GRF": "Gretna Fire",
85 "SPF": "Springfield Fire",
86 "LVF": "La Vista Fire",
87}
88
89# The feed's other vehicle category, "Traffic", is crashes, parking and DUI
90# calls -- reactive, not officer-initiated.
91SARPY_STOP_CATEGORY = "Proactive Policing - Vehicle Stop"
92
93OVERPASS = "https://overpass-api.de/api/interpreter"
94# Douglas and Sarpy counties in Nebraska plus Council Bluffs across the river.
95BBOX = (40.95, -96.35, 41.45, -95.65)
96OVERPASS_QUERY = f"""
97[out:json][timeout:120];
98(
99 node["man_made"="surveillance"]["surveillance:type"="ALPR"]{BBOX};
100 node["man_made"="surveillance"]["surveillance:zone"="traffic"]["brand"~"Flock",i]{BBOX};
101);
102out body;
103"""
104
105
106def get(url, params, retries=4):
107 body = urllib.parse.urlencode(params).encode()
108 for attempt in range(retries):
109 try:
110 req = urllib.request.Request(url, data=body,
111 headers={"User-Agent": "omaha-incidents/1.0"})
112 with urllib.request.urlopen(req, timeout=120, context=SSL_CTX) as r:
113 payload = json.load(r)
114 except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as e:
115 if attempt == retries - 1:
116 raise
117 time.sleep(2 ** attempt)
118 continue
119 if "error" in payload:
120 raise RuntimeError(f"{url}: {payload['error']}")
121 return payload
122 raise AssertionError("unreachable")
123
124
125def query_all(service, since):
126 """Page through an ArcGIS layer, yielding feature attribute dicts."""
127 where = "1=1"
128 if since is not None:
129 stamp = since.astimezone(service["literal_tz"]).strftime("%Y-%m-%d %H:%M:%S")
130 where = f"{service['date_field']} >= TIMESTAMP '{stamp}'"
131 offset = 0
132 while True:
133 page = get(service["url"] + "/query", {
134 "where": where,
135 "outFields": "*",
136 "returnGeometry": "true",
137 "outSR": "4326",
138 "orderByFields": f"{service['oid']} ASC",
139 "resultOffset": offset,
140 "resultRecordCount": service["page"],
141 "f": "json",
142 })
143 feats = page.get("features", [])
144 if not feats:
145 return
146 for f in feats:
147 yield f
148 offset += len(feats)
149 print(f" {offset} rows", end="\r", file=sys.stderr, flush=True)
150 if not page.get("exceededTransferLimit") and len(feats) < service["page"]:
151 return
152
153
154def local_iso(epoch_ms):
155 if epoch_ms is None:
156 return None
157 dt = datetime.fromtimestamp(epoch_ms / 1000, timezone.utc)
158 return dt.astimezone(LOCAL).strftime("%Y-%m-%dT%H:%M:%S")
159
160
161def upsert(conn, rows):
162 conn.executemany(
163 """INSERT INTO incidents
164 (source, source_key, agency, case_id, occurred_at, category,
165 call_type, disposition, offense_desc, is_stop, address, lat, lon)
166 VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)
167 ON CONFLICT(source, source_key) DO UPDATE SET
168 agency=excluded.agency, case_id=excluded.case_id,
169 occurred_at=excluded.occurred_at, category=excluded.category,
170 call_type=excluded.call_type, disposition=excluded.disposition,
171 offense_desc=excluded.offense_desc, is_stop=excluded.is_stop,
172 address=excluded.address, lat=excluded.lat, lon=excluded.lon""",
173 rows)
174 return len(rows)
175
176
177def ingest_opd(conn, since):
178 rows = []
179 for f in query_all(OPD, since):
180 a = f["attributes"]
181 occurred = local_iso(a["dteMidpoint"])
182 if occurred is None:
183 continue
184 rows.append(("opd", str(a["PK"]), "Omaha PD", a.get("RB"), occurred,
185 a.get("NIBRSCategory"), None, None, None, 0,
186 a.get("AddressBlock"), a.get("LatBlock"), a.get("LonBlock")))
187 return upsert(conn, rows)
188
189
190def ingest_sarpy(conn, since):
191 rows, unmapped = [], set()
192 for f in query_all(SARPY, since):
193 a = f["attributes"]
194 occurred = local_iso(a["IncidentDate"])
195 if occurred is None:
196 continue
197 iid = a["IncidentId"]
198 prefix = iid[:3]
199 agency = SARPY_AGENCIES.get(prefix)
200 if agency is None:
201 unmapped.add(prefix)
202 agency = f"Unmapped {prefix}"
203 g = f.get("geometry") or {}
204 rows.append(("sarpy", iid, agency, iid, occurred, a.get("Category"),
205 a.get("CadTypeDesc"), a.get("CadDisposition"),
206 a.get("StatuteDesc"),
207 int(a.get("Category") == SARPY_STOP_CATEGORY),
208 a.get("BlkAddress"), g.get("y"), g.get("x")))
209 n = upsert(conn, rows)
210 if unmapped:
211 print(f" unmapped IncidentId prefixes: {sorted(unmapped)}")
212 return n
213
214
215def ingest_cbpd(conn, since):
216 rows = []
217 for f in query_all(CBPD, since):
218 a = f["attributes"]
219 occurred = local_iso(a["cfs_datetime"])
220 if occurred is None:
221 continue
222 g = f.get("geometry") or {}
223 code = a.get("incident_code")
224 # Council Bluffs withholds the street address; the point is still exact.
225 rows.append(("cbpd", a["cfs_number"], "Council Bluffs PD",
226 a.get("case_number") or a.get("cfs_number"), occurred,
227 a.get("incident_category"), code, a.get("disp_code"), None,
228 int(code == CBPD_STOP_CODE),
229 None, g.get("y"), g.get("x")))
230 return upsert(conn, rows)
231
232
233def ingest_opd_csv(conn, _since):
234 """Backfill the 2015-2023 CSV archive. Statute text goes to offense_desc;
235 category stays NULL because it is not a NIBRS category."""
236 rows = []
237 for path in sorted((ROOT / "raw_data").glob("Incidents_*.csv")):
238 with path.open(newline="") as fh:
239 if fh.readline().startswith("version https://git-lfs"):
240 print(f" {path.name}: git-lfs pointer, run 'git lfs pull'")
241 continue
242 fh.seek(0)
243 for i, r in enumerate(csv.reader(fh)):
244 if len(r) < 8 or r[0] == "RB Number":
245 continue
246 rb, date, tm, desc, loc, district, lat, lon = r[:8]
247 try:
248 when = datetime.strptime(f"{date} {tm}", "%m/%d/%Y %H:%M")
249 except ValueError:
250 continue
251 rows.append(("opd_csv", f"{path.stem}:{i}", "Omaha PD", rb,
252 when.strftime("%Y-%m-%dT%H:%M:%S"), None, None, None,
253 desc, 0, loc,
254 float(lat) if lat else None,
255 float(lon) if lon else None))
256 return upsert(conn, rows)
257
258
259def ingest_alpr(conn, _since):
260 payload = get(OVERPASS, {"data": OVERPASS_QUERY})
261 now = datetime.now(LOCAL).strftime("%Y-%m-%dT%H:%M:%S")
262 rows = []
263 for e in payload["elements"]:
264 t = e.get("tags", {})
265 rows.append((e["id"], e["lat"], e["lon"],
266 t.get("manufacturer") or t.get("brand"),
267 t.get("operator"),
268 t.get("direction") or t.get("camera:direction"),
269 now, now, json.dumps(t, sort_keys=True)))
270 conn.executemany(
271 """INSERT INTO alpr_cameras
272 (osm_id, lat, lon, manufacturer, operator, direction,
273 first_seen, last_seen, tags)
274 VALUES (?,?,?,?,?,?,?,?,?)
275 ON CONFLICT(osm_id) DO UPDATE SET
276 lat=excluded.lat, lon=excluded.lon,
277 manufacturer=excluded.manufacturer, operator=excluded.operator,
278 direction=excluded.direction, last_seen=excluded.last_seen,
279 tags=excluded.tags""",
280 rows)
281 return len(rows)
282
283
284def migrate(conn):
285 """Add columns introduced after a database was first built. Runs before
286 schema.sql so its indexes can reference the new columns."""
287 have = {r[1] for r in conn.execute("PRAGMA table_info(incidents)")}
288 if have and "is_stop" not in have:
289 conn.execute("ALTER TABLE incidents ADD COLUMN is_stop INTEGER NOT NULL"
290 " DEFAULT 0")
291 conn.execute("UPDATE incidents SET is_stop = 1 WHERE category = ?",
292 (SARPY_STOP_CATEGORY,))
293 conn.commit()
294
295
296SOURCES = {
297 "opd": ingest_opd,
298 "sarpy": ingest_sarpy,
299 "cbpd": ingest_cbpd,
300 "alpr": ingest_alpr,
301 "opd_csv": ingest_opd_csv,
302}
303
304
305def main():
306 p = argparse.ArgumentParser(description=__doc__,
307 formatter_class=argparse.RawDescriptionHelpFormatter)
308 p.add_argument("sources", nargs="*", choices=list(SOURCES),
309 help="default: opd sarpy cbpd alpr")
310 p.add_argument("--since-days", type=int, default=30,
311 help="only pull incidents this recent (default 30); "
312 "ignored by alpr and opd_csv")
313 p.add_argument("--full", action="store_true",
314 help="pull the complete feed instead of --since-days")
315 args = p.parse_args()
316 sources = args.sources or ["opd", "sarpy", "cbpd", "alpr"]
317
318 since = None if args.full else datetime.now(timezone.utc) - timedelta(days=args.since_days)
319
320 DB.parent.mkdir(exist_ok=True)
321 conn = sqlite3.connect(DB)
322 migrate(conn)
323 conn.executescript((ROOT / "schema.sql").read_text())
324 for name in sources:
325 start = time.monotonic()
326 print(f" {name}: pulling...")
327 n = SOURCES[name](conn, since)
328 conn.commit()
329 print(f" {name}: {n} rows in {time.monotonic() - start:.1f}s")
330 conn.close()
331
332
333if __name__ == "__main__":
334 main()