scripts/update_public_transport_bus_stops.py

   1 #!/usr/bin/python
   2 import argparse
   3 import re
   4 import sys
   5 from copy import deepcopy
   6 from typing import List, Iterable, Optional
   7
   8 import geojson
   9 import jsonschema
  10 from geojson import GeoJSON
  11 from pyproj import CRS, Geod
  12 from termcolor import cprint  # python3-termcolor
  13
  14 from wrpylib.cli_tools import unified_diff, input_yes_no_quit, Choice
  15 from wrpylib.json_tools import order_json_keys, format_json
  16 from wrpylib.mwapi import WikiSite, page_json
  17 from wrpylib.sledrun_json import Sledrun, Position, PublicTransportStop
  18
  19
  20 def point_feature_distance(geod: Geod, position: Position, feature: GeoJSON) -> float:
  21     return geod.line_length(
  22         [position['longitude'], feature['geometry']['coordinates'][0]],
  23         [position['latitude'], feature['geometry']['coordinates'][1]])
  24
  25
  26 def update_sledrun(site: WikiSite, bus_stop_geojson: GeoJSON, title: str):
  27     cprint(title, 'green')
  28     sledrun_json_page = site.query_page(f'{title}/Rodelbahn.json')
  29     sledrun: Sledrun = page_json(sledrun_json_page)
  30     sledrun_orig = deepcopy(sledrun)
  31
  32     for pt_stop in sledrun.get('public_transport_stops', []):
  33         pt_stop: PublicTransportStop = pt_stop
  34         if 'ifopt_stop_id' in pt_stop:  # e.g. "at:47:61646"
  35             continue
  36         if 'vvt_stop_id' in pt_stop:  # e.g. 61646 -> "at:47:61646"
  37             pt_stop['ifopt_stop_id'] = f'at:47:{pt_stop["vvt_stop_id"]}'
  38             continue
  39         if 'vao_ext_id' in pt_stop:  # e.g. '476164600' -> "at:47:61646"
  40             if match := re.match(r'(47)(\d{5})00', pt_stop['vao_ext_id']):
  41                 g1, g2 = match.groups()
  42                 pt_stop['ifopt_stop_id'] = f"at:{g1}:{g2}"
  43                 continue
  44         if monitor_template := pt_stop.get('monitor_template'):
  45             if monitor_template.get('name') == "Fahrplan Abfahrtsmonitor VVT":
  46                 vvt_stop_id = int(monitor_template['parameter'][2]['value'])
  47                 pt_stop['ifopt_stop_id'] = f'at:47:{vvt_stop_id}'
  48                 continue
  49         if position_elevation := pt_stop.get('position'):
  50             if position := position_elevation.get('position'):
  51                 bus_stop_feature_list = bus_stop_geojson['features']
  52                 geod = CRS("EPSG:4326").get_geod()
  53                 closest_bus_stop_feature = min(bus_stop_feature_list,
  54                                                key=lambda f: point_feature_distance(geod, position, f))
  55                 distance_m = point_feature_distance(geod, position, closest_bus_stop_feature)
  56                 if distance_m < 30:
  57                     name1 = pt_stop["name"] if "name" in pt_stop else pt_stop.get("name_local", "(unnamed stop)")
  58                     name2 = closest_bus_stop_feature['properties']["hst_name"]
  59                     choice = input_yes_no_quit(f'Assign "{name1}" to "{name2}" [yes, no, quit]? ', None)
  60                     if choice == Choice.no:
  61                         return
  62                     elif choice == Choice.quit:
  63                         sys.exit(0)
  64
  65                     pt_stop['ifopt_stop_id'] = closest_bus_stop_feature['properties']['hst_globid']
  66
  67     if sledrun == sledrun_orig:
  68         return
  69
  70     jsonschema.validate(instance=sledrun, schema=site.sledrun_schema())
  71     sledrun_ordered = order_json_keys(sledrun, site.sledrun_schema())
  72     assert sledrun_ordered == sledrun
  73     sledrun_orig_str = format_json(sledrun_orig)
  74     sledrun_str = format_json(sledrun_ordered)
  75
  76     unified_diff(sledrun_orig_str, sledrun_str)
  77     choice = input_yes_no_quit('Do you accept the changes [yes, no, quit]? ', None)
  78     if choice == Choice.no:
  79         return
  80     elif choice == Choice.quit:
  81         sys.exit(0)
  82
  83     site(
  84         'edit',
  85         pageid=sledrun_json_page['pageid'],
  86         text=sledrun_str,
  87         summary='IFOPT Nummer zu Haltestellen ergänzt.',
  88         bot=1,
  89         baserevid=sledrun_json_page['revisions'][0]['revid'],
  90         nocreate=1,
  91         token=site.token(),
  92     )
  93
  94
  95 def get_all_sledrun_titles(site: WikiSite) -> Iterable[str]:
  96     for result in site.query(list='categorymembers', cmtitle='Kategorie:Rodelbahn', cmlimit='max'):
  97         for page in result['categorymembers']:
  98             yield page['title']
  99
 100 def update_public_transport_bus_stops(ini_files: List[str], bus_stop_file: str, sledrun_title: Optional[str]):
 101     with open(bus_stop_file) as fp:
 102         bus_stop_geojson = geojson.load(fp)
 103
 104     site = WikiSite(ini_files)
 105     if sledrun_title is None:
 106         for sledrun_title in get_all_sledrun_titles(site):
 107             update_sledrun(site, bus_stop_geojson, sledrun_title)
 108     else:
 109         update_sledrun(site, bus_stop_geojson, sledrun_title)
 110
 111
 112 def main():
 113     parser = argparse.ArgumentParser(description='Update public transport bus stop information in sledrun JSON files.')
 114     parser.add_argument('--sledrun', help='If given, work on a single sled run page, otherwise at the whole category.')
 115     parser.add_argument('bus_stop_file', help='GeoJSON file with bus stops.')
 116     parser.add_argument('inifile', nargs='+', help='inifile.ini, see: https://www.winterrodeln.org/trac/wiki/ConfigIni')
 117     args = parser.parse_args()
 118     update_public_transport_bus_stops(args.inifile, args.bus_stop_file, args.sledrun)
 119
 120
 121 if __name__ == '__main__':
 122     main()