From 193765d8c89c76e54109ea343217ba273f2a9f9d Mon Sep 17 00:00:00 2001 From: makefu Date: Wed, 30 Aug 2017 16:06:09 +0200 Subject: vvs_efa: init --- vvs_efa/README.md | 61 ++++++++ vvs_efa/VVS_EFA.py | 358 ++++++++++++++++++++++++++++++++++++++++++++++ vvs_efa/__init__.py | 0 vvs_efa/tests/__init__.py | 39 +++++ 4 files changed, 458 insertions(+) create mode 100644 vvs_efa/README.md create mode 100644 vvs_efa/VVS_EFA.py create mode 100644 vvs_efa/__init__.py create mode 100644 vvs_efa/tests/__init__.py diff --git a/vvs_efa/README.md b/vvs_efa/README.md new file mode 100644 index 0000000..0dae508 --- /dev/null +++ b/vvs_efa/README.md @@ -0,0 +1,61 @@ +##VVS API Wrapper + +This was created for easy access to the VVS EFA (Elektronische Fahrplanauskunft) in Python. + +### Example + +```python +# coding: utf-8 +import datetime as dt +from vvs_efa import VVS_EFA + +vvs_efa = VVS_EFA.EFA() +connections = vvs_efa.get_next_connections("Stadtbibliothek", "Feuersee", dt.datetime(2015, 7, 7, 7, 20), True) + + +print connections[0].origin.name +print connections[0].destination.name +print connections[0].time_of_departure.strftime("%d.%m.%Y - %H:%M") + " - " + connections[0].time_of_arrival.strftime("%d.%m.%Y - %H:%M") +print str(connections[0].fare) + "€ / " + str(connections[0].zones) + " Zone(n)" +print "" +for leg in connections[0].legs: + print leg.time_of_departure.strftime("%H:%M") + " - " + leg.time_of_arrival.strftime("%H:%M") + ": " + print leg.origin.name + " - " + leg.destination.name + " ("+ leg.line + ")" + "\n" + + +charlottenplatz = VVS_EFA.Stop("Charlottenplatz") +charlottenplatz_departures = charlottenplatz.get_next_connections(dt.datetime(2015, 7, 9, 18, 10), True) + +print charlottenplatz_departures[0].stop_id +print charlottenplatz_departures[0].stop_name +print charlottenplatz_departures[0].scheduled_datetime.strftime("%d.%m.%Y - %H:%M") +print charlottenplatz_departures[0].realtime_datetime.strftime("%d.%m.%Y - %H:%M") +print charlottenplatz_departures[0].serving_line.key +print charlottenplatz_departures[0].serving_line.number +print charlottenplatz_departures[0].serving_line.direction +print charlottenplatz_departures[0].serving_line.direction_from +print charlottenplatz_departures[0].serving_line.line_type + +``` + +results in: + +``` +Stuttgart, Stadtbibliothek +Stuttgart, Feuersee +06.07.2015 - 07:20 - 06.07.2015 - 07:31 +2.3€ / 2 Zone(n) +07:20 - 07:22: Stadtbibliothek - Hauptbf (A.-Klett-Pl.) (U12) +07:24 - 07:28: Hauptbf (Arnulf-Klett-Platz) - Stuttgart Hauptbahnhof (tief) () +07:28 - 07:31: Stuttgart Hauptbahnhof (tief) - Feuersee (S6) + +5006075 +Charlottenplatz +09.07.2015 - 18:05 +09.07.2015 - 18:11 +143 +43 +Feuersee +Killesberg +Bus +``` diff --git a/vvs_efa/VVS_EFA.py b/vvs_efa/VVS_EFA.py new file mode 100644 index 0000000..3a3ca3d --- /dev/null +++ b/vvs_efa/VVS_EFA.py @@ -0,0 +1,358 @@ +# coding: utf-8 + +import datetime as dt +import requests +import pytz +class EFA(object): + + def __init__(self): + + pass + + @classmethod + def convert_name_to_id(cls, name, mobile=False): + if name.strip() != "": #check if string is empty -> return None + if mobile: + request_url = "http://m.vvs.de/jqm/controller/XSLT_STOPFINDER_REQUEST" + parameters = {} + parameters["name_sf"] = name + parameters["language"] = "de" + parameters["stateless"] = "1" + parameters["locationServerActive"] = "1" + parameters["anyObjFilter_sf"] = "0" + parameters["anyMaxSizeHitList"] = "500" + parameters["outputFormat"] = "JSON" + parameters["SpEncId"] = "0" + parameters["type_sf"] = "any" + parameters["anySigWhenPerfectNoOtherMatches"] = "1" + parameters["convertAddressesITKernel2LocationServer"] = "1" + parameters["convertCoord2LocationServer"] = "1" + parameters["convertCrossingsITKernel2LocationServer"] = "1" + parameters["convertStopsPTKernel2LocationServer"] = "1" + parameters["convertPOIsITKernel2LocationServer"] = "1" + parameters["tryToFindLocalityStops"] = "1" + parameters["w_objPrefAl"] = "12" + parameters["w_regPrefAm"] = "1" + + else: + request_url = "http://www2.vvs.de/vvs/XSLT_STOPFINDER_REQUEST" + parameters = {} + parameters["suggest_macro"] = "vvs" + parameters["name_sf"] = name + + req = requests.post(request_url, parameters) + + points = req.json()["stopFinder"]["points"] + + + best_point_index = 0 + + if len(points) > 1: + for index, point in enumerate(points): + if point["best"] == "1": + best_point_index = index + break + + return points[best_point_index]["stateless"] + + elif len(points) == 1: + return points["point"]["stateless"] #if one point found (no array to iterate!) -> directly return + + else: + return None #no points found -> return None + + else: + return None + + + def get_next_connections(self, origin, destination, datetime, departure, search_by_name = True): + + """Get connections for parameter set: + origin: name of origin + destination: name of destination + time: datetime object of departure/arrival + departure: boolean --> true: time of departure; false: time of arrival + search_by_name: boolean --> true: search by name; false: search by id""" + + if search_by_name: + id_origin = self.convert_name_to_id(origin) + id_destination = self.convert_name_to_id(destination) + if id_origin == None: + raise TypeError('Origin not valid or not found.') + if id_destination == None: + raise TypeError('Destination not valid or not found.') + + else: + id_origin = origin + id_destination = destination + + request_url = "http://m.vvs.de/jqm/controller/XSLT_TRIP_REQUEST2" + + parameters = {} + + #parameters["genC"] = "1" + #parameters["itOptionsActive"] = "1" + #parameters["locationServerActive"] = "1" + #parameters["ptOptionsActive"] = "1" + parameters["stateless"] = "1" + parameters["name_origin"] = id_origin + parameters["name_destination"] = id_destination + parameters["type_destination"] = "any" + parameters["type_origin"] = "any" + parameters["use_realtime"] = "1" + parameters["itdTime"] = datetime.strftime("%H%M") #"2105" + parameters["date"] = datetime.strftime("%Y%m%d") #"20141216" + + if departure: + parameters["itdTripDateTimeDepArr"] = "dep" + else: + parameters["itdTripDateTimeDepArr"] = "arr" + + parameters["routeType"] = "LEASTTIME" + parameters["changeSpeed"] = "normal" + + #parameters["includedMeans"] = "checkbox" #does only work when inclMOT_x set + #parameters["inclMOT_0"] = "1" + #parameters["inclMOT_1"] = "1" + #parameters["inclMOT_2"] = "1" + #parameters["inclMOT_3"] = "1" + #parameters["inclMOT_4"] = "1" + #parameters["inclMOT_5"] = "1" + #parameters["inclMOT_6"] = "1" + #parameters["inclMOT_7"] = "1" + #parameters["inclMOT_8"] = "1" + #parameters["inclMOT_9"] = "1" + #parameters["inclMOT_10"] = "1" + #parameters["inclMOT_11"] = "1" + #parameters["inclMOT_12"] = "1" + #parameters["inclMOT_13"] = "1" + #parameters["inclMOT_14"] = "1" + #parameters["inclMOT_15"] = "1" + #parameters["inclMOT_16"] = "1" + #parameters["inclMOT_17"] = "1" + + parameters["outputFormat"] = "JSON" + + #parameters["SpEncId"] = "0" + #parameters["anySigWhenPerfectNoOtherMatches"] = "1" + #parameters["convertStopsPTKernel2LocationServer"] = "1" + #parameters["convertPOIsITKernel2LocationServer"] = "1" + #parameters["verifyAnyLocViaLocServer"] = "1" + #parameters["w_objPrefAl"] = "12" + #parameters["w_regPrefAm"] = "1" + #parameters["calcOneDirection"] = "1" #if not set, first trip is last trip before date/time for departure, else first trip after date/time + #parameters["searchLimitMinutes"] = "360" + + + req = requests.post(request_url, parameters) + + data = req.json() + trips = data["trips"] + trip_objs = [] + + if trips != None: + for trip in trips: + #Trip attributes + #trip_duration = trip["duration"] + trip_fare = float(trip["itdFare"]["fares"]["fare"]["fareAdult"]) + trip_zones = abs(int(trip["itdFare"]["tariffZones"]["tariffZone"]["toPR"]) - int(trip["itdFare"]["tariffZones"]["tariffZone"]["fromPR"])) + trip_origin = Location(name = data["origin"]["points"]["point"]["name"], loc_id = data["origin"]["points"]["point"]["stateless"], loc_type = data["origin"]["points"]["point"]["anyType"]) + trip_destination = Location(name = data["destination"]["points"]["point"]["name"], loc_id = data["destination"]["points"]["point"]["stateless"], loc_type = data["destination"]["points"]["point"]["anyType"]) + + #print "Dauer: %s \nPreis: %s\n" % (trip_duration, trip_fare) + + trip_legs = [] + for leg in trip["legs"]: + origin = Location(name=leg["points"][0]["name"], loc_id=leg["points"][0]["ref"]["id"]) + destination = Location(name=leg["points"][1]["name"], loc_id=leg["points"][1]["ref"]["id"]) + + origin_realtime_date = leg["points"][0]["dateTime"]["rtDate"] #17.12.2014 + origin_realtime_time = leg["points"][0]["dateTime"]["rtTime"] #16:35 + + destination_realtime_date = leg["points"][1]["dateTime"]["rtDate"] + destination_realtime_time = leg["points"][1]["dateTime"]["rtTime"] + berlin = pytz.timezone('Europe/Berlin') + origin_departure_real_datetime = berlin.localize( + dt.datetime.strptime(origin_realtime_date + "-" + + origin_realtime_time, + "%d.%m.%Y-%H:%M")).replace(tzinfo=None) + destination_arrival_real_datetime = berlin.localize( + dt.datetime.strptime(destination_realtime_date + "-" + + destination_realtime_time, + "%d.%m.%Y-%H:%M")).replace(tzinfo=None) + + duration = leg["timeMinute"] + mode = leg["mode"]["product"] + line = leg["mode"]["number"] + direction = leg["mode"]["destination"] + + l = Leg(origin, destination, origin_departure_real_datetime, destination_arrival_real_datetime, mode, line, direction) + + trip_legs.append(l) + + #print "%s-%s: (%s min)\n %s - %s" %(origin_departure_real_time, destination_arrival_real_time, duration, origin, destination) + + + trip_time_of_departure = trip_legs[0].time_of_departure + trip_time_of_arrival = trip_legs[-1].time_of_arrival + + trip_obj = Trip(trip_origin, trip_destination, trip_time_of_departure, trip_time_of_arrival, trip_legs, trip_fare, trip_zones) + trip_objs.append(trip_obj) + + else: + raise ValueError("No trips found") + + + return trip_objs + + +class Trip(object): + + def __init__(self, origin, destination, time_of_departure, time_of_arrival, legs, fare, zones): + """origin: location object of origin + destination: location object of destination + time_of_departure: datetime object of trip departure time / date + time_of_arrival: datetime object of trip arrival time / date + legs: list of Leg objects + fare: fare of trip (float) + zones: number of zones travelled (int)""" + + self.origin = origin + self.destination = destination + self.time_of_departure = time_of_departure + self.time_of_arrival = time_of_arrival + self.legs = legs + self.fare = fare + self.zones = zones + + + +class Leg(object): + + def __init__(self, origin, destination, time_of_departure, time_of_arrival, mode, line, direction): + """origin: leg origin (station / address / etc.) + destination: leg destination (station / address / etc.) + time_of_departure: departure datetime object + time_of_arrival: arrival datetime object + mode: mode of transportation (U-Bahn / S-Bahn / Bus / Walking) + line: U-Bahn / S-Bahn / Bus Line + direction: ulitmate destination of U-Bahn / S-Bahn / Bus Line""" + + self.origin = origin + self.destination = destination + self.time_of_departure = time_of_departure + self.time_of_arrival = time_of_arrival + self.mode = mode + self.line = line + self.direction = direction + + +class Location(object): + + def __init__(self, name, loc_id, loc_type="any"): + """type: type of location (station / address / etc.) + name: full name of location + id: id of location""" + + self.loc_type = loc_type + self.name = name + self.loc_id = loc_id + + +class Stop(object): + + def __init__(self, name): + self.id = VVS_EFA.convert_name_to_id(name) + if self.id == None: + raise TypeError("Station could not be found.") + + def get_next_connections(self, datetime, departure, limit=20): + + """datetime: datetime object of time & date + departure: boolean, true -> time of departure, false --> time of arrival + limit: int, limit of results fetched, default: 20""" + + if self.id != None: + + parameters = {} + + parameters["deleteAssignedStops_dm"] = "1" + parameters["itOptionsActive"] = "1" + parameters["limit"] = str(limit) + parameters["locationServerActive"] = "1" + parameters["mode"] = "direct" + parameters["name_dm"] = self.id + parameters["ptOptionsActive"] = "1" + parameters["stateless"] = "1" + parameters["type_dm"] = "stop" + parameters["useAllStops"] = "1" + parameters["useRealtime"] = "1" + parameters["itdTime"] = datetime.strftime("%H%M") + parameters["itdDate"] = datetime.strftime("%Y%m%d") + + if departure: + parameters["itdTripDateTimeDepArr"] = "dep" + else: + parameters["itdTripDateTimeDepArr"] = "arr" + + parameters["outputFormat"] = "JSON" + parameters["SpEncId"] = "0" + parameters["anySigWhenPerfectNoOtherMatches"] = "1" + parameters["convertStopsPTKernel2LocationServer"] = "1" + parameters["convertPOIsITKernel2LocationServer"] = "1" + parameters["verifyAnyLocViaLocServer"] = "1" + parameters["w_objPrefAl"] = "12" + parameters["w_regPrefAm"] = "1" + + request_url = "http://m.vvs.de/jqm/controller/XSLT_DM_REQUEST" + req = requests.post(request_url, parameters) + data = req.json() + + departure_list = data["departureList"] + connections = [] + for list_entry in departure_list: + serving_line_data = list_entry["servingLine"] + serving_line = ServingLine(serving_line_data["key"], serving_line_data["number"], serving_line_data["direction"], + serving_line_data["directionFrom"], serving_line_data["name"]) + + stop_id = list_entry["stopID"] + stop_name = list_entry["stopName"] + scheduled_datetime = dt.datetime(int(list_entry["dateTime"]["year"]), int(list_entry["dateTime"]["month"]), + int(list_entry["dateTime"]["day"]), int(list_entry["dateTime"]["hour"]), int(list_entry["dateTime"]["minute"])) + + try: #check if realtime info available + realtime_datetime = dt.datetime(int(list_entry["realDateTime"]["year"]), int(list_entry["realDateTime"]["month"]), + int(list_entry["realDateTime"]["day"]), int(list_entry["realDateTime"]["hour"]), int(list_entry["realDateTime"]["minute"])) + + except: #if not: return None for realtime_datetime + realtime_datetime = None + + connection = Connection(stop_id, stop_name, scheduled_datetime, realtime_datetime, serving_line) + connections.append(connection) + + return connections + + else: + return None + + + +class Connection(object): + """docstring for Connection""" + def __init__(self, stop_id, stop_name, scheduled_datetime, realtime_datetime, serving_line): + """realtime_datetime: datetime object of projected departure/arrival. None if not available""" + self.stop_id = stop_id + self.stop_name = stop_name + self.scheduled_datetime = scheduled_datetime + self.realtime_datetime = realtime_datetime + self.serving_line = serving_line + + +class ServingLine(object): + """docstring for ServingLine""" + def __init__(self, key, number, direction, direction_from, line_type): + self.key = key + self.number = number + self.direction = direction + self.direction_from = direction_from + self.line_type = line_type diff --git a/vvs_efa/__init__.py b/vvs_efa/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/vvs_efa/tests/__init__.py b/vvs_efa/tests/__init__.py new file mode 100644 index 0000000..bbb1bcf --- /dev/null +++ b/vvs_efa/tests/__init__.py @@ -0,0 +1,39 @@ +# coding: utf-8 + +import unittest +import datetime as dt +from vvs_efa import VVS_EFA + +vvs_efa = VVS_EFA.VVS_EFA() + +class TestConvertNameToId(unittest.TestCase): + + def test_stadtbibliothek_name(self): + self.assertEqual(vvs_efa.convertNameToId("Stadtbibliothek"), "5006116") + + def test_hauptbahnhof_name(self): + self.assertEqual(vvs_efa.convertNameToId("Hauptbahnhof"), "5006118") + + def test_name_with_umlaut(self): + self.assertEqual(vvs_efa.convertNameToId("Möhringen"), "5006169") + + def test_name_with_scharfs(self): + self.assertEqual(vvs_efa.convertNameToId("Vaihinger Straße"), "5000170") + + def test_stadtbibliothek_name_mobile(self): + self.assertEqual(vvs_efa.convertNameToId("Stadtbibliothek", True), "5006116") + + def test_hauptbahnhof_name_mobile(self): + self.assertEqual(vvs_efa.convertNameToId("Hauptbahnhof", True), "5006118") + + def test_empty_name(self): + self.assertEqual(vvs_efa.convertNameToId(""), None) + +class TestGetNextConnections(unittest.TestCase): + + def test_invalid_origin(self): + with self.assertRaises(TypeError): + vvs_efa.getNextConnections("", "Feuersee", dt.datetime(2015, 7, 13, 7, 20), True) + +if __name__ == '__main__': + unittest.main() -- cgit v1.2.3