From 3b4fef0ca20b98c670acbaa2899bdb1b35eb12d7 Mon Sep 17 00:00:00 2001 From: Scott Gasch Date: Sat, 4 Jul 2020 11:51:23 -0700 Subject: [PATCH] Fix and re-enable the MyQ renderer. Small format tweaks to the Twitter renderer. --- myq_renderer.py | 324 ++++++++++---------------------------------- renderer_catalog.py | 6 +- twitter_renderer.py | 4 +- 3 files changed, 77 insertions(+), 257 deletions(-) diff --git a/myq_renderer.py b/myq_renderer.py index fa60642..333c356 100644 --- a/myq_renderer.py +++ b/myq_renderer.py @@ -1,251 +1,91 @@ -#!/usr/local/bin/python +#!/usr/local/bin/python3.7 -import requests -import os -import json -import time +import pymyq +from aiohttp import ClientSession +import asyncio import constants import datetime +from dateutil.parser import parse import file_writer -import globals -import http.client -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 - - def get_name(self): - return self.myq_device_descr - - 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") - - def lamp_on(self): - self.change_lamp_state("on") - - def lamp_off(self): - self.change_lamp_state("off") - - 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 get_lamp_status(self): - state = self.check_lamp_state() - if state == "0": - return "off" - elif state == "1": - return "on" - 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 - -# 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 - } - ) - auth = login.json() - self.myq_security_token = auth.get('SecurityToken') - return True - - def get_raw_device_list(self): - devices = requests.get( - 'https://{host_uri}/{device_list_endpoint}'.format( - host_uri=HOST_URI, - device_list_endpoint=DEVICE_LIST_ENDPOINT), - headers={ - 'MyQApplicationId': APP_ID, - 'SecurityToken': self.myq_security_token - }) - return devices.json().get('Devices') - - def get_devices(self): - return self.myq_device_list - - def update_devices(self): - devices = self.get_raw_device_list() - if devices is None: - return False - - for dev in devices: - if dev["MyQDeviceTypeId"] != 7: continue - identifier = str(dev["MyQDeviceId"]) - update_ts = None - state = None - name = None - for attr in dev["Attributes"]: - key = attr["AttributeDisplayName"] - value = attr["Value"] - if (key == "doorstate"): - state = value - ts = int(attr["UpdatedTime"]) / 1000.0 - update_ts = datetime.datetime.fromtimestamp(ts) - elif (key == "desc"): - name = value - if (identifier in self.myq_device_list): - self.myq_device_list[identifier].update( - update_ts, state) - else: - device = MyQDoor(identifier, - None, - name, - update_ts, - state, - self.myq_security_token) - self.myq_device_list[identifier] = device - return True - class garage_door_renderer(renderer.debuggable_abstaining_renderer): def __init__(self, name_to_timeout_dict): super(garage_door_renderer, self).__init__(name_to_timeout_dict, False) - self.myq_service = MyQService(secrets.myq_username, - secrets.myq_password) - self.myq_service.login() - self.myq_service.update_devices() + self.doors = None def debug_prefix(self): return "myq" def periodic_render(self, key): - self.debug_print("*** Executing action %s ***" % key) if key == "Poll MyQ": - return self.myq_service.update_devices() + asyncio.get_event_loop().run_until_complete(self.poll_myq()) elif key == "Update Page": return self.update_page() else: raise error("Unknown operaiton") - def do_door(self, name): - doors = self.myq_service.get_devices() + async def poll_myq(self): + async with ClientSession() as websession: + myq = await pymyq.login(secrets.myq_username, + secrets.myq_password, + websession) + self.doors = myq.devices + + def update_page(self): + f = file_writer.file_writer(constants.myq_pagename) now = datetime.datetime.now() - is_night = now.hour <= 7 or now.hour >= 21 - for key in doors: - door = doors[key] - if door.get_name() == name: - ts = door.get_update_ts() - delta = now - ts - d = int(delta.total_seconds()) - days = divmod(d, constants.seconds_per_day) + f.write(""" +

Garage Door Status

+ +
+ + +""" % now) + html = self.do_door("Near House") + if html == None: + return False + f.write(html) + + html = self.do_door("Middle Door") + if html == None: + return False + f.write(html) + f.write(""" + +
""") + f.close() + return True + + def get_state_icon(self, state): + if state == "open": + return "/icons/garage_open.png" + elif state == "closed": + return "/icons/garage_closed.png" + elif state == "opening": + return "/icons/garage_opening.png" + elif state == "closing": + return "/icons/garage_closing.png" + else: + return str(state) + ", an unknown state for the door." + + def do_door(self, name): + for key in self.doors: + door = self.doors[key] + if door.name == name: + j = self.doors[key].json + state = j["state"]["door_state"] + + # "last_update": "2020-07-04T18:11:34.2981419Z" + raw = j["state"]["last_update"] + ts = parse(raw) + tz_info = ts.tzinfo + now = datetime.datetime.now(tz_info) + delta = (now - ts).total_seconds() + now = datetime.datetime.now() + is_night = now.hour <= 7 or now.hour >= 21 + days = divmod(delta, constants.seconds_per_day) hours = divmod(days[1], constants.seconds_per_hour) minutes = divmod(hours[1], constants.seconds_per_minute) width = 0 @@ -267,34 +107,14 @@ class garage_door_renderer(renderer.debuggable_abstaining_renderer): for %d day(s), %02d:%02d. """ % (name, - door.get_state_icon(), + self.get_state_icon(state), width, color, - door.get_status(), + state, days[0], hours[0], minutes[0]) return None - def update_page(self): - f = file_writer.file_writer(constants.myq_pagename) - now = datetime.datetime.now() - f.write(""" -

Garage Door Status

- -
- - -""" % now) - html = self.do_door("Near House") - if html == None: - return False - f.write(html) - - html = self.do_door("Middle Door") - if html == None: - return False - f.write(html) - f.write(""" - -
""") - f.close() - return True +# Test +#x = garage_door_renderer({"Test" : 1}) +#x.periodic_render("Poll MyQ") +#x.periodic_render("Update Page") diff --git a/renderer_catalog.py b/renderer_catalog.py index b3b3e11..3cd8406 100644 --- a/renderer_catalog.py +++ b/renderer_catalog.py @@ -43,9 +43,9 @@ __registry = [ "Shuffle Events" : (always)}), # pollen_renderer.pollen_count_renderer( # {"Poll" : (hours * 1)}), -# myq_renderer.garage_door_renderer( -# {"Poll MyQ" : (minutes * 5), -# "Update Page" : (minutes * 5)}), + myq_renderer.garage_door_renderer( + {"Poll MyQ" : (minutes * 5), + "Update Page" : (minutes * 5)}), bellevue_reporter_rss_renderer.bellevue_reporter_rss_renderer( {"Fetch News" : (hours * 1), "Shuffle News" : (always)}, diff --git a/twitter_renderer.py b/twitter_renderer.py index 304cf81..8a82e5e 100644 --- a/twitter_renderer.py +++ b/twitter_renderer.py @@ -72,8 +72,8 @@ class twitter_renderer(renderer.debuggable_abstaining_renderer): handle = self.handles_by_author[author] tweets = self.tweets_by_author[author] already_seen = set() - f = file_writer.file_writer('twitter_5_none.html') - f.write('
') + f = file_writer.file_writer('twitter_10_none.html') + f.write('\n' % (author, handle)) f.write('
') f.write('

%s (@%s)

') f.write('
\n') -- 2.45.0