PythonRMV/pyrmv/raw/himSearch.py
2022-09-22 19:27:34 +02:00

121 lines
5.6 KiB
Python

from requests import get
from typing import Dict, Union
from xmltodict import parse as xmlparse
from datetime import datetime
try:
from typing import Literal
except ImportError:
from typing_extensions import Literal
# 2.37. HIM Search (himsearch)
def himSearch(accessId: str,
json: bool = True,
dateB: str = None,
dateE: str = None,
timeB: str = None,
timeE: str = None,
weekdays: Dict[str, bool] = None,
himIds: Union[str, list] = None,
hierarchicalView: bool = False,
operators: Union[str, list] = None,
categories: Union[str, list] = None,
channels: Union[str, list] = None,
companies: Union[str, list] = None,
lines: Union[str, list] = None,
lineids: Union[str, list] = None,
stations: Union[str, list] = None,
fromstation: str = None,
tostation: str = None,
bothways: bool = None,
trainnames: Union[str, list] = None,
metas: Union[str, list] = None,
himcategory: str = None,
himtags: Union[str, list] = None,
regions: Union[str, list] = None,
himtext: Union[str, list] = None,
himtexttags: Union[str, list] = None,
additionalfields: Union[str, list] = None,
poly: bool = False,
searchmode: Literal["MATCH", "NOMATCH", "TFMATCH"] = None,
affectedJourneyMode: Literal["ALL", "OFF"] = None,
affectedJourneyStopMode: Literal["ALL", "IMP", "OFF"] = None,
orderBy: Union[str, list] = None,
minprio: Union[str, int] = None,
maxprio: Union[str, int] = None
) -> dict:
"""The himSearch will return a list of HIM messages if matched by the given criteria as well as affected
products if any.
Read more about this in section 2.37. "HIM Search (himsearch)" of HAFAS ReST Documentation.
### Args:
* accessId (str): _description_
* json (bool, optional): _description_. Defaults to True.
* dateB (str, optional): _description_. Defaults to None.
* dateE (str, optional): _description_. Defaults to None.
* timeB (str, optional): _description_. Defaults to None.
* timeE (str, optional): _description_. Defaults to None.
* weekdays (Dict[str, bool], optional): _description_. Defaults to None.
* himIds (Union[str, list], optional): _description_. Defaults to None.
* hierarchicalView (bool, optional): _description_. Defaults to False.
* operators (Union[str, list], optional): _description_. Defaults to None.
* categories (Union[str, list], optional): _description_. Defaults to None.
* channels (Union[str, list], optional): _description_. Defaults to None.
* companies (Union[str, list], optional): _description_. Defaults to None.
* lines (Union[str, list], optional): _description_. Defaults to None.
* lineids (Union[str, list], optional): _description_. Defaults to None.
* stations (Union[str, list], optional): _description_. Defaults to None.
* fromstation (str, optional): _description_. Defaults to None.
* tostation (str, optional): _description_. Defaults to None.
* bothways (bool, optional): _description_. Defaults to None.
* trainnames (Union[str, list], optional): _description_. Defaults to None.
* metas (Union[str, list], optional): _description_. Defaults to None.
* himcategory (str, optional): _description_. Defaults to None.
* himtags (Union[str, list], optional): _description_. Defaults to None.
* regions (Union[str, list], optional): _description_. Defaults to None.
* himtext (Union[str, list], optional): _description_. Defaults to None.
* himtexttags (Union[str, list], optional): _description_. Defaults to None.
* additionalfields (Union[str, list], optional): _description_. Defaults to None.
* poly (bool, optional): _description_. Defaults to False.
* searchmode (Literal["MATCH", "NOMATCH", "TFMATCH"], optional): _description_. Defaults to None.
* affectedJourneyMode (Literal["ALL", "OFF"], optional): _description_. Defaults to None.
* affectedJourneyStopMode (Literal["ALL", "IMP", "OFF"], optional): _description_. Defaults to None.
* orderBy (Union[str, list], optional): _description_. Defaults to None.
* minprio (Union[str, int], optional): _description_. Defaults to None.
* maxprio (Union[str, int], optional): _description_. Defaults to None.
### Returns:
* dict: Output from RMV. Dict will contain "errorCode" and "errorText" if exception occurs.
"""
if json:
headers = {"Accept": "application/json"}
else:
headers = {"Accept": "application/xml"}
payload = {}
for var, val in locals().items():
if str(var) in ["dateB", "dateE"]:
if val != None:
if isinstance(val, datetime):
payload[str(var)] = val.strftime("%Y-%m-%d")
else:
payload[str(var)] = val
elif str(var) in ["timeB", "timeE"]:
if val != None:
if isinstance(val, datetime):
payload[str(var)] = val.strftime("%H:%M")
else:
payload[str(var)] = val
elif str(var) not in ["json", "headers", "payload"]:
if val != None:
payload[str(var)] = val
output = get("https://www.rmv.de/hapi/himsearch", params=payload, headers=headers)
if json:
return output.json()
else:
return xmlparse(output.content)