diff --git a/pywink/device_types.py b/pywink/device_types.py deleted file mode 100644 index 67c65c3..0000000 --- a/pywink/device_types.py +++ /dev/null @@ -1,8 +0,0 @@ -LIGHT_BULB = 'light_bulb' -BINARY_SWITCH = 'binary_switch' -SENSOR_POD = 'sensor_pod' -LOCK = 'lock' -EGG_TRAY = 'eggtray' -GARAGE_DOOR = 'garage_door' -POWER_STRIP = 'powerstrip' -SIREN = 'siren' diff --git a/src/pywink/__init__.py b/src/pywink/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pywink/api.py b/src/pywink/api.py new file mode 100644 index 0000000..0ea9de8 --- /dev/null +++ b/src/pywink/api.py @@ -0,0 +1,193 @@ +import json + +import requests + +from pywink.devices import types as device_types +from pywink.devices.factory import build_device +from pywink.devices.sensors import WinkSensorPod, WinkHumiditySensor, WinkBrightnessSensor, WinkSoundPresenceSensor, \ + WinkTemperatureSensor, WinkVibrationPresenceSensor +from pywink.devices.types import DEVICE_ID_KEYS + +BASE_URL = "https://winkapi.quirky.com" + + +API_HEADERS = {} + + +class WinkApiInterface(object): + + def set_device_state(self, device, state, id_override=None): + """ + :type device: WinkDevice + :param state: a boolean of true (on) or false ('off') + :return: The JSON response from the API (new device state) + """ + _id = device.device_id() + if id_override: + _id = id_override + url_string = "{}/{}/{}".format(BASE_URL, + device.objectprefix, _id) + arequest = requests.put(url_string, + data=json.dumps(state), headers=API_HEADERS) + return arequest.json() + + +def set_bearer_token(token): + global API_HEADERS + + API_HEADERS = { + "Content-Type": "application/json", + "Authorization": "Bearer {}".format(token) + } + + +def get_bulbs(): + return get_devices(device_types.LIGHT_BULB) + + +def get_switches(): + return get_devices(device_types.BINARY_SWITCH) + + +def get_sensors(): + return get_devices(device_types.SENSOR_POD) + + +def get_locks(): + return get_devices(device_types.LOCK) + + +def get_eggtrays(): + return get_devices(device_types.EGG_TRAY) + + +def get_garage_doors(): + return get_devices(device_types.GARAGE_DOOR) + + +def get_powerstrip_outlets(): + return get_devices(device_types.POWER_STRIP) + + +def get_sirens(): + return get_devices(device_types.SIREN) + + +def get_devices(device_type): + arequest_url = "{}/users/me/wink_devices".format(BASE_URL) + response = requests.get(arequest_url, headers=API_HEADERS) + if response.status_code == 200: + response_dict = response.json() + filter_key = DEVICE_ID_KEYS.get(device_type) + return get_devices_from_response_dict(response_dict, + filter_key=filter_key) + + if response.status_code == 401: + raise WinkAPIException("401 Response from Wink API. Maybe Bearer token is expired?") + else: + raise WinkAPIException("Unexpected") + + +def get_devices_from_response_dict(response_dict, filter_key, api_interface): + """ + :rtype: list of WinkDevice + """ + items = response_dict.get('data') + + devices = [] + + keys = DEVICE_ID_KEYS.values() + if filter_key: + keys = [DEVICE_ID_KEYS.get(filter_key)] + + for item in items: + for key in keys: + if not __device_is_visible(item, key): + continue + + if key == "powerstrip_id": + devices.extend(__get_outlets_from_powerstrip(item, api_interface)) + continue # Don't capture the powerstrip itself as a device, only the individual outlets + + if key == "sensor_pod_id": + subsensors = _get_subsensors_from_sensor_pod(item, api_interface) + if subsensors: + devices.extend(subsensors) + + devices.append(build_device(item, api_interface)) + + return devices + + +def _get_subsensors_from_sensor_pod(item, api_interface): + + capabilities = [cap['field'] for cap in item.get('capabilities', {}).get('fields', [])] + if not capabilities: + return + + subsensors = [] + + if WinkHumiditySensor.CAPABILITY in capabilities: + subsensors.append(WinkHumiditySensor(item, api_interface)) + + if WinkBrightnessSensor.CAPABILITY in capabilities: + subsensors.append(WinkBrightnessSensor(item, api_interface)) + + if WinkSoundPresenceSensor.CAPABILITY in capabilities: + subsensors.append(WinkSoundPresenceSensor(item, api_interface)) + + if WinkTemperatureSensor.CAPABILITY in capabilities: + subsensors.append(WinkTemperatureSensor(item, api_interface)) + + if WinkVibrationPresenceSensor.CAPABILITY in capabilities: + subsensors.append(WinkVibrationPresenceSensor(item, api_interface)) + + if WinkSensorPod.CAPABILITY in capabilities: + subsensors.append(WinkSensorPod(item, api_interface)) + + return subsensors + + +def __get_outlets_from_powerstrip(item, api_interface): + outlets = item['outlets'] + return [build_device(outlet, api_interface) for outlet in outlets if __device_is_visible(outlet, 'outlet_id')] + + +def __device_is_visible(item, key): + is_correctly_structured = bool(item.get(key)) + is_visible = not item.get('hidden_at') + return is_correctly_structured and is_visible + + + + + +def refresh_state_at_hub(device): + """ + Tell hub to query latest status from device and upload to Wink. + PS: Not sure if this even works.. + :type device: WinkDevice + """ + url_string = "{}/{}/{}/refresh".format(BASE_URL, + device.objectprefix, + device.device_id()) + requests.get(url_string, headers=API_HEADERS) + + +def get_device_state(device): + """ + :type device: WinkDevice + """ + url_string = "{}/{}/{}".format(BASE_URL, + device.objectprefix, device.parent_id()) + arequest = requests.get(url_string, headers=API_HEADERS) + return arequest.json() + + +def is_token_set(): + """ Returns if an auth token has been set. """ + return bool(API_HEADERS) + + +class WinkAPIException(Exception): + pass diff --git a/src/pywink/devices/__init__.py b/src/pywink/devices/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/pywink/devices/base.py b/src/pywink/devices/base.py new file mode 100644 index 0000000..fa2861f --- /dev/null +++ b/src/pywink/devices/base.py @@ -0,0 +1,45 @@ + +class WinkDevice(object): + + def __init__(self, device_state_as_json, api_interface, objectprefix=None): + """ + :type api_interface pywink.api.WinkApiInterface: + :return: + """ + self.api_interface = api_interface + self.objectprefix = objectprefix + self.json_state = device_state_as_json + + def __str__(self): + return "%s %s %s" % (self.name(), self.device_id(), self.state()) + + def __repr__(self): + return "".format(name=self.name(), + device=self.device_id(), + state=self.state()) + + def name(self): + return self.json_state.get('name', "Unknown Name") + + def state(self): + raise NotImplementedError("Must implement state") + + def device_id(self): + raise NotImplementedError("Must implement device_id") + + @property + def _last_reading(self): + return self.json_state.get('last_reading') or {} + + def _update_state_from_response(self, response_json): + """ + :param response_json: the json obj returned from query + :return: + """ + self.json_state = response_json.get('data') + + def update_state(self): + """ Update state with latest info from Wink API. """ + response = self.api_interface.get_device_state(self) + self._update_state_from_response(response) + diff --git a/src/pywink/devices/factory.py b/src/pywink/devices/factory.py new file mode 100644 index 0000000..d2ade52 --- /dev/null +++ b/src/pywink/devices/factory.py @@ -0,0 +1,32 @@ +from pywink.devices.base import WinkDevice +from pywink.devices.sensors import WinkSensorPod +from pywink.devices.standard import WinkBulb, WinkBinarySwitch, WinkPowerStripOutlet, WinkLock, \ + WinkEggTray, WinkGarageDoor, WinkSiren + + +def build_device(device_state_as_json, api_interface): + + new_object = None + + # pylint: disable=redefined-variable-type + # These objects all share the same base class: WinkDevice + + if "light_bulb_id" in device_state_as_json: + new_object = WinkBulb(device_state_as_json, api_interface) + elif "sensor_pod_id" in device_state_as_json: + new_object = WinkSensorPod(device_state_as_json, api_interface) + elif "binary_switch_id" in device_state_as_json: + new_object = WinkBinarySwitch(device_state_as_json, api_interface) + elif "outlet_id" in device_state_as_json: + new_object = WinkPowerStripOutlet(device_state_as_json, api_interface) + elif "lock_id" in device_state_as_json: + new_object = WinkLock(device_state_as_json, api_interface) + elif "eggtray_id" in device_state_as_json: + new_object = WinkEggTray(device_state_as_json, api_interface) + elif "garage_door_id" in device_state_as_json: + new_object = WinkGarageDoor(device_state_as_json, api_interface) + elif "siren_id" in device_state_as_json: + new_object = WinkSiren(device_state_as_json, api_interface) + + return new_object or WinkDevice(device_state_as_json, api_interface) + diff --git a/src/pywink/devices/sensors.py b/src/pywink/devices/sensors.py new file mode 100644 index 0000000..7c28bce --- /dev/null +++ b/src/pywink/devices/sensors.py @@ -0,0 +1,145 @@ +# -*- coding: utf-8 -*- +from pywink.devices.base import WinkDevice + + +class _WinkCapabilitySensor(WinkDevice): + + def __init__(self, device_state_as_json, api_interface, capability, unit): + super(_WinkCapabilitySensor, self).__init__(device_state_as_json, api_interface, + objectprefix="sensor_pods") + self._capability = capability + self.unit = unit + + def __repr__(self): + return "".format(name=self.name(), + dev_id=self.device_id(), + reading=self._last_reading.get(self._capability), + unit='' if self.unit is None else self.unit) + + def state(self): + return self._last_reading.get('connection', False) + + def last_reading(self): + return self._last_reading.get(self._capability) + + def capability(self): + return self._capability + + def device_id(self): + root_name = self.json_state.get('sensor_pod_id', self.name()) + return '{}+{}'.format(root_name, self._capability) + + +class WinkSensorPod(_WinkCapabilitySensor): + """ represents a wink.py sensor + json_obj holds the json stat at init (if there is a refresh it's updated) + it's the native format for this objects methods + and looks like so: + """ + CAPABILITY = 'opened' + + def __init__(self, device_state_as_json, api_interface): + super(WinkSensorPod, self).__init__(device_state_as_json, api_interface, self.CAPABILITY, None) + + def __repr__(self): + return "" % (self.name(), + self.device_id(), self.state()) + + def state(self): + if 'opened' in self._last_reading: + return self._last_reading['opened'] + elif 'motion' in self._last_reading: + return self._last_reading['motion'] + return False + + def device_id(self): + return self.json_state.get('sensor_pod_id', self.name()) + + + +class WinkHumiditySensor(_WinkCapabilitySensor): + + CAPABILITY = 'humidity' + UNIT = '%' + + def __init__(self, device_state_as_json, api_interface): + super(WinkHumiditySensor, self).__init__(device_state_as_json, api_interface, + self.CAPABILITY, + self.UNIT) + + def humidity_percentage(self): + """ + :return: The relative humidity detected by the sensor (0% to 100%) + :rtype: int + """ + return self.last_reading() + + +class WinkBrightnessSensor(_WinkCapabilitySensor): + + CAPABILITY = 'brightness' + UNIT = '%' + + def __init__(self, device_state_as_json, api_interface): + super(WinkBrightnessSensor, self).__init__(device_state_as_json, api_interface, + self.CAPABILITY, + self.UNIT) + + def brightness_percentage(self): + """ + :return: The percentage of brightness as determined by the device. + :rtype: int + """ + return self.last_reading() + + +class WinkSoundPresenceSensor(_WinkCapabilitySensor): + + CAPABILITY = 'loudness' + + def __init__(self, device_state_as_json, api_interface): + super(WinkSoundPresenceSensor, self).__init__(device_state_as_json, api_interface, + self.CAPABILITY, + None) + + def loudness_boolean(self): + """ + :return: True if sound is detected. False if sound is below detection threshold (varies by device) + :rtype: bool + """ + return self.last_reading() + + +class WinkTemperatureSensor(_WinkCapabilitySensor): + + CAPABILITY = 'temperature' + UNIT = u'\N{DEGREE SIGN}' + + def __init__(self, device_state_as_json, api_interface): + super(WinkTemperatureSensor, self).__init__(device_state_as_json, api_interface, + self.CAPABILITY, + self.UNIT) + + def temperature_float(self): + """ + :return: A float indicating the temperature. Units are determined by the sensor. + :rtype: float + """ + return self.last_reading() + + +class WinkVibrationPresenceSensor(_WinkCapabilitySensor): + + CAPABILITY = 'vibration' + + def __init__(self, device_state_as_json, api_interface): + super(WinkVibrationPresenceSensor, self).__init__(device_state_as_json, api_interface, + self.CAPABILITY, + None) + + def vibration_boolean(self): + """ + :return: Returns True if vibration is detected. + :rtype: bool + """ + return self.last_reading() diff --git a/pywink/__init__.py b/src/pywink/devices/standard.py similarity index 57% rename from pywink/__init__.py rename to src/pywink/devices/standard.py index d0f67a2..92bfa90 100644 --- a/pywink/__init__.py +++ b/src/pywink/devices/standard.py @@ -2,104 +2,9 @@ Objects for interfacing with the Wink API """ import logging -import json import time -import requests -from pywink import device_types - -BASE_URL = "https://winkapi.quirky.com" - -HEADERS = {} - -DEVICE_ID_KEYS = { - device_types.BINARY_SWITCH: 'binary_switch_id', - device_types.EGG_TRAY: 'eggtray_id', - device_types.GARAGE_DOOR: 'garage_door_id', - device_types.LIGHT_BULB: 'light_bulb_id', - device_types.LOCK: 'lock_id', - device_types.POWER_STRIP: 'powerstrip_id', - device_types.SENSOR_POD: 'sensor_pod_id', - device_types.SIREN: 'siren_id' -} - - -class WinkDevice(object): - @staticmethod - def factory(device_state_as_json): - - new_object = None - - # pylint: disable=redefined-variable-type - # These objects all share the same base class: WinkDevice - - if "light_bulb_id" in device_state_as_json: - new_object = WinkBulb(device_state_as_json) - elif "sensor_pod_id" in device_state_as_json: - new_object = WinkSensorPod(device_state_as_json) - elif "binary_switch_id" in device_state_as_json: - new_object = WinkBinarySwitch(device_state_as_json) - elif "outlet_id" in device_state_as_json: - new_object = WinkPowerStripOutlet(device_state_as_json) - elif "lock_id" in device_state_as_json: - new_object = WinkLock(device_state_as_json) - elif "eggtray_id" in device_state_as_json: - new_object = WinkEggTray(device_state_as_json) - elif "garage_door_id" in device_state_as_json: - new_object = WinkGarageDoor(device_state_as_json) - elif "siren_id" in device_state_as_json: - new_object = WinkSiren(device_state_as_json) - - return new_object or WinkDevice(device_state_as_json) - - def __init__(self, device_state_as_json, objectprefix=None): - self.objectprefix = objectprefix - self.json_state = device_state_as_json - - def __str__(self): - return "%s %s %s" % (self.name(), self.device_id(), self.state()) - - def __repr__(self): - return "".format(name=self.name(), - device=self.device_id(), - state=self.state()) - - def name(self): - return self.json_state.get('name', "Unknown Name") - - def state(self): - raise NotImplementedError("Must implement state") - - def device_id(self): - raise NotImplementedError("Must implement device_id") - - @property - def _last_reading(self): - return self.json_state.get('last_reading') or {} - - def _update_state_from_response(self, response_json): - """ - :param response_json: the json obj returned from query - :return: - """ - self.json_state = response_json.get('data') - - def update_state(self): - """ Update state with latest info from Wink API. """ - url_string = "{}/{}/{}".format(BASE_URL, - self.objectprefix, self.device_id()) - arequest = requests.get(url_string, headers=HEADERS) - self._update_state_from_response(arequest.json()) - - def refresh_state_at_hub(self): - """ - Tell hub to query latest status from device and upload to Wink. - PS: Not sure if this even works.. - """ - url_string = "{}/{}/{}/refresh".format(BASE_URL, - self.objectprefix, - self.device_id()) - requests.get(url_string, headers=HEADERS) +from pywink.devices.base import WinkDevice class WinkEggTray(WinkDevice): @@ -108,8 +13,8 @@ class WinkEggTray(WinkDevice): it's the native format for this objects methods """ - def __init__(self, device_state_as_json, objectprefix="eggtrays"): - super(WinkEggTray, self).__init__(device_state_as_json, + def __init__(self, device_state_as_json, api_interface, objectprefix="eggtrays"): + super(WinkEggTray, self).__init__(device_state_as_json, api_interface, objectprefix=objectprefix) def __repr__(self): @@ -126,40 +31,14 @@ def device_id(self): return self.json_state.get('eggtray_id', self.name()) -class WinkSensorPod(WinkDevice): - """ represents a wink.py sensor - json_obj holds the json stat at init (if there is a refresh it's updated) - it's the native format for this objects methods - and looks like so: - """ - - def __init__(self, device_state_as_json, objectprefix="sensor_pods"): - super(WinkSensorPod, self).__init__(device_state_as_json, - objectprefix=objectprefix) - - def __repr__(self): - return "" % (self.name(), - self.device_id(), self.state()) - - def state(self): - if 'opened' in self._last_reading: - return self._last_reading['opened'] - elif 'motion' in self._last_reading: - return self._last_reading['motion'] - return False - - def device_id(self): - return self.json_state.get('sensor_pod_id', self.name()) - - class WinkBinarySwitch(WinkDevice): """ represents a wink.py switch json_obj holds the json stat at init (if there is a refresh it's updated) it's the native format for this objects methods """ - def __init__(self, device_state_as_json, objectprefix="binary_switches"): - super(WinkBinarySwitch, self).__init__(device_state_as_json, + def __init__(self, device_state_as_json, api_interface, objectprefix="binary_switches"): + super(WinkBinarySwitch, self).__init__(device_state_as_json, api_interface, objectprefix=objectprefix) # Tuple (desired state, time) self._last_call = (0, None) @@ -188,12 +67,8 @@ def set_state(self, state, **kwargs): :param state: a boolean of true (on) or false ('off') :return: nothing """ - url_string = "{}/{}/{}".format(BASE_URL, - self.objectprefix, self.device_id()) - values = {"desired_state": {"powered": state}} - arequest = requests.put(url_string, - data=json.dumps(values), headers=HEADERS) - self._update_state_from_response(arequest.json()) + response = self.api_interface.set_device_state(self, state) + self._update_state_from_response(response) self._last_call = (time.time(), state) @@ -232,8 +107,8 @@ class WinkBulb(WinkBinarySwitch): """ json_state = {} - def __init__(self, device_state_as_json): - super().__init__(device_state_as_json, + def __init__(self, device_state_as_json, api_interface): + super().__init__(device_state_as_json, api_interface, objectprefix="light_bulbs") def device_id(self): @@ -275,7 +150,6 @@ def set_state(self, state, brightness=None, CIE 1931 x,y color coordinates :return: nothing """ - url_string = "{}/light_bulbs/{}".format(BASE_URL, self.device_id()) values = { "desired_state": { "powered": state @@ -300,10 +174,8 @@ def set_state(self, state, brightness=None, values["desired_state"]["color_x"] = next(color_xy_iter) values["desired_state"]["color_y"] = next(color_xy_iter) - url_string = "{}/light_bulbs/{}".format(BASE_URL, self.device_id()) - arequest = requests.put(url_string, - data=json.dumps(values), headers=HEADERS) - self._update_state_from_response(arequest.json()) + response = self.api_interface.set_device_state(self, values) + self._update_state_from_response(response) self._last_call = (time.time(), state) @@ -319,8 +191,8 @@ class WinkLock(WinkDevice): it's the native format for this objects methods """ - def __init__(self, device_state_as_json, objectprefix="locks"): - super(WinkLock, self).__init__(device_state_as_json, + def __init__(self, device_state_as_json, api_interface, objectprefix="locks"): + super(WinkLock, self).__init__(device_state_as_json, api_interface, objectprefix=objectprefix) # Tuple (desired state, time) self._last_call = (0, None) @@ -345,13 +217,9 @@ def set_state(self, state): :param state: a boolean of true (on) or false ('off') :return: nothing """ - url_string = "{}/{}/{}".format(BASE_URL, - self.objectprefix, self.device_id()) values = {"desired_state": {"locked": state}} - arequest = requests.put(url_string, - data=json.dumps(values), headers=HEADERS) - self._update_state_from_response(arequest.json()) - + response = self.api_interface.set_device_state(self, values) + self._update_state_from_response(response) self._last_call = (time.time(), state) def wait_till_desired_reached(self): @@ -386,8 +254,8 @@ class WinkPowerStripOutlet(WinkBinarySwitch): and looks like so: """ - def __init__(self, device_state_as_json, objectprefix="powerstrips"): - super(WinkPowerStripOutlet, self).__init__(device_state_as_json, + def __init__(self, device_state_as_json, api_interface, objectprefix="powerstrips"): + super(WinkPowerStripOutlet, self).__init__(device_state_as_json, api_interface, objectprefix=objectprefix) # Tuple (desired state, time) self._last_call = (0, None) @@ -418,10 +286,8 @@ def _update_state_from_response(self, response_json): def update_state(self): """ Update state with latest info from Wink API. """ - url_string = "{}/{}/{}".format(BASE_URL, - self.objectprefix, self.parent_id()) - arequest = requests.get(url_string, headers=HEADERS) - self._update_state_from_response(arequest.json()) + response = self.api_interface.get_device_state(self) + self._update_state_from_response(response) def index(self): return self.json_state.get('outlet_index', None) @@ -440,16 +306,13 @@ def set_state(self, state, **kwargs): :param state: a boolean of true (on) or false ('off') :return: nothing """ - url_string = "{}/{}/{}".format(BASE_URL, - self.objectprefix, self.parent_id()) if self.index() == 0: values = {"outlets": [{"desired_state": {"powered": state}}, {}]} else: values = {"outlets": [{}, {"desired_state": {"powered": state}}]} - arequest = requests.put(url_string, - data=json.dumps(values), headers=HEADERS) - self._update_state_from_response(arequest.json()) + response = self.api_interface.set_device_state(self, values, id_override=self.parent_id()) + self._update_state_from_response(response) self._last_call = (time.time(), state) @@ -485,8 +348,8 @@ class WinkGarageDoor(WinkDevice): and looks like so: """ - def __init__(self, device_state_as_json, objectprefix="garage_doors"): - super(WinkGarageDoor, self).__init__(device_state_as_json, + def __init__(self, device_state_as_json, api_interface, objectprefix="garage_doors"): + super(WinkGarageDoor, self).__init__(device_state_as_json, api_interface, objectprefix=objectprefix) # Tuple (desired state, time) self._last_call = (0, None) @@ -510,10 +373,9 @@ def set_state(self, state): :param state: a number of 1 ('open') or 0 ('close') :return: nothing """ - url_string = "{}/{}/{}".format(BASE_URL, self.objectprefix, self.device_id()) values = {"desired_state": {"position": state}} - arequest = requests.put(url_string, data=json.dumps(values), headers=HEADERS) - self._update_state_from_response(arequest.json()) + response = self.api_interface.set_device_state(self, values) + self._update_state_from_response(response) self._last_call = (time.time(), state) @@ -549,8 +411,8 @@ class WinkSiren(WinkBinarySwitch): it's the native format for this objects methods """ - def __init__(self, device_state_as_json, objectprefix="sirens"): - super(WinkSiren, self).__init__(device_state_as_json, + def __init__(self, device_state_as_json, api_interface, objectprefix="sirens"): + super(WinkSiren, self).__init__(device_state_as_json, api_interface, objectprefix=objectprefix) # Tuple (desired state, time) self._last_call = (0, None) @@ -561,94 +423,3 @@ def __repr__(self): def device_id(self): return self.json_state.get('siren_id', self.name()) - - -def get_devices(device_type): - arequest_url = "{}/users/me/wink_devices".format(BASE_URL) - response = requests.get(arequest_url, headers=HEADERS) - if response.status_code == 200: - response_dict = response.json() - filter_key = DEVICE_ID_KEYS.get(device_type) - return get_devices_from_response_dict(response_dict, - filter_key=filter_key) - - if response.status_code == 401: - raise WinkAPIException("401 Response from Wink API. Maybe Bearer token is expired?") - else: - raise WinkAPIException("Unexpected") - - -def get_devices_from_response_dict(response_dict, filter_key): - items = response_dict.get('data') - - devices = [] - - keys = DEVICE_ID_KEYS.values() - if filter_key: - keys = [DEVICE_ID_KEYS.get(filter_key)] - - for item in items: - for key in keys: - value_at_key = item.get(key) - if value_at_key is not None and item.get("hidden_at") is None: - if key == "powerstrip_id": - outlets = item['outlets'] - for outlet in outlets: - value_at_key = outlet.get('outlet_id') - if (value_at_key is not None and - outlet.get("hidden_at") is None): - devices.append(WinkDevice.factory(outlet)) - else: - devices.append(WinkDevice.factory(item)) - - return devices - - -def get_bulbs(): - return get_devices(device_types.LIGHT_BULB) - - -def get_switches(): - return get_devices(device_types.BINARY_SWITCH) - - -def get_sensors(): - return get_devices(device_types.SENSOR_POD) - - -def get_locks(): - return get_devices(device_types.LOCK) - - -def get_eggtrays(): - return get_devices(device_types.EGG_TRAY) - - -def get_garage_doors(): - return get_devices(device_types.GARAGE_DOOR) - - -def get_powerstrip_outlets(): - return get_devices(device_types.POWER_STRIP) - - -def get_sirens(): - return get_devices(device_types.SIREN) - - -def is_token_set(): - """ Returns if an auth token has been set. """ - return bool(HEADERS) - - -def set_bearer_token(token): - global HEADERS - - HEADERS = { - "Content-Type": "application/json", - "Authorization": "Bearer {}".format(token) - } - - -class WinkAPIException(Exception): - pass diff --git a/src/pywink/devices/types.py b/src/pywink/devices/types.py new file mode 100644 index 0000000..bbb21ad --- /dev/null +++ b/src/pywink/devices/types.py @@ -0,0 +1,20 @@ +LIGHT_BULB = 'light_bulb' +BINARY_SWITCH = 'binary_switch' +SENSOR_POD = 'sensor_pod' +LOCK = 'lock' +EGG_TRAY = 'eggtray' +GARAGE_DOOR = 'garage_door' +POWER_STRIP = 'powerstrip' +SIREN = 'siren' + +DEVICE_ID_KEYS = { + BINARY_SWITCH: 'binary_switch_id', + EGG_TRAY: 'eggtray_id', + GARAGE_DOOR: 'garage_door_id', + LIGHT_BULB: 'light_bulb_id', + LOCK: 'lock_id', + POWER_STRIP: 'powerstrip_id', + SENSOR_POD: 'sensor_pod_id', + SIREN: 'siren_id' +} + diff --git a/tests/api_responses/quirky_spotter.json b/tests/api_responses/quirky_spotter.json new file mode 100644 index 0000000..e9d4352 --- /dev/null +++ b/tests/api_responses/quirky_spotter.json @@ -0,0 +1,123 @@ +{ + "data": [ + { + "last_event": { + "brightness_occurred_at": 1445973676.198793, + "loudness_occurred_at": 1453186523.6125298, + "vibration_occurred_at": 1453186429.210991 + }, + "desired_state": {}, + "last_reading": { + "battery": 0.85, + "battery_updated_at": 1453187132.7789793, + "brightness": 1, + "brightness_updated_at": 1453187132.7789793, + "external_power": true, + "external_power_updated_at": 1453187132.7789793, + "humidity": 48, + "humidity_updated_at": 1453187132.7789793, + "loudness": false, + "loudness_updated_at": 1453187132.7789793, + "temperature": 5, + "temperature_updated_at": 1453187132.7789793, + "vibration": false, + "vibration_updated_at": 1453187132.7789793, + "brightness_true": "N/A", + "brightness_true_updated_at": 1445973676.198793, + "loudness_true": "N/A", + "loudness_true_updated_at": 1453186523.6125298, + "vibration_true": "N/A", + "vibration_true_updated_at": 1453186429.210991, + "connection": true, + "connection_updated_at": 1453187132.7789793, + "agent_session_id": null, + "agent_session_id_updated_at": null, + "desired_battery_updated_at": null, + "desired_brightness_updated_at": null, + "desired_external_power_updated_at": null, + "desired_humidity_updated_at": null, + "desired_loudness_updated_at": null, + "desired_temperature_updated_at": null, + "desired_vibration_updated_at": null, + "loudness_changed_at": 1453186586.7168324, + "loudness_true_changed_at": 1453186523.6125298, + "vibration_changed_at": 1453186528.978827, + "vibration_true_changed_at": 1453186429.210991, + "temperature_changed_at": 1453186523.6125298, + "humidity_changed_at": 1453187132.7789793, + "brightness_changed_at": 1445973676.198793, + "brightness_true_changed_at": 1445973676.198793, + "battery_changed_at": 1452267645.8792017 + }, + "sensor_pod_id": "72503", + "uuid": "0d889d64-e77b-48a3-a132-475626f8ab1f", + "name": "Well Spotter", + "locale": "en_us", + "units": {}, + "created_at": 1432962859, + "hidden_at": null, + "capabilities": { + "fields": [ + { + "field": "battery", + "type": "percentage", + "mutability": "read-only" + }, + { + "field": "brightness", + "type": "percentage", + "mutability": "read-only" + }, + { + "field": "external_power", + "type": "boolean", + "mutability": "read-only" + }, + { + "field": "humidity", + "type": "percentage", + "mutability": "read-only" + }, + { + "field": "loudness", + "type": "boolean", + "mutability": "read-only" + }, + { + "field": "temperature", + "type": "float", + "mutability": "read-only" + }, + { + "field": "vibration", + "type": "boolean", + "mutability": "read-only" + } + ] + }, + "triggers": [], + "manufacturer_device_model": "quirky_ge_spotter", + "manufacturer_device_id": null, + "device_manufacturer": "quirky_ge", + "model_name": "Spotter", + "upc_id": "25", + "upc_code": "814434018858", + "gang_id": null, + "hub_id": null, + "local_id": null, + "radio_type": null, + "linked_service_id": null, + "lat_lng": [ + null, + null + ], + "location": "", + "mac_address": "0c2a5905a5a2", + "serial": "ABAB00010864" + } + ], + "errors": [], + "pagination": { + "count": 24 + } +} diff --git a/tests/api_responses/quirky_spotter_2.json b/tests/api_responses/quirky_spotter_2.json new file mode 100644 index 0000000..51295f9 --- /dev/null +++ b/tests/api_responses/quirky_spotter_2.json @@ -0,0 +1,116 @@ +{ + "data": [ + { + "last_event": { + "brightness_occurred_at": 1450698768.6228566, + "loudness_occurred_at": 1453188090.419577, + "vibration_occurred_at": 1453187050.929243 + }, + "desired_state": {}, + "last_reading": { + "battery": 0.86, + "battery_updated_at": 1453188090.419577, + "brightness": 0, + "brightness_updated_at": 1453188090.419577, + "external_power": true, + "external_power_updated_at": 1453188090.419577, + "humidity": 27, + "humidity_updated_at": 1453188090.419577, + "loudness": 1, + "loudness_updated_at": 1453188090.419577, + "temperature": 16, + "temperature_updated_at": 1453188090.419577, + "vibration": false, + "vibration_updated_at": 1453188090.419577, + "brightness_true": "N/A", + "brightness_true_updated_at": 1450698768.6228566, + "loudness_true": "N/A", + "loudness_true_updated_at": 1453188090.419577, + "vibration_true": "N/A", + "vibration_true_updated_at": 1453187050.929243, + "connection": true, + "connection_updated_at": 1453188090.419577, + "agent_session_id": null, + "agent_session_id_updated_at": null, + "humidity_changed_at": 1453187780.769285, + "battery_changed_at": 1453014608.7718346, + "temperature_changed_at": 1453169998.9940693, + "vibration_changed_at": 1453187056.2631874, + "loudness_changed_at": 1453188090.419577, + "loudness_true_changed_at": 1453188090.419577, + "vibration_true_changed_at": 1453187050.929243, + "brightness_changed_at": 1450698773.854434, + "brightness_true_changed_at": 1450698768.6228566 + }, + "sensor_pod_id": "84197", + "uuid": "c34335fe-208a-491d-b4d6-685e609e0088", + "name": "Spotter", + "locale": "en_us", + "units": {}, + "created_at": 1436338918, + "hidden_at": null, + "capabilities": { + "fields": [ + { + "field": "battery", + "type": "percentage", + "mutability": "read-only" + }, + { + "field": "brightness", + "type": "percentage", + "mutability": "read-only" + }, + { + "field": "external_power", + "type": "boolean", + "mutability": "read-only" + }, + { + "field": "humidity", + "type": "percentage", + "mutability": "read-only" + }, + { + "field": "loudness", + "type": "boolean", + "mutability": "read-only" + }, + { + "field": "temperature", + "type": "float", + "mutability": "read-only" + }, + { + "field": "vibration", + "type": "boolean", + "mutability": "read-only" + } + ] + }, + "triggers": [], + "manufacturer_device_model": "quirky_ge_spotter", + "manufacturer_device_id": null, + "device_manufacturer": "quirky_ge", + "model_name": "Spotter", + "upc_id": "25", + "upc_code": "814434018858", + "gang_id": null, + "hub_id": null, + "local_id": null, + "radio_type": null, + "linked_service_id": null, + "lat_lng": [ + null, + null + ], + "location": "", + "mac_address": "0v2a6905b738", + "serial": "ABAB00006476" + } + ], + "errors": [], + "pagination": { + "count": 24 + } +} diff --git a/tests/init_test.py b/tests/init_test.py index 3a83b39..8e19ef3 100644 --- a/tests/init_test.py +++ b/tests/init_test.py @@ -2,15 +2,22 @@ import mock import unittest -from pywink import WinkBulb, get_devices_from_response_dict, device_types, WinkGarageDoor, WinkPowerStripOutlet, \ - WinkLock, WinkBinarySwitch, WinkSensorPod, WinkEggTray, WinkSiren +from pywink.api import get_devices_from_response_dict, WinkApiInterface +from pywink.devices import types as device_types +from pywink.devices.sensors import WinkSensorPod +from pywink.devices.standard import WinkBulb, WinkGarageDoor, WinkPowerStripOutlet, WinkSiren, WinkLock, \ + WinkBinarySwitch, WinkEggTray class LightSetStateTests(unittest.TestCase): + def setUp(self): + super(LightSetStateTests, self).setUp() + self.api_interface = WinkApiInterface() + @mock.patch('requests.put') def test_should_send_correct_color_xy_values_to_wink_api(self, put_mock): - bulb = WinkBulb({}) + bulb = WinkBulb({}, self.api_interface) color_x = 0.75 color_y = 0.25 bulb.set_state(True, color_xy=[color_x, color_y]) @@ -21,7 +28,7 @@ def test_should_send_correct_color_xy_values_to_wink_api(self, put_mock): @mock.patch('requests.put') def test_should_send_correct_color_temperature_values_to_wink_api(self, put_mock): - bulb = WinkBulb({}) + bulb = WinkBulb({}, self.api_interface) arbitrary_kelvin_color = 4950 bulb.set_state(True, color_kelvin=arbitrary_kelvin_color) sent_data = json.loads(put_mock.call_args[1].get('data')) @@ -30,7 +37,7 @@ def test_should_send_correct_color_temperature_values_to_wink_api(self, put_mock @mock.patch('requests.put') def test_should_only_send_color_xy_if_both_color_xy_and_color_temperature_are_given(self, put_mock): - bulb = WinkBulb({}) + bulb = WinkBulb({}, self.api_interface) arbitrary_kelvin_color = 4950 bulb.set_state(True, color_kelvin=arbitrary_kelvin_color, color_xy=[0, 1]) sent_data = json.loads(put_mock.call_args[1].get('data')) @@ -121,12 +128,16 @@ def test_should_show_powered_state_as_false_if_device_is_disconnected(self): """ response_dict = json.loads(response) - devices = get_devices_from_response_dict(response_dict, device_types.POWER_STRIP) + devices = get_devices_from_response_dict(response_dict, device_types.POWER_STRIP, None) self.assertFalse(devices[0].state()) class WinkAPIResponseHandlingTests(unittest.TestCase): + def setUp(self): + super(WinkAPIResponseHandlingTests, self).setUp() + self.api_interface = mock.MagicMock() + def test_should_handle_light_bulb_response(self): response = """ { @@ -172,7 +183,7 @@ def test_should_handle_light_bulb_response(self): } """ response_dict = json.loads(response) - devices = get_devices_from_response_dict(response_dict, device_types.LIGHT_BULB) + devices = get_devices_from_response_dict(response_dict, device_types.LIGHT_BULB, self.api_interface) self.assertEqual(1, len(devices)) self.assertIsInstance(devices[0], WinkBulb) @@ -245,7 +256,7 @@ def test_should_handle_garage_door_opener_response(self): } """ response_dict = json.loads(response) - devices = get_devices_from_response_dict(response_dict, device_types.GARAGE_DOOR) + devices = get_devices_from_response_dict(response_dict, device_types.GARAGE_DOOR, self.api_interface) self.assertEqual(1, len(devices)) self.assertIsInstance(devices[0], WinkGarageDoor) @@ -337,7 +348,7 @@ def test_should_handle_power_strip_response(self): } """ response_dict = json.loads(response) - devices = get_devices_from_response_dict(response_dict, device_types.POWER_STRIP) + devices = get_devices_from_response_dict(response_dict, device_types.POWER_STRIP, self.api_interface) self.assertEqual(2, len(devices)) self.assertIsInstance(devices[0], WinkPowerStripOutlet) self.assertIsInstance(devices[1], WinkPowerStripOutlet) @@ -412,7 +423,7 @@ def test_should_handle_siren_response(self): """ response_dict = json.loads(response) - devices = get_devices_from_response_dict(response_dict, device_types.SIREN) + devices = get_devices_from_response_dict(response_dict, device_types.SIREN, self.api_interface) self.assertEqual(1, len(devices)) self.assertIsInstance(devices[0], WinkSiren) @@ -559,7 +570,7 @@ def test_should_handle_lock_response(self): """ response_dict = json.loads(response) - devices = get_devices_from_response_dict(response_dict, device_types.LOCK) + devices = get_devices_from_response_dict(response_dict, device_types.LOCK, self.api_interface) self.assertEqual(1, len(devices)) self.assertIsInstance(devices[0], WinkLock) @@ -631,7 +642,7 @@ def test_should_handle_binary_switch_response(self): """ response_dict = json.loads(response) - devices = get_devices_from_response_dict(response_dict, device_types.BINARY_SWITCH) + devices = get_devices_from_response_dict(response_dict, device_types.BINARY_SWITCH, self.api_interface) self.assertEqual(1, len(devices)) self.assertIsInstance(devices[0], WinkBinarySwitch) @@ -708,7 +719,7 @@ def test_should_handle_sensor_pod_response(self): """ response_dict = json.loads(response) - devices = get_devices_from_response_dict(response_dict, device_types.SENSOR_POD) + devices = get_devices_from_response_dict(response_dict, device_types.SENSOR_POD, self.api_interface) self.assertEqual(1, len(devices)) self.assertIsInstance(devices[0], WinkSensorPod) @@ -798,6 +809,6 @@ def test_should_handle_egg_tray_response(self): """ response_dict = json.loads(response) - devices = get_devices_from_response_dict(response_dict, device_types.EGG_TRAY) + devices = get_devices_from_response_dict(response_dict, device_types.EGG_TRAY, self.api_interface) self.assertEqual(1, len(devices)) self.assertIsInstance(devices[0], WinkEggTray) diff --git a/tests/sensor_test.py b/tests/sensor_test.py new file mode 100644 index 0000000..954dc78 --- /dev/null +++ b/tests/sensor_test.py @@ -0,0 +1,74 @@ +import json +import unittest + +from pywink import device_types +from pywink.api import get_devices_from_response_dict +from pywink.devices.sensors import WinkBrightnessSensor, WinkHumiditySensor, WinkSoundPresenceSensor, \ + WinkVibrationPresenceSensor, WinkTemperatureSensor + + +class SensorTests(unittest.TestCase): + + def test_quirky_spotter_api_response_should_create_unique_one_primary_sensor_and_five_subsensors(self): + with open('api_responses/quirky_spotter.json') as spotter_file: + response_dict = json.load(spotter_file) + + sensors = get_devices_from_response_dict(response_dict, device_types.SENSOR_POD, None) + self.assertEquals(1 + 5, len(sensors)) + + def test_alternative_quirky_spotter_api_response_should_create_one_primary_sensor_and_five_subsensors(self): + with open('api_responses/quirky_spotter_2.json') as spotter_file: + response_dict = json.load(spotter_file) + + sensors = get_devices_from_response_dict(response_dict, device_types.SENSOR_POD, None) + self.assertEquals(1 + 5, len(sensors)) + + def test_brightness_should_have_correct_value(self): + with open('api_responses/quirky_spotter.json') as spotter_file: + response_dict = json.load(spotter_file) + + sensors = get_devices_from_response_dict(response_dict, device_types.SENSOR_POD, None) + """:type : list of WinkBrightnessSensor""" + brightness_sensor = [sensor for sensor in sensors if sensor.capability() is WinkBrightnessSensor.CAPABILITY][0] + expected_brightness = 1 + self.assertEquals(expected_brightness, brightness_sensor.brightness_percentage()) + + def test_humidity_should_have_correct_value(self): + with open('api_responses/quirky_spotter.json') as spotter_file: + response_dict = json.load(spotter_file) + + sensors = get_devices_from_response_dict(response_dict, device_types.SENSOR_POD, None) + """:type : list of WinkHumiditySensor""" + humidity_sensor = [sensor for sensor in sensors if sensor.capability() is WinkHumiditySensor.CAPABILITY][0] + expected_humidity = 48 + self.assertEquals(expected_humidity, humidity_sensor.humidity_percentage()) + + def test_loudness_should_have_correct_value(self): + with open('api_responses/quirky_spotter.json') as spotter_file: + response_dict = json.load(spotter_file) + + sensors = get_devices_from_response_dict(response_dict, device_types.SENSOR_POD, None) + """:type : list of WinkSoundPresenceSensor""" + sound_sensor = [sensor for sensor in sensors if sensor.capability() is WinkSoundPresenceSensor.CAPABILITY][0] + expected_sound_presence = False + self.assertEquals(expected_sound_presence, sound_sensor.loudness_boolean()) + + def test_vibration_should_have_correct_value(self): + with open('api_responses/quirky_spotter.json') as spotter_file: + response_dict = json.load(spotter_file) + + sensors = get_devices_from_response_dict(response_dict, device_types.SENSOR_POD, None) + """:type : list of WinkVibrationPresenceSensor""" + sound_sensor = [sensor for sensor in sensors if sensor.capability() is WinkVibrationPresenceSensor.CAPABILITY][0] + expected_vibrartion_presence = False + self.assertEquals(expected_vibrartion_presence, sound_sensor.vibration_boolean()) + + def test_temperature_should_have_correct_value(self): + with open('api_responses/quirky_spotter.json') as spotter_file: + response_dict = json.load(spotter_file) + + sensors = get_devices_from_response_dict(response_dict, device_types.SENSOR_POD, None) + """:type : list of WinkTemperatureSensor""" + sound_sensor = [sensor for sensor in sensors if sensor.capability() is WinkTemperatureSensor.CAPABILITY][0] + expected_temperature = 5 + self.assertEquals(expected_temperature, sound_sensor.temperature_float())