ich bin neu hier im Forum und habe leider sehr wenig Erfahrung in Python-Programmierung.
Ich beabsichtige die Pünktlichkeit einzelner Linien des Nahverkehrs zu analysieren. Dafür eignet sich die frei zugängliche elektronische Fahrplanauskunft des VRR.
Im Netz habe ich einen für einen etwas anderen Ansatz funktionierenden Code gefunden. Hierbei ging es im Wesentlichen um Fahrtausfälle.
Wenn man die aktuelle Abfrage mit screen startet, sind alle Informationen im screen vorhanden.
Folgendermaßen stelle ich mir die Linienanalyse vor:
• Im Code können die Haltestellen der Linie hinterlegt werden (funktioniert schon)
• Es sollen nur Fahrten mit Ist-Zeiten ausgegeben werden (aktuell werden Fahrten mit Ist- und ohne Ist-Zeiten ausgegeben)
• Es sollen nur Fahrten, die in 0 bis 10 Minuten abfahren ausgegeben werden (aktuell werden alle Fahrten ausgegeben)
• Es sollen nur die Abfahrten der untersuchten Linie ausgegeben werden (aktuell werden alle Linien der Haltestelle ausgegeben)
• Ausgegeben sollen die Daten der Abfrage in eine csv-Datei (aktuell wird die Abfrage nur im screen angezeigt)
• Zur Analyse sind je Datenreihe erforderlich: „Abfragezeit“, „Haltestellen-Nr.“, „Haltestellenname“, „Liniennummer“, „Linienzieltext“, „Abfahrt in Minuten“, „Verspätungsminuten“
• Es soll diese Abfrage alle 10 Minuten automatisiert ausgeführt werden und das Ergebnis in die Auswertungsdatei angefügt werden.
Ich würde mich freuen, wenn ich hier im Forum Hilfe beim Erweitern des Python-Codes erhalte.
Viele Grüße
Agascha
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
So sieht der Grundlagencode bisher aus
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
import asyncio
from aiohttp import ClientSession
from time import sleep
from typing import NamedTuple, List
from datetime import datetime
import xml.etree.ElementTree as ET
from colorama import init, Fore
from csv import writer
from async_retrying import retry
screen = True #False = ohne screen und mit Datei / True = mit screen und ohne Datei
csvfn = './Fahrtausfälle.csv'
csvln = './Linienauswerung.csv'
class Departure(NamedTuple):
stop: str
countdown: int
linenum: str
direction: str
plantime: tuple
isrealtime: bool
delay: int
rtstatus: str
genattr: List[str]
@retry(attempts=10)
async def dmreq(session, ifopt, limit=20):
payload = {'name_dm':ifopt, 'type_dm':'any', 'mode':'direct', 'useRealtime':'1', 'limit':str(limit)} #, 'includedMeans':['5', '6', '7']}
#payload['itdTime']='1200'
#payload['itdDate']='180109'
async with session.get('http://openservice-test.vrr.de/vrr/XML_DM_REQUEST', params=payload) as r:
content = await r.read()
root = ET.fromstring(content)
#tree = ET.parse(...)
#root = tree.getroot()
return (ifopt, root)
sem = asyncio.Semaphore(10)
async def safe_dmreq(*args, **kwa):
async with sem:
return await dmreq(*args, **kwa)
def deps(root):
departures = []
for departure in root.iter('itdDeparture'):
stop = "".join(root.find('itdDepartureMonitorRequest').find('itdOdv').find('itdOdvName').find('odvNameElem').itertext())
servingline = departure.find('itdServingLine')
countdown = int(departure.attrib['countdown'])
linenum = servingline.attrib['number']
direction = servingline.attrib['direction']
realtime = bool(int(servingline.attrib['realtime']))
delay = 0
rtstatus = ""
itdtime = departure.find('itdDateTime').find('itdTime')
plantime = (int(itdtime.attrib['hour']), int(itdtime.attrib['minute']))
#if countdown > 59:
#realtime = False #False
if realtime:
delay = int(servingline.find('itdNoTrain').attrib['delay'])
rtstatus = departure.attrib.get('realtimeStatus', "")
genattr = []
ga = departure.find('genAttrList')
if ga:
for gae in ga:
gaename = gae.findtext('name', default="")
gaevalue = gae.findtext('value', default="")
genattr.append(gaename+"="+gaevalue)
departures.append(Departure(stop=stop,
countdown=countdown,
linenum=linenum,
direction=direction,
plantime=plantime,
isrealtime=realtime,
delay=delay,
rtstatus=rtstatus,
genattr=genattr))
return departures
def depstrings(departures, directionwidth=30):
strings = []
for dep in departures:
color = Fore.RESET
currentstring = ""
if not dep.isrealtime:
color = Fore.WHITE
elif dep.delay > 5 or dep.delay == -9999:
color = Fore.RED
elif dep.delay > 1:
color = Fore.YELLOW
else:
color = Fore.GREEN
currentstring += ('{:>4}'.format(dep.linenum) + " " + ('{:<'+str(directionwidth)+'}').format(dep.direction[:directionwidth]) + " ")
currentstring += color
if dep.countdown == 0:
currentstring += "jetzt"
elif dep.delay == -9999:
currentstring +=" xxx "
elif dep.countdown > 59:
currentstring += tt(dep.plantime)
elif dep.countdown > 9:
currentstring += (str(dep.countdown) + "min")
else:
currentstring += (" " + str(dep.countdown) + "min")
if dep.isrealtime and dep.delay != -9999:
currentstring += (color + "(+" + str(dep.delay) + ")" + Fore.RESET)
else:
currentstring += Fore.RESET
strings.append([dep.countdown, currentstring])
strings.sort(key=lambda x: x[0])
return strings
def tt(timetuple):
return str(timetuple[0]).zfill(2) + ":" + str(timetuple[1]).zfill(2)
async def loopreqs(ifopts):
alldeps = {}
tasks = []
async with ClientSession() as session:
for ifopt in ifopts:
#try:
task = asyncio.ensure_future(safe_dmreq(session, ifopt, 25)) #25 gleich Reihen
tasks.append(task)
responses = await asyncio.gather(*tasks)
for ifopt, root in responses:
alldeps[ifopt] = deps(root)
return alldeps
if __name__ == "__main__":
init() # colorama
# Haltestellen
ifopts = ['de
5447', 'de5001', 'de5002', 'de10049', 'de5003']rtmax = 35
ausfaelle = []
ausfalldeps = []
rtabfahrtenmax = {}
loop = asyncio.get_event_loop()
future = asyncio.ensure_future(loopreqs(ifopts))
alldeps = loop.run_until_complete(future)
for ifopt in alldeps:
departures = alldeps[ifopt]
newausfall = False
for departure in departures:
if departure.countdown <= rtmax:
if departure.linenum not in rtabfahrtenmax:
rtabfahrtenmax[departure.linenum] = {}
if departure.direction not in rtabfahrtenmax[departure.linenum]:
rtabfahrtenmax[departure.linenum][departure.direction] = []
rtabfahrtenmax[departure.linenum][departure.direction].append(departure.isrealtime)
if departure.delay == -9999:
newausfall = True
ausfaelle.append(f"{departure.stop} ({ifopt}): {departure.linenum} => {departure.direction} ({tt(departure.plantime)})")
ausfalldeps.append((ifopt, departure))
# Bildschirmausgabe
if screen:
if departures:
print(f"=== {departures[0].stop} ({ifopt}) ===")
else:
print(f"Keine Abfahrten für {ifopt} gefunden")
strs = depstrings(departures)
for s in strs:
print(s[1])
print("")
if newausfall:
print("==== Ausfälle ====", "\n".join(ausfaelle), "", sep="\n")
# sleep(0.1)
if screen:
print("Sammlung beendet\n")
print(f"Echtzeitabfahrten (<= {rtmax} min) nach Linie und Richtung:")
for linenum, directions in sorted(rtabfahrtenmax.items()):
print(linenum)
for direction, rta in sorted(directions.items()):
rtabfahrten = rta.count(True)
alleabfahrten = len(rta)
print(f" - {direction}: {rtabfahrten}/{alleabfahrten} ({round((rtabfahrten/alleabfahrten)*100)} %)")
print("")
nowtimestr = datetime.now().strftime("%Y-%m-%d %H:%M")
if screen:
print(f"Gefundene Ausfälle ~{nowtimestr}:", ("\n"+"\n".join(ausfaelle)) if ausfaelle else " keine", sep="")
else:
with open(csvfn, 'a') as f:
writer(f).writerows(sorted(list([(nowtimestr, dep.stop, ifopt, dep.linenum, dep.direction, tt(dep.plantime), dep.rtstatus, "|".join(dep.genattr)) for ifopt, dep in ausfalldeps]), key=lambda x: (x[3], x[4], x[5])))
