3 from abc import ABC, abstractmethod
9 from typing import Any, Callable, List, Optional, Set, Tuple
11 from pyutils.datetimes import datetime_utils
13 import kiosk_constants
17 logger = logging.getLogger(__file__)
21 """Base class of a thing that chooses pages"""
26 def get_page_list(self) -> List[str]:
28 valid_filename = re.compile("([^_]+)_(\d+)_([^\.]+)\.html")
32 for f in os.listdir(kiosk_constants.pages_dir)
33 if os.path.isfile(os.path.join(kiosk_constants.pages_dir, f))
36 result = re.match(valid_filename, page)
37 if result is not None:
38 if result.group(3) != "none":
39 freshness_requirement = int(result.group(3))
41 os.path.getmtime(os.path.join(kiosk_constants.pages_dir, page))
43 age = now - last_modified
44 if age > freshness_requirement:
46 f'chooser: "{page}" is too old.'
50 f'chooser: candidate page: "{page}"'
52 filenames.append(page)
56 def choose_next_page(self) -> Any:
60 class weighted_random_chooser(chooser):
61 """Chooser that does it via weighted RNG."""
63 def __init__(self, filter_list: Optional[List[Callable[[str], bool]]]) -> None:
65 self.last_choice = None
66 self.valid_filename = re.compile("([^_]+)_(\d+)_([^\.]+)\.html")
67 self.pages: Optional[List[str]] = None
69 self.filter_list: List[Callable[[str], bool]] = []
70 if filter_list is not None:
71 self.filter_list.extend(filter_list)
72 self.filter_list.append(self.dont_choose_page_twice_in_a_row_filter)
74 def dont_choose_page_twice_in_a_row_filter(self, choice: str) -> bool:
75 if self.last_choice is not None and choice == self.last_choice:
77 self.last_choice = choice
80 def choose_next_page(self) -> Any:
81 if self.pages is None or self.count % 100 == 0:
82 logger.info('chooser: refreshing the candidate pages list.')
83 self.pages = self.get_page_list()
87 for page in self.pages:
88 result = re.match(self.valid_filename, page)
89 if result is not None:
90 weight = int(result.group(2))
91 weights.append(weight)
92 total_weight += weight
97 random_pick = random.randrange(0, total_weight - 1)
99 for x in range(0, len(weights)):
101 if so_far > random_pick:
103 choice = self.pages[x]
105 # Allow filters list to suppress pages.
106 choice_is_filtered = False
107 for f in self.filter_list:
109 choice_is_filtered = True
111 if choice_is_filtered:
119 class weighted_random_chooser_with_triggers(weighted_random_chooser):
120 """Same as WRC but has trigger events"""
124 trigger_list: Optional[List[trigger.trigger]],
125 filter_list: List[Callable[[str], bool]],
127 super().__init__(filter_list)
128 self.trigger_list: List[trigger.trigger] = []
129 if trigger_list is not None:
130 self.trigger_list.extend(trigger_list)
131 self.page_queue: Set[Tuple[str, int]] = set(())
133 def check_for_triggers(self) -> bool:
135 for t in self.trigger_list:
136 x = t.get_triggered_page_list()
137 if x is not None and len(x) > 0:
139 self.page_queue.add(y)
140 logger.info(f'chooser: noticed active trigger {y}')
144 def choose_next_page(self) -> Tuple[str, bool]:
145 if self.pages is None or self.count % 100 == 0:
146 logger.info('chooser: refreshing the candidates page list')
147 self.pages = self.get_page_list()
149 triggered = self.check_for_triggers()
151 # First try to satisfy from the page queue.
152 if len(self.page_queue) > 0:
153 logger.info('chooser: page queue has entries; pulling choice from there.')
156 for t in self.page_queue:
157 if priority is None or t[1] > priority:
160 assert(page is not None)
161 assert(priority is not None)
162 self.page_queue.remove((page, priority))
163 return (page, triggered)
165 # Always show the clock in the middle of the night.
166 now = datetime_utils.now_pacific()
168 for page in self.pages:
172 # Fall back on weighted random choice.
173 return (weighted_random_chooser.choose_next_page(self), False)
177 # def filter_news_during_dinnertime(page):
178 # now = datetime.datetime.now()
179 # is_dinnertime = now.hour >= 17 and now.hour <= 20
180 # return not is_dinnertime or not (
183 # or "mynorthwest" in page
184 # or "seattle" in page
185 # or "stranger" in page
186 # or "twitter" in page
189 # x = weighted_random_chooser_with_triggers([], [ filter_news_during_dinnertime ])
190 # print(x.choose_next_page())