#!/usr/bin/env python3
from datetime import datetime
+import difflib
import gc
import linecache
import os
+import re
import sys
from threading import Thread
import time
import traceback
import tracemalloc
-from typing import Optional
+from typing import Optional, List
+from queue import Queue, Empty
+
+import astral # type: ignore
+from astral.sun import sun # type: ignore
import constants
import renderer
import renderer
import renderer_catalog
import chooser
+import listen
import logging
+import pvporcupine
import trigger_catalog
import utils
-def filter_news_during_dinnertime(page: str) -> bool:
- now = datetime.now()
- is_dinnertime = now.hour >= 17 and now.hour <= 20
- return not is_dinnertime or not (
- "cnn" in page
- or "news" in page
- or "mynorthwest" in page
- or "seattle" in page
- or "stranger" in page
- or "twitter" in page
- or "wsj" in page
- )
-
-
def thread_janitor() -> None:
tracemalloc.start()
tracemalloc_target = 0.0
time.sleep(10.0)
-def thread_change_current() -> None:
+def guess_page(command: str, page_chooser: chooser.chooser) -> str:
+ best_page = None
+ best_score = None
+ for page in page_chooser.get_page_list():
+ page = page.replace('(', ' ')
+ page = page.replace('_', ' ')
+ page = page.replace(')', ' ')
+ page = page.replace('.html', '')
+ page = page.replace('CNNNews', 'news')
+ page = page.replace('CNNTechnology', 'technology')
+ page = page.replace('gocostco', 'costco list')
+ page = page.replace('gohardware', 'hardware list')
+ page = page.replace('gohouse', 'house list honey do')
+ page = page.replace('gcal', 'google calendar events')
+ page = page.replace('mynorthwest', 'northwest news')
+ page = page.replace('myq', 'myq garage door status')
+ page = page.replace('gomenu', 'dinner menu')
+ page = page.replace('wsdot', 'traffic')
+ page = page.replace('gomenu', 'dinner menu')
+ page = page.replace('WSJNews', 'news')
+ page = page.replace('telma', 'telma cabin')
+ page = page.replace('WSJBusiness', 'business news')
+ page = re.sub(r'[0-9]+', '', page)
+ score = SequenceMatcher(None, command, page).ratio()
+ if best_score is None or score > best_score:
+ best_page = page
+ assert best_page is not None
+ return best_page
+
+
+def process_command(command: str, page_history: List[str]) -> str:
+ page = None
+ if 'hold' in command:
+ page = page_history[0]
+ elif 'back' in command:
+ page = page_history[1]
+ elif 'skip' in command:
+ while True:
+ (page, _) = page_chooser.choose_next_page()
+ if page != page_history[0]:
+ break
+ elif 'weather' in command:
+ if 'telma' in command or 'cabin' in command:
+ page = 'weather-telma_3_10800.html'
+ elif 'stevens' in command:
+ page = 'weather-stevens_3_10800.html'
+ else:
+ page = 'weather-home_3_10800.html'
+ elif 'cabin' in command:
+ if 'list' in command:
+ page = 'Cabin-(gocabin)_2_3600.html'
+ else:
+ page = 'hidden/cabin_driveway.html'
+ elif 'news' in command or 'headlines' in command:
+ page = 'cnn-CNNNews_4_25900.html'
+ elif 'clock' in command or 'time' in command:
+ page = 'clock_10_none.html'
+ elif 'countdown' in command or 'countdowns' in command:
+ page = 'countdown_3_7200.html'
+ elif 'costco' in command:
+ page = 'Costco-(gocostco)_2_3600.html'
+ elif 'calendar' in command or 'events' in command:
+ page = 'gcal_3_86400.html'
+ elif 'countdown' in command or 'countdowns' in command:
+ page = 'countdown_3_7200.html'
+ elif 'grocery' in command or 'groceries' in command:
+ page = 'Grocery-(gogrocery)_2_3600.html'
+ elif 'hardware' in command:
+ page = 'Hardware-(gohardware)_2_3600.html'
+ elif 'garage' in command:
+ page = 'myq_4_300.html'
+ elif 'menu' in command:
+ page = 'Menu-(gomenu)_2_3600.html'
+ elif 'cron' in command or 'health' in command:
+ page = 'periodic-health_6_300.html'
+ elif 'photo' in command or 'picture' in command:
+ page = 'photo_23_3600.html'
+ elif 'quote' in command or 'quotation' in command or 'quotes' in command:
+ page = 'quotes_4_10800.html'
+ elif 'stevens' in command:
+ page = 'stevens-conditions_1_86400.html'
+ elif 'stock' in command or 'stocks' in command:
+ page = 'stock_3_86400.html'
+ elif 'twitter' in command:
+ page = 'twitter_10_3600.html'
+ elif 'traffic' in command:
+ page = 'wsdot-bridges_3_none.html'
+ elif 'front' in command and 'door' in command:
+ page = 'hidden/frontdoor.html'
+ elif 'driveway' in command:
+ page = 'hidden/driveway.html'
+ elif 'backyard' in command:
+ page = 'hidden/backyard.html'
+ else:
+ page = guess_page(command, page_chooser)
+ assert page is not None
+ return page
+
+
+def thread_change_current(command_queue: Queue) -> None:
+ page_history = [ "", "" ]
+ swap_page_target = 0.0
+
+ def filter_news_during_dinnertime(page: str) -> bool:
+ now = datetime.now()
+ is_dinnertime = now.hour >= 17 and now.hour <= 20
+ return not is_dinnertime or not (
+ "cnn" in page
+ or "news" in page
+ or "mynorthwest" in page
+ or "seattle" in page
+ or "stranger" in page
+ or "twitter" in page
+ or "wsj" in page
+ )
page_chooser = chooser.weighted_random_chooser_with_triggers(
trigger_catalog.get_triggers(), [filter_news_during_dinnertime]
)
- swap_page_target = 0.0
- last_page = ""
+
while True:
now = time.time()
- (page, triggered) = page_chooser.choose_next_page()
+
+ # Check for a verbal command.
+ command = None
+ try:
+ command = command_queue.get(block=False)
+ except Exception:
+ command = None
+ pass
+ if command is not None:
+ triggered = True
+ page = process_command(command, page_history)
+
+ # Else pick a page randomly.
+ else:
+ while True:
+ (page, triggered) = page_chooser.choose_next_page()
+ if triggered or page != page_history[0]:
+ break
if triggered:
print("chooser[%s] - WE ARE TRIGGERED." % utils.timestamp())
- if page != last_page:
+ if page != page_history[0] or (swap_page_target - now < 10.0):
print(
"chooser[%s] - EMERGENCY PAGE %s LOAD NEEDED"
% (utils.timestamp(), page)
)
try:
- f = open(os.path.join(constants.pages_dir, "current.shtml"), "w")
- emit_wrapped(f, page)
- f.close()
+ with open(os.path.join(constants.pages_dir, "current.shtml"), "w") as f:
+ emit_wrapped(f, page, override_refresh_sec = 40, command = command)
+ page_history.insert(0, page)
+ page_history = page_history[0:10]
+ swap_page_target = now + 40
except:
print("chooser[%s] - page does not exist?!" % (utils.timestamp()))
continue
- last_page = page
- swap_page_target = now + constants.refresh_period_sec
# Also notify XMLHTTP clients that they need to refresh now.
path = os.path.join(constants.pages_dir, "reload_immediately.html")
- f = open(path, "w")
- f.write("Reload, suckers!")
- f.close()
+ with open(path, "w") as f:
+ f.write("Reload, suckers!")
# Fix this hack... maybe read the webserver logs and see if it
# actually was picked up?
- time.sleep(0.750)
+ time.sleep(0.75)
os.remove(path)
elif now >= swap_page_target:
- if page == last_page:
- print(
- (
- "chooser[%s] - nominal choice got the same page..."
- % (utils.timestamp())
- )
- )
- continue
+ assert page != page_history[0]
print("chooser[%s] - nominal choice of %s" % (utils.timestamp(), page))
try:
- f = open(os.path.join(constants.pages_dir, "current.shtml"), "w")
- emit_wrapped(f, page)
- f.close()
+ with open(os.path.join(constants.pages_dir, "current.shtml"), "w") as f:
+ emit_wrapped(f, page)
+ page_history.insert(0, page)
+ page_history = page_history[0:10]
+ swap_page_target = now + constants.refresh_period_sec
except:
print("chooser[%s] - page does not exist?!" % (utils.timestamp()))
continue
- last_page = page
- swap_page_target = now + constants.refresh_period_sec
time.sleep(1)
-def emit_wrapped(f, filename) -> None:
+def emit_wrapped(f,
+ filename: str,
+ *,
+ override_refresh_sec: int = None,
+ command: str = None) -> None:
def pick_background_color() -> str:
+ city = astral.LocationInfo(
+ "Bellevue", "USA", "US/Pacific", 47.610, -122.201
+ )
+ s = sun(city.observer, date=self.dt, tzinfo=pytz.timezone("US/Pacific"))
+ sunrise_mod = utils.minute_number(s["sunrise"].hour, s["sunrise"].minute)
+ sunset_mod = utils.minute_number(s["sunset"].hour, s["sunset"].minute)
now = datetime.now()
- if now.hour <= 6 or now.hour >= 21:
+ now_mod = utils.minute_number(now.hour, now.minute)
+ if now_mod < sunrise_mod or now_mod > (sunset_mod + 45):
return "E6B8B8"
- elif now.hour == 7 or now.hour == 20:
+ elif now_mod < (sunrise_mod + 45) or now_mod > (sunset_mod + 120):
return "EECDCD"
else:
return "FFFFFF"
def get_refresh_period() -> float:
+ if override_refresh_sec is not None:
+ return float(override_refresh_sec * 1000.0)
now = datetime.now()
if now.hour < 7:
- return constants.refresh_period_night_sec * 1000
+ return float(constants.refresh_period_night_sec * 1000.0)
else:
- return constants.refresh_period_sec * 1000
+ return float(constants.refresh_period_sec * 1000.0)
age = utils.describe_age_of_file_briefly(f"pages/{filename}")
bgcolor = pick_background_color()
+ if command is None:
+ pageid = filename
+ else:
+ pageid = f'"{command}" -> {filename}'
+
f.write(
"""
<HEAD>
constants.hostname,
bgcolor,
filename,
- filename,
+ pageid,
age,
)
)
if __name__ == "__main__":
logging.basicConfig()
+ command_queue: Queue = Queue()
changer_thread: Optional[Thread] = None
renderer_thread: Optional[Thread] = None
janitor_thread: Optional[Thread] = None
+ hotword_thread: Optional[Thread] = None
while True:
+ if hotword_thread is None or not hotword_thread.is_alive():
+ keyword_paths = [pvporcupine.KEYWORD_PATHS[x] for x in ["bumblebee"]]
+ sensitivities = [0.7] * len(keyword_paths)
+ listener = listen.HotwordListener(
+ command_queue,
+ keyword_paths,
+ sensitivities,
+ )
+ hotword_thread = Thread(target=listener.listen_forever, args=())
+ hotword_thread.start()
if changer_thread is None or not changer_thread.is_alive():
print(
f"MAIN[{utils.timestamp()}] - (Re?)initializing chooser thread... (wtf?!)"
)
- changer_thread = Thread(target=thread_change_current, args=())
+ changer_thread = Thread(target=thread_change_current, args=(command_queue,))
changer_thread.start()
if renderer_thread is None or not renderer_thread.is_alive():
print(
projects / kiosk.git / commitdiff