X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=renderer.py;h=fa95e346ec7ceb4ffabfd1179057f9fd512bb796;hb=6195654b2405928398dd981233115504162d1020;hp=491d34d52f9b32a9090294b40c21358e42f892fd;hpb=e932e65d4847a4777ddae297f9e52349285221ed;p=kiosk.git diff --git a/renderer.py b/renderer.py index 491d34d..fa95e34 100644 --- a/renderer.py +++ b/renderer.py @@ -1,31 +1,44 @@ -import time +#!/usr/bin/env python3 + +from abc import ABC, abstractmethod from datetime import datetime +from decorators import invocation_logged +import time +from typing import Dict, List, Set + -class renderer(object): +class renderer(ABC): """Base class for something that can render.""" + + @abstractmethod def render(self): pass + @abstractmethod def get_name(self): - return self.__class__.__name__ + pass + class abstaining_renderer(renderer): """A renderer that doesn't do it all the time.""" - def __init__(self, name_to_timeout_dict): - self.name_to_timeout_dict = name_to_timeout_dict; + + def __init__(self, name_to_timeout_dict: Dict[str, int]) -> None: + self.name_to_timeout_dict = name_to_timeout_dict self.last_runs = {} for key in name_to_timeout_dict: self.last_runs[key] = 0 - def should_render(self, keys_to_skip): + def should_render(self, keys_to_skip: Set[str]) -> str: now = time.time() for key in self.name_to_timeout_dict: - if (((now - self.last_runs[key]) > self.name_to_timeout_dict[key]) and - key not in keys_to_skip): + if ( + (now - self.last_runs[key]) > self.name_to_timeout_dict[key] + ) and key not in keys_to_skip: return key return None - def render(self): + @invocation_logged + def render(self) -> None: tries_per_key = {} keys_to_skip = set() while True: @@ -39,32 +52,40 @@ class abstaining_renderer(renderer): tries_per_key[key] = 0 if tries_per_key[key] >= 3: - print('renderer: Too many failures/retries for "%s.%s", ' + - ', giving up for now' % (self.get_name(), key)) + print( + 'renderer: Too many failures for "%s.%s", giving up' + % (self.get_name(), key) + ) keys_to_skip.add(key) else: msg = 'renderer: executing "%s.%s"' % (self.get_name(), key) - if (tries_per_key[key] > 1): - msg = msg + " (retry %d)" % tries_per_key[key] + if tries_per_key[key] > 1: + msg = msg + " (retry #%d)" % tries_per_key[key] print(msg) - if (self.periodic_render(key)): + if self.periodic_render(key): self.last_runs[key] = time.time() - def periodic_render(self, key): + @invocation_logged + @abstractmethod + def periodic_render(self, key) -> bool: pass + def get_name(self) -> str: + return self.__class__.__name__ + + class debuggable_abstaining_renderer(abstaining_renderer): - def __init__(self, name_to_timeout_dict, debug): - super(debuggable_abstaining_renderer, self).__init__(name_to_timeout_dict); + def __init__(self, name_to_timeout_dict: Dict[str, int], debug: bool) -> None: + super(debuggable_abstaining_renderer, self).__init__(name_to_timeout_dict) self.debug = debug - def debug_prefix(self): + def debug_prefix(self) -> str: return self.get_name() - def being_debugged(self): + def being_debugged(self) -> bool: return self.debug - def debug_print(self, template, *args): + def debug_print(self, template: str, *args) -> None: try: if self.being_debugged(): if args: