epaper/modules/sncf.py

from datetime import datetime, timedelta, timezone
import base64
import json
import logging
import os
import os.path
import urllib.error
import urllib.parse
import urllib.request
import re

from PIL import Image, ImageDraw, ImageFont, ImageOps

class SNCFAPI:

    CLEANR = re.compile('<.*?>')

    def __init__(self, config):
        self.baseurl = "https://www.sncf.com/api/iv"
        self.auth = base64.b64encode(b"admin:$2y$10$QvxWSS4f5DIFSkAmuBoV9OJG3M2bhec9d3F2.YnULBMtpzKAq2KS.").decode()

        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):
        height = int(size * 0.531)
        img = Image.open(os.path.join(self._config.icons_dir, "sncf.png")).resize((size, height))
        r,g,b,a = img.split()
        rgb_image = Image.merge('RGB', (r,g,b))
        inverted_image = ImageOps.invert(rgb_image)
        r2,g2,b2 = inverted_image.split()

        return Image.merge('RGBA', (r2,g2,b2,a))

    def get_train_status(self, numero, date):
        cache_file = self._cached_file % (int(numero), date.strftime("%Y-%m-%d"))
        # Read the mod time
        statinfo = None
        try:
            statinfo = os.stat(cache_file)
        except:
            pass

        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")),
                                         headers={
                                             'Authorization': "Basic " + self.auth,
                                             '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",
                                         })
            try:
                with urllib.request.urlopen(req) as f:
                    with open(cache_file, 'wb') as fd:
                        fd.write(f.read())
            except ConnectionResetError:
                pass
            except urllib.error.URLError:
                pass

        try:
            statinfo = os.stat(cache_file)
        except:
            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")))
            logging.exception(Exception("File too old to predict SNCF issue"))
            return None

        # Retrieve cached data
        res = {}
        with open(cache_file) as f:
            res = json.load(f)

        try:
            return res["reponseRechercherListeCirculations"]["reponse"]["listeResultats"]["resultat"][0]["donnees"]["listeCirculations"]["circulation"][0]
        except:
            return None

    def get_weather(self, region):
        cache_file = self._cached_file % (0, region)
        # Read the mod time
        statinfo = None
        try:
            statinfo = os.stat(cache_file)
        except:
            pass

        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),
                                         headers={
                                             'Authorization': "Basic " + self.auth,
                                             '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",
                                         })
            try:
                with urllib.request.urlopen(req, timeout=3) as f:
                    with open(cache_file, 'wb') as fd:
                        fd.write(f.read())
            except ConnectionResetError as e:
                logging.exception(e)
            except urllib.error.URLError as e:
                logging.exception(e)
            except TimeoutError as e:
                logging.exception(e)

        try:
            statinfo = os.stat(cache_file)
        except:
            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):
            raise Exception("File too old")

        # Retrieve cached data
        res = {}
        with open(cache_file) as f:
            res = json.load(f)

        return res


class SNCFWeatherModule:

    def __init__(self):
        pass

    def gen_alerts(self, config, region):
        alerts = []

        weather = SNCFAPI(config).get_weather(region)
        for alert in weather:
            if alert["type"] != "perturbation":
                continue

            yield {
                "description": re.sub(SNCFAPI.CLEANR, '\n', alert["content"]).replace('Restez informés sur le fil', '').replace('Twitter @train_nomad', '').strip(),
                "icon": SNCFAPI.get_icon,
            }