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 position_elevation := pt_stop.get('position'):
  45             if position := position_elevation.get('position'):
  46                 bus_stop_feature_list = bus_stop_geojson['features']
  47                 geod = CRS("EPSG:4326").get_geod()
  48                 closest_bus_stop_feature = min(bus_stop_feature_list,
  49                                                key=lambda f: point_feature_distance(geod, position, f))
  50                 distance_m = point_feature_distance(geod, position, closest_bus_stop_feature)
  51                 if distance_m < 30:
  52                     name1 = pt_stop["name"] if "name" in pt_stop else pt_stop.get("name_local", "(unnamed stop)")
  53                     name2 = closest_bus_stop_feature['properties']["hst_name"]
  54                     choice = input_yes_no_quit(f'Assign "{name1}" to "{name2}" [yes, no, quit]?', None)
  55                     if choice == Choice.no:
  56                         return
  57                     elif choice == Choice.quit:
  58                         sys.exit(0)
  59
  60                     pt_stop['ifopt_stop_id'] = closest_bus_stop_feature['properties']['hst_globid']
  61
  62     if sledrun == sledrun_orig:
  63         return
  64
  65     jsonschema.validate(instance=sledrun, schema=site.sledrun_schema())
  66     sledrun_ordered = order_json_keys(sledrun, site.sledrun_schema())
  67     assert sledrun_ordered == sledrun
  68     sledrun_orig_str = format_json(sledrun_orig)
  69     sledrun_str = format_json(sledrun_ordered)
  70
  71     unified_diff(sledrun_orig_str, sledrun_str)
  72     choice = input_yes_no_quit('Do you accept the changes [yes, no, quit]? ', None)
  73     if choice == Choice.no:
  74         return
  75     elif choice == Choice.quit:
  76         sys.exit(0)
  77
  78     site(
  79         'edit',
  80         pageid=sledrun_json_page['pageid'],
  81         text=sledrun_str,
  82         summary='IFOPT Nummer zu Haltestellen ergänzt.',
  83         bot=1,
  84         baserevid=sledrun_json_page['revisions'][0]['revid'],
  85         nocreate=1,
  86         token=site.token(),
  87     )
  88
  89
  90 def get_all_sledrun_titles(site: WikiSite) -> Iterable[str]:
  91     for result in site.query(list='categorymembers', cmtitle='Kategorie:Rodelbahn', cmlimit='max'):
  92         for page in result['categorymembers']:
  93             yield page['title']
  94
  95 def update_public_transport_bus_stops(ini_files: List[str], bus_stop_file: str, sledrun_title: Optional[str]):
  96     with open(bus_stop_file) as fp:
  97         bus_stop_geojson = geojson.load(fp)
  98
  99     site = WikiSite(ini_files)
 100     if sledrun_title is None:
 101         for sledrun_title in get_all_sledrun_titles(site):
 102             update_sledrun(site, bus_stop_geojson, sledrun_title)
 103     else:
 104         update_sledrun(site, bus_stop_geojson, sledrun_title)
 105
 106
 107 def main():
 108     parser = argparse.ArgumentParser(description='Update public transport bus stop information in sledrun JSON files.')
 109     parser.add_argument('--sledrun', help='If given, work on a single sled run page, otherwise at the whole category.')
 110     parser.add_argument('bus_stop_file', help='GeoJSON file with bus stops.')
 111     parser.add_argument('inifile', nargs='+', help='inifile.ini, see: https://www.winterrodeln.org/trac/wiki/ConfigIni')
 112     args = parser.parse_args()
 113     update_public_transport_bus_stops(args.inifile, args.bus_stop_file, args.sledrun)
 114
 115
 116 if __name__ == '__main__':
 117     main()