Speed Camera Data Downloader and POI Finder

blitzer.py

#!/usr/bin/env python3
"""
Speed Camera Data Downloader and POI Finder

A simple Python tool to download speed camera and POI data and find nearby points of interest
based on your current coordinates.

License: CC-0 / Public Domain

Usage:
    # Download databases and find POIs near coordinates
    python blitzer.py --lat 52.520008 --lon 13.404954 --radius 5

    # Force download of databases
    python blitzer.py --download

    # Limit the number of results
    python blitzer.py --lat 52.520008 --lon 13.404954 --limit 5

    # Output results in JSON format
    python blitzer.py --lat 52.520008 --lon 13.404954 --json

Requirements:
    - Python 3.11+
    - requests library (`pip install requests`)

Data:
    The tool downloads and uses two SQLite databases:
    - mobile.sqlite: Contains mobile speed cameras and other temporary POIs
    - static.sqlite: Contains static speed cameras and other permanent POIs
"""

import argparse
import itertools
import json
import math
import os
import sqlite3
import sys

import requests


def log(*args):
    print(*args, file=sys.stderr)


def download_file(url, filename):
    try:
        response = requests.get(url)
        with open(filename, "wb") as f:
            f.write(response.content)
        log(f"Downloaded {filename} ({len(response.content) // 1024} KB)")
    except Exception as e:
        log(f"Error downloading {filename}: {e}")
        sys.exit(1)


def download_databases():
    log("Downloading databases...")
    download_file("https://data-ssl.atudo.net/output/mobile.php?v=4&key=", "mobile.sqlite")
    download_file("https://data-ssl.atudo.net/output/static.php?v=4", "static.sqlite")


def calculate_distance(lat1, lon1, lat2, lon2):
    """Calculate the distance between two points in kilometers using the Haversine formula."""
    lat1_rad = math.radians(lat1)
    lon1_rad = math.radians(lon1)
    lat2_rad = math.radians(lat2)
    lon2_rad = math.radians(lon2)

    dlon = lon2_rad - lon1_rad
    dlat = lat2_rad - lat1_rad
    a = math.sin(dlat / 2) ** 2 + math.cos(lat1_rad) * math.cos(lat2_rad) * math.sin(dlon / 2) ** 2
    c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
    distance = 6371 * c  # Earth radius in kilometers

    return distance


def custom_text_factory(text):
    if not isinstance(text, bytes):
        return text
    try:
        return text.decode("utf-8")
    except UnicodeDecodeError:
        # fall back to latin-1
        return text.decode("latin-1")


def query_database(db_file, table_name, lat, lon, radius):
    try:
        conn = sqlite3.connect(db_file)
        conn.text_factory = custom_text_factory
        cursor = conn.cursor()

        # get all POIs
        cursor.execute(f"SELECT lat, long, type, street, spd FROM {table_name}")
        db_pois = cursor.fetchall()

        # get type names
        cursor.execute("SELECT id, name FROM type")
        types = {row[0]: row[1] for row in cursor.fetchall()}

        conn.close()

        # deserialize and filter records
        for poi in db_pois:
            poi_lat, poi_lon, poi_type, poi_street, poi_speed = poi
            type_name = types.get(poi_type, "Unknown")

            distance = calculate_distance(lat, lon, poi_lat, poi_lon)
            if distance > radius:
                continue

            yield {
                "type": "mobile" if table_name == "blitzermob" else "static",
                "category": type_name,
                "street": poi_street,
                "speed": poi_speed,
                "distance": round(distance, 2),
                "lat": poi_lat,
                "lon": poi_lon,
            }
    except Exception as e:
        log(f"Error querying {db_file} database: {e}")
        sys.exit(1)


def find_nearby_pois(lat, lon, radius=10, limit=None):
    """Find POIs within the specified radius (in kilometers) of the given coordinates."""

    # ensure database files exist
    if not os.path.exists("mobile.sqlite") or not os.path.exists("static.sqlite"):
        log("Databases not found. Downloading...")
        download_databases()

    # query both database files
    pois = itertools.chain(
        query_database("mobile.sqlite", "blitzermob", lat, lon, radius),
        query_database("static.sqlite", "blitzerstat", lat, lon, radius),
    )

    # sort by distance
    pois = sorted(pois, key=lambda x: x["distance"])

    for i, poi in enumerate(pois):
        if limit is not None and i >= limit:
            break
        yield poi


def main():
    parser = argparse.ArgumentParser(description="Download and query speed camera data for nearby POIs.")
    parser.add_argument("--lat", type=float, help="Your current latitude")
    parser.add_argument("--lon", type=float, help="Your current longitude")
    parser.add_argument(
        "--radius",
        type=float,
        default=10,
        help="Search radius in kilometers (default: 10)",
    )
    parser.add_argument("--download", action="store_true", help="Force download of databases")
    parser.add_argument(
        "--limit",
        type=int,
        default=10,
        help="Limit the number of results (default: 10)",
    )
    parser.add_argument("--json", action="store_true", help="Output results in JSON format")

    args = parser.parse_args()

    if args.download:
        download_databases()
        if not args.lat or not args.lon:
            return

    # if coordinates are provided, find nearby POIs
    if args.lat and args.lon:
        pois = list(find_nearby_pois(args.lat, args.lon, args.radius, args.limit))

        if not pois:
            print(f"No POIs found within {args.radius} km of your location.")
            return

        # output in JSON, if requested
        if args.json:
            print(json.dumps(pois, indent=2))
            return

        # output in human-readable format
        print(f"\nFound {len(pois)} POIs within {args.radius} km of your location:")
        print("-" * 80)

        for i, poi in enumerate(pois):
            print(f"{i+1}. {poi['category']} ({poi['type']}) - {poi['street']}")
            print(f"   Speed: {poi['speed'] or 'N/A'}, Distance: {poi['distance']} km")
            print(f"   Coordinates: {poi['lat']}, {poi['lon']}")
            print("-" * 80)
    else:
        # if no coordinates are provided, show usage
        if not args.download:
            parser.print_help()


if __name__ == "__main__":
    main()