X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=myq_renderer.py;h=ebfac233927a8038cda3daca83a2fe23fb984d78;hb=addd4980077f6e3857c5c035b49784dc3ceca49a;hp=a4c29ff5160100ee24994f21dcf02596dedbcbc4;hpb=56167e8962b01d8e6bde57436f9ec04dd169666e;p=kiosk.git diff --git a/myq_renderer.py b/myq_renderer.py index a4c29ff..ebfac23 100644 --- a/myq_renderer.py +++ b/myq_renderer.py @@ -1,300 +1,127 @@ -#!/usr/local/bin/python +#!/usr/bin/env python3 -import requests -import os -import json -import time -import constants +from aiohttp import ClientSession +import asyncio import datetime -import file_writer -import globals -import httplib -import json -import renderer -import secrets - -APP_ID = "Vj8pQggXLhLy0WHahglCD4N1nAkkXQtGYpq2HrHD7H1nvmbT55KqtN6RSF4ILB/i" -HOST_URI = "myqexternal.myqdevice.com" -LOGIN_ENDPOINT = "api/v4/User/Validate" -DEVICE_LIST_ENDPOINT = "api/v4/UserDeviceDetails/Get" -DEVICE_SET_ENDPOINT = "api/v4/DeviceAttribute/PutDeviceAttribute" - -class MyQDoor: - myq_device_id = None - myq_lamp_device_id = None - myq_device_descr = None - myq_device_update_ts = None - myq_device_state = None - myq_security_token = None - - def __init__(self, device_id, lamp_id, descr, update_ts, state, token): - self.myq_device_id = device_id - self.myq_lamp_device_id = lamp_id - self.myq_device_descr = descr - self.myq_device_update_ts = update_ts - self.myq_device_state = state - self.myq_security_token = token - - def update(self, update_ts, state): - self.myq_device_update_ts = update_ts - self.myq_device_state = state +from dateutil.parser import parse +import pymyq # type: ignore +from typing import Dict, Optional - def get_name(self): - return self.myq_device_descr +from pyutils.datetimes import datetime_utils - def get_update_ts(self): - return self.myq_device_update_ts - - def open(self): - self.change_door_state("open") - - def close(self): - self.change_door_state("close") +import kiosk_constants +import file_writer +import renderer +import kiosk_secrets as secrets - def lamp_on(self): - self.change_lamp_state("on") - def lamp_off(self): - self.change_lamp_state("off") +class garage_door_renderer(renderer.abstaining_renderer): + def __init__(self, name_to_timeout_dict: Dict[str, int]) -> None: + super().__init__(name_to_timeout_dict) + self.doors: Optional[Dict] = None + self.last_update: Optional[datetime.datetime] = None - def get_status(self): - state = self.myq_device_state - if state == "1": - return "open" - elif state == "2": - return "closed" - elif state == "3": - return "stopped" - elif state == "4": - return "opening" - elif state == "5": - return "closing" - elif state == "8": - return "moving" - elif state == "9": - return "open" - else: - return str(state) + ", an unknown state for the door." - - def get_state_icon(self): - state = self.myq_device_state - if (state == "1" or state == "9"): - return "/icons/garage_open.png" - elif (state == "2"): - return "/icons/garage_closed.png" - elif (state == "4"): - return "/icons/garage_opening.png" - elif (state == "5" or state == "8"): - return "/icons/garage_closing.png" - else: - return str(state) + ", an unknown state for the door." + def debug_prefix(self) -> str: + return "myq" - def get_lamp_status(self): - state = self.check_lamp_state() - if state == "0": - return "off" - elif state == "1": - return "on" + def periodic_render(self, key: str) -> bool: + if key == "Poll MyQ": + self.last_update = datetime.datetime.now() + return asyncio.run(self.poll_myq()) + elif key == "Update Page": + return self.update_page() else: - return "unknown" - - def change_device_state(self, payload): - device_action = requests.put( - 'https://{host_uri}/{device_set_endpoint}'.format( - host_uri=HOST_URI, - device_set_endpoint=DEVICE_SET_ENDPOINT), - data=payload, - headers={ - 'MyQApplicationId': APP_ID, - 'SecurityToken': self.myq_security_token - } - ) - return device_action.status_code == 200 - - def change_lamp_state(self, command): - newstate = 1 if command.lower() == "on" else 0 - payload = { - "attributeName": "desiredlightstate", - "myQDeviceId": self.myq_lamp_device_id, - "AttributeValue": newstate - } - return self.change_device_state(payload) - - def change_door_state(self, command): - open_close_state = 1 if command.lower() == "open" else 0 - payload = { - 'attributeName': 'desireddoorstate', - 'myQDeviceId': self.myq_device_id, - 'AttributeValue': open_close_state, - } - return self.change_device_state(payload) - - def check_door_state(self): - return self.myq_device_state + raise Exception("Unknown operaiton") -# def check_lamp_state(self): -# return self.check_device_state(self.myq_lamp_device_id, "lightstate") - - def __repr__(self): - return "MyQ device(%s/%s), last update %s, current state %s" % ( - self.myq_device_descr, - self.myq_device_id, - self.myq_device_update_ts, - self.get_status()); - -class MyQService: - myq_security_token = "" - myq_device_list = {} - - def __init__(self, username, password): - self.username = username - self.password = password - - def login(self): - params = { - 'username': self.username, - 'password': self.password - } - login = requests.post( - 'https://{host_uri}/{login_endpoint}'.format( - host_uri=HOST_URI, - login_endpoint=LOGIN_ENDPOINT), - json=params, - headers={ - 'MyQApplicationId': APP_ID - } + async def poll_myq(self) -> bool: + async with ClientSession() as 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) -> bool: + with file_writer.file_writer(kiosk_constants.myq_pagename) as f: + f.write( + f""" +