- 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()
-
- 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()
- elif key == "Update Page":
- return self.update_page()
- else:
- raise error("Unknown operaiton")
-
- def do_door(self, name):
- doors = self.myq_service.get_devices()
- 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)
- hours = divmod(days[1], constants.seconds_per_hour)
- minutes = divmod(hours[1], constants.seconds_per_minute)