#!/usr/bin/python import argparse import configparser import re import sys from dataclasses import dataclass from datetime import date, timedelta, datetime from itertools import groupby from typing import List, Iterable, Optional, Set, Tuple, Dict import isodate import jsonschema from termcolor import cprint # python3-termcolor from wrpylib.cli_tools import unified_diff, input_yes_no_quit, Choice from wrpylib.json_tools import order_json_keys, format_json from wrpylib.lib_update_public_transport import default_query_date, vao_ext_id_to_ifopt_stop_id from wrpylib.mwapi import WikiSite, page_json from wrpylib.vao import Vao from wrpylib.wrvalidators import LonLat class VaoError(RuntimeError): pass @dataclass class StopInfo: name: str dist_m: int # Luftlinie lon_lat: LonLat line_ids: Set[str] ext_id: str @dataclass class LineInfo: vao_line_id: str # e.g. 'vvt-8-325-T-j23-1' line: Optional[str] # e.g. '325T', some trains don't have line labels. category: str # e.g. 'Anrufsammeltaxi' operator: str # e.g. 'Quaxis Taxi & Busreisen' def get_vvt_stop_id(ext_id: str) -> Optional[int]: vvt_stop_id_match = re.match(r'^47(\d{5})00$', ext_id) if vvt_stop_id_match: return int(vvt_stop_id_match.group(1)) return None def try_vao_nearby_stops(vao: Vao, lon_lat: LonLat) -> Tuple[List[StopInfo], Dict[str, LineInfo]]: """may throw VaoError with JSON decoded response as argument""" parameter = { 'originCoordLat': lon_lat.lat, 'originCoordLong': lon_lat.lon, } response = vao.get('location.nearbystops', parameter).json() stop_info_list: List[StopInfo] = [] line_info_dict: Dict[str, LineInfo] = {} for stop in response.get('stopLocationOrCoordLocation', []): sl = stop['StopLocation'] line_ids = set() for ps in sl['productAtStop']: vao_line_id = ps['lineId'] line_ids.add(vao_line_id) line_info_dict[vao_line_id] = LineInfo( vao_line_id=vao_line_id, line=ps['line'], category=ps['catOutL'], operator=ps['operator'], ) ext_id = sl['mainMastExtId'] stop_info_list.append(StopInfo(sl['name'], sl['dist'], LonLat(sl['mainMastLon'], sl['mainMastLat']), line_ids, ext_id)) return stop_info_list, line_info_dict def merge_stops(stops: List[StopInfo]) -> StopInfo: mean_dist = int(round(sum(stop.dist_m for stop in stops) / len(stops))) all_lines = set().union(*[stop.line_ids for stop in stops]) stop = stops[0] return StopInfo(stop.name, mean_dist, stop.lon_lat, all_lines, stop.ext_id) def unique_nearby_stops(stops: Iterable[StopInfo]) -> List[StopInfo]: stops = sorted(stops, key=lambda stop: stop.ext_id) stops = [merge_stops(list(i)) for k, i in groupby(stops, key=lambda stop: stop.ext_id)] return sorted(stops, key=lambda stop: stop.dist_m) @dataclass class WalkDistInfo: dist_m: int time_min: int def try_vao_walk_distance(vao: Vao, origin: LonLat, dest: LonLat) -> WalkDistInfo: """may throw VaoError with JSON decoded response as argument""" parameter = { 'originCoordLat': origin.lat, 'originCoordLong': origin.lon, 'destCoordLat': dest.lat, 'destCoordLong': dest.lon, 'groupFilter': 'API_WALK', } 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'] return WalkDistInfo(dist_m, duration_minutes) raise VaoError(response) @dataclass class Departure: vao_line_id: str # e.g. 'vvt-8-325-T-j23-1' time: datetime direction: str # e.g. 'Innsbruck Hauptbahnhof' flag: str # e.g. 'H' or 'R' detail_ref: str def try_vao_departure_board(vao: Vao, ext_id: str, journey_date: datetime, journey_minutes: int, line_info_dict: Dict[str, LineInfo]) -> List[Departure]: parameter = { 'id': ext_id, 'date': journey_date.date().isoformat(), 'time': journey_date.time().isoformat('minutes'), 'duration': journey_minutes, } response = vao.get('departureBoard', parameter).json() departure_list = [] for departure in response.get('Departure', []): ps = departure['ProductAtStop'] time = datetime.fromisoformat(f'{departure["date"]}T{departure["time"]}') vao_line_id = ps['lineId'] departure_list.append(Departure(vao_line_id, time, departure['direction'], departure['directionFlag'], departure['JourneyDetailRef']['ref'])) line_info_dict[vao_line_id] = LineInfo( vao_line_id=vao_line_id, line=ps['line'], category=ps['catOutL'], operator=ps['operator'], ) return departure_list @dataclass class Arrival: vao_line_id: str # e.g. 'vvt-8-325-T-j23-1' time: datetime origin: str # e.g. 'Innsbruck Hauptbahnhof' detail_ref: str def try_vao_arrival_board(vao: Vao, ext_id: str, journey_date: datetime, journey_minutes: int, line_info_dict: Dict[str, LineInfo]) -> List[Arrival]: parameter = { 'id': ext_id, 'date': journey_date.date().isoformat(), 'time': journey_date.time().isoformat('minutes'), 'duration': journey_minutes, } response = vao.get('arrivalBoard', parameter).json() arrival_list = [] for arrival in response.get('Arrival', []): ps = arrival['ProductAtStop'] time = datetime.fromisoformat(f'{arrival["date"]}T{arrival["time"]}') vao_line_id = ps['lineId'] arrival_list.append(Arrival(vao_line_id, time, arrival['origin'], arrival['JourneyDetailRef']['ref'])) line_info_dict[vao_line_id] = LineInfo( vao_line_id=vao_line_id, line=ps['line'], category=ps['catOutL'], operator=ps['operator'], ) return arrival_list def decode_service_days(value: str, begin: date) -> Iterable[date]: days_of_service = ''.join(f'{int(i, 16):04b}' for i in value) days_of_service = [bool(int(i)) for i in days_of_service] for i, day_has_service in enumerate(days_of_service): if day_has_service: yield begin + timedelta(days=i) @dataclass class JourneyDetail: planning_period_begin: date planning_period_end: date day_of_operation: date s_days_r: str s_days_i: str s_days_b: str service_days: List[date] first_stop_name: str last_stop_name: str def try_vao_journey_detail(vao: Vao, journey_detail_ref: str) -> JourneyDetail: parameter = { 'id': journey_detail_ref, 'rtMode': 'OFF' } response = vao.get('journeyDetail', parameter).json() sd = response['ServiceDays'][0] stops = response['Stops']['Stop'] planning_period_begin = date.fromisoformat(sd['planningPeriodBegin']) s_days_b = sd['sDaysB'] service_days = list(decode_service_days(s_days_b, planning_period_begin)) return JourneyDetail(planning_period_begin, date.fromisoformat(sd['planningPeriodEnd']), response['dayOfOperation'], sd['sDaysR'], sd['sDaysI'], s_days_b, service_days, stops[0]['name'], stops[-1]['name']) @dataclass class StopWithDist: stop: StopInfo dist: WalkDistInfo def is_redundant_stop(stop: StopWithDist, stops: List[StopWithDist]) -> bool: for s in stops: assert s != stop if stop.stop.line_ids.issubset(s.stop.line_ids): return True return False def remove_redundant_stops(stops: List[StopWithDist]) -> List[StopWithDist]: stops = stops.copy() result = [] while stops: stop = stops.pop() if not is_redundant_stop(stop, stops): result.insert(0, stop) return result def update_sledrun(vao: Vao, site: WikiSite, title: str, query_date: date, missing_only: bool, no_schedules: bool): sledrun_json_page = site.query_page(f'{title}/Rodelbahn.json') sledrun_json = page_json(sledrun_json_page) sledrun_json_orig = sledrun_json.copy() if missing_only: if pt_stops := sledrun_json.get('public_transport_stops'): if len(pt_stops) > 0 and pt_stops[0].get('ifopt_stop_id'): return pos = sledrun_json.get('bottom', {}).get('position') if not pos: return lon_lat = LonLat(pos['longitude'], pos['latitude']) nearby_stops, line_info_dict = try_vao_nearby_stops(vao, lon_lat) nearby_stops = unique_nearby_stops(nearby_stops) stops_with_dists = [StopWithDist(stop, try_vao_walk_distance(vao, stop.lon_lat, lon_lat)) for stop in nearby_stops] stops_with_dists = sorted(stops_with_dists, key=lambda s: s.dist.dist_m) stops_with_dists = remove_redundant_stops(stops_with_dists) journey_date = datetime(query_date.year, query_date.month, query_date.day) journey_minutes = 1439 public_transport_stops = [] lines = [] for stop_with_dist in stops_with_dists: if not no_schedules: departures = try_vao_departure_board(vao, stop_with_dist.stop.ext_id, journey_date, journey_minutes, line_info_dict) arrivals = try_vao_arrival_board(vao, stop_with_dist.stop.ext_id, journey_date, journey_minutes, line_info_dict) # journey_detail_ref = arrivals[41].detail_ref # journey_detail = try_vao_journey_detail(vao, journey_detail_ref) vao_line_ids = set(a.vao_line_id for a in arrivals).union(d.vao_line_id for d in departures) \ .union(stop_with_dist.stop.line_ids) lines = [] for vao_line_id in vao_line_ids: line_info = line_info_dict[vao_line_id] departure = [] for direction in set(d.direction for d in departures if d.vao_line_id == vao_line_id): departure.append({ "direction": direction, "datetime": [d.time.isoformat(timespec='minutes') for d in departures if d.vao_line_id == vao_line_id and d.direction == direction] }) arrival = [] for origin in set(a.origin for a in arrivals if a.vao_line_id == vao_line_id): arrival.append({ "origin": origin, "datetime": [a.time.isoformat(timespec='minutes') for a in arrivals if a.vao_line_id == vao_line_id and a.origin == origin] }) schedules = [{ 'service_date': journey_date.date().isoformat(), "day_type": "work_day", "departure": departure, "arrival": arrival, }] line = { "vao_line_id": vao_line_id, "line": line_info.line, "category": line_info.category, "operator": line_info.operator, "schedules": schedules, } lines.append(line) public_transport_stop = { "name": stop_with_dist.stop.name, "vao_ext_id": stop_with_dist.stop.ext_id, "position": { "position": { "longitude": stop_with_dist.stop.lon_lat.lon, "latitude": stop_with_dist.stop.lon_lat.lat } }, "walk_distance": stop_with_dist.dist.dist_m, "walk_time": stop_with_dist.dist.time_min, } if not no_schedules: public_transport_stop["lines"] = lines if ifopt_stop_id := vao_ext_id_to_ifopt_stop_id(stop_with_dist.stop.ext_id): public_transport_stop['ifopt_stop_id'] = ifopt_stop_id vvt_stop_id = get_vvt_stop_id(stop_with_dist.stop.ext_id) if vvt_stop_id is not None: public_transport_stop["vvt_stop_id"] = vvt_stop_id public_transport_stops.append(public_transport_stop) if len(public_transport_stops) > 0 or 'public_transport_stops' not in sledrun_json: sledrun_json['public_transport_stops'] = public_transport_stops 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_orig_str = format_json(sledrun_json_orig) sledrun_json_str = format_json(sledrun_json_ordered) cprint(title, 'green') unified_diff(sledrun_json_orig_str, sledrun_json_str) choice = input_yes_no_quit('Do you accept the changes [yes, no, quit]? ', None) if choice == Choice.no: return if choice == Choice.quit: sys.exit(0) site( 'edit', pageid=sledrun_json_page['pageid'], text=sledrun_json_str, summary=f'Informationen zu Öffentlichen Verkehrsmitteln zu {title} aktualisiert (dank VAO).', # minor=1, bot=1, baserevid=sledrun_json_page['revisions'][0]['revid'], nocreate=1, token=site.token(), ) def update_public_transport(ini_files: List[str], query_date: date, sledrun: Optional[str], missing_only: bool, no_schedules: bool): config = configparser.ConfigParser() config.read(ini_files) site = WikiSite(ini_files) vao = Vao(config.get('vao', 'access_id')) if sledrun is not None: update_sledrun(vao, site, sledrun, query_date, missing_only, no_schedules) return for result in site.query(list='categorymembers', cmtitle='Kategorie:Rodelbahn', cmlimit='max'): for page in result['categorymembers']: sledrun = page['title'] print(sledrun) update_sledrun(vao, site, sledrun, query_date, missing_only, no_schedules) def main(): query_date = default_query_date(date.today()) parser = argparse.ArgumentParser(description='Update public transport information in sledrun JSON files.') parser.add_argument('--sledrun', help='If given, work on a single sled run page, otherwise at the whole category.') parser.add_argument('--missing', action='store_true', help='If given, only work on stops that have no ifopt_stop_id.') parser.add_argument('--no-schedules', action='store_true', help='If given, don\'t add schedules.') parser.add_argument('--date', type=date.fromisoformat, default=query_date, help='Working week date to query the database.') parser.add_argument('inifile', nargs='+', help='inifile.ini, see: https://www.winterrodeln.org/trac/wiki/ConfigIni') args = parser.parse_args() update_public_transport(args.inifile, args.date, args.sledrun, args.missing, args.no_schedules) if __name__ == '__main__': main()