3 from abc import ABC, abstractmethod
9 from typing import Any, Callable, List, Optional, Set, Tuple
11 from pyutils import logging_utils
12 from pyutils.datetimes import datetime_utils
14 import kiosk_constants
18 logger = logging.getLogger(__name__)
22 """Base class of a thing that chooses pages"""
27 @logging_utils.LoggingContext(logger, prefix="chooser:")
28 def get_page_list(self) -> List[str]:
30 valid_filename = re.compile("([^_]+)_(\d+)_([^\.]+)\.html")
34 for f in os.listdir(kiosk_constants.pages_dir)
35 if os.path.isfile(os.path.join(kiosk_constants.pages_dir, f))
38 result = re.match(valid_filename, page)
39 if result is not None:
40 if result.group(3) != "none":
41 freshness_requirement = int(result.group(3))
43 os.path.getmtime(os.path.join(kiosk_constants.pages_dir, page))
45 age = now - last_modified
46 if age > freshness_requirement:
47 logger.warning(f'"{page}" is too old.')
49 logger.info(f'candidate page: "{page}"')
50 filenames.append(page)
54 def choose_next_page(self) -> Any:
58 class weighted_random_chooser(chooser):
59 """Chooser that does it via weighted RNG."""
61 def __init__(self, filter_list: Optional[List[Callable[[str], bool]]]) -> None:
63 self.last_choice = None
64 self.valid_filename = re.compile("([^_]+)_(\d+)_([^\.]+)\.html")
65 self.pages: Optional[List[str]] = None
67 self.filter_list: List[Callable[[str], bool]] = []
68 if filter_list is not None:
69 self.filter_list.extend(filter_list)
71 @logging_utils.LoggingContext(logger, prefix="chooser:")
72 def choose_next_page(self) -> Any:
73 if self.pages is None or self.count % 100 == 0:
74 logger.info("refreshing the candidate pages list.")
75 self.pages = self.get_page_list()
79 for page in self.pages:
80 result = re.match(self.valid_filename, page)
81 if result is not None:
82 weight = int(result.group(2))
83 weights.append(weight)
84 total_weight += weight
89 random_pick = random.randrange(0, total_weight - 1)
91 for x in range(0, len(weights)):
93 if so_far > random_pick:
95 choice = self.pages[x]
97 # Allow filters list to suppress pages.
98 choice_is_filtered = False
99 for f in self.filter_list:
101 choice_is_filtered = True
103 if choice_is_filtered:
111 class weighted_random_chooser_with_triggers(weighted_random_chooser):
112 """Same as WRC but has trigger events"""
116 trigger_list: Optional[List[trigger.trigger]],
117 filter_list: List[Callable[[str], bool]],
119 super().__init__(filter_list)
120 self.trigger_list: List[trigger.trigger] = []
121 if trigger_list is not None:
122 self.trigger_list.extend(trigger_list)
123 self.page_queue: Set[Tuple[str, int]] = set(())
125 @logging_utils.LoggingContext(logger, prefix="chooser:")
126 def check_for_triggers(self) -> bool:
128 for t in self.trigger_list:
129 x = t.get_triggered_page_list()
130 if x is not None and len(x) > 0:
132 self.page_queue.add(y)
133 logger.info(f"noticed active trigger {y}")
137 @logging_utils.LoggingContext(logger, prefix="chooser:")
138 def choose_next_page(self) -> Tuple[str, bool]:
139 if self.pages is None or self.count % 100 == 0:
140 logger.info("refreshing the candidates page list")
141 self.pages = self.get_page_list()
143 triggered = self.check_for_triggers()
145 # First try to satisfy from the page queue.
146 if len(self.page_queue) > 0:
147 logger.info("page queue has entries; pulling choice from there.")
150 for t in self.page_queue:
151 if priority is None or t[1] > priority:
154 assert page is not None
155 assert priority is not None
156 self.page_queue.remove((page, priority))
157 return (page, triggered)
159 # Always show the clock in the middle of the night.
160 now = datetime_utils.now_pacific()
162 for page in self.pages:
166 # Fall back on weighted random choice.
167 return (weighted_random_chooser.choose_next_page(self), False)
171 # def filter_news_during_dinnertime(page):
172 # now = datetime.datetime.now()
173 # is_dinnertime = now.hour >= 17 and now.hour <= 20
174 # return not is_dinnertime or not (
177 # or "mynorthwest" in page
178 # or "seattle" in page
179 # or "stranger" in page
180 # or "twitter" in page
183 # x = weighted_random_chooser_with_triggers([], [ filter_news_during_dinnertime ])
184 # print(x.choose_next_page())