Divided raw and normal methods for stops

This commit is contained in:
Profitroll 2022-09-24 13:44:24 +02:00
parent 52aea964b1
commit cc6324cc88
6 changed files with 143 additions and 32 deletions

View File

@ -1 +1,3 @@
from .trip_find import trip_find
from .stop_by_coords import stop_by_coords
from .stop_by_name import stop_by_name

View File

@ -0,0 +1,59 @@
from typing import List, Union
from pyrmv.classes.Stop import Stop
from pyrmv.raw.stop_by_coords import stop_by_coords as raw_stop_by_coords
try:
from typing import Literal
except ImportError:
from typing_extensions import Literal
def stop_by_coords(
access_id: str,
coords_lat: Union[str, float],
coords_lon: Union[str, float],
lang: Literal["de", "da", "en", "es", "fr", "hu", "it", "nl", "no", "pl", "sv", "tr"] = "en",
radius: Union[int, float] = 1000,
max_number: int = 10,
stop_type: Literal["S", "P", "SP", "SE", "SPE"] = "S",
selection_mode: Literal["SLCT_N", "SLCT_A"] = None, # type: ignore
) -> List[Stop]:
"""Method returns a list of stops around a given center coordinate.
The returned results are ordered by their distance to the center coordinate.
More detailed request is available as raw.stop_by_coords(), however returns dict instead of List[Stop].
### Args:
* access_id (`str`): Access ID for identifying the requesting client. Get your key on [RMV website](https://opendata.rmv.de/site/start.html).
* coords_lat (`Union[str, float]`): Latitude of centre coordinate.
* coords_lon (`Union[str, float]`): Longitude of centre coordinate.
* lang (`Literal["de","da","en","es","fr","hu","it","nl","no","pl","sv","tr"]`, **optional**): The language of response. Defaults to "en".
* radius (`Union[int, float]`, **optional**): Search radius in meter around the given coordinate if any. Defaults to 1000.
* max_number (`int`, **optional**): Maximum number of returned stops. Defaults to 10.
* stop_type (`Literal["S", "P", "SP", "SE", "SPE"]`, **optional**): Type filter for location types. Defaults to "S".
* selection_mode (`Literal["SLCT_N", "SLCT_A"]`, **optional**): Selection mode for locations. Defaults to None.
### Returns:
* dict: Output from RMV. Dict will contain "errorCode" and "errorText" if exception occurs.
"""
stops = []
stops_raw = raw_stop_by_coords(
accessId=access_id,
originCoordLat=coords_lat,
originCoordLong=coords_lon,
lang=lang,
radius=radius,
maxNo=max_number,
stopType=stop_type,
locationSelectionMode=selection_mode
)
for stop in stops_raw["stopLocationOrCoordLocation"]:
if "StopLocation" in stop:
stops.append(Stop(stop["StopLocation"]))
elif "CoordLocation" in stop:
stops.append(Stop(stop["CoordLocation"]))
return stops

View File

@ -0,0 +1,72 @@
from typing import List, Union
from pyrmv.classes.Stop import Stop
from pyrmv.raw.stop_by_name import stop_by_name as raw_stop_by_name
try:
from typing import Literal
except ImportError:
from typing_extensions import Literal
def stop_by_name(
access_id: str,
query: str,
lang: Literal["de", "da", "en", "es", "fr", "hu", "it", "nl", "no", "pl", "sv", "tr"] = "en",
max_number: int = 10,
stop_type: Literal["A", "ALL", "AP", "P", "S", "SA", "SP"] = "ALL",
selection_mode: Literal["SLCT_N", "SLCT_A"] = None, # type: ignore
coord_lat: Union[str, float] = None, # type: ignore
coord_lon: Union[str, float] = None, # type: ignore
radius: Union[int, float] = 1000,
refine_id: str = None, # type: ignore
stations: Union[str, list] = None, # type: ignore
filter_mode: Literal["DIST_PERI", "EXCL_PERI", "SLCT_PERI"] = "DIST_PERI"
) -> List[Stop]:
"""Method can be used to perform a pattern matching of a user input and to retrieve a list
of possible matches in the journey planner database. Possible matches might be stops/stations,
points of interest and addresses.
More detailed request is available as raw.stop_by_name(), however returns dict instead of List[Stop].
### Args:
* access_id (str): Access ID for identifying the requesting client. Get your key on [RMV website](https://opendata.rmv.de/site/start.html).
* query (str): Search for that token.
* lang (Literal["de","da","en","es","fr","hu","it","nl","no","pl","sv","tr"], optional): The language of response. Defaults to "en".
* max_number (int, optional): Maximum number of returned stops. In range 1-1000. Defaults to 10.
* stop_type (Literal["A", "ALL", "AP", "P", "S", "SA", "SP"], optional): Type filter for location types. Defaults to "ALL".
* selection_mode (str, optional): Selection mode for locations. "SLCT_N": Not selectable, "SLCT_A": Selectable. Defaults to None.
* coord_lat (Union[str, float], optional): Latitude of centre coordinate. Defaults to None.
* coord_lon (Union[str, float], optional): Longitude of centre coordinate. Defaults to None.
* radius (Union[int, float], optional): Search radius in meter around the given coordinate if any. Defaults to 1000.
* refine_id (str, optional): In case of an refinable location, this value takes the ID of the refinable one of a previous result. Defaults to None.
* stations (Union[str, list], optional): Filter for stations. Matches if the given value is prefix of any station in the itinerary. As a list or as a string separated by comma. Defaults to None.
* filter_mode (Literal["DIST_PERI", "EXCL_PERI", "SLCT_PERI"], optional): Filter modes for nearby searches. Defaults to "DIST_PERI".
### Returns:
* dict: Output from RMV. Dict will contain "errorCode" and "errorText" if exception occurs.
"""
stops = []
stops_raw = raw_stop_by_name(
accessId=access_id,
inputString=query,
lang=lang,
maxNo=max_number,
stopType=stop_type,
locationSelectionMode=selection_mode,
coordLat=coord_lat,
coordLong=coord_lon,
radius=radius,
refineId=refine_id,
stations=stations,
filterMode=filter_mode
)
for stop in stops_raw["stopLocationOrCoordLocation"]:
if "StopLocation" in stop:
stops.append(Stop(stop["StopLocation"]))
elif "CoordLocation" in stop:
stops.append(Stop(stop["CoordLocation"]))
return stops

View File

@ -74,7 +74,7 @@ def trip_find(
stop/station IDs or coordinates based on addresses and points of interest validated by the location service or
coordinates freely defined by the client.
Read more about this in section 2.12. "Trip Search (trip)" of HAFAS ReST Documentation.
More detailed request is available as raw.trip_find(), 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).

View File

@ -16,7 +16,6 @@ def stop_by_coords(accessId: str,
originCoordLong: Union[str, float],
lang: Literal["de", "da", "en", "es", "fr", "hu", "it", "nl", "no", "pl", "sv", "tr"] = "en",
json: bool = True,
raw_response: bool = False,
radius: Union[int, float] = 1000,
maxNo: int = 10,
stopType: Literal["S", "P", "SP", "SE", "SPE"] = "S",
@ -25,7 +24,7 @@ def stop_by_coords(accessId: str,
meta: str = None, # type: ignore
sattributes: Union[str, list] = None, # type: ignore
sinfotexts: Union[str, list] = None # type: ignore
) -> Union[dict, List[Stop]]:
) -> dict:
"""The location.nearbystops service returns a list of stops around a given center coordinate (within a
radius of 1000m). The returned results are ordered by their distance to the center coordinate.
@ -37,7 +36,6 @@ def stop_by_coords(accessId: str,
* originCoordLong (Union[str, float]): Longitude of centre coordinate.
* lang (Literal["de","da","en","es","fr","hu","it","nl","no","pl","sv","tr"], optional): The language of response. Defaults to "en".
* json (bool, optional): Whether response should be retrieved as JSON. XML is returned if False. Only matters if raw_response is True. Defaults to True.
* raw_response (bool, optional): Whether response should be returned as `dict` instead of `List[Stop]`. Defaults to False.
* radius (Union[int, float], optional): Search radius in meter around the given coordinate if any. Defaults to 1000.
* maxNo (int, optional): Maximum number of returned stops. Defaults to 10.
* stopType (Literal["S", "P", "SP", "SE", "SPE"], optional): Type filter for location types. Defaults to "S".
@ -67,16 +65,7 @@ def stop_by_coords(accessId: str,
find_exception(output.json())
if raw_response:
if json:
return output.json()
else:
return xmlparse(output.content)
if json:
return output.json()
else:
stops = []
for stop in output.json()["stopLocationOrCoordLocation"]:
if "StopLocation" in stop:
stops.append(Stop(stop["StopLocation"]))
elif "CoordLocation" in stop:
stops.append(Stop(stop["CoordLocation"]))
return stops
return xmlparse(output.content)

View File

@ -15,7 +15,6 @@ def stop_by_name(accessId: str,
inputString: str,
lang: Literal["de", "da", "en", "es", "fr", "hu", "it", "nl", "no", "pl", "sv", "tr"] = "en",
json: bool = True,
raw_response: bool = False,
maxNo: int = 10,
stopType: Literal["A", "ALL", "AP", "P", "S", "SA", "SP"] = "ALL",
locationSelectionMode: Literal["SLCT_N", "SLCT_A"] = None, # type: ignore
@ -28,7 +27,7 @@ def stop_by_name(accessId: str,
stations: Union[str, list] = None, # type: ignore
sattributes: Union[str, list] = None, # type: ignore
filterMode: Literal["DIST_PERI", "EXCL_PERI", "SLCT_PERI"] = "DIST_PERI"
) -> Union[dict, List[Stop]]:
) -> dict:
"""The location.name service can be used to perform a pattern matching of a user input and to retrieve a list
of possible matches in the journey planner database. Possible matches might be stops/stations, points of
interest and addresses.
@ -44,7 +43,6 @@ def stop_by_name(accessId: str,
* inputString (str): Search for that token.
* lang (Literal["de","da","en","es","fr","hu","it","nl","no","pl","sv","tr"], optional): The language of response. Defaults to "en".
* json (bool, optional): Whether response should be retrieved as JSON. XML is returned if False. Only matters if raw_response is True. Defaults to True.
* raw_response (bool, optional): Whether response should be returned as `dict` instead of `List[Stop]`. Defaults to False.
* maxNo (int, optional): Maximum number of returned stops. In range 1-1000. Defaults to 10.
* stopType (Literal["A", "ALL", "AP", "P", "S", "SA", "SP"], optional): Type filter for location types. Defaults to "ALL".
* locationSelectionMode (str, optional): Selection mode for locations. "SLCT_N": Not selectable, "SLCT_A": Selectable. Defaults to None.
@ -81,16 +79,7 @@ def stop_by_name(accessId: str,
find_exception(output.json())
if raw_response:
if json:
return output.json()
else:
return xmlparse(output.content)
if json:
return output.json()
else:
stops = []
for stop in output.json()["stopLocationOrCoordLocation"]:
if "StopLocation" in stop:
stops.append(Stop(stop["StopLocation"]))
elif "CoordLocation" in stop:
stops.append(Stop(stop["CoordLocation"]))
return stops
return xmlparse(output.content)