From a39c42d00746e7e75c220806d1152d0d3c36713d Mon Sep 17 00:00:00 2001 From: Mathis Dirksen-Thedens Date: Thu, 13 Feb 2025 08:42:53 +0100 Subject: [PATCH] =?UTF-8?q?Add=20a=20profile=20for=20GVH=20(Gro=C3=9Fraumv?= =?UTF-8?q?erkehr=20Hannover)=20and=20tests=20for=20it=20plus=20some=20doc?= =?UTF-8?q?umentation.=20#54?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - the GVH doesn't use plain HAFAS but a HAMM - the location ID format is different from HAFAS standard and doesn't contain names and coordinates - the data format was reverse engineered from the web app at https://gvh.hafas.de --- docs/usage/profiles.rst | 32 ++++ pyhafas/profile/__init__.py | 1 + pyhafas/profile/gvh/__init__.py | 54 +++++++ pyhafas/profile/gvh/helper/parse_lid.py | 40 +++++ pyhafas/profile/gvh/helper/station_names.py | 9 ++ pyhafas/profile/gvh/requests/journey.py | 50 +++++++ pyhafas/profile/gvh/requests/journeys.py | 139 ++++++++++++++++++ pyhafas/profile/gvh/requests/station_board.py | 101 +++++++++++++ tests/gvh/__init__.py | 0 tests/gvh/request/__init__.py | 0 tests/gvh/request/arrivals_test.py | 24 +++ tests/gvh/request/departures_test.py | 25 ++++ tests/gvh/request/journey_test.py | 34 +++++ tests/gvh/request/locations_test.py | 15 ++ tests/gvh/request/nearby_test.py | 12 ++ 15 files changed, 536 insertions(+) create mode 100644 pyhafas/profile/gvh/__init__.py create mode 100644 pyhafas/profile/gvh/helper/parse_lid.py create mode 100644 pyhafas/profile/gvh/helper/station_names.py create mode 100644 pyhafas/profile/gvh/requests/journey.py create mode 100644 pyhafas/profile/gvh/requests/journeys.py create mode 100644 pyhafas/profile/gvh/requests/station_board.py create mode 100644 tests/gvh/__init__.py create mode 100644 tests/gvh/request/__init__.py create mode 100644 tests/gvh/request/arrivals_test.py create mode 100644 tests/gvh/request/departures_test.py create mode 100644 tests/gvh/request/journey_test.py create mode 100644 tests/gvh/request/locations_test.py create mode 100644 tests/gvh/request/nearby_test.py diff --git a/docs/usage/profiles.rst b/docs/usage/profiles.rst index 391affa..f657378 100644 --- a/docs/usage/profiles.rst +++ b/docs/usage/profiles.rst @@ -74,6 +74,38 @@ Specialities * The `max_trips` filter in station board (departures/arrival) requests seems not to work +Großraumverkehr Hannover (GVH) +------------------------------ + +Usage +^^^^^^ +.. code:: python + + from pyhafas.profile import GVHProfile + client = HafasClient(GVHProfile()) + +Available Products +^^^^^^^^^^^^^^^^^^ + +===================== =================== +pyHaFAS Internal Name Example Train Types +===================== =================== +ice ICE +ic-ec IC, EC +re-rb RE, RB +s-bahn S-Bahn +stadtbahn U-Bahn +bus Bus +on-demand Bedarfsverkehr +===================== =================== + +Default Products +^^^^^^^^^^^^^^^^ +All available products specified above are enabled by default. + +Noteworthy +^^^^^^^^^^ +The location IDs are different from standard HAFAS and don't contain names and coordinates. Nahverkehr Sachsen-Anhalt (NASA) --------------------------------------- diff --git a/pyhafas/profile/__init__.py b/pyhafas/profile/__init__.py index 8078538..841d29f 100644 --- a/pyhafas/profile/__init__.py +++ b/pyhafas/profile/__init__.py @@ -1,6 +1,7 @@ from .interfaces import ProfileInterface # isort:skip from .base import BaseProfile from .db import DBProfile +from .gvh import GVHProfile from .vsn import VSNProfile from .rkrp import RKRPProfile from .nasa import NASAProfile diff --git a/pyhafas/profile/gvh/__init__.py b/pyhafas/profile/gvh/__init__.py new file mode 100644 index 0000000..57918d2 --- /dev/null +++ b/pyhafas/profile/gvh/__init__.py @@ -0,0 +1,54 @@ +import pytz + +from pyhafas.profile import BaseProfile +from pyhafas.profile.gvh.helper.parse_lid import GVHParseLidHelper +from pyhafas.profile.gvh.requests.journey import GVHJourneyRequest +from pyhafas.profile.gvh.requests.journeys import GVHJourneysRequest +from pyhafas.profile.gvh.requests.station_board import GVHStationBoardRequest + + +class GVHProfile(BaseProfile, GVHParseLidHelper, GVHStationBoardRequest, GVHJourneysRequest, GVHJourneyRequest): + """ + Profile of the HaFAS of Großraumverkehr Hannover (GVH) - regional in Hannover area + """ + baseUrl = "https://gvh.hafas.de/hamm" + defaultUserAgent = "Mozilla/5.0 (X11; Linux x86_64; rv:135.0) Gecko/20100101 Firefox/135.0" + + locale = 'de-DE' + timezone = pytz.timezone('Europe/Berlin') + + requestBody = { + 'client': { + 'id': 'HAFAS', + 'l': 'vs_webapp', + 'name': 'webapp', + 'type': 'WEB', + 'v': '10109' + }, + 'ver': '1.62', + 'lang': 'deu', + 'auth': { + 'type': 'AID', + 'aid': 'IKSEvZ1SsVdfIRSK' + } + } + + availableProducts = { + "ice": [1], + "ic-ec": [2, 4], + "re-rb": [8], + "s-bahn": [16], + "stadtbahn": [256], + "bus": [32], + "on-demand": [512] + } + + defaultProducts = [ + "ice", + "ic-ec", + "re-rb", + "s-bahn", + "stadtbahn", + "bus", + "on-demand" + ] diff --git a/pyhafas/profile/gvh/helper/parse_lid.py b/pyhafas/profile/gvh/helper/parse_lid.py new file mode 100644 index 0000000..011ffe6 --- /dev/null +++ b/pyhafas/profile/gvh/helper/parse_lid.py @@ -0,0 +1,40 @@ +from pyhafas.profile import ProfileInterface +from pyhafas.profile.base import BaseParseLidHelper +from pyhafas.types.fptf import Station + + +class GVHParseLidHelper(BaseParseLidHelper): + def parse_lid(self: ProfileInterface, lid: str) -> dict: + """ + Converts the LID given by HaFAS + + This implementation only returns the LID inside a dict + because GVH doesn't have normal HaFAS IDs but only HAMM IDs. + + :param lid: Location identifier (given by HaFAS) + :return: Dict wrapping the given LID + """ + return {"lid": lid} + + def parse_lid_to_station( + self: ProfileInterface, + lid: str, + name: str = "", + latitude: float = 0, + longitude: float = 0) -> Station: + """ + Parses the LID given by HaFAS to a station object + + :param lid: Location identifier (given by HaFAS) + :param name: Station name (optional, if not given, empty string is used) + :param latitude: Latitude of the station (optional, if not given, 0 is used) + :param longitude: Longitude of the station (optional, if not given, 0 is used) + :return: Parsed LID as station object + """ + return Station( + id=lid, + lid=lid, + name=name, + latitude=latitude, + longitude=longitude + ) diff --git a/pyhafas/profile/gvh/helper/station_names.py b/pyhafas/profile/gvh/helper/station_names.py new file mode 100644 index 0000000..a09c464 --- /dev/null +++ b/pyhafas/profile/gvh/helper/station_names.py @@ -0,0 +1,9 @@ +from typing import Union + + +def find(station_name_by_lid: dict[str, str], lid: str, id: str) -> Union[str, None]: + to_search = lid if lid else id + for entry in station_name_by_lid.items(): + if to_search.startswith(entry[0]): + return entry[1] + return None \ No newline at end of file diff --git a/pyhafas/profile/gvh/requests/journey.py b/pyhafas/profile/gvh/requests/journey.py new file mode 100644 index 0000000..587d717 --- /dev/null +++ b/pyhafas/profile/gvh/requests/journey.py @@ -0,0 +1,50 @@ +from io import UnsupportedOperation + +from pyhafas.profile import ProfileInterface +from pyhafas.profile.base import BaseJourneyRequest +from pyhafas.profile.gvh.helper.station_names import find +from pyhafas.profile.interfaces.requests.journey import JourneyRequestInterface +from pyhafas.types.fptf import Journey +from pyhafas.types.hafas_response import HafasResponse + + +class GVHJourneyRequest(BaseJourneyRequest): + def format_journey_request( + self: ProfileInterface, + journey: Journey) -> dict: + """ + Creates the HAFAS / HAMM request for refreshing journey details + :param journey: Id of the journey (ctxRecon) + :return: Request for HAFAS (KVB-deployment) + """ + return { + 'req': { + 'outReconL': [{ + 'ctx': journey.id + }] + }, + 'meth': 'Reconstruction' + } + + def parse_journey_request(self: ProfileInterface, data: HafasResponse) -> Journey: + """ + Parses the HaFAS response for a journey request + :param data: Formatted HaFAS response + :return: List of Journey objects + """ + date = self.parse_date(data.res['outConL'][0]['date']) + + # station details + station_name_by_lid = dict() + for loc in data.common['locL']: + station_name_by_lid[loc['lid']] = loc['name'] + + journey = Journey(data.res['outConL'][0]['recon']['ctx'], date=date, + duration=self.parse_timedelta(data.res['outConL'][0]['dur']), + legs=self.parse_legs(data.res['outConL'][0], data.common, date)) + for leg in journey.legs: + leg.origin.name = find(station_name_by_lid, leg.origin.lid, leg.origin.id) + leg.destination.name = find(station_name_by_lid, leg.destination.lid, leg.destination.id) + for stopover in leg.stopovers: + stopover.stop.name = find(station_name_by_lid, stopover.stop.lid, stopover.stop.id) + return journey diff --git a/pyhafas/profile/gvh/requests/journeys.py b/pyhafas/profile/gvh/requests/journeys.py new file mode 100644 index 0000000..10ddd00 --- /dev/null +++ b/pyhafas/profile/gvh/requests/journeys.py @@ -0,0 +1,139 @@ +import datetime +from typing import Dict, List, Union + +from pyhafas.profile import ProfileInterface +from pyhafas.profile.base import BaseJourneysRequest +from pyhafas.profile.gvh.helper.station_names import find +from pyhafas.profile.interfaces.requests.journeys import \ + JourneysRequestInterface +from pyhafas.types.fptf import Journey, Station, Leg +from pyhafas.types.hafas_response import HafasResponse + + +class GVHJourneysRequest(BaseJourneysRequest): + def format_journeys_request( + self: ProfileInterface, + origin: Station, + destination: Station, + via: List[Station], + date: datetime.datetime, + min_change_time: int, + max_changes: int, + products: Dict[str, bool], + max_journeys: int + ) -> dict: + """ + Creates the HaFAS request body for a journeys request + + :param origin: Origin station + :param destination: Destionation station + :param via: Via stations, maybe empty list) + :param date: Date and time to search journeys for + :param min_change_time: Minimum transfer/change time at each station + :param max_changes: Maximum number of changes + :param products: Allowed products (a product is a mean of transport like ICE,IC) + :param max_journeys: Maximum number of returned journeys + :return: Request body for HaFAS + """ + return { + 'req': { + 'arrLocL': [{ + 'lid': destination.lid if destination.lid else destination.id + }], + 'viaLocL': [{ + 'loc': { + 'lid': via_station.lid if via_station.lid else via_station.id + } + } for via_station in via], + 'depLocL': [{ + 'lid': origin.lid if origin.lid else origin.id + }], + 'outDate': date.strftime("%Y%m%d"), + 'outTime': date.strftime("%H%M%S"), + 'jnyFltrL': [ + self.format_products_filter(products) + ], + 'minChgTime': min_change_time, + 'maxChg': max_changes, + 'numF': max_journeys, + }, + 'meth': 'TripSearch' + } + + def format_search_from_leg_request( + self: ProfileInterface, + origin: Leg, + destination: Station, + via: List[Station], + min_change_time: int, + max_changes: int, + products: Dict[str, bool], + ) -> dict: + """ + Creates the HaFAS request body for a journeys request + + :param origin: Origin leg + :param destination: Destionation station + :param via: Via stations, maybe empty list) + :param min_change_time: Minimum transfer/change time at each station + :param max_changes: Maximum number of changes + :param products: Allowed products (a product is a mean of transport like ICE,IC) + :return: Request body for HaFAS + """ + return { + 'req': { + 'arrLocL': [{ + 'lid': destination.lid if destination.lid else destination.id + }], + 'viaLocL': [{ + 'loc': { + 'lid': via_station.lid if via_station.lid else via_station.id + } + } for via_station in via], + 'locData': { + 'loc': { + 'lid': origin.lid if origin.lid else origin.id + }, + 'type': 'DEP', + 'date': origin.departure.strftime("%Y%m%d"), + 'time': origin.departure.strftime("%H%M%S") + }, + 'jnyFltrL': [ + self.format_products_filter(products) + ], + 'minChgTime': min_change_time, + 'maxChg': max_changes, + 'jid': origin.id, + 'sotMode': 'JI' + }, + 'meth': 'SearchOnTrip' + } + + def parse_journeys_request( + self: ProfileInterface, + data: HafasResponse) -> List[Journey]: + """ + Parses the HaFAS response for a journeys request + + :param data: Formatted HaFAS response + :return: List of Journey objects + """ + journeys = [] + + # station details + station_name_by_lid = dict() + for loc in data.common['locL']: + station_name_by_lid[loc['lid']] = loc['name'] + + # journeys + for jny in data.res['outConL']: + date = self.parse_date(jny['date']) + journey = Journey(jny['recon']['ctx'], date=date, duration=self.parse_timedelta(jny['dur']), + legs=self.parse_legs(jny, data.common, date)) + for leg in journey.legs: + leg.origin.name = find(station_name_by_lid, leg.origin.lid, leg.origin.id) + leg.destination.name = find(station_name_by_lid, leg.destination.lid, leg.destination.id) + for stopover in leg.stopovers: + stopover.stop.name = find(station_name_by_lid, stopover.stop.lid, stopover.stop.id) + journeys.append(journey) + return journeys diff --git a/pyhafas/profile/gvh/requests/station_board.py b/pyhafas/profile/gvh/requests/station_board.py new file mode 100644 index 0000000..8d0e73d --- /dev/null +++ b/pyhafas/profile/gvh/requests/station_board.py @@ -0,0 +1,101 @@ +import datetime +from typing import Dict, Optional, List + +from pyhafas.profile import ProfileInterface +from pyhafas.profile.base import BaseStationBoardRequest +from pyhafas.types.fptf import Station, StationBoardLeg +from pyhafas.types.hafas_response import HafasResponse +from pyhafas.types.station_board_request import StationBoardRequestType + + +class GVHStationBoardRequest(BaseStationBoardRequest): + def format_station_board_request( + self: ProfileInterface, + station: Station, + request_type: StationBoardRequestType, + date: datetime.datetime, + max_trips: int, + duration: int, + products: Dict[str, bool], + direction: Optional[Station] + ) -> dict: + """ + Creates the HaFAS request for a station board request (departure/arrival) + + :param station: Station to get departures/arrivals for + :param request_type: ARRIVAL or DEPARTURE + :param date: Date and time to get departures/arrival for + :param max_trips: Maximum number of trips that can be returned + :param products: Allowed products (e.g. ICE,IC) + :param duration: Time in which trips are searched + :param direction: Direction (end) station of the train. If none, filter will not be applied + :return: Request body for HaFAS + """ + return { + 'req': { + 'type': request_type.value, + 'stbLoc': { + 'lid': station.lid if station.lid else station.id + }, + 'dirLoc': { + 'lid': direction.lid if direction.lid else direction.id + } if direction is not None else None, + 'maxJny': max_trips, + 'date': date.strftime("%Y%m%d"), + 'time': date.strftime("%H%M%S"), + 'dur': duration, + 'jnyFltrL': [ + self.format_products_filter(products) + ], + }, + 'meth': 'StationBoard' + } + + def parse_station_board_request( + self: ProfileInterface, + data: HafasResponse, + departure_arrival_prefix: str) -> List[StationBoardLeg]: + """ + Parses the HaFAS data for a station board request + + :param data: Formatted HaFAS response + :param departure_arrival_prefix: Prefix for specifying whether its for arrival or departure (either "a" or "d") + :return: List of StationBoardLeg objects + """ + legs = [] + if not data.res.get('jnyL', False): + return legs + else: + for raw_leg in data.res['jnyL']: + date = self.parse_date(raw_leg['date']) + + try: + platform = raw_leg['stbStop'][departure_arrival_prefix + 'PltfR']['txt'] if \ + raw_leg['stbStop'].get(departure_arrival_prefix + 'PltfR') is not None else \ + raw_leg['stbStop'][departure_arrival_prefix + 'PltfS']['txt'] + except KeyError: + platform = raw_leg['stbStop'].get( + departure_arrival_prefix + 'PlatfR', + raw_leg['stbStop'].get( + departure_arrival_prefix + 'PlatfS', + None)) + + legs.append(StationBoardLeg( + id=raw_leg['jid'], + name=data.common['prodL'][raw_leg['prodX']]['name'], + direction=raw_leg.get('dirTxt'), + date_time=self.parse_datetime( + raw_leg['stbStop'][departure_arrival_prefix + 'TimeS'], + date + ), + station=self.parse_lid_to_station(data.common['locL'][raw_leg['stbStop']['locX']]['lid'], + name=data.common['locL'][raw_leg['stbStop']['locX']]['name']), + platform=platform, + delay=self.parse_datetime( + raw_leg['stbStop'][departure_arrival_prefix + 'TimeR'], + date) - self.parse_datetime( + raw_leg['stbStop'][departure_arrival_prefix + 'TimeS'], + date) if raw_leg['stbStop'].get(departure_arrival_prefix + 'TimeR') is not None else None, + cancelled=bool(raw_leg['stbStop'].get(departure_arrival_prefix + 'Cncl', False)) + )) + return legs diff --git a/tests/gvh/__init__.py b/tests/gvh/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/gvh/request/__init__.py b/tests/gvh/request/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/gvh/request/arrivals_test.py b/tests/gvh/request/arrivals_test.py new file mode 100644 index 0000000..9778fec --- /dev/null +++ b/tests/gvh/request/arrivals_test.py @@ -0,0 +1,24 @@ +import datetime + +from pyhafas import HafasClient +from pyhafas.profile.gvh import GVHProfile + + +def test_gvh_arrivals_request(): + client = HafasClient(GVHProfile()) + + # Lister Platz + arrivals_lister_platz = client.arrivals( + station="de:03241:2021", + date=datetime.datetime.now(), + max_trips=5 + ) + assert arrivals_lister_platz and len(arrivals_lister_platz) >= 1 + + # Hauptbahnhof + arrivals_hauptbahnhof = client.arrivals( + station="de:03241:31", + date=datetime.datetime.now(), + max_trips=5 + ) + assert arrivals_hauptbahnhof and len(arrivals_hauptbahnhof) >= 1 diff --git a/tests/gvh/request/departures_test.py b/tests/gvh/request/departures_test.py new file mode 100644 index 0000000..c8c126c --- /dev/null +++ b/tests/gvh/request/departures_test.py @@ -0,0 +1,25 @@ +import datetime + +from pyhafas import HafasClient +from pyhafas.profile.gvh import GVHProfile +from pyhafas.types.fptf import Station + + +def test_gvh_departures_request(): + client = HafasClient(GVHProfile()) + + # Lister Platz + departures_lister_platz = client.departures( + station="de:03241:2021", + date=datetime.datetime.now(), + max_trips=5 + ) + assert departures_lister_platz and len(departures_lister_platz) >= 1 + + # Hauptbahnhof + departures_hauptbahnhof = client.departures( + station="de:03241:31", + date=datetime.datetime.now(), + max_trips=5 + ) + assert departures_hauptbahnhof and len(departures_hauptbahnhof) >= 1 diff --git a/tests/gvh/request/journey_test.py b/tests/gvh/request/journey_test.py new file mode 100644 index 0000000..5c381b8 --- /dev/null +++ b/tests/gvh/request/journey_test.py @@ -0,0 +1,34 @@ +import datetime +from typing import List + +from pyhafas import HafasClient +from pyhafas.profile.gvh import GVHProfile +from pyhafas.types.fptf import Journey + + +def test_gvh_journey_request(): + client = HafasClient(GVHProfile()) + journeys = client.journeys( + origin="de:03241:31", # Hauptbahnhof + destination="de:03241:2021", # Lister Platz + date=datetime.datetime.now(), + min_change_time=0, + max_changes=-1 + ) + + assert isinstance(journeys, List) + assert len(journeys) >= 1 + assert len(journeys[0].legs) >= 1 + assert journeys[0].legs[0].origin.name + assert journeys[0].legs[0].destination.name + for stopover in journeys[0].legs[0].stopovers: + assert stopover.stop.name + + journey = client.journey( + journey=journeys[0].id + ) + assert isinstance(journey, Journey) + assert journey.legs[0].origin.name + assert journey.legs[0].destination.name + for stopover in journey.legs[0].stopovers: + assert stopover.stop.name \ No newline at end of file diff --git a/tests/gvh/request/locations_test.py b/tests/gvh/request/locations_test.py new file mode 100644 index 0000000..b7210bd --- /dev/null +++ b/tests/gvh/request/locations_test.py @@ -0,0 +1,15 @@ +from pyhafas import HafasClient +from pyhafas.profile.gvh import GVHProfile + + +def test_gvh_locations_request(): + client = HafasClient(GVHProfile()) + + locations_lister_platz = client.locations(term="Lister Platz") + assert len(locations_lister_platz) >= 1 + + locations_hauptbahnhof = client.locations(term="Hannover Hauptbahnhof") + assert len(locations_hauptbahnhof) >= 1 + + locations_aegi = client.locations(term="Aegidientorplatz") + assert len(locations_aegi) >= 1 diff --git a/tests/gvh/request/nearby_test.py b/tests/gvh/request/nearby_test.py new file mode 100644 index 0000000..4bd64a8 --- /dev/null +++ b/tests/gvh/request/nearby_test.py @@ -0,0 +1,12 @@ +from pyhafas import HafasClient +from pyhafas.profile.gvh import GVHProfile +from pyhafas.types.nearby import LatLng +from tests.distance import calculate_distance_in_meters + + +def test_gvh_nearby_request(): + pos = LatLng(50.940614, 6.958120) + client = HafasClient(GVHProfile()) + stations = client.nearby(pos) + assert stations + assert calculate_distance_in_meters(pos.latitude, pos.longitude, stations[0].latitude, stations[0].longitude) < 200