X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=chooser.py;h=3fecc09717ff47c43988d76ac167b79cc2e4ef3d;hb=73e4d75ceabe5546f3966cfdcd1f705c77be17f7;hp=d5c6482f331e818f7863eae63087c94f3e1ad077;hpb=c06bfef53f70551e7920bc4facce27f47b89e2ba;p=kiosk.git diff --git a/chooser.py b/chooser.py index d5c6482..3fecc09 100644 --- a/chooser.py +++ b/chooser.py @@ -8,7 +8,7 @@ import random import re import sys import time -from typing import Callable, List +from typing import Any, Callable, List, Optional, Set, Tuple import constants import trigger @@ -28,7 +28,7 @@ class chooser(ABC): ] for page in pages: result = re.match(valid_filename, page) - if result != None: + if result is not None: print(f'chooser: candidate page: "{page}"') if result.group(3) != "none": freshness_requirement = int(result.group(3)) @@ -43,21 +43,21 @@ class chooser(ABC): return filenames @abstractmethod - def choose_next_page(self) -> str: + def choose_next_page(self) -> Any: pass class weighted_random_chooser(chooser): """Chooser that does it via weighted RNG.""" - def __init__(self, filter_list: List[Callable[[str], bool]]) -> None: + def __init__(self, filter_list: Optional[List[Callable[[str], bool]]]) -> None: self.last_choice = "" self.valid_filename = re.compile("([^_]+)_(\d+)_([^\.]+)\.html") - self.pages = None + self.pages: Optional[List[str]] = None self.count = 0 - self.filter_list = filter_list - if filter_list is None: - self.filter_list = [] + self.filter_list: List[Callable[[str], bool]] = [] + if filter_list is not None: + self.filter_list.extend(filter_list) self.filter_list.append(self.dont_choose_page_twice_in_a_row_filter) def dont_choose_page_twice_in_a_row_filter(self, choice: str) -> bool: @@ -66,20 +66,20 @@ class weighted_random_chooser(chooser): self.last_choice = choice return True - def choose_next_page(self) -> str: - if self.pages == None or self.count % 100 == 0: + def choose_next_page(self) -> Any: + if self.pages is None or self.count % 100 == 0: self.pages = self.get_page_list() total_weight = 0 weights = [] for page in self.pages: result = re.match(self.valid_filename, page) - if result != None: + if result is not None: weight = int(result.group(2)) weights.append(weight) total_weight += weight if total_weight <= 0: - raise error + raise Exception while True: random_pick = random.randrange(0, total_weight - 1) @@ -110,46 +110,55 @@ class weighted_random_chooser_with_triggers(weighted_random_chooser): def __init__( self, - trigger_list: List[trigger.trigger], + trigger_list: Optional[List[trigger.trigger]], filter_list: List[Callable[[str], bool]], ) -> None: weighted_random_chooser.__init__(self, filter_list) - self.trigger_list = trigger_list - if trigger_list is None: - self.trigger_list = [] - self.page_queue = set(()) + self.trigger_list: List[trigger.trigger] = [] + if trigger_list is not None: + self.trigger_list.extend(trigger_list) + self.page_queue: Set[Tuple[str, int]] = set(()) def check_for_triggers(self) -> bool: triggered = False for t in self.trigger_list: x = t.get_triggered_page_list() - if x != None and len(x) > 0: + if x is not None and len(x) > 0: for y in x: self.page_queue.add(y) triggered = True return triggered - def choose_next_page(self) -> str: - if self.pages == None or self.count % 100 == 0: + def choose_next_page(self) -> Tuple[str, bool]: + if self.pages is None or self.count % 100 == 0: self.pages = self.get_page_list() triggered = self.check_for_triggers() # First try to satisfy from the page queue. + now = datetime.datetime.now() if len(self.page_queue) > 0: print("chooser: Pulling page from queue...") page = None priority = None for t in self.page_queue: - if priority == None or t[1] > priority: + if priority is None or t[1] > priority: page = t[0] priority = t[1] + assert(page is not None) + assert(priority is not None) self.page_queue.remove((page, priority)) - return page, triggered + return (page, triggered) + + # Always show the clock in the middle of the night. + elif now.hour < 7: + for page in self.pages: + if "clock" in page: + return (page, False) # Fall back on weighted random choice. else: - return weighted_random_chooser.choose_next_page(self), False + return (weighted_random_chooser.choose_next_page(self), False) # Test