Create script to update the public transport stops.

[philipp/winterrodeln/wrpylib.git] / scripts / update_public_transport_bus_stops.py

diff --git a/scripts/update_public_transport_bus_stops.py b/scripts/update_public_transport_bus_stops.py
new file mode 100644 (file)
index 0000000..d11e50c
--- /dev/null
+++ b/scripts/update_public_transport_bus_stops.py
@@ -0,0 +1,117 @@
+#!/usr/bin/python
+import argparse
+import re
+import sys
+from copy import deepcopy
+from typing import List, Iterable, Optional
+
+import geojson
+import jsonschema
+from geojson import GeoJSON
+from pyproj import CRS, Geod
+from termcolor import cprint  # python3-termcolor
+
+from wrpylib.cli_tools import unified_diff, input_yes_no_quit, Choice
+from wrpylib.json_tools import order_json_keys, format_json
+from wrpylib.mwapi import WikiSite, page_json
+from wrpylib.sledrun_json import Sledrun, Position, PublicTransportStop
+
+
+def point_feature_distance(geod: Geod, position: Position, feature: GeoJSON) -> float:
+    return geod.line_length(
+        [position['longitude'], feature['geometry']['coordinates'][0]],
+        [position['latitude'], feature['geometry']['coordinates'][1]])
+
+
+def update_sledrun(site: WikiSite, bus_stop_geojson: GeoJSON, title: str):
+    cprint(title, 'green')
+    sledrun_json_page = site.query_page(f'{title}/Rodelbahn.json')
+    sledrun: Sledrun = page_json(sledrun_json_page)
+    sledrun_orig = deepcopy(sledrun)
+
+    for pt_stop in sledrun.get('public_transport_stops', []):
+        pt_stop: PublicTransportStop = pt_stop
+        if 'ifopt_stop_id' in pt_stop:  # e.g. "at:47:61646"
+            continue
+        if 'vvt_stop_id' in pt_stop:  # e.g. 61646 -> "at:47:61646"
+            pt_stop['ifopt_stop_id'] = f'at:47:{pt_stop["vvt_stop_id"]}'
+            continue
+        if 'vao_ext_id' in pt_stop:  # e.g. '476164600' -> "at:47:61646"
+            if match := re.match(r'(47)(\d{5})00', pt_stop['vao_ext_id']):
+                g1, g2 = match.groups()
+                pt_stop['ifopt_stop_id'] = f"at:{g1}:{g2}"
+                continue
+        if position_elevation := pt_stop.get('position'):
+            if position := position_elevation.get('position'):
+                bus_stop_feature_list = bus_stop_geojson['features']
+                geod = CRS("EPSG:4326").get_geod()
+                closest_bus_stop_feature = min(bus_stop_feature_list,
+                                               key=lambda f: point_feature_distance(geod, position, f))
+                distance_m = point_feature_distance(geod, position, closest_bus_stop_feature)
+                if distance_m < 30:
+                    name1 = pt_stop["name"] if "name" in pt_stop else pt_stop.get("name_local", "(unnamed stop)")
+                    name2 = closest_bus_stop_feature['properties']["hst_name"]
+                    choice = input_yes_no_quit(f'Assign "{name1}" to "{name2}" [yes, no, quit]?', None)
+                    if choice == Choice.no:
+                        return
+                    elif choice == Choice.quit:
+                        sys.exit(0)
+
+                    pt_stop['ifopt_stop_id'] = closest_bus_stop_feature['properties']['hst_globid']
+
+    if sledrun == sledrun_orig:
+        return
+
+    jsonschema.validate(instance=sledrun, schema=site.sledrun_schema())
+    sledrun_ordered = order_json_keys(sledrun, site.sledrun_schema())
+    assert sledrun_ordered == sledrun
+    sledrun_orig_str = format_json(sledrun_orig)
+    sledrun_str = format_json(sledrun_ordered)
+
+    unified_diff(sledrun_orig_str, sledrun_str)
+    choice = input_yes_no_quit('Do you accept the changes [yes, no, quit]? ', None)
+    if choice == Choice.no:
+        return
+    elif choice == Choice.quit:
+        sys.exit(0)
+
+    site(
+        'edit',
+        pageid=sledrun_json_page['pageid'],
+        text=sledrun_str,
+        summary='IFOPT Nummer zu Haltestellen ergänzt.',
+        bot=1,
+        baserevid=sledrun_json_page['revisions'][0]['revid'],
+        nocreate=1,
+        token=site.token(),
+    )
+
+
+def get_all_sledrun_titles(site: WikiSite) -> Iterable[str]:
+    for result in site.query(list='categorymembers', cmtitle='Kategorie:Rodelbahn', cmlimit='max'):
+        for page in result['categorymembers']:
+            yield page['title']
+
+def update_public_transport_bus_stops(ini_files: List[str], bus_stop_file: str, sledrun_title: Optional[str]):
+    with open(bus_stop_file) as fp:
+        bus_stop_geojson = geojson.load(fp)
+
+    site = WikiSite(ini_files)
+    if sledrun_title is None:
+        for sledrun_title in get_all_sledrun_titles(site):
+            update_sledrun(site, bus_stop_geojson, sledrun_title)
+    else:
+        update_sledrun(site, bus_stop_geojson, sledrun_title)
+
+
+def main():
+    parser = argparse.ArgumentParser(description='Update public transport bus stop information in sledrun JSON files.')
+    parser.add_argument('--sledrun', help='If given, work on a single sled run page, otherwise at the whole category.')
+    parser.add_argument('bus_stop_file', help='GeoJSON file with bus stops.')
+    parser.add_argument('inifile', nargs='+', help='inifile.ini, see: https://www.winterrodeln.org/trac/wiki/ConfigIni')
+    args = parser.parse_args()
+    update_public_transport_bus_stops(args.inifile, args.bus_stop_file, args.sledrun)
+
+
+if __name__ == '__main__':
+    main()