Add a profile for GVH (Großraumverkehr Hannover) and tests for it. #54

The GVH doesn't use plain HAFAS but a HAMM, so e.g. the ID format is different. The data format was reverse engineered from the web app at: https://gvh.hafas.de

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"
+    ]

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
+        )

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

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

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

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