import datetime_utils
-import constants
+import kiosk_constants
import trigger
filenames = []
pages = [
f
- for f in os.listdir(constants.pages_dir)
- if os.path.isfile(os.path.join(constants.pages_dir, f))
+ for f in os.listdir(kiosk_constants.pages_dir)
+ if os.path.isfile(os.path.join(kiosk_constants.pages_dir, f))
]
for page in pages:
result = re.match(valid_filename, page)
if result.group(3) != "none":
freshness_requirement = int(result.group(3))
last_modified = int(
- os.path.getmtime(os.path.join(constants.pages_dir, page))
+ os.path.getmtime(os.path.join(kiosk_constants.pages_dir, page))
)
age = now - last_modified
if age > freshness_requirement:
#!/usr/bin/env python3
-import constants
import os
from uuid import uuid4
+import kiosk_constants
class file_writer:
"""Helper context to write a pages file."""
def __init__(self, filename: str, *, transformations=[]):
temp = "temp-" + str(uuid4())
- self.temp_filename = os.path.join(constants.pages_dir, temp)
- self.full_filename = os.path.join(constants.pages_dir, filename)
+ self.temp_filename = os.path.join(kiosk_constants.pages_dir, temp)
+ self.full_filename = os.path.join(kiosk_constants.pages_dir, filename)
self.xforms = [file_writer.remove_tricky_unicode]
self.xforms.extend(transformations)
self.f = None
import gdata_oauth
import pytz
-import constants
+import kiosk_constants
import file_writer
import globals
import renderer
x = int(delta.total_seconds())
if x > 0:
identifier = "id%d" % count
- days = divmod(x, constants.seconds_per_day)
- hours = divmod(days[1], constants.seconds_per_hour)
- minutes = divmod(hours[1], constants.seconds_per_minute)
+ days = divmod(x, kiosk_constants.seconds_per_day)
+ hours = divmod(days[1], kiosk_constants.seconds_per_hour)
+ minutes = divmod(hours[1], kiosk_constants.seconds_per_minute)
g.write(
f'<li><SPAN id="%s">%d days, %02d:%02d</SPAN> until %s</li>\n'
% (
return False
def look_for_triggered_events(self) -> bool:
- with file_writer.file_writer(constants.gcal_imminent_pagename) as f:
+ with file_writer.file_writer(kiosk_constants.gcal_imminent_pagename) as f:
f.write("<h1>Imminent Upcoming Calendar Events:</h1>\n<hr>\n")
f.write("<center><table width=99%>\n")
now = datetime.datetime.now(pytz.timezone("US/Pacific"))
continue
delta = eventstamp - now
x = int(delta.total_seconds())
- if x > -120 and x < 4 * constants.seconds_per_minute:
- days = divmod(x, constants.seconds_per_day)
- hours = divmod(days[1], constants.seconds_per_hour)
- minutes = divmod(hours[1], constants.seconds_per_minute)
+ if x > -120 and x < 4 * kiosk_constants.seconds_per_minute:
+ days = divmod(x, kiosk_constants.seconds_per_day)
+ hours = divmod(days[1], kiosk_constants.seconds_per_hour)
+ minutes = divmod(hours[1], kiosk_constants.seconds_per_minute)
eventstamp = event.start_time
name = event.friendly_name()
calendar = event.calendar
import gkeepapi # type: ignore
-import constants
+import kiosk_constants
import file_writer
import renderer
import kiosk_secrets as secrets
f.write("</DIV>")
else:
logger.debug(f"Note is empty, deleting {filename}.")
- _ = os.path.join(constants.pages_dir, filename)
+ _ = os.path.join(kiosk_constants.pages_dir, filename)
try:
os.remove(_)
except:
import datetime_utils
import file_utils
-import constants
+import kiosk_constants
import file_writer
import renderer_catalog
import chooser
break
elif 'internal' in command:
if 'render' in command:
- page = constants.render_stats_pagename
+ page = kiosk_constants.render_stats_pagename
else:
- page = constants.render_stats_pagename
+ page = kiosk_constants.render_stats_pagename
elif 'weather' in command:
if 'telma' in command or 'cabin' in command:
page = 'weather-telma_3_10800.html'
page_chooser = chooser.weighted_random_chooser_with_triggers(
trigger_catalog.get_triggers(), [filter_news_during_dinnertime]
)
- current_file = os.path.join(constants.pages_dir, "current.shtml")
- emergency_file = os.path.join(constants.pages_dir, "reload_immediately.html")
+ current_file = os.path.join(kiosk_constants.pages_dir, "current.shtml")
+ emergency_file = os.path.join(kiosk_constants.pages_dir, "reload_immediately.html")
# Main chooser loop
while True:
if triggered:
if page != page_history[0] or (swap_page_target - now) < 10.0:
logger.info(f'chooser: An emergency page reload to {page} is needed at this time.')
- swap_page_target = now + constants.emergency_refresh_period_sec
+ swap_page_target = now + kiosk_constants.emergency_refresh_period_sec
# Set current.shtml to the right page.
try:
emit(
f,
page,
- override_refresh_sec = constants.emergency_refresh_period_sec,
+ override_refresh_sec = kiosk_constants.emergency_refresh_period_sec,
command = command
)
logger.debug(f'chooser: Wrote {current_file}.')
logger.info(
f'chooser: Nominal choice of {page} as the next to show.'
)
- swap_page_target = now + constants.refresh_period_sec
+ swap_page_target = now + kiosk_constants.refresh_period_sec
try:
with open(current_file, "w") as f:
emit(f, page)
return float(override_refresh_sec * 1000.0)
now = datetime.now(tz=pytz.timezone("US/Pacific"))
if now.hour < 6:
- return float(constants.refresh_period_night_sec * 1000.0)
+ return float(kiosk_constants.refresh_period_night_sec * 1000.0)
else:
- return float(constants.refresh_period_sec * 1000.0)
+ return float(kiosk_constants.refresh_period_sec * 1000.0)
age = file_utils.describe_file_ctime(f"pages/{filename}")
bgcolor = pick_background_color()
setInterval(check_reload, 500);
</SCRIPT>
</HEAD>
-""" % constants.root_url)
+""" % kiosk_constants.root_url)
f.write(f'<BODY BGCOLOR="#{bgcolor}">')
f.write(
"""
logger.info(
'renderer: Updating internal render statistics page.'
)
- with file_writer.file_writer(constants.render_stats_pagename) as f:
+ with file_writer.file_writer(kiosk_constants.render_stats_pagename) as f:
f.write(
'''
<CENTER>
# Update a page about internal stats of renderers.
renderer_update_internal_stats_page(last_render, render_counts, render_times)
logger.info('renderer: having a little nap...')
- time.sleep(constants.render_period_sec)
+ time.sleep(kiosk_constants.render_period_sec)
@bootstrap.initialize
# Have a little break and then check to make sure all threads are still alive.
logger.debug('watchdog: having a little nap.')
- time.sleep(constants.check_threads_period_sec)
+ time.sleep(kiosk_constants.check_threads_period_sec)
if __name__ == "__main__":
--- /dev/null
+#!/usr/bin/env python3
+
+hostname = 'kiosk.house'
+pages_dir = '/var/www/html/kiosk'
+root_url = f'http://{hostname}/kiosk'
+
+refresh_period_sec = 22.0
+emergency_refresh_period_sec = 45.0
+refresh_period_night_sec = 600.0
+render_period_sec = 30.0
+check_threads_period_sec = 60.0
+
+seconds_per_minute = 60
+seconds_per_hour = seconds_per_minute * 60
+seconds_per_day = seconds_per_hour * 24
+
+myq_pagename = "myq_4_300.html"
+render_stats_pagename = 'internal/render-stats_1_1000.html'
+gcal_imminent_pagename = "hidden/gcal-imminent_0_none.html"
import datetime_utils
-import constants
+import kiosk_constants
import file_writer
import renderer
import kiosk_secrets as secrets
return len(self.doors) > 0
def update_page(self) -> bool:
- with file_writer.file_writer(constants.myq_pagename) as f:
+ with file_writer.file_writer(kiosk_constants.myq_pagename) as f:
f.write(
f"""
<H1>Garage Door Status</H1>
#!/usr/bin/env python3
-import constants
+import kiosk_constants
import globals
import trigger
from typing import List, Optional, Tuple
def get_triggered_page_list(self) -> Optional[List[Tuple[str, int]]]:
if globals.get("myq_triggered"):
print("****** MyQ garage door is open page trigger ******")
- return [(constants.myq_pagename, trigger.trigger.PRIORITY_HIGH)]
+ return [(kiosk_constants.myq_pagename, trigger.trigger.PRIORITY_HIGH)]
else:
return None
import bellevue_city_calendar_renderer
import bellevue_reporter_rss_renderer
-import constants
+import kiosk_constants
import cnn_rss_renderer
import gdata_oauth
import gcal_renderer
import myq_renderer
import reddit_renderer
import seattletimes_rss_renderer
+import stevens_renderer
import kiosk_secrets as secrets
-import stranger_renderer
import stock_renderer
import twitter_renderer
import urbanist_renderer
seconds = 1
minutes = 60
-hours = constants.seconds_per_hour
+hours = kiosk_constants.seconds_per_hour
always = seconds * 1
["/rss/RSSMarketsMain.xml", "/rss/WSJcomUSBusiness.xml"],
"WSJBusiness",
),
+ stevens_renderer.stevens_renderer(
+ {'Unused': (minutes * 30)}
+ ),
google_news_rss_renderer.google_news_rss_renderer(
{"Fetch News": (minutes * 30), "Shuffle News": (always)},
"news.google.com",
#!/usr/bin/env python3
-import http.client
-from typing import List, Dict
-import xml.etree.ElementTree as ET
+import datetime
+import json
+import logging
+import requests
+from typing import Dict
-import renderer
+import datetime_utils
import file_writer
+import renderer
+
+
+logger = logging.getLogger(__file__)
+
+class stevens_renderer(renderer.abstaining_renderer):
+ URL = 'https://wsdot.com/Travel/Real-time/Service/api/MountainPass/Details/10'
-class stevens_pass_conditions_renderer(renderer.debuggable_abstaining_renderer):
- """Renders a page about Stevens Pass conditions."""
-
- def __init__(
- self, name_to_timeout_dict: Dict[str, int], feed_site: str, feed_uris: List[str]
- ) -> None:
- super(stevens_pass_conditions_renderer, self).__init__(
- name_to_timeout_dict, False
- )
- self.feed_site = feed_site
- self.feed_uris = feed_uris
-
- def debug_prefix(self) -> str:
- return "stevens"
-
- def periodic_render(self, key: str) -> bool:
- with file_writer.file_writer("stevens-conditions_1_86400.html") as f:
- for uri in self.feed_uris:
- self.conn = http.client.HTTPSConnection(self.feed_site)
- self.conn.request("GET", uri, None, {"Accept-Charset": "utf-8"})
- response = self.conn.getresponse()
- if response.status == 200:
- raw = response.read()
- rss = ET.fromstring(raw)
- channel = rss[0]
- for item in channel.getchildren():
- if item.tag == "title":
- f.write(f"<h1>{item.text}</h1><hr>")
- f.write(
- '<IMG WIDTH=512 ALIGN=RIGHT HEIGHT=382 SRC="https://images.wsdot.wa.gov/nc/002vc06430.jpg?t=637059938785646824" style="padding:8px;">'
- )
- elif item.tag == "item":
- for x in item.getchildren():
- if x.tag == "description":
- text = x.text
- if text is not None:
- text = text.replace(
- "<strong>Stevens Pass US2</strong><br/>", ""
- )
- text = text.replace("<br/><br/>", "<BR>")
- text = text.replace(
- "<strong>Elevation Meters:</strong>1238<BR>", ""
- )
- else:
- text = ""
- f.write(f"<P>\n{text}\n")
- return True
+ def __init__(self, name_to_timeout_dict: Dict[str, int]) -> None:
+ super().__init__(name_to_timeout_dict)
+
+ def render_conditions(mp: Dict[str, str], conditions: Dict[str, str]) -> str:
+ ret = f'''
+<TABLE>
+<TR>
+ <TD><B>temperature:</B></TD>
+ <TD>{conditions['temperature']}°{conditions['temperatureUnit'][0]}</TD>
+</TR>
+<TR>
+ <TD><B>weather:</B></TD>
+ <TD>{conditions['weather']}</TD>
+</TR>
+<TR>
+ <TD><B>road:</B></TD>
+ <TD>{conditions['roadCondition']}</TD>
+</TR>'''
+ if 'restrictionOne' in conditions or 'restrictionTwo' in conditions:
+ ret += '''
+<TR>
+ <TD><B>restrictions:</B></TD>
+ <TD>'''
+ if 'restrictionOne' in conditions:
+ ret += f'''
+ {conditions['restrictionOne']['travelDirectionName']}:
+ {conditions['restrictionOne']['publicPage']}<BR>'''
+ if 'restrictionTwo' in conditions:
+ ret += f'''
+ {conditions['restrictionTwo']['travelDirectionName']}:
+ {conditions['restrictionTwo']['publicPage']}'''
+ ret += '</TR></TABLE>'
+ return ret
+
+ def render_forecast(forecasts: Dict[str, str]) -> str:
+ ret = '<TABLE>'
+ fc = forecasts['forecast']['forecastData']
+ for n, f in enumerate(fc):
+ color = ''
+ if n % 2 == 0:
+ color = ' BGCOLOR="#dfefff"'
+ ret += f'''
+<TR>
+ <TD{color}><B>{f['periodText']}</B></TD>
+ <TD{color}>{f['forecastText']}</TD>
+</TR>'''
+ ret += '</TABLE>'
+ return ret
+
+ def render_image(cameras: Dict[str, str]) -> str:
+ for camera in cameras:
+ if camera['cameraId'] == 8063:
+ return f'''
+<CENTER>
+ <IMG SRC={camera['cameraUrl']} WIDTH={camera['width']}>
+ <BR>
+ <B>{camera['cameraLabel']} ({camera['direction']})</B>
+</CENTER>'''
+ return ''
+
+ def periodic_render(self, unused: str) -> bool:
+ page = requests.get(stevens_renderer.URL)
+ if page.status_code == 200:
+ contents = json.loads(page.content)
+ mp = contents['mountainPass']
+ conditions = contents['condition']
+ cameras = contents['cameras']
+ forecasts = contents['stationForecasts'][0]
+ now = datetime_utils.now_pacific()
+ tss = conditions['displayDate']
+ tss = tss.replace('Z', '+00:00')
+ ts = datetime.datetime.strptime(tss, '%Y-%m-%dT%H:%M:%S.%f%z')
+ tss = datetime_utils.describe_timedelta_briefly(now - ts)
+ with file_writer.file_writer('stevens-conditions_5_3000.html') as f:
+ f.write(f'''
+<H2>Stevens Pass Conditions ~{tss} ago:</H2>
+<HR>
+<TABLE WIDTH=90%>
+<TR>
+ <TD>
+ {stevens_renderer.render_conditions(mp, conditions)}
+ </TD>
+ <TD>
+ {stevens_renderer.render_image(cameras)}
+ </TD>
+</TR>
+<TR>
+ <TD COLSPAN=2>
+ {stevens_renderer.render_forecast(forecasts)}
+ </TD>
+</TR>
+</TABLE>''')
+ return True
return False
+
+test = stevens_renderer({"Test", 123})
+test.periodic_render("Test")
projects / kiosk.git / commitdiff