#!/usr/bin/env python3 from __future__ import annotations import argparse import hashlib import json import os import re import sqlite3 from pathlib import Path ROOT = Path(__file__).resolve().parents[1] OUTPUT = ROOT / "ReleaseData" / "production" WEBSITE_DATA = ROOT / "Website" / "data" DATABASE = OUTPUT / "evcharge-offline.sqlite" MANIFEST = OUTPUT / "station-manifest.json" APPROVAL = OUTPUT / "approval-record.json" NOTICE = OUTPUT / "ODbL-NOTICE.txt" SOURCE_ID = "openstreetmap_odbl" ATTRIBUTION = "© OpenStreetMap contributors" ATTRIBUTION_URL = "https://www.openstreetmap.org/copyright" LICENSE_NAME = "Open Database License (ODbL) 1.0" LICENSE_URL = "https://opendatacommons.org/licenses/odbl/1-0/" def sha256_bytes(payload: bytes) -> str: return hashlib.sha256(payload).hexdigest() def normalized_brand(tags: dict[str, str]) -> str: raw = ( tags.get("brand") or tags.get("operator") or tags.get("network") or "OpenStreetMap" ).strip() folded = raw.casefold() canonical = { "zes": "ZES", "eşarj": "Eşarj", "eşarj şarj istasyonu": "Eşarj", "sharz": "SHARZ", "sharz.net": "SHARZ", "trugo": "Trugo", "voltrun": "Voltrun", "wat mobilite": "WAT Mobilite", "tesla": "Tesla Supercharger", "tesla supercharger": "Tesla Supercharger", } return canonical.get(folded, raw) def parse_connector_count(value: str | None) -> int: if not value: return 0 normalized = value.strip().lower() if not re.fullmatch(r"\d+(?:\s*;\s*\d+)*", normalized): return 0 return sum(int(item) for item in re.findall(r"\d+", normalized)) def connector_tag_present(value: str | None) -> bool: if not value: return False normalized = value.strip().lower() if normalized in {"no", "false", "0"}: return False return normalized in {"yes", "true"} or parse_connector_count(value) > 0 def connector_types(tags: dict[str, str]) -> list[str]: result = [] if connector_tag_present(tags.get("socket:type2_combo")): result.append("ccs2") if connector_tag_present(tags.get("socket:type2")): result.append("type2") if connector_tag_present(tags.get("socket:chademo")): result.append("chademo") return result def socket_slots(tags: dict[str, str]) -> tuple[int, bool]: connector_total = sum( parse_connector_count(tags.get(key)) for key in ( "socket:type2_combo", "socket:type2", "socket:chademo", ) ) if connector_total > 0: return connector_total, False capacity_value = (tags.get("capacity") or "").strip() if re.fullmatch(r"\d+", capacity_value): capacity = int(capacity_value) if 1 <= capacity <= 64: return capacity, False if capacity > 64: return 0, True return 0, bool(capacity_value) def coordinates(element: dict) -> tuple[float, float] | None: if "lat" in element and "lon" in element: return float(element["lat"]), float(element["lon"]) center = element.get("center") or {} if "lat" in center and "lon" in center: return float(center["lat"]), float(center["lon"]) return None def is_green(tags: dict[str, str]) -> bool: candidates = ( tags.get("renewable_energy"), tags.get("source:energy"), tags.get("energy_source"), ) return any( value and any(token in value.casefold() for token in ("renewable", "solar", "wind")) for value in candidates ) def build(input_path: Path) -> None: source_bytes = input_path.read_bytes() source_sha = sha256_bytes(source_bytes) payload = json.loads(source_bytes) observed_at = payload["osm3s"]["timestamp_osm_base"] generated_at = observed_at rows = [] brands = set() connector_station_count = 0 total_slots = 0 ignored_ambiguous_capacity_records = 0 for element in payload.get("elements", []): point = coordinates(element) if point is None: continue tags = element.get("tags") or {} latitude, longitude = point brand = normalized_brand(tags) title = ( tags.get("name") or tags.get("operator") or tags.get("brand") or f"Charging station {element['type']} {element['id']}" ).strip() connectors = connector_types(tags) if connectors: connector_station_count += 1 slots, ignored_capacity = socket_slots(tags) total_slots += slots ignored_ambiguous_capacity_records += int(ignored_capacity) station_key = f"osm:{element['type']}:{element['id']}" canonical_record = json.dumps( { "type": element["type"], "id": element["id"], "lat": latitude, "lon": longitude, "tags": tags, }, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") rows.append( ( station_key, SOURCE_ID, f"{element['type']}/{element['id']}", title, brand, latitude, longitude, int(is_green(tags)), "unknown", "station", "historical_snapshot", observed_at, slots, 0, "|" + "|".join(connectors) + "|" if connectors else "", sha256_bytes(canonical_record), ) ) brands.add(brand) rows.sort(key=lambda item: item[0]) OUTPUT.mkdir(parents=True, exist_ok=True) if DATABASE.exists(): DATABASE.unlink() connection = sqlite3.connect(DATABASE) connection.executescript( """ PRAGMA application_id = 1163281224; PRAGMA user_version = 1; CREATE TABLE bundle_metadata ( key TEXT PRIMARY KEY, value TEXT NOT NULL ) WITHOUT ROWID; CREATE TABLE stations ( station_key TEXT PRIMARY KEY, source_id TEXT NOT NULL, source_station_id TEXT NOT NULL, title TEXT NOT NULL, brand TEXT NOT NULL, latitude REAL NOT NULL, longitude REAL NOT NULL, is_green INTEGER NOT NULL, station_availability TEXT NOT NULL, availability_scope TEXT NOT NULL, availability_freshness TEXT NOT NULL, availability_observed_at TEXT NOT NULL, socket_slot_count INTEGER NOT NULL, socket_reference_count INTEGER NOT NULL, connector_types TEXT NOT NULL, source_record_sha256 TEXT NOT NULL ) WITHOUT ROWID; CREATE TABLE brand_catalog ( brand TEXT PRIMARY KEY ) WITHOUT ROWID; CREATE INDEX idx_stations_geo ON stations(latitude, longitude, station_key); CREATE INDEX idx_stations_brand ON stations(brand, station_key); CREATE INDEX idx_stations_availability ON stations(station_availability, station_key); """ ) connection.executemany( "INSERT INTO stations VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", rows, ) connection.executemany( "INSERT INTO brand_catalog VALUES (?)", [(brand,) for brand in sorted(brands, key=str.casefold)], ) metadata = { "bundle_format": "tr.evcharge.offline-stations", "schema_version": "1", "profile": "openstreetmap-odbl-release", "observed_at": observed_at, "source_snapshot_sha256": source_sha, } connection.executemany( "INSERT INTO bundle_metadata VALUES (?, ?)", sorted(metadata.items()), ) connection.commit() connection.execute("VACUUM") integrity = connection.execute("PRAGMA quick_check").fetchone()[0] connection.close() if integrity != "ok": raise RuntimeError(f"SQLite integrity failure: {integrity}") database_sha = sha256_bytes(DATABASE.read_bytes()) approval = { "approvedAt": generated_at, "approvalBasis": ( "OpenStreetMap data is available free of charge under the open " "ODbL 1.0 license; no paid OSM data license is purchased." ), "databaseSha256": database_sha, "licenseName": LICENSE_NAME, "licenseURL": LICENSE_URL, "sourceId": SOURCE_ID, "sourceSnapshotSha256": source_sha, "status": "approved_open_license", } APPROVAL.write_text( json.dumps(approval, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n", encoding="utf-8", ) manifest = { "availability": { "isLive": False, "observedAt": observed_at, "reportedAvailableStationCount": 0, "reportedUnavailableStationCount": 0, "scope": "station", }, "bundleFormat": "tr.evcharge.offline-stations", "capabilities": { "connectorTypes": connector_station_count > 0, "stationAvailability": False, }, "counts": { "brands": len(brands), "socketReferences": 0, "socketSlots": total_slots, "stations": len(rows), }, "dataQuality": { "connectorTaggedStationRecords": connector_station_count, "ignoredAmbiguousCapacityRecords": ( ignored_ambiguous_capacity_records ), "mappedChargingPointSemantics": ( "Explicit socket counts, otherwise pure numeric capacity " "between 1 and 64. Power-like or ambiguous values are ignored." ), }, "files": [ { "bytes": DATABASE.stat().st_size, "path": DATABASE.name, "sha256": database_sha, } ], "intendedUse": "public_app_distribution", "licenseGate": { "allowedUse": [ "internal_technical_evaluation", "local_development", "automated_tests", "app_store_distribution", "public_redistribution", ], "approvalRecordPath": APPROVAL.name, "blockedUse": [], "failureMode": "fail_closed", "redistributionAllowed": True, "releaseEligible": True, "requiredBeforeRelease": [ "visible_openstreetmap_attribution", "odbl_database_notice", ], "status": "approved", }, "profile": "openstreetmap-odbl-release", "schemaVersion": 1, "source": { "accessCost": "free_of_charge", "attributionText": ATTRIBUTION, "attributionURL": ATTRIBUTION_URL, "databaseDownloadURL": "https://volnavi.fikiryazilim.com/data/evcharge-offline.sqlite", "displayName": "OpenStreetMap contributors", "kind": "open_data", "licenseClass": "open_license", "licenseName": LICENSE_NAME, "licenseURL": LICENSE_URL, "observedAt": observed_at, "sourceId": SOURCE_ID, "sourceEvidencePath": os.path.relpath(input_path, OUTPUT), "sourceEvidenceURL": "https://volnavi.fikiryazilim.com/data/osm-source-minimal.json", "sourceSnapshotSha256": source_sha, "paidCommercialLicensePurchased": False, }, } MANIFEST.write_text( json.dumps(manifest, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n", encoding="utf-8", ) NOTICE.write_text( "Volnavi station catalog - OpenStreetMap notice\n\n" f"{ATTRIBUTION}\n" "OpenStreetMap data is used free of charge under an open license. " "Volnavi does not purchase a paid OSM data license.\n" f"Source and attribution: {ATTRIBUTION_URL}\n" f"Database license: {LICENSE_NAME}\n" f"License text: {LICENSE_URL}\n\n" "The bundled charging-station database is derived from OpenStreetMap " "data observed at " f"{observed_at}. Volnavi does not claim that OpenStreetMap station " "records are live or complete.\n" "The derived database and deterministic production method are kept " "accessible to meet the ODbL attribution and share-alike obligations.\n", encoding="utf-8", ) WEBSITE_DATA.mkdir(parents=True, exist_ok=True) (WEBSITE_DATA / DATABASE.name).write_bytes(DATABASE.read_bytes()) (WEBSITE_DATA / MANIFEST.name).write_text( MANIFEST.read_text(encoding="utf-8"), encoding="utf-8", ) (WEBSITE_DATA / APPROVAL.name).write_text( APPROVAL.read_text(encoding="utf-8"), encoding="utf-8", ) (WEBSITE_DATA / NOTICE.name).write_text( NOTICE.read_text(encoding="utf-8"), encoding="utf-8", ) (WEBSITE_DATA / "osm-source-minimal.json").write_bytes(input_path.read_bytes()) build_directory = WEBSITE_DATA / "build" build_directory.mkdir(parents=True, exist_ok=True) for script_name in ( "generate_osm_release_dataset.py", "sanitize_osm_snapshot.py", ): source_script = ROOT / "Scripts" / script_name (build_directory / script_name).write_bytes(source_script.read_bytes()) print(f"stations={len(rows)}") print(f"brands={len(brands)}") print(f"connector_stations={connector_station_count}") print(f"socket_slots={total_slots}") print( "ignored_ambiguous_capacity_records=" f"{ignored_ambiguous_capacity_records}" ) print(f"database_sha256={database_sha}") print(f"source_sha256={source_sha}") def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--input", type=Path, required=True) args = parser.parse_args() build(args.input) return 0 if __name__ == "__main__": raise SystemExit(main())