159 lines
5.3 KiB
Python
Executable File
159 lines
5.3 KiB
Python
Executable File
#!/usr/bin/python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
from subprocess import check_output
|
|
import requests
|
|
import json
|
|
import sys
|
|
from datetime import datetime
|
|
import time
|
|
import pytz
|
|
|
|
timezone = pytz.timezone('Europe/Berlin')
|
|
statusData = None #initialize variables...
|
|
tripData = None
|
|
|
|
tlApiToken="[YOUR API TOKEN]
|
|
|
|
# Request the raw data from ICE-Portal and raise an exception, when it is not avialable.
|
|
def get_api_db():
|
|
global statusData
|
|
global tripData
|
|
try:
|
|
statusData_response = requests.get("https://iceportal.de/api1/rs/status", timeout=5)
|
|
statusData = statusData_response.json()
|
|
tripData_response = requests.get("https://iceportal.de/api1/rs/tripInfo/trip", timeout=5)
|
|
tripData = tripData_response.json()
|
|
return 1
|
|
except requests.exceptions.RequestException:
|
|
return 0
|
|
|
|
# Request data from travelynx
|
|
def get_api_tl():
|
|
global tripDataTl
|
|
tripDataTl_response = requests.get(f"https://travelynx.de/api/v1/status/{tlApiToken}", timeout=5)
|
|
tripDataTl = tripDataTl_response.json()
|
|
|
|
def get_api_cd():
|
|
global statusDataCD
|
|
global tripDataCD
|
|
try:
|
|
statusDataCD_response = requests.get("http://cdwifi.cz/portal/api/vehicle/realtime", timeout=5)
|
|
statusDataCD = statusDataCD_response.json()
|
|
return 1
|
|
except requests.exceptions.RequestException:
|
|
return 0
|
|
|
|
|
|
|
|
# Look for the current SSID. Replace wlp2s0 with the real name of your wifi interface and check, that the path to iwlist is correct (just type 'which iwlist' in the terminal)
|
|
def get_ssid():
|
|
try:
|
|
output = check_output(['/usr/sbin/iwgetid']).decode('utf-8')
|
|
ssid=output.split('"')[1]
|
|
return ssid
|
|
except:
|
|
pass
|
|
|
|
|
|
# Get Speed from ICE-Portal
|
|
def get_speed_db():
|
|
try:
|
|
speed= str(statusData["speed"])
|
|
if speed==None:
|
|
speed="No Speed"
|
|
return speed
|
|
|
|
except:
|
|
return "No Speed"
|
|
|
|
def get_speed_cd():
|
|
try:
|
|
speed=str(statusDataCD['speed'])
|
|
if speed==None:
|
|
speed="No Speed"
|
|
return speed
|
|
except:
|
|
return "No Speed"
|
|
|
|
|
|
|
|
# Figure out, whats the next station and filter the output for this.
|
|
def get_next_db():
|
|
if tripData["trip"]["stopInfo"] == None :
|
|
return "No TripInfo"
|
|
else:
|
|
next_stop=tripData["trip"]["stopInfo"]["actualNext"]
|
|
for i in tripData['trip']['stops']:
|
|
if i['station']['evaNr'] == next_stop:
|
|
nextStationName = i['station']['name']
|
|
arrivalDelay = i['timetable']['arrivalDelay']
|
|
utime=i['timetable']['actualArrivalTime']/1000
|
|
arrivalTime = datetime.fromtimestamp(utime, tz=timezone).strftime('%H:%M')
|
|
nextStationTrack = str(i['track']['actual'])
|
|
break
|
|
tripinfo = nextStationName+" "+ arrivalTime+" ("+ arrivalDelay+") @ "+ nextStationTrack
|
|
return tripinfo
|
|
|
|
def get_next_tl():
|
|
try:
|
|
for i in tripDataTl['intermediateStops']:
|
|
if i['realArrival'] >= time.time():
|
|
nextStationName = i['name']
|
|
uScheduledTime=i['scheduledArrival']
|
|
uActualTime=i['realArrival']
|
|
arrivalDelay="{:+.0f}".format(((uActualTime-uScheduledTime)/60))
|
|
arrivalTime = datetime.fromtimestamp(uActualTime, tz=timezone).strftime('%H:%M')
|
|
tripinfo = nextStationName+" "+arrivalTime+" ("+arrivalDelay+")"
|
|
break
|
|
except:
|
|
nextStationName=tripDataTl['toStation']['name']
|
|
tripinfo=nextStationName
|
|
return tripinfo
|
|
|
|
|
|
# Get the Trainnumber and final station of your current train
|
|
def get_trainInfo_db():
|
|
if tripData['trip']["trainType"]==None:
|
|
return "No TrainInfo"
|
|
else:
|
|
trainType = tripData["trip"]["trainType"]
|
|
trainNumber = tripData["trip"]["vzn"]
|
|
finalStationName = tripData["trip"]["stopInfo"]["finalStationName"]
|
|
return trainType+" "+trainNumber+" "+ finalStationName
|
|
|
|
# Get the Trainnumber and final station of your current travelynx trip
|
|
def get_trainInfo_tl():
|
|
trainType = tripDataTl["train"]["type"]
|
|
trainNumber = tripDataTl["train"]["no"]
|
|
fromStationName = tripDataTl["fromStation"]["name"]
|
|
toStationName = tripDataTl["toStation"]["name"]
|
|
toStationTrack = tripDataTl["toStation"]["platform"]
|
|
uScheduledTime=tripDataTl['toStation']['scheduledTime']
|
|
uActualTime=tripDataTl['toStation']['realTime']
|
|
arrivalDelay="{:+.0f}".format(((uActualTime-uScheduledTime)/60))
|
|
arrivalTime = datetime.fromtimestamp(uActualTime, tz=timezone).strftime('%H:%M')
|
|
return trainType+" "+trainNumber+" | "+fromStationName+" → "+ toStationName+" "+arrivalTime+" ("+arrivalDelay+") @ "+ toStationTrack
|
|
|
|
|
|
def main():
|
|
# Look for WIFIonICE or WIFI@DB otherwise do nothing...
|
|
if get_ssid() == "WIFIonICE":
|
|
if get_api_db() == 1:
|
|
print(get_trainInfo_db()+" | "+"SPEED: "+get_speed_db()+" km/h | "+"NEXT: "+get_next_db())
|
|
else:
|
|
get_api_tl()
|
|
print(get_trainInfo_tl()+" | NEXT: "+get_next_tl())
|
|
|
|
if get_ssid() == "WIFI@DB" or "RRX Hotspot":
|
|
get_api_tl()
|
|
print(get_trainInfo_tl()+" | NEXT: "+get_next_tl())
|
|
|
|
if get_ssid() == "CDWiFi":
|
|
get_api_tl()
|
|
get_api_cd()
|
|
print(get_trainInfo_tl()+" | "+"SPEED: "+get_speed_cd()+" km/h | NEXT: "+get_next_tl())
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|