mvg

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.

Disclaimer

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

(from https://www.mvg.de/impressum.html, accessed on 04. Feb 2023)

Why another MVG package?

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.

Installation

Install from the Python Package Index (PyPI) using pip:

pip install mvg

Basic Usage

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.

A basic example looks like this:

from mvg import MvgApi

station = MvgApi.station('Universität, München')
if station:
    mvgapi = MvgApi(station['id'])
    departures = mvgapi.departures()
    print(station, departures)

Available Stations and Lines

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.

Filters

The results from .departures(limit, offset, transport_types) can be filtered using the following arguments:

limit limits the output to the given number of departures, defaults to 10
offset adds an offset (e.g. walking distance to the station) in minutes, defaults to 0
transport_types filters the result by a list of transport types (e.g. [TransportType.UBAHN])

A filtered example looks like this:

from mvg import MvgApi, TransportType

station = MvgApi.station('Universität, München')
if station:
    mvgapi = MvgApi(station['id'])
    departures = mvgapi.departures(
        limit=3,
        offset=5,
        transport_types=[TransportType.UBAHN])
    print(station, departures)

Example results

station() or nearby() results a dict:

{
'id': 'de:09162:70',
'name': 'Universität',
'place': 'München'
'latitude': 48.15007,
'longitude': 11.581
}

departures() results a list of dict:

[{
'time': 1668524580,
'planned': 1668524460,
'line': 'U3',
'destination': 'Fürstenried West',
'type': 'U-Bahn',
'icon': 'mdi:subway',
'cancelled': False,
'messages': []
}, ... ]

Advanced Usage: Asynchronous Methods

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:

import asyncio
from mvg import MvgApi

async def demo() -> None:
    station = await MvgApi.station_async('Universität, München')
    if station:
        departures = MvgApi.departures_async(station['id'])
        print(station, await departures)
loop = asyncio.get_event_loop()
loop.run_until_complete(demo())

Module Reference

An unofficial interface to timetable information of the Münchner Verkehrsgesellschaft (MVG).

class mvg.MvgApi(station: str)[source]

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
ValueError – raised on bad station id format

departures(limit: int = 10, offset: int = 0, transport_types: list[TransportType] | None = None) → list[dict[str, Any]][source]

Retreive the next departures.

Parameters:

limit – limit of departures, defaults to 10
offset – offset (e.g. walking distance to the station) in minutes, defaults to 0
transport_types – filter by transport type, defaults to None

Raises:

MvgApiError – raised on communication failure or unexpected result

Returns:

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': []
}, ... ]

async static departures_async(station_id: str, limit: int = 10, offset: int = 0, transport_types: list[TransportType] | None = None) → list[dict[str, Any]][source]

Retreive the next departures for a station by station id.

Parameters:

station_id – the global station id (‘de:09162:70’)
limit – limit of departures, defaults to 10
offset – offset (e.g. walking distance to the station) in minutes, defaults to 0
transport_types – filter by transport type, defaults to None

Raises:

MvgApiError – raised on communication failure or unexpected result
ValueError – raised on bad station id format

Returns:

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': []
}, ... ]

static lines() → list[dict[str, Any]][source]

Retrieve a list of all lines.

Raises:: MvgApiError – raised on communication failure or unexpected result
Returns:: a list of lines as dictionary

async static lines_async() → list[dict[str, Any]][source]

Retrieve a list of all lines.

Raises:: MvgApiError – raised on communication failure or unexpected result
Returns:: a list of lines as dictionary

static nearby(latitude: float, longitude: float) → dict[str, str] | None[source]

Find the nearest station by coordinates.

Parameters:

latitude – coordinate in decimal degrees
longitude – coordinate in decimal degrees

Raises:

MvgApiError – raised on communication failure or unexpected result

Returns:

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}

async static nearby_async(latitude: float, longitude: float) → dict[str, str] | None[source]

Find the nearest station by coordinates.

Parameters:

latitude – coordinate in decimal degrees
longitude – coordinate in decimal degrees

Raises:

MvgApiError – raised on communication failure or unexpected result

Returns:

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}

static station(query: str) → dict[str, str] | None[source]

Find a station 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
Returns:: 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}

async static station_async(query: str) → dict[str, str] | None[source]

Find a station 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
Returns:: 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}

async static station_ids_async() → list[str][source]

Retrieve a list of all valid station ids.

Raises:: MvgApiError – raised on communication failure or unexpected result
Returns:: station ids as a list

static stations() → list[dict[str, Any]][source]

Retrieve a list of all stations.

Raises:: MvgApiError – raised on communication failure or unexpected result
Returns:: a list of stations as dictionary

async static stations_async() → list[dict[str, Any]][source]

Retrieve a list of all stations.

Raises:: MvgApiError – raised on communication failure or unexpected result
Returns:: a list of stations as dictionary

static valid_station_id(station_id: str, validate_existance: bool = False) → bool[source]

Check if the station id is a global station ID according to VDV Recommendation 432.

Parameters:

station_id – a global station id (e.g. ‘de:09162:70’)
validate_existance – validate the existance in a list from the API

Returns:

True if valid, False if Invalid

exception mvg.MvgApiError[source]: Failed communication with MVG API.

class mvg.TransportType(value)[source]

MVG products defined by the API with name and icon.

BAHN: tuple[str, str] = ('Bahn', 'mdi:train')

BUS: tuple[str, str] = ('Bus', 'mdi:bus')

REGIONAL_BUS: tuple[str, str] = ('Regionalbus', 'mdi:bus')

SBAHN: tuple[str, str] = ('S-Bahn', 'mdi:subway-variant')

SCHIFF: tuple[str, str] = ('Schiff', 'mdi:ferry')

SEV: tuple[str, str] = ('SEV', 'mdi:taxi')

TRAM: tuple[str, str] = ('Tram', 'mdi:tram')

UBAHN: tuple[str, str] = ('U-Bahn', 'mdi:subway')

classmethod all() → list[TransportType][source]: Return a list of all products.