#!/usr/bin/python import argparse import configparser import re import sys import time from copy import deepcopy from datetime import timedelta from typing import List, NamedTuple, Optional import isodate import jsonschema from osgeo import ogr from osgeo.ogr import Layer, Geometry, wkbPoint, Feature from osgeo.osr import SpatialReference, CoordinateTransformation, OAMS_TRADITIONAL_GIS_ORDER from wrpylib.json_tools import order_json_keys, format_json from wrpylib.mwapi import WikiSite, page_json from wrpylib.vao import Vao class DistInfo(NamedTuple): city_name: str geoname_id: int duration_minutes: int dist_m: int co2_kg: float class VaoError(RuntimeError): pass def try_vao_car_distance(vao: Vao, parking_lon: float, parking_lat: float, city: Feature) -> DistInfo: """may throw VaoError with JSON decoded response as argument""" geometry = city.GetGeometryRef() point = geometry.GetPoint(0) city_lon, city_lat, _ = point parameter = { 'originCoordLat': city_lat, 'originCoordLong': city_lon, 'destCoordLat': parking_lat, 'destCoordLong': parking_lon, 'groupFilter': 'API_CAR', 'totalCar': '1|evnt=0,aevnt=0,tsta=0,htsta=0,getInitEndTimes=0', } response = vao.trip(parameter).json() trip = response.get('Trip', []) if trip: leg_list = trip[0].get('LegList', {}).get('Leg', []) if leg_list: leg = leg_list[0] duration = isodate.parse_duration(leg['duration']) duration_minutes = int(round(duration / timedelta(minutes=1))) dist_m = leg['dist'] co2_kg = trip[0]['Eco']['co2'] return DistInfo(city['name'], city['geonames_id'], duration_minutes, dist_m, co2_kg) raise VaoError(response) def vao_car_distance(vao: Vao, parking_lon: float, parking_lat: float, city: Feature, retry_count: int = 2) -> DistInfo: for c in range(retry_count): try: return try_vao_car_distance(vao, parking_lon, parking_lat, city) except VaoError as vao_error: response = vao_error.args[0] if response.get('errorCode') is not None: print(response['errorCode'], response.get('errorText'), f'(attempt {c+1}/{retry_count})') if response['errorCode'] == 'SVC_NO_RESULT': time.sleep(2.) continue else: print('Unexpected result from VAO') sys.exit(1) def distance_meter(a: Geometry, b: Geometry) -> float: spatial_reference_ll = SpatialReference() spatial_reference_ll.ImportFromEPSG(4326) spatial_reference_ll.SetAxisMappingStrategy(OAMS_TRADITIONAL_GIS_ORDER) spatial_reference_m = SpatialReference() spatial_reference_m.ImportFromProj4(f'+proj=merc +lat_ts={(a.GetY() + b.GetY()) / 2}') ll_to_m = CoordinateTransformation(spatial_reference_ll, spatial_reference_m) a_m = a.Clone() a_m.Transform(ll_to_m) b_m = b.Clone() b_m.Transform(ll_to_m) return a_m.Distance(b_m) def dist_info_to_dict(dist_info: DistInfo) -> dict: return { 'km': round(dist_info.dist_m / 1000, 1), 'route': dist_info.city_name, 'minutes': dist_info.duration_minutes, 'geonames_id': dist_info.geoname_id, 'onward_co2_kg': round(dist_info.co2_kg, 1), } def update_sledrun(vao: Vao, db_cities: Layer, site: WikiSite, title: str): sledrun_json_page = site.query_page(f'{title}/Rodelbahn.json') sledrun_json = page_json(sledrun_json_page) sledrun_json_orig = sledrun_json.copy() car_parking = sledrun_json.get('car_parking') if not car_parking: print(' (no parking)') print('') return parking = car_parking[0].get('position', {}).get('position') if not parking: return parking_lon = parking['longitude'] parking_lat = parking['latitude'] car_distance_list = deepcopy(sledrun_json.get('car_distances', [])) if len([car_distance for car_distance in car_distance_list if car_distance.get('geonames_id') is not None]) > 0: return db_cities.SetSpatialFilter(None) db_cities.SetAttributeFilter(None) for car_distance in car_distance_list: if car_distance.get('geonames_id') is None: name = car_distance['route'] match = re.match(r'([-\w. ]+)\(?.*$', name) if match is not None: name = match.group(1).strip() candidates = [city for city in db_cities if city['name'] == name] if len(candidates) == 1: city = candidates[0] dist_info = vao_car_distance(vao, parking_lon, parking_lat, city) if dist_info is not None: dist_info_dict = dist_info_to_dict(dist_info) car_distance.update(dist_info_dict) spatial_reference_ll = SpatialReference() spatial_reference_ll.ImportFromEPSG(4326) spatial_reference_ll.SetAxisMappingStrategy(OAMS_TRADITIONAL_GIS_ORDER) spatial_reference_m = SpatialReference() spatial_reference_m.ImportFromProj4(f'+proj=merc +lat_ts={parking_lat}') ll_to_m = CoordinateTransformation(spatial_reference_ll, spatial_reference_m) m_to_ll = CoordinateTransformation(spatial_reference_m, spatial_reference_ll) parking_ll = Geometry(wkbPoint) parking_ll.AddPoint(parking_lon, parking_lat) parking_m = parking_ll.Clone() parking_m.Transform(ll_to_m) max_dist_m = 60000 bound_m = parking_m.Buffer(max_dist_m) bound_ll = bound_m.Clone() bound_ll.Transform(m_to_ll) db_cities.SetSpatialFilter(bound_ll) db_cities.SetAttributeFilter('level<=2') close_cities = [city for city in db_cities] close_cities = sorted(close_cities, key=lambda city: distance_meter(city.GetGeometryRef(), parking_ll)) max_number_distances = 3 car_distances_to_append: List[dict] = [] for city in close_cities: if len(car_distances_to_append) >= max_number_distances: if car_distances_to_append[2]['km'] * 1000 < distance_meter(city.GetGeometryRef(), parking_ll): break print(city['name']) car_distance = next((cd for cd in car_distance_list if cd.get('geonames_id') == city['geonames_id']), None) if car_distance is None: car_distance = vao_car_distance(vao, parking_lon, parking_lat, city) if car_distance is not None: car_distance = dist_info_to_dict(car_distance) if car_distance is not None: car_distances_to_append.append(car_distance) car_distances_to_append = sorted(car_distances_to_append, key=lambda di: di['km']) car_distances_to_append = sorted(car_distances_to_append, key=lambda di: di['km'])[:max_number_distances] for car_distance in car_distances_to_append: if len([cd for cd in car_distance_list if cd.get('geonames_id') == car_distance['geonames_id']]) == 0: car_distance_list.append(car_distance) car_distance_list = sorted(car_distance_list, key=lambda di: di['km']) sledrun_json['car_distances'] = car_distance_list if sledrun_json == sledrun_json_orig: return jsonschema.validate(instance=sledrun_json, schema=site.sledrun_schema()) sledrun_json_ordered = order_json_keys(sledrun_json, site.sledrun_schema()) assert sledrun_json_ordered == sledrun_json sledrun_json_str = format_json(sledrun_json_ordered) site( 'edit', pageid=sledrun_json_page['pageid'], text=sledrun_json_str, summary=f'Entfernungen zu {title} aktualisiert (dank VAO).', # minor=1, bot=1, baserevid=sledrun_json_page['revisions'][0]['revid'], nocreate=1, token=site.token(), ) def update_car_distances(ini_files: List[str]): ogr.UseExceptions() config = configparser.ConfigParser() config.read(ini_files) host = config.get('mysql', 'host') dbname = config.get('mysql', 'dbname') user = config.get('mysql', 'user_name') passwd = config.get('mysql', 'user_pass') cities_wr_source = ogr.Open(f'MySQL:{dbname},"host={host}","user={user}","password={passwd}","tables=wrcity"') db_cities = cities_wr_source.GetLayerByIndex(0) site = WikiSite(ini_files) vao = Vao(config.get('vao', 'access_id')) for result in site.query(list='categorymembers', cmtitle='Kategorie:Rodelbahn', cmlimit='max'): for page in result['categorymembers']: print(page['title']) if page['title'] in ['Anzère', 'Hochhäderich (Falkenhütte)', 'Hochlitten-Moosalpe', 'Saas-Fee']: continue update_sledrun(vao, db_cities, site, page['title']) def main(): parser = argparse.ArgumentParser(description='Update car distance information in sledrun JSON files.') parser.add_argument('inifile', nargs='+', help='inifile.ini, see: https://www.winterrodeln.org/trac/wiki/ConfigIni') args = parser.parse_args() update_car_distances(args.inifile) if __name__ == '__main__': main()