PKNNRMVtransport/__init__.py"""Define module-level imports.""" # pylint: disable=C0103 from .rmvtransport import RMVtransport # noqa __version__ = "0.2.7" PKǀvN.'uGGRMVtransport/const.py"""Constants.""" from typing import List, Dict PRODUCTS: Dict[str, int] = { "ICE": 1, "IC": 2, "EC": 2, "RB": 4, "RE": 4, "S": 8, "U-Bahn": 16, "Tram": 32, "Bus": 64, "Bus2": 128, "Fähre": 256, "Taxi": 512, "Bahn": 1024, } ALL_PRODUCTS: List[str] = list(PRODUCTS.keys()) PKSZMIjRMVtransport/errors.py"""Define package errors.""" class RMVtransportError(Exception): """General error exception occurred.""" pass class RMVtransportApiConnectionError(RMVtransportError): """When a connection error is encountered.""" pass PKǀvNXRMVtransport/rmvjourney.py"""This class represents a single journey.""" from datetime import datetime, timedelta import html from typing import List, Dict, Any, Optional from lxml import objectify # type: ignore from .const import PRODUCTS class RMVJourney: """A journey object to hold information about a journey.""" # pylint: disable=I1101 def __init__(self, journey: objectify.ObjectifiedElement, now: datetime) -> None: """Initialize the journey object.""" self.journey: objectify.ObjectifiedElement = journey self.now: datetime = now self.attr_types = self.journey.JourneyAttributeList.xpath("*/Attribute/@type") self.name: str = self._extract("NAME") self.number: str = self._extract("NUMBER") self.product: str = self._extract("CATEGORY") self.train_id: str = self.journey.get("trainId") self.departure: datetime = self._departure() self.delay: int = self._delay() self.real_departure_time: datetime = self._real_departure_time() self.real_departure: int = self._real_departure() self.direction = self._extract("DIRECTION") self.info = self._info() self.info_long = self._info_long() self.platform = self._platform() self.stops = self._pass_list() self.icon = self._icon() def _platform(self) -> Optional[str]: """Extract platform.""" try: return str(self.journey.MainStop.BasicStop.Dep.Platform.text) except AttributeError: return None def _delay(self) -> int: """Extract departure delay.""" try: return int(self.journey.MainStop.BasicStop.Dep.Delay.text) except AttributeError: return 0 def _departure(self) -> datetime: """Extract departure time.""" departure_time = datetime.strptime( self.journey.MainStop.BasicStop.Dep.Time.text, "%H:%M" ).time() if departure_time > (self.now - timedelta(hours=1)).time(): return datetime.combine(self.now.date(), departure_time) return datetime.combine(self.now.date() + timedelta(days=1), departure_time) def _real_departure_time(self) -> datetime: """Calculate actual departure time.""" return self.departure + timedelta(minutes=self.delay) def _real_departure(self) -> int: """Calculate actual minutes left for departure.""" return round((self.real_departure_time - self.now).seconds / 60) def _extract(self, attribute) -> str: """Extract train information.""" attr_data = self.journey.JourneyAttributeList.JourneyAttribute[ self.attr_types.index(attribute) ].Attribute attr_variants = attr_data.xpath("AttributeVariant/@type") data = attr_data.AttributeVariant[attr_variants.index("NORMAL")].Text.pyval return str(data) def _info(self) -> Optional[str]: """Extract journey information.""" try: return str(html.unescape(self.journey.InfoTextList.InfoText.get("text"))) except AttributeError: return None def _info_long(self) -> Optional[str]: """Extract journey information.""" try: return str( html.unescape(self.journey.InfoTextList.InfoText.get("textL")).replace( "
", "\n" ) ) except AttributeError: return None def _pass_list(self) -> List[Dict[str, Any]]: """Extract next stops along the journey.""" stops: List[Dict[str, Any]] = [] for stop in self.journey.PassList.BasicStop: index = stop.get("index") station = stop.Location.Station.HafasName.Text.text station_id = stop.Location.Station.ExternalId.text stops.append({"index": index, "stationId": station_id, "station": station}) return stops def _icon(self) -> str: """Extract product icon.""" pic_url = "https://www.rmv.de/auskunft/s/n/img/products/%i_pic.png" return pic_url % PRODUCTS[self.product] PK}?yNǯTRMVtransport/rmvtransport.py"""A module to query bus and train departure times.""" import asyncio import urllib.request import urllib.parse from datetime import datetime import logging from typing import List, Dict, Any, Optional, Union import aiohttp import async_timeout from lxml import objectify, etree # type: ignore from .errors import RMVtransportError, RMVtransportApiConnectionError from .rmvjourney import RMVJourney from .const import PRODUCTS, ALL_PRODUCTS _LOGGER = logging.getLogger(__name__) BASE_URI: str = "http://www.rmv.de/auskunft/bin/jp/" QUERY_PATH: str = "query.exe/" GETSTOP_PATH: str = "ajax-getstop.exe/" STBOARD_PATH: str = "stboard.exe/" class RMVtransport: """Connection data and travel information.""" def __init__(self, session: aiohttp.ClientSession, timeout: int = 10) -> None: """Initialize connection data.""" self._session: aiohttp.ClientSession = session self._timeout: int = timeout self.now: datetime self.station: str self.station_id: str self.direction_id: Optional[str] self.products_filter: str self.max_journeys: int self.obj: objectify.ObjectifiedElement # pylint: disable=I1101 self.journeys: List[RMVJourney] = [] async def get_departures( self, station_id: str, direction_id: Optional[str] = None, max_journeys: int = 20, products: Optional[List[str]] = None, ) -> Dict[str, Any]: """Fetch data from rmv.de.""" self.station_id: str = station_id self.direction_id: str = direction_id self.max_journeys: int = max_journeys self.products_filter: str = _product_filter(products or ALL_PRODUCTS) base_url: str = _base_url() params: Dict[str, Union[str, int]] = { "selectDate": "today", "time": "now", "input": self.station_id, "maxJourneys": self.max_journeys, "boardType": "dep", "productsFilter": self.products_filter, "disableEquivs": "discard_nearby", "output": "xml", "start": "yes", } if self.direction_id: params["dirInput"] = self.direction_id url = base_url + urllib.parse.urlencode(params) try: with async_timeout.timeout(self._timeout): async with self._session.get(url) as response: _LOGGER.debug(f"Response from RMV API: {response.status}") xml = await response.read() _LOGGER.debug(xml) except (asyncio.TimeoutError, aiohttp.ClientError): _LOGGER.error("Can not load data from RMV API") raise RMVtransportApiConnectionError() # pylint: disable=I1101 try: self.obj = objectify.fromstring(xml) except (TypeError, etree.XMLSyntaxError): _LOGGER.debug(f"Get from string: {xml[:100]}") print(f"Get from string: {xml}") raise RMVtransportError() try: self.now = self.current_time() self.station = self._station() except (TypeError, AttributeError): _LOGGER.debug( f"Time/Station TypeError or AttributeError {objectify.dump(self.obj)}" ) raise RMVtransportError() self.journeys.clear() try: for journey in self.obj.SBRes.JourneyList.Journey: self.journeys.append(RMVJourney(journey, self.now)) except AttributeError: _LOGGER.debug(f"Extract journeys: {objectify.dump(self.obj.SBRes)}") raise RMVtransportError() return self.data() def data(self) -> Dict[str, Any]: """Return travel data.""" data: Dict[str, Any] = {} data["station"] = self.station data["stationId"] = self.station_id data["filter"] = self.products_filter journeys = [] for j in sorted(self.journeys, key=lambda k: k.real_departure)[ : self.max_journeys ]: journeys.append( { "product": j.product, "number": j.number, "trainId": j.train_id, "direction": j.direction, "departure_time": j.real_departure_time, "minutes": j.real_departure, "delay": j.delay, "stops": [s["station"] for s in j.stops], "info": j.info, "info_long": j.info_long, "icon": j.icon, } ) data["journeys"] = journeys return data def _station(self) -> str: """Extract station name.""" return str(self.obj.SBRes.SBReq.Start.Station.HafasName.Text.pyval) def current_time(self) -> datetime: """Extract current time.""" _date = datetime.strptime(self.obj.SBRes.SBReq.StartT.get("date"), "%Y%m%d") _time = datetime.strptime(self.obj.SBRes.SBReq.StartT.get("time"), "%H:%M") return datetime.combine(_date.date(), _time.time()) def output(self) -> None: """Pretty print travel times.""" print("%s - %s" % (self.station, self.now)) print(self.products_filter) for j in sorted(self.journeys, key=lambda k: k.real_departure)[ : self.max_journeys ]: print("-------------") print(f"{j.product}: {j.number} ({j.train_id})") print(f"Richtung: {j.direction}") print(f"Abfahrt in {j.real_departure} min.") print(f"Abfahrt {j.departure.time()} (+{j.delay})") print(f"Nächste Haltestellen: {([s['station'] for s in j.stops])}") if j.info: print(f"Hinweis: {j.info}") print(f"Hinweis (lang): {j.info_long}") print(f"Icon: {j.icon}") def _product_filter(products) -> str: """Calculate the product filter.""" _filter = 0 for product in {PRODUCTS[p] for p in products}: _filter += product return format(_filter, "b")[::-1] def _base_url() -> str: """Build base url.""" _lang: str = "d" _type: str = "n" _with_suggestions: str = "?" return BASE_URI + STBOARD_PATH + _lang + _type + _with_suggestions PKpL(l\''&PyRMVtransport-0.2.7.dist-info/LICENSEMIT License Copyright (c) 2018 cgtobi Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!HPO$PyRMVtransport-0.2.7.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,szd&Y)r$[)T&UrPK!H;+I 'PyRMVtransport-0.2.7.dist-info/METADATAVmo6_%l1$M2k lI!(J:l(#);Jrib{ ,e Y@7a)pj%zDϼiL99π2,A:Cw2b -U'.F2ʐ-=yAɟOmnRן}H_HW5M841==w&\Yt?o ,<)ƂPgm r=$or ]oosc"Vh1.]˪`,Ic:)ш칍H.=0Sd[ _D̂gv+78+4ξWيd X[;3~Ub+(dY̷;4`V5 Vz6\!yA07W$<8P!9 gsS̵LT&Eu1Pϻ~u$`|kDJޒ,^o/XƂfbl!`os7|ӄʥ e6: UeI%ӔQ*U6Fu}?2鿿_L^Zp<`YX%8U;pF\1rc!ABe!@8d ',F\mk^U1mg9^BؼH0Zj1WXC^ "ۆbYX&DY2'+,r=6 3KyՌ@[&o4 JiÜt{Gۻ]@DC!_pV,UH5P ýVOǂx0㊨C1G=2jӭ15hT:35K ft?q0D;+E^%e+JNš֦DAΆMgҝCrWj6-{b:iv?ܭ͝xԩM4=kkFit*[ܛ0/2ft"!U$qIcz=!2* X^]?D17x%Z[T&<gwDXE,q86yBJ\Db3;sγYY.f$4[VgWPK!H".%PyRMVtransport-0.2.7.dist-info/RECORDһv@g 7)P`hf@D"<9MJaookøe3+bB9(/MBl=kO*pB!. ogXCk+R$kqsI} _}h#I +ƲnS^Es'U#Y?>knG+N+66`yއkM*gb