nemunaire/epaper
1
0
Fork 0
Code Issues Pull Requests Releases Activity

Use the new SNCF API
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
nemunaire 2024-11-08 17:44:37 +01:00
parent cf1db8f746
commit d8c1ce419d
2 changed files with 32 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
modules/ical.py
View File

@ -141,34 +141,29 @@ class IcalModule:
"end": end, "end": end,
} }
new_start = datetime.fromisoformat(train_start_station["depart"]["dateHeureReelle"]) new_start = datetime.fromisoformat(start.strftime("%Y%m%dT") + train_start_station["departure_time"])
if start.astimezone(pytz.timezone('Europe/Paris')).strftime("%d %H:%M") != new_start.strftime("%d %H:%M"): if start.astimezone(pytz.timezone('Europe/Paris')).strftime("%d %H:%M") != new_start.strftime("%d %H:%M"):
evt["new_start"] = new_start evt["new_start"] = new_start
if train_end_station: if train_end_station:
new_end = datetime.fromisoformat(train_end_station["arrivee"]["dateHeureReelle"]) new_end = datetime.fromisoformat(end.strftime("%Y%m%dT") + train_end_station["arrival_time"])
if end.astimezone(pytz.timezone('Europe/Paris')).strftime("%d %H:%M") != new_end.strftime("%d %H:%M"): if end.astimezone(pytz.timezone('Europe/Paris')).strftime("%d %H:%M") != new_end.strftime("%d %H:%M"):
evt["new_end"] = new_end evt["new_end"] = new_end
if len(train_conjoncturels) > 0: if len(train_conjoncturels) > 0:
evt["alert"] = train_conjoncturels["messagesConjoncturels"][0]["titre"] evt["alert"] = train_conjoncturels["messagesConjoncturels"][0]["titre"]
elif len(train_situations) > 0: elif len(train_situations) > 0:
evt["alert"] = train_situations[0]["libelleSituation"] ret = []
for msg in train_situations[0]["messages"]:
ret.append(msg["text"])
if new_start > datetime.now(tz=pytz.timezone('Europe/Paris')): evt["alert"] = ", ".join(ret)
if "evenement" in train_start_station["depart"]:
evt["alert"] = train_start_station["depart"]["evenement"]["texte"] if new_start.astimezone(pytz.timezone('Europe/Paris')) > datetime.now(tz=pytz.timezone('Europe/Paris')):
if "voie" in train_start_station and "numero" in train_start_station["voie"]:
evt["info"] = "Départ voie " + train_start_station["voie"]["numero"]
if place is not None: if place is not None:
evt["info"] = ((evt["info"] + "\n") if "info" in evt else "") + place evt["info"] = ((evt["info"] + "\n") if "info" in evt else "") + place
elif train_end_station is not None: elif train_end_station is not None:
evt["info"] = "Arrivée à " + new_end.strftime("%H:%M") evt["info"] = "Arrivée à " + new_end.strftime("%H:%M")
if "evenement" in train_end_station["arrivee"]:
evt["alert"] = train_end_station["arrivee"]["evenement"]["texte"]
evt["info"] += " (+%d')" % train_end_station["arrivee"]["evenement"]["retard"]["duree"]
if "voie" in train_end_station and "numero" in train_end_station["voie"]:
evt["info"] += " voie " + train_end_station["voie"]["numero"]
elif now is not None and time.mktime(component.decoded("DTEND").timetuple()) < time.mktime(now.timetuple()): elif now is not None and time.mktime(component.decoded("DTEND").timetuple()) < time.mktime(now.timetuple()):
continue continue
@ -213,35 +208,36 @@ class IcalModule:
if numero_train is None: if numero_train is None:
return None, None, None, None, place return None, None, None, None, place
numero_train = "3341"
start_time = evt.decoded("DTSTART") start_time = evt.decoded("DTSTART")
now = datetime.now(tz=pytz.timezone('Europe/Paris')) now = datetime.now(tz=pytz.timezone('Europe/Paris'))
start_time = datetime.fromisoformat("2024-11-08T06:59:00Z")
from .sncf import SNCFAPI from .sncf import SNCFAPI
status = SNCFAPI(config).get_train_status(numero_train, start_time) status = SNCFAPI(config).get_train_status(numero_train, start_time)
if status is None: if status is None or status["vehicle_journeys"] is None or len(status["vehicle_journeys"][0]) == 0:
return None, None, None, None, place return None, None, None, None, place
situations = [] disruptions = []
if "situation" in status and len(status["situation"]) > 0: for d1 in status["vehicle_journeys"][0]["disruptions"]:
situations = status["situation"] for d2 in status["disruptions"]:
if d1["id"] == d2["id"]:
conjoncturels = [] disruptions.append(d2)
if "listeMessagesConjoncturels" in status and len(status["listeMessagesConjoncturels"]) > 0:
conjoncturels = status["listeMessagesConjoncturels"]
if ville_arrivee is None: if ville_arrivee is None:
return situations, conjoncturels, None, None, place return disruptions, [], None, None, place
start_station = None start_station = None
end_station = None end_station = None
for arret in status["listeArretsDesserte"]["arret"]: for arret in status["vehicle_journeys"][0]["stop_times"]:
if arret["emplacement"]["libelle"].startswith(ville_arrivee): if arret["stop_point"]["name"].startswith(ville_arrivee):
end_station = arret end_station = arret
if arret["emplacement"]["libelle"].startswith(ville_depart): if arret["stop_point"]["name"].startswith(ville_depart):
start_station = arret start_station = arret
return situations, conjoncturels, start_station, end_station, place return disruptions, [], start_station, end_station, place
def event_coming(self, config): def event_coming(self, config):
now = time.mktime(datetime.now(tz=pytz.timezone('Europe/Paris')).timetuple()) now = time.mktime(datetime.now(tz=pytz.timezone('Europe/Paris')).timetuple())
@ -336,7 +332,7 @@ class IcalModule:
) )
if "alert" in evt: if "alert" in evt:
draw.text( draw.text(
(2 + fnt_R.getbbox(evt["start"].astimezone(pytz.timezone('Europe/Paris')).strftime("%H:%M "))[0], align+line_height*0.6), (2 + fnt_R.getbbox(evt["start"].astimezone(pytz.timezone('Europe/Paris')).strftime("%H:%M "))[2], align+line_height*0.6),
evt["alert"], evt["alert"],
fill="black", anchor="lt", font=fnt_R fill="black", anchor="lt", font=fnt_R
) )

17
modules/sncf.py
View File

@ -15,9 +15,9 @@ class SNCFAPI:
CLEANR = re.compile('<.*?>') CLEANR = re.compile('<.*?>')
def __init__(self, config): def __init__(self, config, apikey=None):
self.baseurl = "https://www.sncf.com/api/iv" self.baseurl = "https://api.sncf.com/v1"
self.auth = base64.b64encode(b"admin:$2y$10$QvxWSS4f5DIFSkAmuBoV9OJG3M2bhec9d3F2.YnULBMtpzKAq2KS.").decode() self.apikey = apikey or os.environ["TOKEN_SNCF"]
self._config = config self._config = config
self._cached_file = ".sncf-%d-%s.cache" self._cached_file = ".sncf-%d-%s.cache"
@ -45,9 +45,9 @@ class SNCFAPI:
if statinfo is None or datetime.fromtimestamp(statinfo.st_mtime, tz=timezone.utc) + timedelta(minutes=self.cache_timeout) < datetime.now(tz=timezone.utc): if statinfo is None or datetime.fromtimestamp(statinfo.st_mtime, tz=timezone.utc) + timedelta(minutes=self.cache_timeout) < datetime.now(tz=timezone.utc):
# Do the request and save it # Do the request and save it
req = urllib.request.Request(self.baseurl + "/1.0/infoVoy/rechercherListeCirculations?numero=%d&dateCirculation=%s&codeZoneArret&typeHoraire=TEMPS_REEL" % (int(numero), date.strftime("%Y-%m-%d")), req = urllib.request.Request(self.baseurl + "/coverage/sncf/networks/network%%3ASNCF%%3ATER/vehicle_journeys?filter=vehicle_journey.name%%3D%%22%d%%22&since=%sT000000&until=%sT000000" % (int(numero), date.strftime("%Y%m%d"), (date + timedelta(minutes=1440)).strftime("%Y%m%d")),
headers={ headers={
'Authorization': "Basic " + self.auth, 'Authorization': self.apikey,
'Accept': "application/json", 'Accept': "application/json",
'Accept-Language': "fr,fr-FR;q=0.8,en-US;q=0.5,en;q=0.3", 'Accept-Language': "fr,fr-FR;q=0.8,en-US;q=0.5,en;q=0.3",
'User-Agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36", 'User-Agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36",
@ -67,7 +67,7 @@ class SNCFAPI:
pass pass
if statinfo is None or datetime.fromtimestamp(statinfo.st_mtime, tz=timezone.utc) + timedelta(minutes=self.max_cache_timeout) < datetime.now(tz=timezone.utc): if statinfo is None or datetime.fromtimestamp(statinfo.st_mtime, tz=timezone.utc) + timedelta(minutes=self.max_cache_timeout) < datetime.now(tz=timezone.utc):
print(self.baseurl + "/1.0/infoVoy/rechercherListeCirculations?numero=%d&dateCirculation=%s&codeZoneArret&typeHoraire=TEMPS_REEL" % (int(numero), date.strftime("%Y-%m-%d"))) #print(self.baseurl + "/coverage/sncf/networks/network%%3ASNCF%%3ATER/vehicle_journeys?filter=vehicle_journey.name%%3D%%22%d%%22&since=%sT000000&until=%sT000000" % (int(numero), date.strftime("%Y%m%d"), (date + timedelta(minutes=1440)).strftime("%Y%m%d")))
logging.exception(Exception("File too old to predict SNCF issue")) logging.exception(Exception("File too old to predict SNCF issue"))
return None return None
@ -77,13 +77,14 @@ class SNCFAPI:
res = json.load(f) res = json.load(f)
try: try:
return res["reponseRechercherListeCirculations"]["reponse"]["listeResultats"]["resultat"][0]["donnees"]["listeCirculations"]["circulation"][0] return res
except: except:
return None return None
def get_weather(self, region): def get_weather(self, region):
cache_file = self._cached_file % (0, region) cache_file = self._cached_file % (0, region)
# Read the mod time # Read the mod time
return []
statinfo = None statinfo = None
try: try:
statinfo = os.stat(cache_file) statinfo = os.stat(cache_file)
@ -94,7 +95,7 @@ class SNCFAPI:
# Do the request and save it # Do the request and save it
req = urllib.request.Request(self.baseurl + "/edito/bandeaux?region=%s" % (region), req = urllib.request.Request(self.baseurl + "/edito/bandeaux?region=%s" % (region),
headers={ headers={
'Authorization': "Basic " + self.auth, 'Authorization': self.apikey,
'Accept': "application/json", 'Accept': "application/json",
'Accept-Language': "fr,fr-FR;q=0.8,en-US;q=0.5,en;q=0.3", 'Accept-Language': "fr,fr-FR;q=0.8,en-US;q=0.5,en;q=0.3",
'User-Agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36", 'User-Agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36",