diff --git a/.buildinfo b/.buildinfo
new file mode 100644
index 0000000..c4033e2
--- /dev/null
+++ b/.buildinfo
@@ -0,0 +1,4 @@
+# Sphinx build info version 1
+# This file records the configuration used when building these files. When it is not found, a full rebuild will be done.
+config: 62ae94187439c39bd8a16beba3fe40d7
+tags: 645f666f9bcd5a90fca523b33c5a78b7
diff --git a/.doctrees/environment.pickle b/.doctrees/environment.pickle
new file mode 100644
index 0000000..9b4d6ae
Binary files /dev/null and b/.doctrees/environment.pickle differ
diff --git a/.doctrees/index.doctree b/.doctrees/index.doctree
new file mode 100644
index 0000000..f0eeafa
Binary files /dev/null and b/.doctrees/index.doctree differ
diff --git a/.nojekyll b/.nojekyll
new file mode 100644
index 0000000..e69de29
diff --git a/_modules/index.html b/_modules/index.html
new file mode 100644
index 0000000..952239f
--- /dev/null
+++ b/_modules/index.html
@@ -0,0 +1,98 @@
+
+
+
+
+
+
+
+ Overview: module code — mvg documentation
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+"""Provides the class MvgApi."""
+
+from__future__importannotations
+
+importasyncio
+importre
+fromenumimportEnum
+fromhttpimportHTTPStatus
+fromtypingimportAny
+
+importaiohttp
+fromfurlimportfurl
+
+MVGAPI_DEFAULT_LIMIT=10# API defaults to 10, limits to 100
+
+
+classBase(Enum):
+"""MVG APIs base URLs."""
+
+ FIB="https://www.mvg.de/api/bgw-pt/v3"
+ ZDM="https://www.mvg.de/.rest/zdm"
+
+
+classEndpoint(Enum):
+"""MVG API endpoints with URLs and arguments."""
+
+ FIB_LOCATION:tuple[str,list[str]]=("/locations",["query"])
+ FIB_NEARBY:tuple[str,list[str]]=("/stations/nearby",["latitude","longitude"])
+ FIB_DEPARTURE:tuple[str,list[str]]=("/departures",["globalId","limit","offsetInMinutes"])
+ ZDM_STATION_IDS:tuple[str,list[str]]=("/mvgStationGlobalIds",[])
+ ZDM_STATIONS:tuple[str,list[str]]=("/stations",[])
+ ZDM_LINES:tuple[str,list[str]]=("/lines",[])
+
+
+
+[docs]
+classTransportType(Enum):
+"""MVG products defined by the API with name and icon."""
+
+ BAHN:tuple[str,str]=("Bahn","mdi:train")
+ SBAHN:tuple[str,str]=("S-Bahn","mdi:subway-variant")
+ UBAHN:tuple[str,str]=("U-Bahn","mdi:subway")
+ TRAM:tuple[str,str]=("Tram","mdi:tram")
+ BUS:tuple[str,str]=("Bus","mdi:bus")
+ REGIONAL_BUS:tuple[str,str]=("Regionalbus","mdi:bus")
+ SEV:tuple[str,str]=("SEV","mdi:taxi")
+ SCHIFF:tuple[str,str]=("Schiff","mdi:ferry")
+
+
+[docs]
+ @classmethod
+ defall(cls)->list[TransportType]:
+"""Return a list of all products."""
+ return[getattr(TransportType,c.name)forcinclsifc.name!="SEV"]
+
+
+
+
+
+[docs]
+classMvgApiError(Exception):
+"""Failed communication with MVG API."""
+
+
+
+
+[docs]
+classMvgApi:
+"""A class interface to retrieve stations, lines and departures from the MVG.
+
+ The implementation uses the Münchner Verkehrsgesellschaft (MVG) API at https://www.mvg.de.
+ It can be instanciated by station name and place or global station id.
+
+ :param name: name, place ('Universität, München') or global station id (e.g. 'de:09162:70')
+ :raises MvgApiError: raised on communication failure or unexpected result
+ :raises ValueError: raised on bad station id format
+ """
+
+ def__init__(self,station:str)->None:
+"""Initialize the MVG interface."""
+ station=station.strip()
+ ifnotself.valid_station_id(station):
+ msg="Invalid station."
+ raiseValueError(msg)
+
+ station_details=self.station(station)
+ ifstation_details:
+ self.station_id=station_details["id"]
+
+
+[docs]
+ @staticmethod
+ defvalid_station_id(station_id:str,validate_existance:bool=False)->bool:
+"""Check if the station id is a global station ID according to VDV Recommendation 432.
+
+ :param station_id: a global station id (e.g. 'de:09162:70')
+ :param validate_existance: validate the existance in a list from the API
+ :return: True if valid, False if Invalid
+ """
+ valid_format=bool(re.match("de:[0-9]{2,5}:[0-9]+",station_id))
+ ifnotvalid_format:
+ returnFalse
+
+ ifvalidate_existance:
+ try:
+ result=asyncio.run(MvgApi.__api(Base.ZDM,Endpoint.ZDM_STATION_IDS))
+ ifnotisinstance(result,list):
+ msg=f"Bad API call: Expected a list, but got {type(result)}."
+ raiseMvgApiError(msg)
+ except(AssertionError,KeyError)asexc:
+ msg="Bad API call: Could not parse station data."
+ raiseMvgApiError(msg)fromexc
+ else:
+ returnstation_idinresult
+
+ returnTrue
+
+
+ @staticmethod
+ asyncdef__api(base:Base,endpoint:Endpoint|tuple[str,list[str]],args:dict[str,Any]|None=None)->Any:# noqa: ANN401
+"""Call the API endpoint with the given arguments.
+
+ :param base: the API base
+ :param endpoint: the endpoint
+ :param args: a dictionary containing arguments
+ :raises MvgApiError: raised on communication failure or unexpected result
+ :return: the response as JSON object
+ """
+ url=furl(base.value)
+ url/=endpoint.value[0]ifisinstance(endpoint,Endpoint)elseendpoint[0]
+ url.set(query_params=args)
+
+ try:
+ asyncwithaiohttp.ClientSession()assession,session.get(
+ url.url,
+ )asresp:
+ ifresp.status!=HTTPStatus.OK:
+ msg=f"Bad API call: Got response ({resp.status}) from {url.url}."
+ raiseMvgApiError(msg)
+ ifresp.content_type!="application/json":
+ msg=f"Bad API call: Got content type {resp.content_type} from {url.url}."
+ raiseMvgApiError(msg)
+ returnawaitresp.json()
+
+ exceptaiohttp.ClientErrorasexc:
+ msg=f"Bad API call: Got {type(exc)!s} from {url.url}"
+ raiseMvgApiErrorfromexc
+
+
+[docs]
+ @staticmethod
+ asyncdefstation_ids_async()->list[str]:
+"""Retrieve a list of all valid station ids.
+
+ :raises MvgApiError: raised on communication failure or unexpected result
+ :return: station ids as a list
+ """
+ try:
+ result=awaitMvgApi.__api(Base.ZDM,Endpoint.ZDM_STATION_IDS)
+ ifnotisinstance(result,list):
+ msg=f"Bad API call: Expected a list, but got {type(result)}."
+ raiseMvgApiError(msg)
+ returnsorted(result)
+ except(AssertionError,KeyError)asexc:
+ msg="Bad API call: Could not parse station data."
+ raiseMvgApiError(msg)fromexc
+
+
+
+[docs]
+ @staticmethod
+ asyncdefstations_async()->list[dict[str,Any]]:
+"""Retrieve a list of all stations.
+
+ :raises MvgApiError: raised on communication failure or unexpected result
+ :return: a list of stations as dictionary
+ """
+ try:
+ result=awaitMvgApi.__api(Base.ZDM,Endpoint.ZDM_STATIONS)
+ ifnotisinstance(result,list):
+ msg=f"Bad API call: Expected a list, but got {type(result)}."
+ raiseMvgApiError(msg)
+ except(AssertionError,KeyError)asexc:
+ msg="Bad API call: Could not parse station data."
+ raiseMvgApiError(msg)fromexc
+ else:
+ returnresult
+
+
+
+[docs]
+ @staticmethod
+ defstations()->list[dict[str,Any]]:
+"""Retrieve a list of all stations.
+
+ :raises MvgApiError: raised on communication failure or unexpected result
+ :return: a list of stations as dictionary
+ """
+ returnasyncio.run(MvgApi.stations_async())
+
+
+
+[docs]
+ @staticmethod
+ asyncdeflines_async()->list[dict[str,Any]]:
+"""Retrieve a list of all lines.
+
+ :raises MvgApiError: raised on communication failure or unexpected result
+ :return: a list of lines as dictionary
+ """
+ try:
+ result=awaitMvgApi.__api(Base.ZDM,Endpoint.ZDM_LINES)
+ ifnotisinstance(result,list):
+ msg=f"Bad API call: Expected a list, but got {type(result)}."
+ raiseMvgApiError(msg)
+ except(AssertionError,KeyError)asexc:
+ msg="Bad API call: Could not parse station data."
+ raiseMvgApiError(msg)fromexc
+ else:
+ returnresult
+
+
+
+[docs]
+ @staticmethod
+ deflines()->list[dict[str,Any]]:
+"""Retrieve a list of all lines.
+
+ :raises MvgApiError: raised on communication failure or unexpected result
+ :return: a list of lines as dictionary
+ """
+ returnasyncio.run(MvgApi.stations_async())
+
+
+
+[docs]
+ @staticmethod
+ asyncdefstation_async(query:str)->dict[str,str]|None:
+"""Find a station by station name and place or global station id.
+
+ :param name: name, place ('Universität, München') or global station id (e.g. 'de:09162:70')
+ :raises MvgApiError: raised on communication failure or unexpected result
+ :return: the fist matching station as dictionary with keys 'id', 'name', 'place', 'latitude', 'longitude'
+
+ Example result::
+
+ {"id": "de:09162:6", "name": "Hauptbahnhof", "place": "München", "latitude": 48.14003, "longitude": 11.56107}
+ """# noqa: E501
+ query=query.strip()
+ try:
+ # return details from ZDM if query is a station id
+ ifMvgApi.valid_station_id(query):
+ stations_endpoint=Endpoint.ZDM_STATIONS.value[0]
+ station_endpoint=f"{stations_endpoint}/{query}"
+ result=awaitMvgApi.__api(Base.ZDM,(station_endpoint,[]))
+ ifnotisinstance(result,dict):
+ msg=f"Bad API call: Expected a dict, but got {type(result)}."
+ raiseMvgApiError(msg)
+
+ return{
+ "id":result["id"],
+ "name":result["name"],
+ "place":result["place"],
+ "latitude":result["latitude"],
+ "longitude":result["longitude"],
+ }
+
+ # use open search if query is not a station id
+ args=dict.fromkeys(Endpoint.FIB_LOCATION.value[1])
+ args.update({"query":query.strip(),"locationTypes":"STATION"})
+ result=awaitMvgApi.__api(Base.FIB,Endpoint.FIB_LOCATION,args)
+ ifnotisinstance(result,list):
+ msg=f"Bad API call: Expected a list, but got {type(result)}."
+ raiseMvgApiError(msg)
+
+ # return first location if lis is not empty
+ iflen(result)>0:
+ return{
+ "id":result[0]["globalId"],
+ "name":result[0]["name"],
+ "place":result[0]["place"],
+ "latitude":result[0]["latitude"],
+ "longitude":result[0]["longitude"],
+ }
+
+ except(AssertionError,KeyError)asexc:
+ msg="Bad API call: Could not parse station data."
+ raiseMvgApiError(msg)fromexc
+ else:
+ # return None if no station was found
+ returnNone
+
+
+
+[docs]
+ @staticmethod
+ defstation(query:str)->dict[str,str]|None:
+"""Find a station by station name and place or global station id.
+
+ :param name: name, place ('Universität, München') or global station id (e.g. 'de:09162:70')
+ :raises MvgApiError: raised on communication failure or unexpected result
+ :return: the fist matching station as dictionary with keys 'id', 'name', 'place', 'latitude', 'longitude'
+
+ Example result::
+
+ {"id": "de:09162:6", "name": "Hauptbahnhof", "place": "München", "latitude": 48.14003, "longitude": 11.56107}
+ """# noqa: E501
+ returnasyncio.run(MvgApi.station_async(query))
+
+
+
+[docs]
+ @staticmethod
+ asyncdefnearby_async(latitude:float,longitude:float)->dict[str,str]|None:
+"""Find the nearest station by coordinates.
+
+ :param latitude: coordinate in decimal degrees
+ :param longitude: coordinate in decimal degrees
+ :raises MvgApiError: raised on communication failure or unexpected result
+ :return: the fist matching station as dictionary with keys 'id', 'name', 'place', 'latitude', 'longitude'
+
+ Example result::
+
+ {"id": "de:09162:70", "name": "Universität", "place": "München", "latitude": 48.15007, "longitude": 11.581}
+ """
+ try:
+ args=dict.fromkeys(Endpoint.FIB_NEARBY.value[1])
+ args.update({"latitude":latitude,"longitude":longitude})
+ result=awaitMvgApi.__api(Base.FIB,Endpoint.FIB_NEARBY,args)
+ ifnotisinstance(result,list):
+ msg=f"Bad API call: Expected a list, but got {type(result)}."
+ raiseMvgApiError(msg)
+
+ # return first location of type "STATION"
+ forlocationinresult:
+ return{
+ "id":location["globalId"],
+ "name":location["name"],
+ "place":location["place"],
+ "latitude":result[0]["latitude"],
+ "longitude":result[0]["longitude"],
+ }
+
+ except(AssertionError,KeyError)asexc:
+ msg="Bad API call: Could not parse station data."
+ raiseMvgApiError(msg)fromexc
+ else:
+ # return None if no station was found
+ returnNone
+
+
+
+[docs]
+ @staticmethod
+ defnearby(latitude:float,longitude:float)->dict[str,str]|None:
+"""Find the nearest station by coordinates.
+
+ :param latitude: coordinate in decimal degrees
+ :param longitude: coordinate in decimal degrees
+ :raises MvgApiError: raised on communication failure or unexpected result
+ :return: the fist matching station as dictionary with keys 'id', 'name', 'place', 'latitude', 'longitude'
+
+ Example result::
+
+ {"id": "de:09162:70", "name": "Universität", "place": "München", "latitude": 48.15007, "longitude": 11.581}
+ """
+ returnasyncio.run(MvgApi.nearby_async(latitude,longitude))
+
+
+
+[docs]
+ @staticmethod
+ asyncdefdepartures_async(
+ station_id:str,
+ limit:int=MVGAPI_DEFAULT_LIMIT,
+ offset:int=0,
+ transport_types:list[TransportType]|None=None,
+ )->list[dict[str,Any]]:
+"""Retreive the next departures for a station by station id.
+
+ :param station_id: the global station id ('de:09162:70')
+ :param limit: limit of departures, defaults to 10
+ :param offset: offset (e.g. walking distance to the station) in minutes, defaults to 0
+ :param transport_types: filter by transport type, defaults to None
+ :raises MvgApiError: raised on communication failure or unexpected result
+ :raises ValueError: raised on bad station id format
+ :return: a list of departures as dictionary
+
+ Example result::
+
+ [
+ {
+ "time": 1668524580,
+ "planned": 1668524460,
+ "line": "U3",
+ "destination": "Fürstenried West",
+ "type": "U-Bahn",
+ "icon": "mdi:subway",
+ "cancelled": False,
+ "messages": [],
+ },
+ ...,
+ ]
+ """
+ station_id.strip()
+ ifnotMvgApi.valid_station_id(station_id):
+ msg="Invalid format of global staton id."
+ raiseValueError(msg)
+
+ try:
+ args=dict.fromkeys(Endpoint.FIB_DEPARTURE.value[1])
+ args.update({"globalId":station_id,"offsetInMinutes":offset,"limit":limit})
+ iftransport_typesisNone:
+ transport_types=TransportType.all()
+ args.update({"transportTypes":",".join([product.nameforproductintransport_types])})
+ result=awaitMvgApi.__api(Base.FIB,Endpoint.FIB_DEPARTURE,args)
+ ifnotisinstance(result,list):
+ msg=f"Bad API call: Expected a list, but got {type(result)}."
+ raiseMvgApiError(msg)
+
+ departures=[
+ {
+ "time":int(departure["realtimeDepartureTime"]/1000),
+ "planned":int(departure["plannedDepartureTime"]/1000),
+ "line":departure["label"],
+ "destination":departure["destination"],
+ "type":TransportType[departure["transportType"]].value[0],
+ "icon":TransportType[departure["transportType"]].value[1],
+ "cancelled":departure["cancelled"],
+ "messages":departure["messages"],
+ }
+ fordepartureinresult
+ ]
+
+ except(AssertionError,KeyError)asexc:
+ msg="Bad MVG API call: Invalid departure data."
+ raiseMvgApiError(msg)fromexc
+ else:
+ returndepartures
+
+
+
+[docs]
+ defdepartures(
+ self,
+ limit:int=MVGAPI_DEFAULT_LIMIT,
+ offset:int=0,
+ transport_types:list[TransportType]|None=None,
+ )->list[dict[str,Any]]:
+"""Retreive the next departures.
+
+ :param limit: limit of departures, defaults to 10
+ :param offset: offset (e.g. walking distance to the station) in minutes, defaults to 0
+ :param transport_types: filter by transport type, defaults to None
+ :raises MvgApiError: raised on communication failure or unexpected result
+ :return: a list of departures as dictionary
+
+ Example result::
+
+ [
+ {
+ "time": 1668524580,
+ "planned": 1668524460,
+ "line": "U3",
+ "destination": "Fürstenried West",
+ "type": "U-Bahn",
+ "icon": "mdi:subway",
+ "cancelled": False,
+ "messages": [],
+ },
+ ...,
+ ]
+
+ """
+ returnasyncio.run(self.departures_async(self.station_id,limit,offset,transport_types))
This package aims to provide a clean, performant and barrier-free interface to timetable information of the Münchner Verkehrsgesellschaft (MVG), responsible for public transport in Munich. It exports the class MvgApi to retrieve stations, lines and departures from the unofficial JSON API at https://www.mvg.de.
This project is not an official project from the Münchner Verkehrsgesellschaft (MVG). It was developed as a private project from lack of a documented and openly accessible API. It simply reproduces the requests made by https://www.mvg.de to provide a barrier-free access to local timetable information.
+
Therefore, the following usage restrictions from the MVG Imprint do apply to all users of this package:
+
+
Our systems are used for direct customer interaction. The processing of our content or data by third parties requires our express consent. For private, non-commercial purposes, moderate use is tolerated without our explicit consent. Any form of data mining does not constitute moderate use. We reserve the right to revoke this permission in principle or in individual cases. Please direct any questions to: redaktion@mvg.de
The project was inspired by two existing packages:
+
+
The package PyMVGLive from 2017 does provide an interface to the former MVGLive API at mvg-live.de. As of 2022 the MVGLive website does not exist anymore and the package has been archived. Although the old API still works for some stations, it does not for others - mainly due to updated station identifiers. Therefore, the package is considered deprecated and cannot be used for new designs.
+
The newer package mvg-api offers an implementation from 2020 based on the API at www.mvg.de/api/fahrinfo. It considers the updated station identifiers and still works perfectly. This package provides the basis for recent projects such as mvg-cli.
+
+
So why another MVG API package? In the end three reasons were decisive:
+
+
The recent website at uses a new API at www.mvg.de/api/fib/v2, which seems to be more performant than the previous one.
+
None of the existing packages offer asynchronous calls for concurrent code projects.
+
An optimized package was required to develop a Home Assistant integration.
The interface was designed to be simple and intuitive. Basic usage follows these steps:
+
+
Find a station using MvgApi.station(station) by its name and place (e.g. "Universität,München") or its global station identifier (e.g. "de:09162:70").
+
Alternatively, MvgApi.nearby(latitude,longitude) finds the nearest station.
+
Create an API instance using MvgApi(station) by station name and place or its global identifier.
+
Use the method .departures() to retrieve information from the API.
The static methods MvgApi.stations() and MvgApi.lines() expose a list of all available stations and a list of all available lines from designated API endpoints. While these calls are great for reference, they are also quite extensive and should not be used within a frequent query loop.
The class MvgApi internally calls asynchronous methods using asyncio and aiohttp to perform the web requests efficiently. These asynchronous methods are marked by the suffix _async and can be utilized by users in projects with concurrent code.
+
The basic example but with asynchronous calls looks like this:
A class interface to retrieve stations, lines and departures from the MVG.
+
The implementation uses the Münchner Verkehrsgesellschaft (MVG) API at https://www.mvg.de.
+It can be instanciated by station name and place or global station id.
+
+
Parameters:
+
name – name, place (‘Universität, München’) or global station id (e.g. ‘de:09162:70’)
+
+
Raises:
+
+
MvgApiError – raised on communication failure or unexpected result