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