From 1ee8cf418173d9bdfaf1bdd3ef425ef1d284b31a Mon Sep 17 00:00:00 2001 From: Jake Pullen Date: Sat, 25 Jul 2026 09:13:45 +0100 Subject: [PATCH] small test --- main.py | 207 +++++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 205 insertions(+), 2 deletions(-) diff --git a/main.py b/main.py index 959aa39..b48493c 100644 --- a/main.py +++ b/main.py @@ -1,6 +1,209 @@ -def main(): - print("Hello from police-data!") +import requests +import time +import sqlite3 +import threading +import queue +from collections import deque +from typing import List +# -------------------------------- +# Thread-Safe Leaky Bucket Rate Limiter +# -------------------------------- +class LeakyBucketRateLimiter: + def __init__(self, max_rate: float = 15.0): + self.max_rate = max_rate + self.timestamps: deque = deque() + self.lock = threading.Lock() + self.window: float = 1.0 # Sliding window size in seconds + + def acquire(self) -> None: + with self.lock: + now = time.monotonic() + # Remove timestamps outside the current window + while self.timestamps and self.timestamps[0] <= now - self.window: + self.timestamps.popleft() + + # If limit reached, sleep until the oldest request exits the window + if len(self.timestamps) >= self.max_rate: + sleep_time = 1.0 - (now - self.timestamps[0]) + if sleep_time > 0: + time.sleep(sleep_time) + now = time.monotonic() + while self.timestamps and self.timestamps[0] <= now - self.window: + self.timestamps.popleft() + + self.timestamps.append(time.monotonic()) + +# -------------------------------- +# Database Initialization +# -------------------------------- +def init_db(db_path: str = "uk_crimes.db") -> sqlite3.Connection: + conn = sqlite3.connect(db_path) + conn.execute(""" + CREATE TABLE IF NOT EXISTS crimes ( + id INTEGER PRIMARY KEY, + category TEXT, + month TEXT, + latitude REAL, + longitude REAL, + street_name TEXT, + outcome_category TEXT + ) + """) + conn.commit() + return conn + +# -------------------------------- +# UK Grid Polygon Generator +# -------------------------------- +def generate_uk_polygons(lat_min: float = 49.5, lat_max: float = 61.0, + lng_min: float = -13.0, lng_max: float = 5.0, + step: float = 0.5) -> List[str]: + polygons = [] + lats = [l / 10.0 for l in range(int(lat_min * 10), int(lat_max * 10), int(step * 10))] + lns = [ln / 10.0 for ln in range(int(lng_min * 10), int(lng_max * 10), int(step * 10))] + + for i in range(len(lats) - 1): + for j in range(len(lns) - 1): + poly = f"{lats[i]},{lns[j]}:{lats[i]},{lns[j+1]}:{lats[i+1]},{lns[j+1]}:{lats[i+1]},{lns[j]}" + polygons.append(poly) + return polygons + +# -------------------------------- +# Producer Thread: Fetch & Queue +# -------------------------------- +def producer_worker(polygons: List[str], date: str, limiter: LeakyBucketRateLimiter, data_queue: queue.Queue): + total = len(polygons) + for i, poly in enumerate(polygons): + # Skip polygons exceeding the 4094 character limit [1] + if len(poly) > 4090: + continue + + # Enforce strict 15 requests/second limit [1] + limiter.acquire() + + try: + response = requests.get( + "https://data.police.uk/api/crimes-street/all-crime", + params={"date": date, "poly": poly}, + timeout=30 + ) + response.raise_for_status() + data = response.json() + + # API returns 503 if a custom area contains >10,000 crimes [1] + if not isinstance(data, list): + print(f"[PRODUCER] Warning: Received non-list response for poly {poly[:30]}...") + continue + + if not data: + continue + + rows = [] + for crime in data: + loc = crime.get("location", {}) + outcome = crime.get("outcome_status") or {} + rows.append(( + crime.get("id"), + crime.get("category"), + crime.get("month"), + float(loc.get("latitude", 0)), + float(loc.get("longitude", 0)), + loc.get("street", {}).get("name"), + outcome.get("category") + )) + + # Push batch to memory queue (non-blocking if size allows) + data_queue.put(rows) + + except requests.exceptions.RequestException as e: + print(f"[PRODUCER] Network error for poly {poly[:30]}... | {e}") + except Exception as e: + print(f"[PRODUCER] Unexpected error: {e}") + + if (i + 1) % 100 == 0 or i == total - 1: + print(f"[PRODUCER] Progress: {i+1}/{total} fetched | Queue size: {data_queue.qsize()}") + + # Signal completion to consumer + data_queue.put(None) + +# -------------------------------- +# Consumer Thread: Queue to DB +# -------------------------------- +def consumer_worker(data_queue: queue.Queue, conn: sqlite3.Connection): + batch_size = 500 + batch = [] + + while True: + try: + item = data_queue.get(timeout=5.0) + except queue.Empty: + # If queue is empty for 5s, assume production is done + if not batch: + break + else: + # Flush remaining batch + conn.executemany( + "INSERT OR IGNORE INTO crimes (id, category, month, latitude, longitude, street_name, outcome_category) VALUES (?, ?, ?, ?, ?, ?, ?)", + batch + ) + conn.commit() + batch = [] + continue + + if item is None: + # Flush last batch before exiting + if batch: + conn.executemany( + "INSERT OR IGNORE INTO crimes (id, category, month, latitude, longitude, street_name, outcome_category) VALUES (?, ?, ?, ?, ?, ?, ?)", + batch + ) + conn.commit() + data_queue.task_done() + break + + batch.extend(item) + + if len(batch) >= batch_size: + conn.executemany( + "INSERT OR IGNORE INTO crimes (id, category, month, latitude, longitude, street_name, outcome_category) VALUES (?, ?, ?, ?, ?, ?, ?)", + batch + ) + conn.commit() + batch = [] + + data_queue.task_done() + +# -------------------------------- +# Main Orchestrator +# -------------------------------- +def main(): + limiter = LeakyBucketRateLimiter(max_rate=15.0) + conn = init_db("uk_crimes.db") + data_queue = queue.Queue(maxsize=2000) # Backpressure to prevent memory overflow + date = "2024-01" + + print("Generating UK coverage polygons...") + polygons = generate_uk_polygons() + print(f"Total regions to process: {len(polygons)}\n") + + # Start threads + prod_thread = threading.Thread(target=producer_worker, args=(polygons, date, limiter, data_queue), daemon=True) + cons_thread = threading.Thread(target=consumer_worker, args=(data_queue, conn), daemon=True) + + cons_thread.start() + prod_thread.start() + + # Wait for producer to finish + prod_thread.join() + print("\n[MAIN] Producer finished. Waiting for consumer to drain queue...") + + # Wait for consumer to finish + cons_thread.join() + + total = conn.execute('SELECT COUNT(*) FROM crimes').fetchone()[0] + print(f"[MAIN] Completed. Total unique records stored in uk_crimes.db: {total}") + conn.close() if __name__ == "__main__": main()