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