scripts/update_car_distances.py

   1 #!/usr/bin/python
   2 import argparse
   3 import configparser
   4 import re
   5 import sys
   6 from copy import deepcopy
   7 from datetime import timedelta
   8 from typing import List, NamedTuple, Optional
   9
  10 import isodate
  11 import jsonschema
  12 from osgeo import ogr
  13 from osgeo.ogr import Layer, Geometry, wkbPoint, Feature
  14 from osgeo.osr import SpatialReference, CoordinateTransformation, OAMS_TRADITIONAL_GIS_ORDER
  15
  16 from wrpylib.json_tools import order_json_keys, format_json
  17 from wrpylib.mwapi import WikiSite, page_json
  18 from wrpylib.vao import Vao
  19
  20
  21 class DistInfo(NamedTuple):
  22     city_name: str
  23     geoname_id: int
  24     duration_minutes: int
  25     dist_m: int
  26     co2_kg: float
  27
  28
  29 def vao_car_distance(vao: Vao, parking_lon: float, parking_lat: float, city: Feature) -> Optional[DistInfo]:
  30     geometry = city.GetGeometryRef()
  31     point = geometry.GetPoint(0)
  32     city_lon, city_lat, _ = point
  33     parameter = {
  34         'originCoordLat': city_lat,
  35         'originCoordLong': city_lon,
  36         'destCoordLat': parking_lat,
  37         'destCoordLong': parking_lon,
  38         'groupFilter': 'API_CAR',
  39         'totalCar': '1|evnt=0,aevnt=0,tsta=0,htsta=0,getInitEndTimes=0',
  40     }
  41     response = vao.trip(parameter).json()
  42     trip = response.get('Trip', [])
  43     if trip:
  44         leg_list = trip[0].get('LegList', {}).get('Leg', [])
  45         if leg_list:
  46             leg = leg_list[0]
  47             duration = isodate.parse_duration(leg['duration'])
  48             duration_minutes = int(round(duration / timedelta(minutes=1)))
  49             dist_m = leg['dist']
  50             co2_kg = trip[0]['Eco']['co2']
  51             return DistInfo(city['name'], city['geonames_id'], duration_minutes, dist_m, co2_kg)
  52     if response.get('errorCode') is not None:
  53         print(response)
  54         sys.exit(1)
  55     print('Unexpected result from VAO')
  56
  57
  58 def dist_info_to_dict(dist_info: DistInfo) -> dict:
  59     return {
  60         'km': round(dist_info.dist_m / 1000, 1),
  61         'route': dist_info.city_name,
  62         'minutes': dist_info.duration_minutes,
  63         'geonames_id': dist_info.geoname_id,
  64         'onward_co2_kg': round(dist_info.co2_kg, 1),
  65     }
  66
  67
  68 def update_sledrun(vao: Vao, db_cities: Layer, site: WikiSite, title: str):
  69     sledrun_json_page = site.query_page(f'{title}/Rodelbahn.json')
  70     sledrun_json = page_json(sledrun_json_page)
  71
  72     sledrun_json_orig = sledrun_json.copy()
  73
  74     car_parking = sledrun_json.get('car_parking')
  75     if not car_parking:
  76         print('  (no parking)')
  77         print('')
  78         return
  79
  80     parking = car_parking[0].get('position', {}).get('position')
  81     if not parking:
  82         return
  83     parking_lon = parking['longitude']
  84     parking_lat = parking['latitude']
  85
  86     car_distance_list = deepcopy(sledrun_json.get('car_distances', []))
  87     if len([car_distance for car_distance in car_distance_list if car_distance.get('geonames_id') is not None]) > 0:
  88         return
  89
  90     for car_distance in car_distance_list:
  91         if car_distance.get('geonames_id') is None:
  92             name = car_distance['route']
  93             match = re.match(r'([-\w. ]+)\(?.*$', name)
  94             if match is not None:
  95                 name = match.group(1).strip()
  96                 candidates = [city for city in db_cities if city['name'] == name]
  97                 if len(candidates) == 1:
  98                     city = candidates[0]
  99                     dist_info = vao_car_distance(vao, parking_lon, parking_lat, city)
 100                     if dist_info is not None:
 101                         dist_info_dict = dist_info_to_dict(dist_info)
 102                         car_distance.update(dist_info_dict)
 103
 104     spatial_reference_ll = SpatialReference()
 105     spatial_reference_ll.ImportFromEPSG(4326)
 106     spatial_reference_ll.SetAxisMappingStrategy(OAMS_TRADITIONAL_GIS_ORDER)
 107
 108     spatial_reference_m = SpatialReference()
 109     spatial_reference_m.ImportFromProj4(f'+proj=merc +lat_ts={parking_lat}')
 110     # spatial_reference_m.ImportFromProj4(f'+proj=merc')
 111
 112     ll_to_m = CoordinateTransformation(spatial_reference_ll, spatial_reference_m)
 113     m_to_ll = CoordinateTransformation(spatial_reference_m, spatial_reference_ll)
 114
 115     loc_ll = Geometry(wkbPoint)
 116     loc_ll.AddPoint(parking_lon, parking_lat)
 117
 118     loc_m = loc_ll.Clone()
 119     loc_m.Transform(ll_to_m)
 120     max_dist_m = 60000
 121     bound_m = loc_m.Buffer(max_dist_m)
 122     bound_ll = bound_m.Clone()
 123     bound_ll.Transform(m_to_ll)
 124
 125     db_cities.SetSpatialFilter(bound_ll)
 126     db_cities.SetAttributeFilter('level<=2')
 127
 128     car_distances_to_append: List[dict] = []
 129     for city in db_cities:
 130         print(city['name'])
 131         car_distance = next((cd for cd in car_distance_list if cd.get('geonames_id') == city['geonames_id']), None)
 132         if car_distance is None:
 133             car_distance = vao_car_distance(vao, parking_lon, parking_lat, city)
 134             if car_distance is not None:
 135                 car_distance = dist_info_to_dict(car_distance)
 136         if car_distance is not None:
 137             car_distances_to_append.append(car_distance)
 138     car_distances_to_append = sorted(car_distances_to_append, key=lambda di: di['km'])[:3]
 139
 140     for car_distance in car_distances_to_append:
 141         if len([cd for cd in car_distance_list if cd.get('geonames_id') == car_distance['geonames_id']]) == 0:
 142             car_distance_list.append(car_distance)
 143
 144     car_distance_list = sorted(car_distance_list, key=lambda di: di['km'])
 145     sledrun_json['car_distances'] = car_distance_list
 146
 147     if sledrun_json == sledrun_json_orig:
 148         return
 149
 150     jsonschema.validate(instance=sledrun_json, schema=site.sledrun_schema())
 151     sledrun_json_ordered = order_json_keys(sledrun_json, site.sledrun_schema())
 152     assert sledrun_json_ordered == sledrun_json
 153     sledrun_json_str = format_json(sledrun_json_ordered)
 154
 155     site(
 156         'edit',
 157         pageid=sledrun_json_page['pageid'],
 158         text=sledrun_json_str,
 159         summary=f'Entfernungen zu {title} aktualisiert (dank VAO).',
 160         # minor=1,
 161         bot=1,
 162         baserevid=sledrun_json_page['revisions'][0]['revid'],
 163         nocreate=1,
 164         token=site.token(),
 165     )
 166
 167
 168 def update_car_distances(ini_files: List[str]):
 169     ogr.UseExceptions()
 170
 171     config = configparser.ConfigParser()
 172     config.read(ini_files)
 173     host = config.get('mysql', 'host')
 174     dbname = config.get('mysql', 'dbname')
 175     user = config.get('mysql', 'user_name')
 176     passwd = config.get('mysql', 'user_pass')
 177
 178     cities_wr_source = ogr.Open(f'MySQL:{dbname},"host={host}","user={user}","password={passwd}","tables=wrcity"')
 179     db_cities = cities_wr_source.GetLayerByIndex(0)
 180
 181     site = WikiSite(ini_files)
 182     vao = Vao(config.get('vao', 'access_id'))
 183     for result in site.query(list='categorymembers', cmtitle='Kategorie:Rodelbahn', cmlimit='max'):
 184         for page in result['categorymembers']:
 185             print(page['title'])
 186             update_sledrun(vao, db_cities, site, page['title'])
 187
 188
 189 def main():
 190     parser = argparse.ArgumentParser(description='Update car distance information in sledrun JSON files.')
 191     parser.add_argument('inifile', nargs='+', help='inifile.ini, see: https://www.winterrodeln.org/trac/wiki/ConfigIni')
 192     args = parser.parse_args()
 193     update_car_distances(args.inifile)
 194
 195
 196 if __name__ == '__main__':
 197     main()