epaper/modules/sncf.py

from datetime import datetime, timedelta, timezone
import base64
import json
import os
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):
        self.baseurl = "https://www.sncf.com/api/iv"
        self.auth = base64.b64encode(b"admin:$2y$10$QvxWSS4f5DIFSkAmuBoV9OJG3M2bhec9d3F2.YnULBMtpzKAq2KS.").decode()

        self._cached_file = ".sncf-%d-%s.cache"
        self.cache_time = 5

    def get_icon(size):
        height = int(size * 0.531)
        img = Image.open("icons/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_time) < 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})
            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

        # 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_time) < 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})
            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

        # 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, region):
        alerts = []

        weather = SNCFAPI().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,
            }