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

New arguments to handle cache expiration

Browse Source
This commit is contained in:
nemunaire 2022-12-30 19:54:21 +01:00
parent 70f0c81fcc
commit f79022b7a3
5 changed files with 65 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
main.py
View File

@ -45,7 +45,7 @@ class WidgetPlacement:
self.kwargs = kwargs
def main(only_on_coming_evt=False, ignore_module=[]):
def main(only_on_coming_evt=False, ignore_module=[], **config_args):
image = Image.new('1', (480, 800), 255)
#image = Image.new('L', (480, 800), 'white')
draw = ImageDraw.Draw(image)
@ -53,7 +53,7 @@ def main(only_on_coming_evt=False, ignore_module=[]):
shape = []
import modules
config = modules.Config()
config = modules.Config(**config_args)
# alerts
alerts = []
@ -99,14 +99,14 @@ def main(only_on_coming_evt=False, ignore_module=[]):
# RATP weather
from modules.ratp import RATPWeatherModule
shape.append(WidgetPlacement(RATPWeatherModule, size=(int(480/1.6), 94), position=(480-int(480/1.6), 155)))
alerts.append(RATPWeatherModule)
alerts.append({"module": RATPWeatherModule, "args_func": [config]})
# Toolbar
from modules.weather import WeatherToolbarModule
shape.append(WidgetPlacement(WeatherToolbarModule, size=(480, 50), position=(0, 530)))
from modules.sncf import SNCFWeatherModule
alerts.append({"module": SNCFWeatherModule, "args": "normandie"})
alerts.append({"module": SNCFWeatherModule, "args_func": [config, "normandie"]})
from modules.weather import WeatherAlerts
alerts.append(WeatherAlerts)
@ -193,7 +193,16 @@ if __name__ == '__main__':
help='Ignore the given modules')
parser.add_argument('--only-on-coming-evt', '-O', action='store_const', const=True,
help='Refresh screen only if there is upcoming event')
parser.add_argument('--cache-timeout', '-C', type=int, default=90,
help='How many minutes to keep the infos in cache')
parser.add_argument('--max-cache-timeout', type=int, default=120,
help='Maximum time to serve the infos in cache in case of issue')
args = parser.parse_args()
main(args.only_on_coming_evt, args.ignore_module)
main(
args.only_on_coming_evt,
args.ignore_module,
cache_timeout=args.cache_timeout,
max_cache_timeout=args.max_cache_timeout,
)

13
modules/__init__.py
View File

@ -6,7 +6,10 @@ from PIL import Image, ImageDraw, ImageFont
class Config:
def __init__(self, cals_file_list=".cals"):
def __init__(self, cache_timeout, max_cache_timeout, cals_file_list=".cals"):
self.cache_timeout = cache_timeout
self.max_cache_timeout = max_cache_timeout
if os.path.exists('/usr/share/fonts/source-pro/'):
self.fnt_R_path = "/usr/share/fonts/source-pro/SourceSansPro-Regular.otf"
self.fnt_RI_path = "/usr/share/fonts/source-pro/SourceSansPro-It.otf"
@ -140,14 +143,16 @@ class AlertsModule:
align = 7
for alert in self.alerts:
args = []
args_module = []
args_func = []
if isinstance(alert, dict) and "module" in alert:
args = alert["args"] if "args" in alert else []
args_module = alert["args_module"] if "args_module" in alert else []
args_func = alert["args_func"] if "args_func" in alert else []
alert = alert["module"]
if isinstance(alert, type):
try:
for alert in alert(*args).gen_alerts():
for alert in alert(*args_module).gen_alerts(*args_func):
align = self.draw_alert(alert, width, image, draw, fnt_R, fnt_B, align, font_size)
except BaseException as e:
logging.exception(e)

6
modules/ical.py
View File

@ -112,7 +112,7 @@ class IcalModule:
for component in self._get_events(config, toofar):
evt = None
train_situations, train_conjoncturels, train_start_station, train_end_station, place = self.is_train_event(component)
train_situations, train_conjoncturels, train_start_station, train_end_station, place = self.is_train_event(config, component)
if train_start_station is not None:
start = component.decoded("DTSTART")
end = component.decoded("DTEND")
@ -166,7 +166,7 @@ class IcalModule:
if evt is not None:
yield evt
def is_train_event(self, evt):
def is_train_event(self, config, evt):
if "description" not in evt:
return None, None, None, None, None
@ -198,7 +198,7 @@ class IcalModule:
now = datetime.now(tz=pytz.timezone('Europe/Paris'))
from .sncf import SNCFAPI
status = SNCFAPI().get_train_status(numero_train, start_time)
status = SNCFAPI(config).get_train_status(numero_train, start_time)
if status is None:
return None, None, None, None, place

23
modules/ratp.py
View File

@ -55,12 +55,13 @@ class IDFMAPI:
def __init__(self, apikey=None):
def __init__(self, config, apikey=None):
self.baseurl = "https://prim.iledefrance-mobilites.fr/marketplace"
self.apikey = apikey or os.environ["TOKEN_IDFM"]
self._cached_file = ".ratp-%s.cache"
self.cache_time = 5
self.cache_timeout = config.cache_timeout
self.max_cache_timeout = config.max_cache_timeout
def fromIVtoPRIM(src):
return {
@ -138,7 +139,7 @@ class IDFMAPI:
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):
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("https://api-iv.iledefrance-mobilites.fr/disruptions")
try:
@ -148,6 +149,14 @@ class IDFMAPI:
except ConnectionResetError:
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):
raise Exception("File too old")
# Retrieve cached data
res = {}
with open(cache_file) as f:
@ -196,10 +205,10 @@ class RATPWeatherModule:
def __init__(self):
self.major_lines = ["M7", "M5", "M14", "RB", "T3A"]
def gen_alerts(self):
def gen_alerts(self, config):
alerts = []
weather = IDFMAPI().get_weather()
weather = IDFMAPI(config).get_weather()
for mode in weather:
for line in weather[mode]:
if mode[0].upper() + line not in self.major_lines:
@ -239,7 +248,7 @@ class RATPWeatherModule:
image = Image.new('RGB', (width, height), '#fff')
draw = ImageDraw.Draw(image)
weather = IDFMAPI().get_weather()
weather = IDFMAPI(config).get_weather()
align_x = 0
align_y = 0
@ -289,7 +298,7 @@ class RATPNextStopModule:
align = 0
api = IDFMAPI()
api = IDFMAPI(config)
for stop in stops:
tmp = stop.split("/")
mode = tmp[0][0]

29
modules/sncf.py
View File

@ -13,12 +13,13 @@ class SNCFAPI:
CLEANR = re.compile('<.*?>')
def __init__(self):
def __init__(self, config):
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
self.cache_timeout = config.cache_timeout
self.max_cache_timeout = config.max_cache_timeout
def get_icon(size):
height = int(size * 0.531)
@ -39,7 +40,7 @@ class SNCFAPI:
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):
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})
try:
@ -51,6 +52,14 @@ class SNCFAPI:
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):
raise Exception("File too old")
# Retrieve cached data
res = {}
with open(cache_file) as f:
@ -70,7 +79,7 @@ class SNCFAPI:
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):
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})
try:
@ -82,6 +91,14 @@ class SNCFAPI:
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):
raise Exception("File too old")
# Retrieve cached data
res = {}
with open(cache_file) as f:
@ -95,10 +112,10 @@ class SNCFWeatherModule:
def __init__(self):
pass
def gen_alerts(self, region):
def gen_alerts(self, config, region):
alerts = []
weather = SNCFAPI().get_weather(region)
weather = SNCFAPI(config).get_weather(region)
for alert in weather:
if alert["type"] != "perturbation":
continue