hass-sncf/sncf/sncf.py
2022-01-03 20:56:40 +01:00

74 lines
2.6 KiB
Python

import datetime
import requests
class SNCFJourney(dict):
def __repr__(self):
sections = []
for section in self["sections"]:
nice_dep_date = SNCF.api_date_to_datetime(section['departure_date_time']).strftime("%H:%M")
nice_arr_date = SNCF.api_date_to_datetime(section['arrival_date_time']).strftime("%H:%M")
sections.append(f"{nice_dep_date} {section['from']['name']} - {section['to']['name']} {nice_arr_date}")
return "\n".join(sections)
class SNCF:
BASE_URL = "https://api.sncf.com/v1/coverage/sncf"
def __init__(self, token: str):
self.token = token
self.source_stop_point: str = ""
self.dest_stop_point: str = ""
self.allowed_lines: list[str] = []
self.max_transfers: int = 0
self.max_duration_secs: int = 1800
self.prefered_date_format = "%H:%M"
def set_source_stop_point(self, stop_point: str):
self.source_stop_point = stop_point
def set_dest_stop_point(self, stop_point: str):
self.dest_stop_point = stop_point
def set_allowed_lines(self, allowed_lines: list[str]):
self.allowed_lines = allowed_lines
def get_next_journeys(self, count: int = 1):
valid_journeys = []
api_response = self.api_request(f"/journeys?from={self.source_stop_point}&to={self.dest_stop_point}&datetime={self.date_now()}&count={count}")
journeys = api_response["journeys"]
for journey in journeys:
if journey["nb_transfers"] <= self.max_transfers and journey["duration"] <= self.max_duration_secs:
valid_journeys.append(SNCFJourney(journey))
return valid_journeys
def api_request(self, url):
req = requests.get(SNCF.BASE_URL + url, headers={"Authorization": self.token})
req.raise_for_status()
return req.json()
@staticmethod
def date_now() -> str:
return datetime.datetime.now().strftime("%Y%m%dT%H%M%S")
@staticmethod
def api_date_to_datetime(api_date: str) -> datetime:
return datetime.datetime.strptime(api_date, "%Y%m%dT%H%M%S")
def test_api_key(self) -> bool:
try:
self.api_request("")
return True
except requests.HTTPError:
return False
if __name__ == '__main__':
sncf = SNCF("5722269b-2e58-49ee-986b-21741438d5ff")
sncf.set_source_stop_point("stop_point:SNCF:87214056:Train")
sncf.set_dest_stop_point("stop_point:SNCF:87212027:Train")
next_journeys = sncf.get_next_journeys(count=5)
for next_journey in next_journeys:
print(next_journey)
#print(json.dumps(next_schedules, indent=4))