diff options
| author | Ariel Costas Guerrero <ariel@costas.dev> | 2025-12-22 18:18:59 +0100 |
|---|---|---|
| committer | Ariel Costas Guerrero <ariel@costas.dev> | 2025-12-22 18:18:59 +0100 |
| commit | 68f49dec91d68579803d6d579b1f1ecb4fc1dd1f (patch) | |
| tree | a67ec0cc572ae56e85a5dc08b2469bdbcb451135 /src/stop_downloader/vigo/download-stops.py | |
| parent | 4b7eaa318f22d7cc768491c421cb7aeac477f95d (diff) | |
Remove systems we won't need anymore
Diffstat (limited to 'src/stop_downloader/vigo/download-stops.py')
| -rw-r--r-- | src/stop_downloader/vigo/download-stops.py | 260 |
1 files changed, 0 insertions, 260 deletions
diff --git a/src/stop_downloader/vigo/download-stops.py b/src/stop_downloader/vigo/download-stops.py deleted file mode 100644 index fa08019..0000000 --- a/src/stop_downloader/vigo/download-stops.py +++ /dev/null @@ -1,260 +0,0 @@ -# /// script -# requires-python = ">=3.12" -# dependencies = [ -# "PyYAML>=6.0.2", # For YAML support -# ] -# /// -import csv -import json -import os -import sys -import urllib.request -import yaml # Add YAML support for overrides - -OVERRIDES_DIR = "overrides" -OUTPUT_FILE = "../../frontend/public/stops/vigo.json" - -SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) - - -def load_stop_overrides(file_path): - """Load stop overrides from a YAML file""" - if not os.path.exists(file_path): - print(f"Warning: Overrides file {file_path} not found") - return {} - - try: - with open(file_path, "r", encoding="utf-8") as f: - overrides = yaml.safe_load(f) - print(f"Loaded {len(overrides) if overrides else 0} stop overrides") - return overrides or {} - except Exception as e: - print(f"Error loading overrides: {e}", file=sys.stderr) - return {} - - -def apply_overrides(stops, overrides): - """Apply overrides to the stop data and add new stops""" - # Track existing stop IDs - existing_stop_ids = {stop.get("stopId") for stop in stops} - - # Apply overrides to existing stops - for stop in stops: - stop_id = stop.get("stopId") - if stop_id in overrides: - override = overrides[stop_id] - - # Override name if provided - if "name" in override: - stop["name"] = override["name"] - - # Apply or add alternate names - if "alternateNames" in override: - for key, value in override["alternateNames"].items(): - stop["name"][key] = value - - # Apply location override - if "location" in override: - if "latitude" in override["location"]: - stop["latitude"] = override["location"]["latitude"] - if "longitude" in override["location"]: - stop["longitude"] = override["location"]["longitude"] - - # Add amenities - if "amenities" in override: - stop["amenities"] = override["amenities"] - - # Mark stop as hidden if needed - if "hide" in override: - stop["hide"] = override["hide"] - - # Mark stop as cancelled - if "cancelled" in override: - stop["cancelled"] = override["cancelled"] - - if "alert" in override: - stop["alert"] = override["alert"] - - if "title" in override: - stop["title"] = override["title"] - - if "message" in override: - stop["message"] = override["message"] - - # Add new stops (those with "new: true" parameter) - new_stops_added = 0 - for stop_id, override in overrides.items(): - # Check if this is a new stop - if override.get("new") and stop_id not in existing_stop_ids: - # Ensure stop_id is an integer for consistency - stop_id_int = int(stop_id) if isinstance(stop_id, str) else stop_id - - # Create the new stop - new_stop = { - "stopId": stop_id_int, - "name": override.get("name", f"Stop {stop_id_int}"), - "latitude": override.get("location", {}).get("latitude"), - "longitude": override.get("location", {}).get("longitude"), - "lines": override.get("lines", []), - } - - # Add optional fields (excluding the 'new' parameter) - if "alternateNames" in override: - for key, value in override["alternateNames"].items(): - new_stop["name"][key] = value - if "amenities" in override: - new_stop["amenities"] = override["amenities"] - if "cancelled" in override: - new_stop["cancelled"] = override["cancelled"] - if "title" in override: - new_stop["title"] = override["title"] - if "message" in override: - new_stop["message"] = override["message"] - if "alternateCodes" in override: - new_stop["alternateCodes"] = override["alternateCodes"] - - stops.append(new_stop) - new_stops_added += 1 - - if new_stops_added > 0: - print(f"Added {new_stops_added} new stops from overrides") - - return stops - - -def download_stops_vitrasa() -> list[dict]: - url = "https://datos.vigo.org/vci_api_app/api2.jsp?tipo=TRANSPORTE_PARADAS" - req = urllib.request.Request(url) - - try: - with urllib.request.urlopen(req) as response: - # Read the response and decode from ISO-8859-1 to UTF-8 - content = response.read().decode("iso-8859-1") - data = json.loads(content) - - print(f"Downloaded {len(data)} stops") - - # Process the data - processed_stops = [] - for stop in data: - name = stop.get("nombre", "").strip() - # Fix double space equals comma-space: "Castrelos 202" -> "Castrelos, 202"; and remove quotes - name = name.replace(" ", ", ").replace('"', "").replace("'", "") - - processed_stop = { - "stopId": "vitrasa:" + str(stop.get("id")), - "name": name, - "latitude": stop.get("lat"), - "longitude": stop.get("lon"), - "lines": [line.strip() for line in stop.get("lineas", "").split(",")] - if stop.get("lineas") - else [], - } - processed_stops.append(processed_stop) - - return processed_stops - except Exception as e: - print(f"Error processing vigo stops data: {e}", file=sys.stderr) - return [] - - -def download_stops_renfe() -> list[dict]: - url = "https://data.renfe.com/dataset/1146f3f1-e06d-477c-8f74-84f8d0668cf9/resource/b22cd560-3a2b-45dd-a25d-2406941f6fcc/download/listado_completo_av_ld_md.csv" - req = urllib.request.Request(url) - - # CÓDIGO;DESCRIPCION;LATITUD;LONGITUD;DIRECCIÓN;C.P.;POBLACION;PROVINCIA;PAIS - - try: - with urllib.request.urlopen(req) as response: - content = response.read() - data = csv.DictReader( - content.decode("utf-8").splitlines(), - delimiter=";", - fieldnames=[ - "CODE", - "NAME", - "LAT", - "LNG", - "ADDRESS", - "ZIP", - "CITY", - "PROVINCE", - "COUNTRY", - ], - ) - - stops = [row for row in data] - - print(f"Downloaded {len(stops)} stops") - - # Process the data - processed_stops = [] - for stop in stops: - if stop.get("PROVINCE") != "Pontevedra": - continue - - name = stop.get("NAME", "").strip() - - processed_stop = { - "stopId": "renfe:" + str(stop.get("CODE", 0)), - "name": name, - "latitude": float(stop.get("LAT", 0).replace(",", ".")), - "longitude": float(stop.get("LNG", 0).replace(",", ".")), - "lines": [], - } - processed_stops.append(processed_stop) - - print(f"Processed {len(processed_stops)} Renfe stops in Pontevedra") - return processed_stops - except Exception as e: - print(f"Error processing Pontevedra stops data: {e}", file=sys.stderr) - return [] - - -def main(): - print("Fetching stop list data...") - - vigo_stops = download_stops_vitrasa() - renfe_stops = download_stops_renfe() - - all_stops = vigo_stops + (renfe_stops if renfe_stops else []) - - try: - # Load and apply overrides - overrides_dir = os.path.join(SCRIPT_DIR, OVERRIDES_DIR) - # For each YML/YAML file in the overrides directory, load and apply the overrides - for filename in os.listdir(overrides_dir): - if not filename.endswith(".yml") and not filename.endswith(".yaml"): - continue - - print(f"Loading overrides from {filename}") - overrides_file = os.path.join(overrides_dir, filename) - overrides = load_stop_overrides(overrides_file) - all_stops = apply_overrides(all_stops, overrides) - - # Filter out hidden stops - visible_stops = [stop for stop in all_stops if not stop.get("hide")] - print(f"Removed {len(all_stops) - len(visible_stops)} hidden stops") - - # Sort stops by ID ascending - visible_stops.sort(key=lambda x: x["stopId"]) - - output_file = os.path.join(SCRIPT_DIR, OUTPUT_FILE) - - with open(output_file, "w", encoding="utf-8") as f: - json.dump(visible_stops, f, ensure_ascii=False, indent=2) - - print(f"Saved processed stops data to {output_file}") - return 0 - - except Exception as e: - print(f"Error processing stops data: {e}", file=sys.stderr) - # Print full exception traceback - import traceback - - traceback.print_exc() - return 1 - - -if __name__ == "__main__": - sys.exit(main()) |
