Use the new SNCF API

main.py

@@ -120,7 +120,7 @@ def main(only_on_coming_evt=False, ignore_module=[], force_coming_event=True, ex
     shape.append(WidgetPlacement(WeatherToolbarModule, size=(480, 50), position=(0, 530)))
 
     from modules.sncf import SNCFWeatherModule
-    alerts.append({"module": SNCFWeatherModule, "args_func": [config, "normandie"]})
+    alerts.append({"module": SNCFWeatherModule, "args_func": [config, "line%3ASNCF%3AFR%3ALine%3A%3A66849E04-284F-47D7-8795-4C925C6ED66F%3A"]})
     from modules.weather import WeatherAlerts
     alerts.append(WeatherAlerts)
 

modules/ical.py

@@ -141,34 +141,29 @@ class IcalModule:
                     "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"):
                     evt["new_start"] = new_start
 
                 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"):
                         evt["new_end"] = new_end
 
                 if len(train_conjoncturels) > 0:
                     evt["alert"] = train_conjoncturels["messagesConjoncturels"][0]["titre"]
                 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:
                         evt["info"] = ((evt["info"] + "\n") if "info" in evt else "") + place
                 elif train_end_station is not None:
                     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()):
                 continue
@@ -213,35 +208,35 @@ class IcalModule:
         if numero_train is None:
             return None, None, None, None, place
 
+        numero_train = "3341"
+
         start_time = evt.decoded("DTSTART")
         now = datetime.now(tz=pytz.timezone('Europe/Paris'))
 
         from .sncf import SNCFAPI
         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
 
-        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:
-            return situations, conjoncturels, None, None, place
+            return disruptions, [], None, None, place
 
         start_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
-            if arret["emplacement"]["libelle"].startswith(ville_depart):
+            if arret["stop_point"]["name"].startswith(ville_depart):
                 start_station = arret
 
-        return situations, conjoncturels, start_station, end_station, place
+        return disruptions, [], start_station, end_station, place
 
     def event_coming(self, config):
         now = time.mktime(datetime.now(tz=pytz.timezone('Europe/Paris')).timetuple())
@@ -336,7 +331,7 @@ class IcalModule:
                 )
             if "alert" in evt:
                 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"],
                     fill="black", anchor="lt", font=fnt_R
                 )

modules/sncf.py

@@ -15,24 +15,26 @@ class SNCFAPI:
 
     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._cached_file = ".sncf-%d-%s.cache"
         self.cache_timeout = config.cache_timeout
         self.max_cache_timeout = config.max_cache_timeout
 
-    def get_icon(size):
+    def get_icon(config):
-        height = int(size * 0.531)
+        def icon(size=64):
-        img = Image.open(os.path.join(self._config.icons_dir, "sncf.png")).resize((size, height))
+            height = int(size * 0.531)
-        r,g,b,a = img.split()
+            img = Image.open(os.path.join(config.icons_dir, "sncf.png")).resize((size, height))
-        rgb_image = Image.merge('RGB', (r,g,b))
+            r,g,b,a = img.split()
-        inverted_image = ImageOps.invert(rgb_image)
+            rgb_image = Image.merge('RGB', (r,g,b))
-        r2,g2,b2 = inverted_image.split()
+            inverted_image = ImageOps.invert(rgb_image)
+            r2,g2,b2 = inverted_image.split()
 
-        return Image.merge('RGBA', (r2,g2,b2,a))
+            return Image.merge('RGBA', (r2,g2,b2,a))
+        return icon
 
     def get_train_status(self, numero, date):
         cache_file = self._cached_file % (int(numero), date.strftime("%Y-%m-%d"))
@@ -45,9 +47,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):
             # 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={
-                                             'Authorization': "Basic " + self.auth,
+                                             'Authorization': self.apikey,
                                              'Accept': "application/json",
                                              '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",
@@ -67,7 +69,7 @@ class SNCFAPI:
             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):
-            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"))
             return None
 
@@ -77,7 +79,7 @@ class SNCFAPI:
             res = json.load(f)
 
         try:
-            return res["reponseRechercherListeCirculations"]["reponse"]["listeResultats"]["resultat"][0]["donnees"]["listeCirculations"]["circulation"][0]
+            return res
         except:
             return None
 
@@ -92,9 +94,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):
             # Do the request and save it
-            req = urllib.request.Request(self.baseurl + "/edito/bandeaux?region=%s" % (region),
+            req = urllib.request.Request(self.baseurl + "/coverage/sncf/lines/%s/stop_schedules?" % (region),
                                          headers={
-                                             'Authorization': "Basic " + self.auth,
+                                             'Authorization': self.apikey,
                                              'Accept': "application/json",
                                              '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",
@@ -135,11 +137,15 @@ class SNCFWeatherModule:
         alerts = []
 
         weather = SNCFAPI(config).get_weather(region)
-        for alert in weather:
+        for alert in weather["disruptions"]:
-            if alert["type"] != "perturbation":
+            if alert["status"] != "active":
                 continue
 
+            ret = []
+            for msg in alert["messages"]:
+                ret.append(msg["text"])
+
             yield {
-                "description": re.sub(SNCFAPI.CLEANR, '\n', alert["content"]).replace('Restez informés sur le fil', '').replace('Twitter @train_nomad', '').strip(),
+                "description": re.sub(SNCFAPI.CLEANR, '\n', ", ".join(ret)).strip(),
-                "icon": SNCFAPI.get_icon,
+                "icon": SNCFAPI.get_icon(config),
             }