X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;ds=inline;f=myq_renderer.py;h=df696f1f92304ab660f94992aca5df735c44334c;hb=e4dca16bbd329afdb587e8488767d88e17777254;hp=333c356a6e96891a26ddf8c36f743636c785b155;hpb=3b4fef0ca20b98c670acbaa2899bdb1b35eb12d7;p=kiosk.git diff --git a/myq_renderer.py b/myq_renderer.py index 333c356..df696f1 100644 --- a/myq_renderer.py +++ b/myq_renderer.py @@ -1,64 +1,74 @@ -#!/usr/local/bin/python3.7 +#!/usr/bin/env python3 -import pymyq from aiohttp import ClientSession import asyncio -import constants import datetime from dateutil.parser import parse +import pymyq # type: ignore +from typing import Dict, List, Optional + +import constants import file_writer import renderer import secrets +import utils + class garage_door_renderer(renderer.debuggable_abstaining_renderer): - def __init__(self, name_to_timeout_dict): + def __init__(self, name_to_timeout_dict: Dict[str, int]) -> None: super(garage_door_renderer, self).__init__(name_to_timeout_dict, False) - self.doors = None + self.doors: Optional[Dict] = None + self.last_update: Optional[datetime.datetime] = None - def debug_prefix(self): + def debug_prefix(self) -> str: return "myq" - def periodic_render(self, key): + def periodic_render(self, key: str) -> bool: if key == "Poll MyQ": - asyncio.get_event_loop().run_until_complete(self.poll_myq()) + self.last_update = datetime.datetime.now() + return asyncio.run(self.poll_myq()) elif key == "Update Page": return self.update_page() else: - raise error("Unknown operaiton") + raise Exception("Unknown operaiton") - async def poll_myq(self): + async def poll_myq(self) -> bool: async with ClientSession() as websession: - myq = await pymyq.login(secrets.myq_username, - secrets.myq_password, - websession) + myq = await pymyq.login( + secrets.myq_username, secrets.myq_password, websession + ) self.doors = myq.devices + assert(self.doors is not None) + return len(self.doors) > 0 - def update_page(self): - f = file_writer.file_writer(constants.myq_pagename) - now = datetime.datetime.now() - f.write(""" + def update_page(self) -> bool: + with file_writer.file_writer(constants.myq_pagename) as f: + f.write( + f"""