#!/usr/bin/env python3
from datetime import datetime
-import difflib
import gc
import linecache
import os
import traceback
import tracemalloc
from typing import Optional, List
-from queue import Queue, Empty
+from queue import Queue
import astral # type: ignore
from astral.sun import sun # type: ignore
import pytz
-import constants
-import renderer
-import renderer
+import datetime_utils
+import file_utils
+
+import kiosk_constants as constants
import renderer_catalog
import chooser
import listen
import logging
import pvporcupine
import trigger_catalog
-import utils
def thread_janitor() -> None:
break
if triggered:
- print("chooser[%s] - WE ARE TRIGGERED." % utils.timestamp())
+ print("chooser[%s] - WE ARE TRIGGERED." % datetime_utils.timestamp())
if page != page_history[0] or (swap_page_target - now < 10.0):
print(
"chooser[%s] - EMERGENCY PAGE %s LOAD NEEDED"
- % (utils.timestamp(), page)
+ % (datetime_utils.timestamp(), page)
)
try:
with open(os.path.join(constants.pages_dir, "current.shtml"), "w") as f:
page_history = page_history[0:10]
swap_page_target = now + 40
except:
- print("chooser[%s] - page does not exist?!" % (utils.timestamp()))
+ print("chooser[%s] - page does not exist?!" % (datetime_utils.timestamp()))
continue
# Also notify XMLHTTP clients that they need to refresh now.
elif now >= swap_page_target:
assert page != page_history[0]
- print("chooser[%s] - nominal choice of %s" % (utils.timestamp(), page))
+ print("chooser[%s] - nominal choice of %s" % (datetime_utils.timestamp(), page))
try:
with open(os.path.join(constants.pages_dir, "current.shtml"), "w") as f:
emit_wrapped(f, 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()))
+ print("chooser[%s] - page does not exist?!" % (datetime_utils.timestamp()))
continue
time.sleep(1)
"Bellevue", "USA", "US/Pacific", 47.610, -122.201
)
s = sun(city.observer, date=now, 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_mod = utils.minute_number(now.hour, now.minute)
+ sunrise_mod = datetime_utils.minute_number(s["sunrise"].hour, s["sunrise"].minute)
+ sunset_mod = datetime_utils.minute_number(s["sunset"].hour, s["sunset"].minute)
+ now_mod = datetime_utils.minute_number(now.hour, now.minute)
if now_mod < sunrise_mod or now_mod > (sunset_mod + 45):
return "E6B8B8"
elif now_mod < (sunrise_mod + 45) or now_mod > (sunset_mod + 120):
else:
return float(constants.refresh_period_sec * 1000.0)
- age = utils.describe_age_of_file_briefly(f"pages/{filename}")
+ age = file_utils.describe_file_ctime(f"pages/{filename}")
bgcolor = pick_background_color()
if command is None:
pageid = filename
def thread_invoke_renderers() -> None:
while True:
- print(f"renderer[{utils.timestamp()}]: invoking all renderers in catalog...")
+ print(f"renderer[{datetime_utils.timestamp()}]: invoking all renderers in catalog...")
for r in renderer_catalog.get_renderers():
now = time.time()
try:
except Exception as e:
traceback.print_exc(file=sys.stdout)
print(
- f"renderer[{utils.timestamp()}] unknown exception ({e}) in {r.get_name()}, swallowing it."
+ f"renderer[{datetime_utils.timestamp()}] unknown exception ({e}) in {r.get_name()}, swallowing it."
)
delta = time.time() - now
if delta > 1.0:
print(
- f"renderer[{utils.timestamp()}]: Warning: {r.get_name()}'s rendering took {delta:5.2f}s."
+ f"renderer[{datetime_utils.timestamp()}]: Warning: {r.get_name()}'s rendering took {delta:5.2f}s."
)
print(
- f"renderer[{utils.timestamp()}]: thread having a little break for {constants.render_period_sec}s..."
+ f"renderer[{datetime_utils.timestamp()}]: thread having a little break for {constants.render_period_sec}s..."
)
time.sleep(constants.render_period_sec)
hotword_thread.start()
if changer_thread is None or not changer_thread.is_alive():
print(
- f"MAIN[{utils.timestamp()}] - (Re?)initializing chooser thread... (wtf?!)"
+ f"MAIN[{datetime_utils.timestamp()}] - (Re?)initializing chooser thread... (wtf?!)"
)
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(
- f"MAIN[{utils.timestamp()}] - (Re?)initializing render thread... (wtf?!)"
+ f"MAIN[{datetime_utils.timestamp()}] - (Re?)initializing render thread... (wtf?!)"
)
renderer_thread = Thread(target=thread_invoke_renderers, args=())
renderer_thread.start()
if janitor_thread is None or not janitor_thread.is_alive():
print(
- f"MAIN[{utils.timestamp()}] - (Re?)initializing janitor thread... (wtf?!)"
+ f"MAIN[{datetime_utils.timestamp()}] - (Re?)initializing janitor thread... (wtf?!)"
)
janitor_thread = Thread(target=thread_janitor, args=())
janitor_thread.start()