Skip to content

Instantly share code, notes, and snippets.

@Phaeilo

Phaeilo/blitzer.py

Created June 15, 2025 22:16
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save Phaeilo/31f36ac50679aa9aca113621f87b2688 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save Phaeilo/31f36ac50679aa9aca113621f87b2688 to your computer and use it in GitHub Desktop.
Download ZIP
Speed Camera Data Downloader and POI Finder
Raw
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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment