#!/usr/bin/python import argparse import configparser import re import sys 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 def vao_car_distance(vao: Vao, parking_lon: float, parking_lat: float, city: Feature) -> Optional[DistInfo]: 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) if response.get('errorCode') is not None: print(response) sys.exit(1) print('Unexpected result from VAO') 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 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}') # spatial_reference_m.ImportFromProj4(f'+proj=merc') ll_to_m = CoordinateTransformation(spatial_reference_ll, spatial_reference_m) m_to_ll = CoordinateTransformation(spatial_reference_m, spatial_reference_ll) loc_ll = Geometry(wkbPoint) loc_ll.AddPoint(parking_lon, parking_lat) loc_m = loc_ll.Clone() loc_m.Transform(ll_to_m) max_dist_m = 60000 bound_m = loc_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') car_distances_to_append: List[dict] = [] for city in db_cities: 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'])[:3] 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']) 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()