add profile for GVH (Großraumverkehr Hannover)

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

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

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