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
+from typing import List, Iterable, Optional, Set, Tuple, Dict
import isodate
import jsonschema
-from osgeo import ogr
-from osgeo.ogr import Layer
+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
from wrpylib.mwapi import WikiSite, page_json
from wrpylib.vao import Vao
from wrpylib.wrvalidators import LonLat
ext_id: str
+@dataclass
class LineInfo:
vao_line_id: str # e.g. 'vvt-8-325-T-j23-1'
- name: str # e.g. '325T'
+ 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'
return None
-def try_vao_nearby_stops(vao: Vao, lon_lat: LonLat) -> List[StopInfo]:
+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()
- result = []
- for stop in response['stopLocationOrCoordLocation']:
+ stop_info_list: List[StopInfo] = []
+ line_info_dict: Dict[str, LineInfo] = {}
+ for stop in response.get('stopLocationOrCoordLocation', []):
sl = stop['StopLocation']
line_ids = set()
- for p in sl['productAtStop']:
- line_ids.add(p['lineId'])
+ 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']
- result.append(StopInfo(sl['name'], sl['dist'], LonLat(sl['mainMastLon'], sl['mainMastLat']), line_ids, ext_id))
- return result
+ 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:
@dataclass
class Departure:
- line_id: str
- line_name: str
- line_label: str
+ vao_line_id: str # e.g. 'vvt-8-325-T-j23-1'
time: datetime
- direction: str
- flag: str
- category: str
- operator: str
+ 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) -> List[Departure]:
+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(),
'duration': journey_minutes,
}
response = vao.get('departureBoard', parameter).json()
- result = []
+ departure_list = []
for departure in response.get('Departure', []):
ps = departure['ProductAtStop']
time = datetime.fromisoformat(f'{departure["date"]}T{departure["time"]}')
- result.append(Departure(ps['lineId'], ps['name'], ps['displayNumber'], time, departure['direction'],
- departure['directionFlag'], ps['catOutL'], ps['operator'],
- departure['JourneyDetailRef']['ref']))
- return result
+ 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:
- line_id: str
- line_name: str
- line_label: str
+ vao_line_id: str # e.g. 'vvt-8-325-T-j23-1'
time: datetime
- origin: str
- category: str
- operator: str
+ 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) -> List[Arrival]:
+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(),
'duration': journey_minutes,
}
response = vao.get('arrivalBoard', parameter).json()
- result = []
+ arrival_list = []
for arrival in response.get('Arrival', []):
ps = arrival['ProductAtStop']
time = datetime.fromisoformat(f'{arrival["date"]}T{arrival["time"]}')
- result.append(Arrival(ps['lineId'], ps['name'], ps['displayNumber'], time, arrival['origin'],
- ps['catOutL'], ps['operator'], arrival['JourneyDetailRef']['ref']))
- return result
+ 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]:
return result
-def update_sledrun(vao: Vao, db_cities: Layer, site: WikiSite, title: str):
+def update_sledrun(vao: Vao, site: WikiSite, title: str, query_date: date):
sledrun_json_page = site.query_page(f'{title}/Rodelbahn.json')
sledrun_json = page_json(sledrun_json_page)
pos = sledrun_json['bottom']['position']
lon_lat = LonLat(pos['longitude'], pos['latitude'])
- nearby_stops = try_vao_nearby_stops(vao, lon_lat)
+ 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(2023, 1, 4)
+ journey_date = datetime(query_date.year, query_date.month, query_date.day)
journey_minutes = 1439
public_transport_stops = []
for stop_with_dist in stops_with_dists:
- departures = try_vao_departure_board(vao, stop_with_dist.stop.ext_id, journey_date, journey_minutes)
- arrivals = try_vao_arrival_board(vao, stop_with_dist.stop.ext_id, journey_date, journey_minutes)
+ 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)
- line_ids = set(a.line_id for a in arrivals).union(d.line_id for d in departures)
- line_names = {}
- line_labels = {}
- line_category = {}
- line_operator = {}
- for a in arrivals:
- line_names[a.line_id] = a.line_name
- line_labels[a.line_id] = a.line_label
- line_category[a.line_id] = a.category
- line_operator[a.line_id] = a.operator
- for d in departures:
- line_names[d.line_id] = d.line_name
- line_labels[d.line_id] = d.line_label
- line_category[d.line_id] = d.category
- line_operator[d.line_id] = d.operator
+ 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 line_id in line_ids:
+ 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.line_id == line_id):
+ 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.line_id == line_id and d.direction == direction]
+ 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.line_id == line_id):
+ 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.line_id == line_id and a.origin == origin]
+ if a.vao_line_id == vao_line_id and a.origin == origin]
})
schedules = [{
+ 'service_date': journey_date.date().isoformat(),
"day_type": "work_day",
- "begin": journey_date.isoformat(timespec='minutes'),
- "minutes": journey_minutes,
"departure": departure,
"arrival": arrival,
}]
- lines.append({
- "vao_line_id": line_id,
- "name": line_names[line_id],
- "label": line_labels[line_id],
- "category": line_category[line_id],
- "operator": line_operator[line_id],
+ 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,
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'],
)
-def update_public_transport(ini_files: List[str]):
- ogr.UseExceptions()
-
+def update_public_transport(ini_files: List[str], query_date: date, sledrun: Optional[str]):
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'))
- # update_sledrun(vao, None, site, 'Rumer Alm')
- # update_sledrun(vao, None, site, 'Kemater Alm')
- update_sledrun(vao, None, site, 'Nisslalm')
- return
+ if sledrun is not None:
+ update_sledrun(vao, site, sledrun, query_date)
+ return
for result in site.query(list='categorymembers', cmtitle='Kategorie:Rodelbahn', cmlimit='max'):
for page in result['categorymembers']:
- print(page['title'])
- # if page['title'] in ['Anzère', 'Hochhäderich (Falkenhütte)', 'Hochlitten-Moosalpe']:
- # continue
- update_sledrun(vao, db_cities, site, page['title'])
+ sledrun = page['title']
+ print(sledrun)
+ update_sledrun(vao, site, sledrun, query_date)
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('--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)
+ update_public_transport(args.inifile, args.date, args.sledrun)
if __name__ == '__main__':
ToastFreeware / philipp / winterrodeln / wrpylib.git / blobdiff