scripts/update_public_transport.py

   1 #!/usr/bin/python
   2 import argparse
   3 import configparser
   4 import re
   5 import sys
   6 from dataclasses import dataclass
   7 from datetime import date, timedelta, datetime
   8 from itertools import groupby
   9 from typing import List, Iterable, Optional, Set, Tuple, Dict
  10
  11 import isodate
  12 import jsonschema
  13 from termcolor import cprint  # python3-termcolor
  14
  15 from wrpylib.cli_tools import unified_diff, input_yes_no_quit, Choice
  16 from wrpylib.json_tools import order_json_keys, format_json
  17 from wrpylib.lib_update_public_transport import default_query_date, vao_ext_id_to_ifopt_stop_id
  18 from wrpylib.mwapi import WikiSite, page_json
  19 from wrpylib.vao import Vao
  20 from wrpylib.wrvalidators import LonLat
  21
  22
  23 class VaoError(RuntimeError):
  24     pass
  25
  26
  27 @dataclass
  28 class StopInfo:
  29     name: str
  30     dist_m: int  # Luftlinie
  31     lon_lat: LonLat
  32     line_ids: Set[str]
  33     ext_id: str
  34
  35
  36 @dataclass
  37 class LineInfo:
  38     vao_line_id: str  # e.g. 'vvt-8-325-T-j23-1'
  39     line: Optional[str]  # e.g. '325T', some trains don't have line labels.
  40     category: str  # e.g. 'Anrufsammeltaxi'
  41     operator: str  # e.g. 'Quaxis Taxi & Busreisen'
  42
  43
  44 def get_vvt_stop_id(ext_id: str) -> Optional[int]:
  45     vvt_stop_id_match = re.match(r'^47(\d{5})00$', ext_id)
  46     if vvt_stop_id_match:
  47         return int(vvt_stop_id_match.group(1))
  48     return None
  49
  50
  51 def try_vao_nearby_stops(vao: Vao, lon_lat: LonLat) -> Tuple[List[StopInfo], Dict[str, LineInfo]]:
  52     """may throw VaoError with JSON decoded response as argument"""
  53     parameter = {
  54         'originCoordLat': lon_lat.lat,
  55         'originCoordLong': lon_lat.lon,
  56     }
  57     response = vao.get('location.nearbystops', parameter).json()
  58     stop_info_list: List[StopInfo] = []
  59     line_info_dict: Dict[str, LineInfo] = {}
  60     for stop in response.get('stopLocationOrCoordLocation', []):
  61         sl = stop['StopLocation']
  62         line_ids = set()
  63         for ps in sl['productAtStop']:
  64             vao_line_id = ps['lineId']
  65             line_ids.add(vao_line_id)
  66             line_info_dict[vao_line_id] = LineInfo(
  67                 vao_line_id=vao_line_id,
  68                 line=ps['line'],
  69                 category=ps['catOutL'],
  70                 operator=ps['operator'],
  71             )
  72         ext_id = sl['mainMastExtId']
  73         stop_info_list.append(StopInfo(sl['name'], sl['dist'], LonLat(sl['mainMastLon'], sl['mainMastLat']), line_ids, ext_id))
  74     return stop_info_list, line_info_dict
  75
  76
  77 def merge_stops(stops: List[StopInfo]) -> StopInfo:
  78     mean_dist = int(round(sum(stop.dist_m for stop in stops) / len(stops)))
  79     all_lines = set().union(*[stop.line_ids for stop in stops])
  80     stop = stops[0]
  81     return StopInfo(stop.name, mean_dist, stop.lon_lat, all_lines, stop.ext_id)
  82
  83
  84 def unique_nearby_stops(stops: Iterable[StopInfo]) -> List[StopInfo]:
  85     stops = sorted(stops, key=lambda stop: stop.ext_id)
  86     stops = [merge_stops(list(i)) for k, i in groupby(stops, key=lambda stop: stop.ext_id)]
  87     return sorted(stops, key=lambda stop: stop.dist_m)
  88
  89
  90 @dataclass
  91 class WalkDistInfo:
  92     dist_m: int
  93     time_min: int
  94
  95
  96 def try_vao_walk_distance(vao: Vao, origin: LonLat, dest: LonLat) -> WalkDistInfo:
  97     """may throw VaoError with JSON decoded response as argument"""
  98     parameter = {
  99         'originCoordLat': origin.lat,
 100         'originCoordLong': origin.lon,
 101         'destCoordLat': dest.lat,
 102         'destCoordLong': dest.lon,
 103         'groupFilter': 'API_WALK',
 104     }
 105     response = vao.trip(parameter).json()
 106     trip = response.get('Trip', [])
 107     if trip:
 108         leg_list = trip[0].get('LegList', {}).get('Leg', [])
 109         if leg_list:
 110             leg = leg_list[0]
 111             duration = isodate.parse_duration(leg['duration'])
 112             duration_minutes = int(round(duration / timedelta(minutes=1)))
 113             dist_m = leg['dist']
 114             return WalkDistInfo(dist_m, duration_minutes)
 115     raise VaoError(response)
 116
 117
 118 @dataclass
 119 class Departure:
 120     vao_line_id: str  # e.g. 'vvt-8-325-T-j23-1'
 121     time: datetime
 122     direction: str  # e.g. 'Innsbruck Hauptbahnhof'
 123     flag: str  # e.g. 'H' or 'R'
 124     detail_ref: str
 125
 126
 127 def try_vao_departure_board(vao: Vao, ext_id: str, journey_date: datetime, journey_minutes: int,
 128                             line_info_dict: Dict[str, LineInfo]) -> List[Departure]:
 129     parameter = {
 130         'id': ext_id,
 131         'date': journey_date.date().isoformat(),
 132         'time': journey_date.time().isoformat('minutes'),
 133         'duration': journey_minutes,
 134     }
 135     response = vao.get('departureBoard', parameter).json()
 136     departure_list = []
 137     for departure in response.get('Departure', []):
 138         ps = departure['ProductAtStop']
 139         time = datetime.fromisoformat(f'{departure["date"]}T{departure["time"]}')
 140         vao_line_id = ps['lineId']
 141         departure_list.append(Departure(vao_line_id, time, departure['direction'],
 142                               departure['directionFlag'], departure['JourneyDetailRef']['ref']))
 143         line_info_dict[vao_line_id] = LineInfo(
 144             vao_line_id=vao_line_id,
 145             line=ps['line'],
 146             category=ps['catOutL'],
 147             operator=ps['operator'],
 148         )
 149     return departure_list
 150
 151
 152 @dataclass
 153 class Arrival:
 154     vao_line_id: str  # e.g. 'vvt-8-325-T-j23-1'
 155     time: datetime
 156     origin: str  # e.g. 'Innsbruck Hauptbahnhof'
 157     detail_ref: str
 158
 159
 160 def try_vao_arrival_board(vao: Vao, ext_id: str, journey_date: datetime, journey_minutes: int,
 161                           line_info_dict: Dict[str, LineInfo]) -> List[Arrival]:
 162     parameter = {
 163         'id': ext_id,
 164         'date': journey_date.date().isoformat(),
 165         'time': journey_date.time().isoformat('minutes'),
 166         'duration': journey_minutes,
 167     }
 168     response = vao.get('arrivalBoard', parameter).json()
 169     arrival_list = []
 170     for arrival in response.get('Arrival', []):
 171         ps = arrival['ProductAtStop']
 172         time = datetime.fromisoformat(f'{arrival["date"]}T{arrival["time"]}')
 173         vao_line_id = ps['lineId']
 174         arrival_list.append(Arrival(vao_line_id, time, arrival['origin'], arrival['JourneyDetailRef']['ref']))
 175         line_info_dict[vao_line_id] = LineInfo(
 176             vao_line_id=vao_line_id,
 177             line=ps['line'],
 178             category=ps['catOutL'],
 179             operator=ps['operator'],
 180         )
 181     return arrival_list
 182
 183
 184 def decode_service_days(value: str, begin: date) -> Iterable[date]:
 185     days_of_service = ''.join(f'{int(i, 16):04b}' for i in value)
 186     days_of_service = [bool(int(i)) for i in days_of_service]
 187     for i, day_has_service in enumerate(days_of_service):
 188         if day_has_service:
 189             yield begin + timedelta(days=i)
 190
 191
 192 @dataclass
 193 class JourneyDetail:
 194     planning_period_begin: date
 195     planning_period_end: date
 196     day_of_operation: date
 197     s_days_r: str
 198     s_days_i: str
 199     s_days_b: str
 200     service_days: List[date]
 201     first_stop_name: str
 202     last_stop_name: str
 203
 204
 205 def try_vao_journey_detail(vao: Vao, journey_detail_ref: str) -> JourneyDetail:
 206     parameter = {
 207         'id': journey_detail_ref,
 208         'rtMode': 'OFF'
 209     }
 210     response = vao.get('journeyDetail', parameter).json()
 211     sd = response['ServiceDays'][0]
 212     stops = response['Stops']['Stop']
 213     planning_period_begin = date.fromisoformat(sd['planningPeriodBegin'])
 214     s_days_b = sd['sDaysB']
 215     service_days = list(decode_service_days(s_days_b, planning_period_begin))
 216     return JourneyDetail(planning_period_begin, date.fromisoformat(sd['planningPeriodEnd']),
 217                          response['dayOfOperation'], sd['sDaysR'], sd['sDaysI'], s_days_b,
 218                          service_days, stops[0]['name'], stops[-1]['name'])
 219
 220
 221 @dataclass
 222 class StopWithDist:
 223     stop: StopInfo
 224     dist: WalkDistInfo
 225
 226
 227 def is_redundant_stop(stop: StopWithDist, stops: List[StopWithDist]) -> bool:
 228     for s in stops:
 229         assert s != stop
 230         if stop.stop.line_ids.issubset(s.stop.line_ids):
 231             return True
 232     return False
 233
 234
 235 def remove_redundant_stops(stops: List[StopWithDist]) -> List[StopWithDist]:
 236     stops = stops.copy()
 237     result = []
 238     while stops:
 239         stop = stops.pop()
 240         if not is_redundant_stop(stop, stops):
 241             result.insert(0, stop)
 242     return result
 243
 244
 245 def update_sledrun(vao: Vao, site: WikiSite, title: str, query_date: date):
 246     sledrun_json_page = site.query_page(f'{title}/Rodelbahn.json')
 247     sledrun_json = page_json(sledrun_json_page)
 248
 249     sledrun_json_orig = sledrun_json.copy()
 250
 251     pos = sledrun_json['bottom']['position']
 252     lon_lat = LonLat(pos['longitude'], pos['latitude'])
 253     nearby_stops, line_info_dict = try_vao_nearby_stops(vao, lon_lat)
 254     nearby_stops = unique_nearby_stops(nearby_stops)
 255     stops_with_dists = [StopWithDist(stop, try_vao_walk_distance(vao, stop.lon_lat, lon_lat)) for stop in nearby_stops]
 256     stops_with_dists = sorted(stops_with_dists, key=lambda s: s.dist.dist_m)
 257     stops_with_dists = remove_redundant_stops(stops_with_dists)
 258
 259     journey_date = datetime(query_date.year, query_date.month, query_date.day)
 260     journey_minutes = 1439
 261
 262     public_transport_stops = []
 263     for stop_with_dist in stops_with_dists:
 264         departures = try_vao_departure_board(vao, stop_with_dist.stop.ext_id, journey_date, journey_minutes,
 265                                              line_info_dict)
 266         arrivals = try_vao_arrival_board(vao, stop_with_dist.stop.ext_id, journey_date, journey_minutes, line_info_dict)
 267         # journey_detail_ref = arrivals[41].detail_ref
 268         # journey_detail = try_vao_journey_detail(vao, journey_detail_ref)
 269
 270         vao_line_ids = set(a.vao_line_id for a in arrivals).union(d.vao_line_id for d in departures) \
 271             .union(stop_with_dist.stop.line_ids)
 272         lines = []
 273
 274         for vao_line_id in vao_line_ids:
 275             line_info = line_info_dict[vao_line_id]
 276             departure = []
 277             for direction in set(d.direction for d in departures if d.vao_line_id == vao_line_id):
 278                 departure.append({
 279                     "direction": direction,
 280                     "datetime": [d.time.isoformat(timespec='minutes') for d in departures
 281                                  if d.vao_line_id == vao_line_id and d.direction == direction]
 282                 })
 283
 284             arrival = []
 285             for origin in set(a.origin for a in arrivals if a.vao_line_id == vao_line_id):
 286                 arrival.append({
 287                     "origin": origin,
 288                     "datetime": [a.time.isoformat(timespec='minutes') for a in arrivals
 289                                  if a.vao_line_id == vao_line_id and a.origin == origin]
 290                 })
 291
 292             schedules = [{
 293                 'service_date': journey_date.date().isoformat(),
 294                 "day_type": "work_day",
 295                 "departure": departure,
 296                 "arrival": arrival,
 297             }]
 298
 299             line = {
 300                 "vao_line_id": vao_line_id,
 301                 "line": line_info.line,
 302                 "category": line_info.category,
 303                 "operator": line_info.operator,
 304                 "schedules": schedules,
 305             }
 306             lines.append(line)
 307
 308         public_transport_stop = {
 309             "name": stop_with_dist.stop.name,
 310             "vao_ext_id": stop_with_dist.stop.ext_id,
 311             "position": {
 312                 "position": {
 313                     "longitude": stop_with_dist.stop.lon_lat.lon,
 314                     "latitude": stop_with_dist.stop.lon_lat.lat
 315                 }
 316             },
 317             "walk_distance": stop_with_dist.dist.dist_m,
 318             "walk_time": stop_with_dist.dist.time_min,
 319             "lines": lines,
 320         }
 321         if ifopt_stop_id := vao_ext_id_to_ifopt_stop_id(stop_with_dist.stop.ext_id):
 322             public_transport_stop['ifopt_stop_id'] = ifopt_stop_id
 323         vvt_stop_id = get_vvt_stop_id(stop_with_dist.stop.ext_id)
 324         if vvt_stop_id is not None:
 325             public_transport_stop["vvt_stop_id"] = vvt_stop_id
 326
 327         public_transport_stops.append(public_transport_stop)
 328
 329     sledrun_json['public_transport_stops'] = public_transport_stops
 330
 331     if sledrun_json == sledrun_json_orig:
 332         return
 333
 334     jsonschema.validate(instance=sledrun_json, schema=site.sledrun_schema())
 335     sledrun_json_ordered = order_json_keys(sledrun_json, site.sledrun_schema())
 336     assert sledrun_json_ordered == sledrun_json
 337     sledrun_json_orig_str = format_json(sledrun_json_orig)
 338     sledrun_json_str = format_json(sledrun_json_ordered)
 339
 340     cprint(title, 'green')
 341     unified_diff(sledrun_json_orig_str, sledrun_json_str)
 342     choice = input_yes_no_quit('Do you accept the changes [yes, no, quit]? ', None)
 343     if choice == Choice.no:
 344         return
 345
 346     if choice == Choice.quit:
 347         sys.exit(0)
 348
 349     site(
 350         'edit',
 351         pageid=sledrun_json_page['pageid'],
 352         text=sledrun_json_str,
 353         summary=f'Informationen zu Öffentlichen Verkehrsmitteln zu {title} aktualisiert (dank VAO).',
 354         # minor=1,
 355         bot=1,
 356         baserevid=sledrun_json_page['revisions'][0]['revid'],
 357         nocreate=1,
 358         token=site.token(),
 359     )
 360
 361
 362 def update_public_transport(ini_files: List[str], query_date: date, sledrun: Optional[str]):
 363     config = configparser.ConfigParser()
 364     config.read(ini_files)
 365
 366     site = WikiSite(ini_files)
 367     vao = Vao(config.get('vao', 'access_id'))
 368
 369     if sledrun is not None:
 370         update_sledrun(vao, site, sledrun, query_date)
 371         return
 372
 373     for result in site.query(list='categorymembers', cmtitle='Kategorie:Rodelbahn', cmlimit='max'):
 374         for page in result['categorymembers']:
 375             sledrun = page['title']
 376             print(sledrun)
 377             update_sledrun(vao, site, sledrun, query_date)
 378
 379
 380 def main():
 381     query_date = default_query_date(date.today())
 382     parser = argparse.ArgumentParser(description='Update public transport information in sledrun JSON files.')
 383     parser.add_argument('--sledrun', help='If given, work on a single sled run page, otherwise at the whole category.')
 384     parser.add_argument('--date', type=date.fromisoformat, default=query_date,
 385                         help='Working week date to query the database.')
 386     parser.add_argument('inifile', nargs='+', help='inifile.ini, see: https://www.winterrodeln.org/trac/wiki/ConfigIni')
 387     args = parser.parse_args()
 388     update_public_transport(args.inifile, args.date, args.sledrun)
 389
 390
 391 if __name__ == '__main__':
 392     main()