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, missing_only: bool, no_schedules: bool):
 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     if missing_only:
 252         if pt_stops := sledrun_json.get('public_transport_stops'):
 253             if len(pt_stops) > 0 and pt_stops[0].get('ifopt_stop_id'):
 254                 return
 255
 256     pos = sledrun_json.get('bottom', {}).get('position')
 257     if not pos:
 258         return
 259     lon_lat = LonLat(pos['longitude'], pos['latitude'])
 260     nearby_stops, line_info_dict = try_vao_nearby_stops(vao, lon_lat)
 261     nearby_stops = unique_nearby_stops(nearby_stops)
 262     stops_with_dists = [StopWithDist(stop, try_vao_walk_distance(vao, stop.lon_lat, lon_lat)) for stop in nearby_stops]
 263     stops_with_dists = sorted(stops_with_dists, key=lambda s: s.dist.dist_m)
 264     stops_with_dists = remove_redundant_stops(stops_with_dists)
 265
 266     journey_date = datetime(query_date.year, query_date.month, query_date.day)
 267     journey_minutes = 1439
 268
 269     public_transport_stops = []
 270     lines = []
 271     for stop_with_dist in stops_with_dists:
 272         if not no_schedules:
 273             departures = try_vao_departure_board(vao, stop_with_dist.stop.ext_id, journey_date, journey_minutes,
 274                                                  line_info_dict)
 275             arrivals = try_vao_arrival_board(vao, stop_with_dist.stop.ext_id, journey_date, journey_minutes, line_info_dict)
 276             # journey_detail_ref = arrivals[41].detail_ref
 277             # journey_detail = try_vao_journey_detail(vao, journey_detail_ref)
 278
 279             vao_line_ids = set(a.vao_line_id for a in arrivals).union(d.vao_line_id for d in departures) \
 280                 .union(stop_with_dist.stop.line_ids)
 281             lines = []
 282
 283             for vao_line_id in vao_line_ids:
 284                 line_info = line_info_dict[vao_line_id]
 285                 departure = []
 286                 for direction in set(d.direction for d in departures if d.vao_line_id == vao_line_id):
 287                     departure.append({
 288                         "direction": direction,
 289                         "datetime": [d.time.isoformat(timespec='minutes') for d in departures
 290                                      if d.vao_line_id == vao_line_id and d.direction == direction]
 291                     })
 292
 293                 arrival = []
 294                 for origin in set(a.origin for a in arrivals if a.vao_line_id == vao_line_id):
 295                     arrival.append({
 296                         "origin": origin,
 297                         "datetime": [a.time.isoformat(timespec='minutes') for a in arrivals
 298                                      if a.vao_line_id == vao_line_id and a.origin == origin]
 299                     })
 300
 301                 schedules = [{
 302                     'service_date': journey_date.date().isoformat(),
 303                     "day_type": "work_day",
 304                     "departure": departure,
 305                     "arrival": arrival,
 306                 }]
 307
 308                 line = {
 309                     "vao_line_id": vao_line_id,
 310                     "line": line_info.line,
 311                     "category": line_info.category,
 312                     "operator": line_info.operator,
 313                     "schedules": schedules,
 314                 }
 315                 lines.append(line)
 316
 317         public_transport_stop = {
 318             "name": stop_with_dist.stop.name,
 319             "vao_ext_id": stop_with_dist.stop.ext_id,
 320             "position": {
 321                 "position": {
 322                     "longitude": stop_with_dist.stop.lon_lat.lon,
 323                     "latitude": stop_with_dist.stop.lon_lat.lat
 324                 }
 325             },
 326             "walk_distance": stop_with_dist.dist.dist_m,
 327             "walk_time": stop_with_dist.dist.time_min,
 328         }
 329         if not no_schedules:
 330             public_transport_stop["lines"] = lines
 331         if ifopt_stop_id := vao_ext_id_to_ifopt_stop_id(stop_with_dist.stop.ext_id):
 332             public_transport_stop['ifopt_stop_id'] = ifopt_stop_id
 333         vvt_stop_id = get_vvt_stop_id(stop_with_dist.stop.ext_id)
 334         if vvt_stop_id is not None:
 335             public_transport_stop["vvt_stop_id"] = vvt_stop_id
 336
 337         public_transport_stops.append(public_transport_stop)
 338
 339     if len(public_transport_stops) > 0 or 'public_transport_stops' not in sledrun_json:
 340         sledrun_json['public_transport_stops'] = public_transport_stops
 341
 342     if sledrun_json == sledrun_json_orig:
 343         return
 344
 345     jsonschema.validate(instance=sledrun_json, schema=site.sledrun_schema())
 346     sledrun_json_ordered = order_json_keys(sledrun_json, site.sledrun_schema())
 347     assert sledrun_json_ordered == sledrun_json
 348     sledrun_json_orig_str = format_json(sledrun_json_orig)
 349     sledrun_json_str = format_json(sledrun_json_ordered)
 350
 351     cprint(title, 'green')
 352     unified_diff(sledrun_json_orig_str, sledrun_json_str)
 353     choice = input_yes_no_quit('Do you accept the changes [yes, no, quit]? ', None)
 354     if choice == Choice.no:
 355         return
 356
 357     if choice == Choice.quit:
 358         sys.exit(0)
 359
 360     site(
 361         'edit',
 362         pageid=sledrun_json_page['pageid'],
 363         text=sledrun_json_str,
 364         summary=f'Informationen zu Öffentlichen Verkehrsmitteln zu {title} aktualisiert (dank VAO).',
 365         # minor=1,
 366         bot=1,
 367         baserevid=sledrun_json_page['revisions'][0]['revid'],
 368         nocreate=1,
 369         token=site.token(),
 370     )
 371
 372
 373 def update_public_transport(ini_files: List[str], query_date: date, sledrun: Optional[str], missing_only: bool,
 374                             no_schedules: bool):
 375     config = configparser.ConfigParser()
 376     config.read(ini_files)
 377
 378     site = WikiSite(ini_files)
 379     vao = Vao(config.get('vao', 'access_id'))
 380
 381     if sledrun is not None:
 382         update_sledrun(vao, site, sledrun, query_date, missing_only, no_schedules)
 383         return
 384
 385     for result in site.query(list='categorymembers', cmtitle='Kategorie:Rodelbahn', cmlimit='max'):
 386         for page in result['categorymembers']:
 387             sledrun = page['title']
 388             print(sledrun)
 389             update_sledrun(vao, site, sledrun, query_date, missing_only, no_schedules)
 390
 391
 392 def main():
 393     query_date = default_query_date(date.today())
 394     parser = argparse.ArgumentParser(description='Update public transport information in sledrun JSON files.')
 395     parser.add_argument('--sledrun', help='If given, work on a single sled run page, otherwise at the whole category.')
 396     parser.add_argument('--missing', action='store_true',
 397                         help='If given, only work on stops that have no ifopt_stop_id.')
 398     parser.add_argument('--no-schedules', action='store_true', help='If given, don\'t add schedules.')
 399     parser.add_argument('--date', type=date.fromisoformat, default=query_date,
 400                         help='Working week date to query the database.')
 401     parser.add_argument('inifile', nargs='+', help='inifile.ini, see: https://www.winterrodeln.org/trac/wiki/ConfigIni')
 402     args = parser.parse_args()
 403     update_public_transport(args.inifile, args.date, args.sledrun, args.missing, args.no_schedules)
 404
 405
 406 if __name__ == '__main__':
 407     main()