PythonRMV/pyrmv/methods/trip_recon.py

from datetime import datetime
from typing import List, Union
from pyrmv.classes.Journey import Journey
from pyrmv.classes.Trip import Trip
from pyrmv.enums.rt_mode import RealTimeMode
from pyrmv.raw.trip_recon import trip_recon as raw_trip_recon
from pyrmv.utility.find_exception import find_exception

try:
    from typing import Literal
except ImportError:
    from typing_extensions import Literal

def trip_recon(

        access_id: str,
        context: Union[str, Journey],
        date: Union[str, datetime] = None,
        match_real_time: bool = None,
        enable_replacements: bool = None,
        arrival_dev_lower: int = None,
        arrival_dev_upper: int = None,
        departure_dev_lower: int = None,
        departure_dev_upper: int = None,
        passlist: bool = None,
        passing_points: bool = False,
        real_time_mode: Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT] = None,
        tariff: bool = None,
        messages: bool = False
    ) -> List[Trip]:
    """Reconstructing a trip can be achieved using the reconstruction context provided by any trip result in the
    `ctx_recon` attribute of `Trip` object. The result will be a true copy of the original trip search result given
    that the underlying data did not change.  

    More detailed request is available as `raw.trip_recon()`, however returns `dict` instead of `List[Trip]`.  

    ### Args:
        * access_id (`str`): Access ID for identifying the requesting client. Get your key on [RMV website](https://opendata.rmv.de/site/start.html).
        * context (`Union[str, Journey]`): Specifies the reconstruction context.
        * date (`Union[str, datetime]`, **optional**): Sets the start date for which the departures shall be retrieved. Represented in the format `YYYY-MM-DD` or as a datetime object. This parameter will force the service to reconstruct the trip on that specific date. If the trip is not available on that date, because it does not operate, exception SvcNoResultError will be raised. Defaults to `None`.
        * match_real_time (`bool`, **optional**): Whether the realtime type that journeys are based on be considered. Defaults to `None`.
        * enable_replacements (`bool`, **optional**): If set to `True` replaces cancelled journeys with their replacement journeys if possible. Defaults to `None`.
        * arrival_dev_lower (`int`, **optional**): Lower deviation in minutes within interval `[0, 720]` indicating "how much earlier than original arrival". Defaults to `None`.
        * arrival_dev_upper (`int`, **optional**): Upper deviation in minutes within interval `[0, 720]` indicating "how much later than original arrival". Defaults to `None`.
        * departure_dev_lower (`int`, **optional**): Lower deviation in minutes within interval `[0, 720]` indicating "how much earlier than original departure". Defaults to `None`.
        * departure_dev_upper (`int`, **optional**): Upper deviation in minutes within interval `[0, 720]` indicating "how much later than original departure". Defaults to `None`.
        * passlist (`bool`, **optional**): Enables/disables the return of the passlist for each leg of the trip. Defaults to `None`.
        * passing_points (`bool`, **optional**): Enables/disables the return of stops having no alighting and boarding in its passlist for each leg of the trip. Needs passlist parameter enabled. Defaults to `False`.
        * real_time_mode (`Literal[RealTimeMode.FULL, RealTimeMode.INFOS, RealTimeMode.OFF, RealTimeMode.REALTIME, RealTimeMode.SERVER_DEFAULT]`, **optional**): Set the realtime mode to be used. Read more about this in HAFAS ReST Documentation. Defaults to `None`.
        * tariff (`bool`, **optional**): Enables/disables the output of tariff data. The default is configurable via provisioning. Defaults to `None`.
        * messages (`bool`, **optional**): Enables/disables the output of traffic messages. The default is configurable via provisioning. Defaults to `False`.

    ### Returns:
        * List[Trip]: List of `Trip` objects. Empty list if none found.
    """    
    
    if real_time_mode == None:
        real_time_mode = None
    else:
        real_time_mode = real_time_mode.code
    
    if isinstance(context, Journey):
        context = context.ctx_recon

    trips = []
    trips_raw = raw_trip_recon(
        accessId=access_id,
        ctx=context,
        date=date,
        matchRtType=match_real_time,
        enableReplacements=enable_replacements,
        arrL=arrival_dev_lower,
        arrU=arrival_dev_upper,
        depL=departure_dev_lower,
        depU=departure_dev_upper,
        passlist=passlist,
        showPassingPoints=passing_points,
        rtMode=real_time_mode,
        tariff=tariff,
        trafficMessages=messages,
    )

    find_exception(trips_raw)

    for trip in trips_raw["Trip"]:
        trips.append(Trip(trip))

    return trips