2019-05-15 20:21:12 +02:00
#!/usr/bin/env python3
import requests
import json
from typing import List
from pprint import pprint , pformat
2019-05-15 21:31:11 +02:00
import datetime
import pause
2019-05-15 21:42:37 +02:00
import sys
2019-05-15 20:21:12 +02:00
class MOT :
LONG_DISTANCE_TRAIN = 0
REGIONAL_TRAIN = 1
COMMUTER_TRAIN = 2
UNDERGROUND_TRAIN = 3
TRAM = 4
BUS = 15
ELEVATED_TRAIN = 6
ALL_MODES = [ LONG_DISTANCE_TRAIN , REGIONAL_TRAIN , COMMUTER_TRAIN , UNDERGROUND_TRAIN , TRAM , BUS , ELEVATED_TRAIN ]
ALL_LINES = [ ]
TRIP_CANCELLED = - 9999
2019-05-15 21:31:11 +02:00
lines_filter = [
' rbg:70070: :H ' , # U70 -> Düsseldorf Hbf
' rbg:70070: :R ' , # U70 -> Krefeld Rheinstr
' rbg:70076: :H ' , # U76 -> Düsseldorf Hbf
' rbg:70076: :R ' , # U76 -> Krefeld Rheinstr
]
2019-05-15 20:21:12 +02:00
def t ( s : str ) - > str :
"""
Encode a string to be used as a station identifier .
: param s : a string to encode
: return : the encoded string
"""
return s . replace ( ' ' , ' + ' )
def make_request_data ( station_id : int , result_count : int = 8 , modes : List = MOT . ALL_MODES , lines : List [ str ] = ALL_LINES ) - > dict :
"""
Prepare a request data dictionary to put into get_data ( )
: param station_id : an EFA station ID
: param result_count : how many departures to return
: param modes : which modes of transport to use
: param lines : which lines to use ( line identifiers look like ' provider:line ID: :direction ID ' ,
e . g . ' rbg:70070: :H ' for the Rheinbahn U70 to Düsseldorf Hbf .
: return : a dictionary with the data necessary to make a request to the Abfahrtsmonitor API .
"""
"""
The request data dictionary can have the following items :
stationID : a numerical EFA station ID
stationName : ( optional ) the station ' s name
platformVisibility : ( optional ) ? ? ?
transport : a comma - separated list of the modes of transport to be displayed . See the constants for values .
useAllLines : display all available lines or filter them using the linesFilter
linesFilter : a JSON array with the lines to be displayed . See lines_filter for the format
optimizedForStation : ( optional ) ? ? ?
rowCount : the amount of results to be returned
refreshInterval : ( optional ) ( display parameter ) refresh rate in seconds for the browser UI
distance : ( optional ) ( display parameter ) distance from the monitor to the stop
marquee : ( optional ) ( display parameter ) make the path text scroll sideways
sortBy : ( optional ) ? ? ?
"""
request_data = {
' stationId ' : int ( station_id ) ,
' rowCount ' : result_count
}
# sanity check: do the modes exist?
for mode in modes :
if mode not in MOT . ALL_MODES :
raise ValueError ( str ( mode ) + " Unknown transport mode! " )
# Add the list to the data dictionary
request_data [ ' transport ' ] = ' , ' . join ( " {0} " . format ( n ) for n in modes ) . rstrip ( ' , ' )
if lines is ALL_LINES :
request_data [ ' useAllLines ' ] = 1
else :
lines_dictarr = [ { ' data ' : t ( v ) } for v in lines ]
request_data [ ' linesFilter ' ] = json . dumps ( lines_dictarr )
request_data [ ' useAllLines ' ] = 0
# finally, add the HTML naming
request_data = { " table[departure][ {0} ] " . format ( k ) : v for k , v in request_data . items ( ) }
return request_data
def get_data ( request_data : dict , headers : dict = None , cookies : dict = None ) - > dict :
url = ' https://abfahrtsmonitor.vrr.de/backend/api/stations/table '
reply = requests . post ( url , data = request_data , headers = headers , cookies = cookies )
reply . raise_for_status ( )
2019-05-15 21:42:37 +02:00
print ( ' Request time elapsed: ' + str ( reply . elapsed ) , file = sys . stderr )
2019-05-15 20:21:12 +02:00
return reply . json ( )
def is_cancelled ( trip : dict ) - > bool :
if trip [ ' delay ' ] is not None :
return int ( trip [ ' delay ' ] ) == TRIP_CANCELLED
return False
def is_late ( trip : dict ) - > bool :
if trip [ ' delay ' ] is not None :
return int ( trip [ ' delay ' ] ) > 0
return False
def is_early ( trip : dict ) - > bool :
if trip [ ' delay ' ] is not None :
return int ( trip [ ' delay ' ] ) < 0 and int ( trip [ ' delay ' ] ) != TRIP_CANCELLED
return False
# Pretty-print the reply data.
""" print( " Data: " )
pprint ( reply_data ) """
def fixup_data ( d : dict ) - > dict :
for trip in d [ ' departureData ' ] :
if trip [ ' delay ' ] == ' ' :
trip [ ' delay ' ] = None
return d
2019-05-15 21:42:37 +02:00
def print_trip ( trip : dict ) - > None :
2019-05-15 20:21:12 +02:00
trip_part = " The {} : {} {} (???: {} : : {} ) service to {} " . format ( trip [ ' hour ' ] , trip [ ' minute ' ] , trip [ ' lineNumber ' ] , trip [ ' lineCode ' ] , trip [ ' directionCode ' ] , trip [ ' direction ' ] )
if is_cancelled ( trip ) :
2019-05-15 21:42:37 +02:00
print ( trip_part + " is cancelled. " )
2019-05-15 20:21:12 +02:00
elif is_late ( trip ) :
2019-05-15 21:42:37 +02:00
print ( trip_part + " is {} minutes late. " . format ( trip [ ' delay ' ] ) )
2019-05-15 20:21:12 +02:00
elif is_early ( trip ) :
2019-05-15 21:42:37 +02:00
print ( trip_part + " is {} minutes early. " . format ( - trip [ ' delay ' ] ) )
2019-05-15 20:21:12 +02:00
2019-05-15 21:31:11 +02:00
def get_next_refresh ( data : dict ) :
times = [ ]
for trip in data [ ' departureData ' ] :
times . append ( trip [ ' orgFullTime ' ] )
times . append ( trip [ ' fullTime ' ] )
times = [ int ( time ) for time in times if int ( time ) > datetime . datetime . now ( ) . timestamp ( ) ]
times . sort ( )
for time in times :
if ( datetime . datetime . fromtimestamp ( time ) - datetime . datetime . now ( ) ) > datetime . timedelta ( seconds = 30 ) :
2019-05-15 21:54:52 +02:00
if ( datetime . datetime . fromtimestamp ( time ) - datetime . datetime . now ( ) ) > datetime . timedelta ( minutes = 5 ) :
return ( datetime . datetime . now ( ) + datetime . timedelta ( minutes = 5 ) ) . timestamp ( )
2019-05-15 21:31:11 +02:00
return time
return ( datetime . datetime . now ( ) + datetime . timedelta ( seconds = 60 ) ) . timestamp ( )
def update ( ) :
reply_data = get_data (
make_request_data (
20021002 ,
8 ,
lines = lines_filter
)
)
reply_data = fixup_data ( reply_data )
for trip in reply_data [ ' departureData ' ] :
2019-05-15 21:42:37 +02:00
print_trip ( trip )
2019-05-15 21:31:11 +02:00
return reply_data
def wait ( ) :
data = update ( )
while True :
next_refresh = get_next_refresh ( data )
2019-05-15 21:42:37 +02:00
print ( " Sleeping until " + datetime . datetime . fromtimestamp ( next_refresh ) . isoformat ( ) , file = sys . stderr )
2019-05-15 21:31:11 +02:00
pause . until ( next_refresh )
data = update ( )
def main ( ) :
wait ( )
main ( )