krz/omaha-metro-blotter
clone: git clone https://gitbay.org/krz/omaha-metro-blotter.git
main: 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 flock Search audits exported from the agencies' Flock transparency portals.
12 Cloudflare serves every non-browser client a challenge, so these are
13 not fetchable from here: save the portal's "Download CSV" into
14 raw_data/flock/<portal-slug>_<date>.csv and this reads what is there.
15 The portals keep a rolling 30 days, so a gap longer than that is
16 permanent.
17 opd_archive
18 OPD's own yearly incident CSVs, 2015-2021. The ArcGIS view only goes
19 back to 2022-01-01, so this is the only route to the earlier years,
20 and it carries the statute description rather than a NIBRS category.
21 Reported crimes only: no stops, no dispositions.
22
23All three ArcGIS services return UTC epochs. Their WHERE literals do not agree:
24OPD and Council Bluffs use UTC, Sarpy uses America/Chicago, so each source
25carries its own literal timezone and page size.
26"""
27
28import argparse
29import csv
30import hashlib
31import json
32import sqlite3
33import ssl
34import sys
35import time
36import urllib.error
37import urllib.parse
38import urllib.request
39
40import certifi
41from datetime import datetime, timedelta, timezone
42from pathlib import Path
43from zoneinfo import ZoneInfo
44
45# Python does not use the macOS keychain, and Council Bluffs' host is not in the
46# bundled trust store on every platform.
47SSL_CTX = ssl.create_default_context(cafile=certifi.where())
48
49ROOT = Path(__file__).parent
50DB = ROOT / "raw_data" / "metro.db"
51LOCAL = ZoneInfo("America/Chicago")
52
53OPD = {
54 "url": "https://services1.arcgis.com/tIBLyYZX96jUntYm/arcgis/rest/services"
55 "/Omaha_Police_Incident_Data_(View)/FeatureServer/0",
56 "date_field": "dteMidpoint",
57 "literal_tz": timezone.utc,
58 "oid": "OBJECTID",
59 "page": 2000,
60}
61
62SARPY = {
63 "url": "https://geodata.sarpy.gov/arcgis/rest/services/PublicSafety"
64 "/PublicCrimeMap/FeatureServer/1",
65 "date_field": "IncidentDate",
66 "literal_tz": LOCAL,
67 "oid": "ObjectID",
68 "page": 2000,
69}
70
71CBPD = {
72 "url": "https://gispublic.councilbluffs-ia.gov/publicserver/rest/services"
73 "/Hosted/Public_Facing_CFS_view/FeatureServer/0",
74 "date_field": "cfs_datetime",
75 "literal_tz": timezone.utc,
76 "oid": "objectid",
77 "page": 1000,
78}
79
80# Council Bluffs files officer-initiated stops under one incident_code.
81CBPD_STOP_CODE = "TRAFFIC : TRAFFIC STOP"
82
83# Sarpy IncidentId prefixes. Fire/EMS agencies share the feed with the police
84# agencies; they are kept so the archive stays complete and filtered in the app.
85SARPY_AGENCIES = {
86 "LBP": "Bellevue PD",
87 "LPP": "Papillion PD",
88 "LLP": "La Vista PD",
89 "LSO": "Sarpy County SO",
90 "LGP": "Gretna PD",
91 "LSP": "Springfield PD",
92 "BVF": "Bellevue Fire",
93 "PAF": "Papillion Fire",
94 "GRF": "Gretna Fire",
95 "SPF": "Springfield Fire",
96 "LVF": "La Vista Fire",
97}
98
99# The feed's other vehicle category, "Traffic", is crashes, parking and DUI
100# calls -- reactive, not officer-initiated.
101SARPY_STOP_CATEGORY = "Proactive Policing - Vehicle Stop"
102
103# OPD publishes a CSV per year at a predictable path, updated daily. The ArcGIS
104# view starts 2022-01-01, so only the years before that are taken from here --
105# ingesting the overlap would double-count every Omaha incident since 2022.
106OPD_ARCHIVE = "https://police-static.cityofomaha.org/crime-data/{y}/Incidents_{y}.csv"
107OPD_ARCHIVE_YEARS = range(2015, 2022)
108OPD_ARCHIVE_COLUMNS = ("RB Number", "Reported Date", "Reported Time",
109 "Statute/Ordinance Description", "Occurred Location",
110 "Occurred District", "Occurred Block LAT",
111 "Occurred Block LON")
112
113OVERPASS = "https://overpass-api.de/api/interpreter"
114# Douglas and Sarpy counties in Nebraska plus Council Bluffs across the river.
115BBOX = (40.95, -96.35, 41.45, -95.65)
116OVERPASS_QUERY = f"""
117[out:json][timeout:120];
118(
119 node["man_made"="surveillance"]["surveillance:type"="ALPR"]{BBOX};
120 node["man_made"="surveillance"]["surveillance:zone"="traffic"]["brand"~"Flock",i]{BBOX};
121);
122out body;
123"""
124
125
126def get(url, params, retries=4):
127 body = urllib.parse.urlencode(params).encode()
128 for attempt in range(retries):
129 try:
130 req = urllib.request.Request(url, data=body,
131 headers={"User-Agent": "omaha-metro-blotter/1.0"})
132 with urllib.request.urlopen(req, timeout=120, context=SSL_CTX) as r:
133 payload = json.load(r)
134 except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as e:
135 if attempt == retries - 1:
136 raise
137 time.sleep(2 ** attempt)
138 continue
139 if "error" in payload:
140 raise RuntimeError(f"{url}: {payload['error']}")
141 return payload
142 raise AssertionError("unreachable")
143
144
145def query_all(service, since):
146 """Page through an ArcGIS layer, yielding feature attribute dicts."""
147 where = "1=1"
148 if since is not None:
149 stamp = since.astimezone(service["literal_tz"]).strftime("%Y-%m-%d %H:%M:%S")
150 where = f"{service['date_field']} >= TIMESTAMP '{stamp}'"
151 offset = 0
152 while True:
153 page = get(service["url"] + "/query", {
154 "where": where,
155 "outFields": "*",
156 "returnGeometry": "true",
157 "outSR": "4326",
158 "orderByFields": f"{service['oid']} ASC",
159 "resultOffset": offset,
160 "resultRecordCount": service["page"],
161 "f": "json",
162 })
163 feats = page.get("features", [])
164 if not feats:
165 return
166 for f in feats:
167 yield f
168 offset += len(feats)
169 print(f" {offset} rows", end="\r", file=sys.stderr, flush=True)
170 if not page.get("exceededTransferLimit") and len(feats) < service["page"]:
171 return
172
173
174def local_iso(epoch_ms):
175 if epoch_ms is None:
176 return None
177 dt = datetime.fromtimestamp(epoch_ms / 1000, timezone.utc)
178 return dt.astimezone(LOCAL).strftime("%Y-%m-%dT%H:%M:%S")
179
180
181# The order every row tuple is built in, and the order digest() hashes.
182COLUMNS = ("source", "source_key", "agency", "case_id", "occurred_at", "category",
183 "call_type", "disposition", "offense_desc", "is_stop", "address",
184 "lat", "lon")
185PAYLOAD = COLUMNS[2:] # everything the feed can change; the first two are the key
186
187
188# Columns whose SQLite affinity rewrites what the feed sent: a JSON lon of -96
189# arrives as int and comes back out of a REAL column as -96.0. Hashing the raw
190# value would mark such a record amended on every run forever, so coerce to the
191# stored representation first.
192REAL_FIELDS = {"lat", "lon"}
193INT_FIELDS = {"is_stop"}
194
195
196def digest(values):
197 """Hash of a record's payload. Feeds amend records after publishing them, so
198 this is what tells an unchanged record from a genuinely new version. Must
199 give the same answer for a value going into the database and coming back."""
200 parts = []
201 for name, v in zip(PAYLOAD, values):
202 if v is None:
203 parts.append("")
204 elif name in REAL_FIELDS:
205 parts.append(repr(float(v)))
206 elif name in INT_FIELDS:
207 parts.append(str(int(v)))
208 else:
209 parts.append(str(v))
210 return hashlib.blake2b("\x1f".join(parts).encode(), digest_size=8).hexdigest()
211
212
213def upsert(conn, rows):
214 """Insert records not seen before; file a changed record as an amendment.
215
216 Takes (values, raw) pairs, where raw is the feature exactly as the feed
217 served it. Nothing in incidents is ever updated: a record whose payload
218 differs from the one on file is appended to incident_amendments, so the
219 version the agency published first stays readable next to what it published
220 later. Every version's raw payload is kept too, so a parse can be redone
221 against what actually arrived."""
222 now = datetime.now(LOCAL).strftime("%Y-%m-%dT%H:%M:%S")
223 marks = ",".join("?" * (len(COLUMNS) + 2))
224 staged = [r + (digest(r[2:]), raw) for r, raw in rows]
225
226 conn.execute("DROP TABLE IF EXISTS temp.incoming")
227 conn.execute(f"CREATE TEMP TABLE incoming ({','.join(COLUMNS)}, digest, raw)")
228 conn.executemany(f"INSERT INTO temp.incoming VALUES ({marks})", staged)
229 conn.execute("CREATE INDEX temp.incoming_key ON incoming (source, source_key)")
230
231 cols = ",".join(COLUMNS)
232 conn.execute(
233 f"""INSERT OR IGNORE INTO incidents ({cols}, digest, first_seen)
234 SELECT {cols}, digest, ? FROM temp.incoming""", (now,))
235 amended = conn.execute(
236 f"""INSERT OR IGNORE INTO incident_amendments ({cols}, digest, seen_at)
237 SELECT i.{', i.'.join(COLUMNS)}, i.digest, ?
238 FROM temp.incoming i
239 JOIN incidents o
240 ON o.source = i.source AND o.source_key = i.source_key
241 WHERE o.digest <> i.digest""", (now,)).rowcount
242
243 # OR IGNORE keyed on the version, so a run that re-serves a known record
244 # stores nothing and the first full run backfills whatever is still served.
245 conn.execute(
246 """INSERT OR IGNORE INTO raw_records (source, source_key, digest,
247 fetched_at, payload)
248 SELECT source, source_key, digest, ?, raw FROM temp.incoming
249 WHERE raw IS NOT NULL""", (now,))
250
251 conn.execute("DROP TABLE temp.incoming")
252 return len(rows), amended
253
254
255def ingest_opd(conn, since):
256 rows = []
257 for f in query_all(OPD, since):
258 a = f["attributes"]
259 occurred = local_iso(a["dteMidpoint"])
260 if occurred is None:
261 continue
262 rows.append((("opd", str(a["PK"]), "Omaha PD", a.get("RB"), occurred,
263 a.get("NIBRSCategory"), None, None, None, 0,
264 a.get("AddressBlock"), a.get("LatBlock"), a.get("LonBlock")),
265 json.dumps(f, sort_keys=True)))
266 return upsert(conn, rows)
267
268
269def ingest_sarpy(conn, since):
270 rows, unmapped = [], set()
271 for f in query_all(SARPY, since):
272 a = f["attributes"]
273 occurred = local_iso(a["IncidentDate"])
274 if occurred is None:
275 continue
276 iid = a["IncidentId"]
277 prefix = iid[:3]
278 agency = SARPY_AGENCIES.get(prefix)
279 if agency is None:
280 unmapped.add(prefix)
281 agency = f"Unmapped {prefix}"
282 g = f.get("geometry") or {}
283 rows.append((("sarpy", iid, agency, iid, occurred, a.get("Category"),
284 a.get("CadTypeDesc"), a.get("CadDisposition"),
285 a.get("StatuteDesc"),
286 int(a.get("Category") == SARPY_STOP_CATEGORY),
287 a.get("BlkAddress"), g.get("y"), g.get("x")),
288 json.dumps(f, sort_keys=True)))
289 result = upsert(conn, rows)
290 if unmapped:
291 print(f" unmapped IncidentId prefixes: {sorted(unmapped)}")
292 return result
293
294
295def ingest_cbpd(conn, since):
296 rows = []
297 for f in query_all(CBPD, since):
298 a = f["attributes"]
299 occurred = local_iso(a["cfs_datetime"])
300 if occurred is None:
301 continue
302 g = f.get("geometry") or {}
303 code = a.get("incident_code")
304 # Council Bluffs withholds the street address; the point is still exact.
305 rows.append((("cbpd", a["cfs_number"], "Council Bluffs PD",
306 a.get("case_number") or a.get("cfs_number"), occurred,
307 a.get("incident_category"), code, a.get("disp_code"), None,
308 int(code == CBPD_STOP_CODE),
309 None, g.get("y"), g.get("x")),
310 json.dumps(f, sort_keys=True)))
311 return upsert(conn, rows)
312
313
314def ingest_opd_archive(conn, _since):
315 """Load OPD's yearly incident CSVs for the years the ArcGIS view predates.
316
317 Keyed on a hash of the row rather than its position in the file: these are
318 closed years and should be stable, but a single row inserted upstream would
319 otherwise shift every key below it and file fifty thousand false
320 amendments."""
321 rows, seen = [], {}
322 for year in OPD_ARCHIVE_YEARS:
323 url = OPD_ARCHIVE.format(y=year)
324 req = urllib.request.Request(url, headers={"User-Agent": "omaha-metro-blotter/1.0"})
325 with urllib.request.urlopen(req, timeout=180, context=SSL_CTX) as r:
326 text = r.read().decode("utf-8-sig", "replace")
327 reader = csv.DictReader(text.splitlines())
328 if tuple(reader.fieldnames or ()) != OPD_ARCHIVE_COLUMNS:
329 print(f" {year}: unexpected columns {reader.fieldnames}")
330 continue
331 n = 0
332 for rec in reader:
333 when = f"{rec['Reported Date']} {rec['Reported Time']}"
334 try:
335 occurred = datetime.strptime(when, "%m/%d/%Y %H:%M:%S")
336 except ValueError:
337 continue
338 raw = json.dumps(rec, sort_keys=True)
339 # a few rows repeat verbatim; number them so each keeps its own key
340 h = hashlib.blake2b(raw.encode(), digest_size=8).hexdigest()
341 # No raw kept: unlike the rolling feeds, whose aged-out records can
342 # never be fetched again, these year files stay downloadable. The
343 # one field not carried into a column is Occurred District.
344 seen[h] = seen.get(h, 0) + 1
345 lat, lon = rec["Occurred Block LAT"], rec["Occurred Block LON"]
346 rows.append(((
347 "opd_archive", f"{h}:{seen[h]}", "Omaha PD", rec["RB Number"],
348 occurred.strftime("%Y-%m-%dT%H:%M:%S"), None, None, None,
349 rec["Statute/Ordinance Description"], 0,
350 rec["Occurred Location"],
351 float(lat) if lat else None, float(lon) if lon else None), None))
352 n += 1
353 print(f" {year}: {n}", end="\r", file=sys.stderr, flush=True)
354 return upsert(conn, rows)
355
356
357FLOCK_COLUMNS = ("id", "userId", "searchDate", "networkCount", "reason")
358FLOCK_DIR = ROOT / "raw_data" / "flock"
359# Council Bluffs is the only metro portal that offers the export at all; Sarpy
360# and Douglas publish counts without one.
361FLOCK_DEFAULT_AGENCY = "council-bluffs-ia-pd"
362
363
364def import_flock(path, agency):
365 """File a portal download under the name ingest_flock expects.
366
367 The portal names every export public_search_audit.csv, with the agency
368 nowhere in the file, so the slug has to be supplied and is worth printing:
369 getting it wrong silently files one agency's searches under another."""
370 src = Path(path).expanduser()
371 with src.open(newline="") as fh:
372 reader = csv.DictReader(fh)
373 if tuple(reader.fieldnames or ()) != FLOCK_COLUMNS:
374 raise SystemExit(f" {src.name}: not a Flock search audit "
375 f"(columns {reader.fieldnames})")
376 rows = list(reader)
377 if not rows:
378 raise SystemExit(f" {src.name}: no rows")
379 span = f"{min(r['searchDate'] for r in rows)[:10]} to " \
380 f"{max(r['searchDate'] for r in rows)[:10]}"
381 FLOCK_DIR.mkdir(parents=True, exist_ok=True)
382 dest = FLOCK_DIR / f"{agency}_{datetime.now(LOCAL):%Y-%m-%d}.csv"
383 dest.write_bytes(src.read_bytes())
384 print(f" {len(rows)} searches, {span}")
385 print(f" filed as {dest.relative_to(ROOT)} under agency '{agency}'")
386
387
388def ingest_flock(conn, _since):
389 """Load Flock transparency-portal search audits from raw_data/flock.
390
391 The agency comes from the filename, since the export itself does not name
392 it. Search ids are stable UUIDs, so re-importing overlapping exports is
393 what keeps the archive whole across the portal's 30-day window."""
394 now = datetime.now(LOCAL).strftime("%Y-%m-%dT%H:%M:%S")
395 rows = []
396 for path in sorted((ROOT / "raw_data" / "flock").glob("*.csv")):
397 agency = path.stem.split("_")[0]
398 with path.open(newline="") as fh:
399 reader = csv.DictReader(fh)
400 if tuple(reader.fieldnames or ()) != FLOCK_COLUMNS:
401 print(f" {path.name}: unexpected columns {reader.fieldnames}")
402 continue
403 for r in reader:
404 rows.append((agency, r["id"], r["searchDate"],
405 int(r["networkCount"]) if r["networkCount"] else None,
406 r["reason"].strip() or None, r["userId"], now))
407 conn.executemany(
408 """INSERT OR IGNORE INTO alpr_searches
409 (agency, search_id, searched_at, network_count, reason, user_id,
410 imported_at) VALUES (?,?,?,?,?,?,?)""", rows)
411 return len(rows), 0
412
413
414def ingest_alpr(conn, _since):
415 payload = get(OVERPASS, {"data": OVERPASS_QUERY})
416 now = datetime.now(LOCAL).strftime("%Y-%m-%dT%H:%M:%S")
417 rows = []
418 for e in payload["elements"]:
419 t = e.get("tags", {})
420 rows.append((e["id"], e["lat"], e["lon"],
421 t.get("manufacturer") or t.get("brand"),
422 t.get("operator"),
423 t.get("direction") or t.get("camera:direction"),
424 now, now, json.dumps(t, sort_keys=True)))
425 conn.executemany(
426 """INSERT INTO alpr_cameras
427 (osm_id, lat, lon, manufacturer, operator, direction,
428 first_seen, last_seen, tags)
429 VALUES (?,?,?,?,?,?,?,?,?)
430 ON CONFLICT(osm_id) DO UPDATE SET
431 lat=excluded.lat, lon=excluded.lon,
432 manufacturer=excluded.manufacturer, operator=excluded.operator,
433 direction=excluded.direction, last_seen=excluded.last_seen,
434 tags=excluded.tags""",
435 rows)
436 return len(rows), 0
437
438
439def migrate(conn):
440 """Add columns introduced after a database was first built. Runs before
441 schema.sql so its views and indexes can reference the new columns."""
442 have = {r[1] for r in conn.execute("PRAGMA table_info(incidents)")}
443 if not have:
444 return
445
446 if "is_stop" not in have:
447 conn.execute("ALTER TABLE incidents ADD COLUMN is_stop INTEGER NOT NULL"
448 " DEFAULT 0")
449 conn.execute("UPDATE incidents SET is_stop = 1 WHERE category = ?",
450 (SARPY_STOP_CATEGORY,))
451 conn.commit()
452
453 if "digest" not in have:
454 # Backfill from the stored values, using the same function ingest uses,
455 # so the next run sees the existing rows as unchanged rather than
456 # amending all of them.
457 conn.execute("ALTER TABLE incidents ADD COLUMN digest TEXT")
458 conn.execute("ALTER TABLE incidents ADD COLUMN first_seen TEXT")
459 payload = ", ".join(PAYLOAD)
460 rows = conn.execute(
461 f"SELECT source, source_key, {payload} FROM incidents").fetchall()
462 conn.executemany(
463 "UPDATE incidents SET digest = ? WHERE source = ? AND source_key = ?",
464 [(digest(r[2:]), r[0], r[1]) for r in rows])
465 conn.commit()
466 print(f" migrated: digested {len(rows)} existing rows")
467
468
469SOURCES = {
470 "opd": ingest_opd,
471 "sarpy": ingest_sarpy,
472 "cbpd": ingest_cbpd,
473 "alpr": ingest_alpr,
474 "flock": ingest_flock,
475 "opd_archive": ingest_opd_archive,
476}
477
478
479def main():
480 p = argparse.ArgumentParser(description=__doc__,
481 formatter_class=argparse.RawDescriptionHelpFormatter)
482 p.add_argument("sources", nargs="*", choices=list(SOURCES),
483 help="default: opd sarpy cbpd alpr flock")
484 p.add_argument("--since-days", type=int, default=30,
485 help="only pull incidents this recent (default 30); "
486 "ignored by alpr, flock and opd_archive")
487 p.add_argument("--full", action="store_true",
488 help="pull the complete feed instead of --since-days")
489 p.add_argument("--import-flock", metavar="CSV",
490 help="file a Flock portal download into raw_data/flock and "
491 "load it")
492 p.add_argument("--agency", default=FLOCK_DEFAULT_AGENCY,
493 help=f"portal slug for --import-flock (default "
494 f"{FLOCK_DEFAULT_AGENCY})")
495 args = p.parse_args()
496 if args.import_flock:
497 import_flock(args.import_flock, args.agency)
498 sources = ["flock"]
499 else:
500 sources = args.sources or ["opd", "sarpy", "cbpd", "alpr", "flock"]
501
502 since = None if args.full else datetime.now(timezone.utc) - timedelta(days=args.since_days)
503
504 DB.parent.mkdir(exist_ok=True)
505 conn = sqlite3.connect(DB)
506 migrate(conn)
507 conn.executescript((ROOT / "schema.sql").read_text())
508 for name in sources:
509 start = time.monotonic()
510 print(f" {name}: pulling...")
511 seen, amended = SOURCES[name](conn, since)
512 conn.commit()
513 note = f", {amended} amended" if amended else ""
514 print(f" {name}: {seen} rows{note} in {time.monotonic() - start:.1f}s")
515 conn.close()
516
517
518if __name__ == "__main__":
519 main()