diff --git a/README.rst b/README.rst index d152438..dc765da 100644 --- a/README.rst +++ b/README.rst @@ -46,3 +46,11 @@ Changelog - Add Type 10 - SkyLift - Handle calls to update shade position during maintenance - Raise error directly on hub calls instead of logger + +**v3.0.0** + +- Major overhaul to incorporate gateway version 3 API. Version can be automatically detected or manually specified. +- UserData class is deprecated and replaced with Hub. +- ShadePosition class now replaces the raw json management of shades in support of cross generational management. +- Schedules / Automations are now supported by the API +- New get_*objecttype* methods available to returned structured data objects for consistent management \ No newline at end of file diff --git a/aiopvapi/__version__.py b/aiopvapi/__version__.py index e5662b7..242ca93 100644 --- a/aiopvapi/__version__.py +++ b/aiopvapi/__version__.py @@ -1,3 +1,3 @@ """Aio PowerView api version.""" -__version__ = "2.0.4" +__version__ = "3.0.0" diff --git a/aiopvapi/automations.py b/aiopvapi/automations.py new file mode 100644 index 0000000..4bdc354 --- /dev/null +++ b/aiopvapi/automations.py @@ -0,0 +1,59 @@ +"""Scenes class managing all scene data.""" + +import logging + +from aiopvapi.helpers.aiorequest import AioRequest +from aiopvapi.helpers.api_base import ApiEntryPoint +from aiopvapi.helpers.constants import ( + ATTR_ID, + ATTR_SCHEDULED_EVENT_DATA, +) +from aiopvapi.resources.automation import Automation + +from aiopvapi.resources.model import PowerviewData + +_LOGGER = logging.getLogger(__name__) + + +class Automations(ApiEntryPoint): + """Powerview Automations""" + + def __init__(self, request: AioRequest) -> None: + self.api_endpoint = "scheduledevents" + if request.api_version >= 3: + self.api_endpoint = "automations" + super().__init__(request, self.api_endpoint) + + def _resource_factory(self, raw): + return Automation(raw, self.request) + + def _loop_raw(self, raw): + if self.api_version < 3: + raw = raw[ATTR_SCHEDULED_EVENT_DATA] + + for _raw in raw: + yield _raw + + def _get_to_actual_data(self, raw): + if self.api_version >= 3: + return raw + return raw.get("scene") + + async def get_automations(self, fetch_scene_data: bool = True) -> PowerviewData: + """Get a list of automations. + + :returns PowerviewData object + :raises PvApiError when an error occurs. + """ + resources = await self.get_resources() + if self.api_version < 3: + resources = resources[ATTR_SCHEDULED_EVENT_DATA] + + processed = {entry[ATTR_ID]: Automation(entry, self.request) for entry in resources} + + if fetch_scene_data is True: + for automation in processed.values(): + await automation.fetch_associated_scene_data() + + _LOGGER.debug("Raw automation data: %s", resources) + return PowerviewData(raw=resources, processed=processed) diff --git a/aiopvapi/example/hub.py b/aiopvapi/example/hub.py index 665446e..b314a9b 100644 --- a/aiopvapi/example/hub.py +++ b/aiopvapi/example/hub.py @@ -19,8 +19,8 @@ async def get_firmware(hub_ip): async def get_user_data(hub_ip): request = AioRequest(hub_ip) hub = Hub(request) - await hub.query_user_data() + await hub.query_firmware() print("UserData") - print("hub name: {}".format(hub.user_data.hub_name)) - pprint(hub.user_data._raw) + print("hub name: {}".format(hub.hub_name)) + pprint(hub._raw) diff --git a/aiopvapi/helpers/aiorequest.py b/aiopvapi/helpers/aiorequest.py index 4fc7f60..9f3ac81 100644 --- a/aiopvapi/helpers/aiorequest.py +++ b/aiopvapi/helpers/aiorequest.py @@ -6,6 +6,9 @@ import aiohttp import async_timeout +from aiopvapi.helpers.constants import FWVERSION +from aiopvapi.helpers.tools import join_path, get_base_path + _LOGGER = logging.getLogger(__name__) @@ -13,32 +16,30 @@ class PvApiError(Exception): """General Api error. Means we have a problem communication with the PowerView hub.""" - pass - class PvApiResponseStatusError(PvApiError): """Wrong http response error.""" -class PvApiConnectionError(PvApiError): - """Problem connecting to PowerView hub.""" +class PvApiMaintenance(PvApiError): + """Hub is undergoing maintenance.""" -async def check_response(response, valid_response_codes): - """Check the response for correctness.""" - if response.status in [204, 423]: - return True - if response.status in valid_response_codes: - _js = await response.json() - return _js - else: - raise PvApiResponseStatusError(response.status) +class PvApiConnectionError(PvApiError): + """Problem connecting to PowerView hub.""" class AioRequest: """Request class managing hub connection.""" - def __init__(self, hub_ip, loop=None, websession=None, timeout=15): + def __init__( + self, + hub_ip, + loop=None, + websession=None, + timeout: int = 15, + api_version: int | None = None, + ) -> None: self.hub_ip = hub_ip self._timeout = timeout if loop: @@ -49,6 +50,45 @@ def __init__(self, hub_ip, loop=None, websession=None, timeout=15): self.websession = websession else: self.websession = aiohttp.ClientSession() + self.api_version: int | None = api_version + self._last_request_status: int = 0 + _LOGGER.debug("Powerview api version: %s", self.api_version) + + @property + def api_path(self) -> str: + """Returns the initial api call path""" + if self.api_version and self.api_version >= 3: + return "home" + return "api" + + async def check_response(self, response, valid_response_codes): + """Check the response for correctness.""" + _val = None + if response.status == 403 and self._last_request_status == 423: + # if last status was hub undergoing maint then it is common + # on reboot for a 403 response. Generally this should raise + # PvApiResponseStatusError but as this is unavoidable we + # class this situation as still undergoing maintenance + _val = False + elif response.status in [204, 423]: + # 423 hub under maintenance, returns data, but not shade + _val = True + elif response.status in valid_response_codes: + _val = await response.json() + + # store the status for next check + self._last_request_status = response.status + + # raise a maintenance error + if isinstance(_val, bool): + raise PvApiMaintenance("Powerview Hub is undergoing maintenance") + + # if none of the above checks passed, raise a response error + if _val is None: + raise PvApiResponseStatusError(response.status) + + # finally, return the result + return _val async def get(self, url: str, params: str = None) -> dict: """ @@ -58,38 +98,43 @@ async def get(self, url: str, params: str = None) -> dict: :param params: :return: """ - _LOGGER.debug("Sending a get request") response = None try: - _LOGGER.debug("Sending GET request to: %s" % url) + _LOGGER.debug("Sending GET request to: %s params: %s", url, params) with async_timeout.timeout(self._timeout): response = await self.websession.get(url, params=params) - return await check_response(response, [200, 204]) + return await self.check_response(response, [200, 204]) except (asyncio.TimeoutError, aiohttp.ClientError) as error: raise PvApiConnectionError( - f"Failed to communicate with PowerView hub: {error}" - ) + "Failed to communicate with PowerView hub" + ) from error finally: if response is not None: await response.release() async def post(self, url: str, data: dict = None): + """ + Post a resource update. + + :param url: + :param data: a Dict. later converted to json. + :return: + """ response = None try: + _LOGGER.debug("Sending POST request to: %s data: %s", url, data) with async_timeout.timeout(self._timeout): - _LOGGER.debug("url: %s", url) - _LOGGER.debug("data: %s", data) response = await self.websession.post(url, json=data) - return await check_response(response, [200, 201]) + return await self.check_response(response, [200, 201]) except (asyncio.TimeoutError, aiohttp.ClientError) as error: raise PvApiConnectionError( - f"Failed to communicate with PowerView hub: {error}" - ) + "Failed to communicate with PowerView hub" + ) from error finally: if response is not None: await response.release() - async def put(self, url: str, data: dict = None): + async def put(self, url: str, data: dict = None, params=None): """ Do a put request. @@ -99,15 +144,19 @@ async def put(self, url: str, data: dict = None): """ response = None try: + _LOGGER.debug( + "Sending PUT request to: %s params: %s data: %s", + url, + params, + data, + ) with async_timeout.timeout(self._timeout): - _LOGGER.debug("url: %s", url) - _LOGGER.debug("data: %s", data) - response = await self.websession.put(url, json=data) - return await check_response(response, [200, 204]) + response = await self.websession.put(url, json=data, params=params) + return await self.check_response(response, [200, 204]) except (asyncio.TimeoutError, aiohttp.ClientError) as error: raise PvApiConnectionError( - f"Failed to communicate with PowerView hub: {error}" - ) + "Failed to communicate with PowerView hub" + ) from error finally: if response is not None: await response.release() @@ -124,13 +173,39 @@ async def delete(self, url: str, params: dict = None): """ response = None try: + _LOGGER.debug("Sending DELETE request to: %s with param %s", url, params) with async_timeout.timeout(self._timeout): response = await self.websession.delete(url, params=params) - return await check_response(response, [200, 204]) + return await self.check_response(response, [200, 204]) except (asyncio.TimeoutError, aiohttp.ClientError) as error: raise PvApiConnectionError( - f"Failed to communicate with PowerView hub: {error}" - ) + "Failed to communicate with PowerView hub" + ) from error finally: if response is not None: await response.release() + + async def set_api_version(self): + """ + Set the API generation based on what the gateway responds to. + """ + _LOGGER.debug("Attempting Gen 2 connection") + try: + await self.get(get_base_path(self.hub_ip, join_path("api", FWVERSION))) + self.api_version = 2 + _LOGGER.debug("Powerview api version changed to %s", self.api_version) + return + except Exception: # pylint: disable=broad-except + _LOGGER.debug("Gen 2 connection failed") + + _LOGGER.debug("Attempting Gen 3 connection") + try: + await self.get(get_base_path(self.hub_ip, join_path("gateway", "info"))) + self.api_version = 3 + _LOGGER.debug("Powerview api version changed to %s", self.api_version) + # TODO: what about dual hubs + return + except Exception as err: # pylint: disable=broad-except + _LOGGER.debug("Gen 3 connection failed %s", err) + + raise PvApiConnectionError("Failed to discover gateway version") diff --git a/aiopvapi/helpers/api_base.py b/aiopvapi/helpers/api_base.py index 8257502..68acd3d 100644 --- a/aiopvapi/helpers/api_base.py +++ b/aiopvapi/helpers/api_base.py @@ -1,9 +1,19 @@ +"""Class containing the api base.""" + import logging -from typing import List from aiopvapi.helpers.aiorequest import AioRequest -from aiopvapi.helpers.constants import ATTR_ID, ATTR_NAME_UNICODE, ATTR_NAME -from aiopvapi.helpers.tools import join_path, get_base_path, base64_to_unicode +from aiopvapi.helpers.constants import ( + ATTR_ID, + ATTR_NAME_UNICODE, + ATTR_NAME, + ATTR_PTNAME, +) +from aiopvapi.helpers.tools import ( + join_path, + get_base_path, + base64_to_unicode, +) _LOGGER = logging.getLogger(__name__) @@ -11,26 +21,59 @@ class ApiBase: """Api base class""" - def __init__(self, request: AioRequest, base_path): + api_endpoint = "" + + def __init__(self, request: AioRequest, api_endpoint: str = "") -> None: self.request = request - self._base_path = get_base_path(request.hub_ip, base_path) + self._api_endpoint = api_endpoint + self._raw_data = None + + @property + def api_version(self) -> int: + """Return the API version of the connected hub""" + return self.request.api_version + + @property + def api_path(self) -> str: + """Returns the initial api call path based on the api version""" + return self.request.api_path + + @property + def base_path(self) -> str: + """Returns the base path of the resource""" + return get_base_path( + self.request.hub_ip, join_path(self.api_path, self._api_endpoint) + ) + + @property + def url(self) -> str: + """Returns the url of the hub""" + return self.base_path + + def _parse(self, *keys, converter=None, data=None): + """Retrieve attributes from data dictionary""" + val = data if data else self._raw_data + try: + for key in keys: + val = val[key] + except KeyError as err: + _LOGGER.error(err) + return None + if converter: + return converter(val) + return val class ApiResource(ApiBase): """Represent a single PowerView resource, i.e. a scene, a shade or a room.""" - def __init__(self, request, api_endpoint, raw_data=None): + def __init__(self, request, api_endpoint, raw_data=None) -> None: super().__init__(request, api_endpoint) - self._id = "unknown" - if raw_data: - self._id = raw_data.get(ATTR_ID) + self._id = "unknown" if raw_data is None else raw_data.get(ATTR_ID) self._raw_data = raw_data - - self._resource_path = join_path(self._base_path, str(self._id)) - _LOGGER.debug( - "Initializing resource. resource path %s", self._resource_path - ) + self._resource_path = join_path(self.base_path, str(self._id)) + _LOGGER.debug("Initializing resource path: %s", self._resource_path) async def delete(self): """Deletes a resource.""" @@ -46,10 +89,18 @@ def name(self): """Name of the resource. If conversion to unicode somehow didn't go well value is returned in base64 encoding.""" return ( - self._raw_data.get(ATTR_NAME_UNICODE) + self._raw_data.get(ATTR_PTNAME) + or self._raw_data.get(ATTR_NAME_UNICODE) + or self._parse(ATTR_NAME, converter=base64_to_unicode, data=self._raw_data) or self._raw_data.get(ATTR_NAME) or "" ) + # resource[ATTR_NAME_UNICODE] = base64_to_unicode(_name) + + @property + def url(self) -> str: + """Return url for the shade.""" + return self._resource_path @property def raw_data(self): @@ -64,21 +115,24 @@ def raw_data(self, data): class ApiEntryPoint(ApiBase): """API entrypoint.""" - @classmethod - def _sanitize_resources(cls, resources): + def __init__(self, request, api_endpoint, use_initial=True) -> None: + super().__init__(request, api_endpoint) + if use_initial: + api_endpoint = join_path(self.api_path, api_endpoint) + + def _sanitize_resources(self, resources: dict): """Loops over incoming data looking for base64 encoded data and converts them to a readable format.""" try: - for resource in cls._loop_raw(resources): - cls._sanitize_resource(resource) + for resource in self._loop_raw(resources): + self._sanitize_resource(resource) except (KeyError, TypeError): - _LOGGER.debug("no shade data available") + _LOGGER.warning("No shade data available") return None @classmethod def _sanitize_resource(cls, resource): - _name = resource.get(ATTR_NAME) if _name: resource[ATTR_NAME_UNICODE] = base64_to_unicode(_name) @@ -88,7 +142,8 @@ async def get_resources(self, **kwargs) -> dict: :raises PvApiError when an error occurs. """ - resources = await self.request.get(self._base_path, **kwargs) + # resources = await self.request.get(self._base_path, **kwargs) + resources = await self.request.get(self.base_path, **kwargs) self._sanitize_resources(resources) return resources @@ -96,20 +151,18 @@ async def get_resource(self, resource_id: int) -> dict: """Get a single resource. :raises PvApiError when a hub connection occurs.""" - resource = await self.request.get( - join_path(self._base_path, str(resource_id)) - ) + resource = await self.request.get(join_path(self.base_path, str(resource_id))) + # resource = await self.request.get(join_path(self._base_path, str(resource_id))) self._sanitize_resource(self._get_to_actual_data(resource)) return resource - async def get_instances(self, **kwargs) -> List[ApiResource]: + async def get_instances(self, **kwargs) -> list[ApiResource]: """Returns a list of resource instances. :raises PvApiError when a hub problem occurs.""" raw_resources = await self.get_resources(**kwargs) _instances = [ - self._resource_factory(_raw) - for _raw in self._loop_raw(raw_resources) + self._resource_factory(_raw) for _raw in self._loop_raw(raw_resources) ] return _instances @@ -122,15 +175,13 @@ async def get_instance(self, resource_id) -> ApiResource: def _resource_factory(self, raw) -> ApiResource: """Converts raw data to a instantiated resource""" - raise NotImplemented + raise NotImplementedError - @staticmethod - def _loop_raw(raw): + def _loop_raw(self, raw): """Loops over raw data""" - raise NotImplemented + raise NotImplementedError - @staticmethod - def _get_to_actual_data(raw): + def _get_to_actual_data(self, raw): """incoming data is wrapped inside a key value pair for real unknown reasons making this a necessary call.""" - raise NotImplemented + raise NotImplementedError diff --git a/aiopvapi/helpers/constants.py b/aiopvapi/helpers/constants.py index db2a9d5..fa30181 100644 --- a/aiopvapi/helpers/constants.py +++ b/aiopvapi/helpers/constants.py @@ -1,42 +1,146 @@ +"""Constants for Hunter Douglas Powerview hub.""" + +# used for is_supported functions +MOTION_VELOCITY = "velocity" +MOTION_JOG = "jog" +MOTION_CALIBRATE = "calibrate" +MOTION_FAVORITE = "heart" +FUNCTION_SET_POWER = "set_power" +FUNCTION_SCHEDULE = "schedule" +FUNCTION_REBOOT = "reboot" +FUNCTION_IDENTIFY = "identify" + +# common across all API versions +HUB_NAME = "hubName" + +ATTR_SCENE = "scene" +ATTR_SHADE = "shade" +ATTR_ROOM = "room" +ATTR_SCENE_MEMBER = "sceneMember" + +ATTR_SHADE_DATA = "shadeData" +ATTR_SCENE_DATA = "sceneData" +SCENE_MEMBER_DATA = "sceneMemberData" +ATTR_ROOM_DATA = "roomData" + ATTR_ICON_ID = "iconId" ATTR_COLOR_ID = "colorId" ATTR_SCENE_ID = "sceneId" ATTR_SHADE_ID = "shadeId" ATTR_ROOM_ID = "roomId" +ATTR_ID = "id" +ATTR_TYPE = "type" ATTR_NAME = "name" ATTR_NAME_UNICODE = "name_unicode" +ATTR_SIGNAL_STRENGTH = "signalStrength" +ATTR_SIGNAL_STRENGTH_MAX = 4 + +FIRMWARE = "firmware" +FIRMWARE_NAME = "name" +FIRMWARE_REVISION = "revision" +FIRMWARE_SUB_REVISION = "subRevision" +FIRMWARE_BUILD = "build" +FIRMWARE_MAINPROCESSOR = "mainProcessor" + +MAC_ADDRESS = "macAddress" +SERIAL_NUMBER = "serialNumber" + +ATTR_CAPABILITIES = "capabilities" +ATTR_POSITIONS = "positions" + +MOTION_STOP = "stop" + +POWERTYPE_HARDWIRED = "Hardwired" +POWERTYPE_BATTERY = "Battery" +POWERTYPE_RECHARGABLE = "Rechargable" + +# using percentage based positions in aiopvapi v3 +MIN_POSITION = 0 +MID_POSITION = 50 +MAX_POSITION = 100 +CLOSED_POSITION = 0 + +# v2 +FWVERSION = "fwversion" +USER_DATA = "userData" + +MAX_POSITION_V2 = 65535 + +SHADE_BATTERY_STRENGTH = "batteryStrength" +SHADE_BATTERY_STRENGTH_MAX = 200 + +POSKIND_PRIMARY = 1 +POSKIND_SECONDARY = 2 +POSKIND_TILT = 3 + +ATTR_BATTERY_KIND = "batteryKind" +BATTERY_KIND_HARDWIRED = 1 +BATTERY_KIND_BATTERY = 2 +BATTERY_KIND_RECHARGABLE = 3 + ATTR_POSKIND1 = "posKind1" ATTR_POSKIND2 = "posKind2" ATTR_POSITION1 = "position1" ATTR_POSITION2 = "position2" -ATTR_POSITION = "position" -ATTR_COMMAND = "command" -ATTR_MOVE = "move" + +ATTR_SCHEDULED_EVENT = "scheduledEvent" +ATTR_SCHEDULED_EVENT_DATA = "scheduledEventData" + +ATTR_SHADE_IDS = "shadeIds" + +POSITIONS_V2 = ( + (ATTR_POSITION1, ATTR_POSKIND1), + (ATTR_POSITION2, ATTR_POSKIND2), +) + +POWERTYPE_MAP_V2 = { + POWERTYPE_HARDWIRED: 1, + POWERTYPE_BATTERY: 2, + POWERTYPE_RECHARGABLE: 3, +} + +# v3 +NETWORK_STATUS = "networkStatus" +CONFIG = "config" + +SHADE_BATTERY_STATUS = "batteryStatus" +SHADE_BATTERY_STATUS_MAX = 3 + +ATTR_PTNAME = "ptName" +ATTR_POWER_TYPE = "powerType" + +ATTR_ROOM_IDS = "roomIds" + +ATTR_PRIMARY = "primary" +ATTR_SECONDARY = "secondary" ATTR_TILT = "tilt" +ATTR_VELOCITY = "velocity" -ATTR_OPEN_POSITION = "open_position" -ATTR_CLOSE_POSITION = "close_position" -ATTR_ALLOWED_POSITIONS = "allowed_positions" +POSITIONS_V3 = ( + ATTR_PRIMARY, + ATTR_SECONDARY, + ATTR_TILT, + ATTR_VELOCITY, +) -ATTR_CAPABILITIES = "capabilities" -ATTR_POSITION_DATA = "positions" -ATTR_SCENE = "scene" -ATTR_SHADE = "shade" -ATTR_ROOM = "room" -ATTR_SCENE_MEMBER = "sceneMember" -ATTR_USER_DATA = "user" -ATTR_TYPE = "type" -ATTR_TYPES = "types" -ATTR_ID = "id" +POWERTYPE_MAP_V3 = { + POWERTYPE_BATTERY: 0, + POWERTYPE_HARDWIRED: 1, + POWERTYPE_RECHARGABLE: 2, +} -MIN_POSITION = 0 -MID_POSITION = 32767 -MAX_POSITION = 65535 +# Legacy (Gen1) where firmware needs to be hardcoded +DEFAULT_LEGACY_MAINPROCESSOR = { + FIRMWARE_REVISION: 0, + FIRMWARE_SUB_REVISION: 1, + FIRMWARE_BUILD: 0, + FIRMWARE_NAME: "PowerView Hub", +} -POSKIND_NONE = 0 -POSKIND_PRIMARY = 1 -POSKIND_SECONDARY = 2 -POSKIND_VANE = 3 -POSKIND_ERROR = 4 +HUB_MODEL_MAPPING = { + "PV_Gen3": "Powerview Generation 3", + "PV Hub2.0": "Powerview Generation 2", + "PowerView Hub": "Powerview Generation 1", +} diff --git a/aiopvapi/helpers/powerview_util.py b/aiopvapi/helpers/powerview_util.py index 47006fd..456657c 100644 --- a/aiopvapi/helpers/powerview_util.py +++ b/aiopvapi/helpers/powerview_util.py @@ -117,7 +117,7 @@ async def add_shade_to_scene(self, shade_id, scene_id, position=None): _shade = await self.get_shade(shade_id) position = await _shade.get_current_position() - await (SceneMembers(self.request)).create_scene_member( + await SceneMembers(self.request).create_scene_member( position, scene_id, shade_id ) diff --git a/aiopvapi/helpers/tools.py b/aiopvapi/helpers/tools.py index e44b3c7..78ccd49 100644 --- a/aiopvapi/helpers/tools.py +++ b/aiopvapi/helpers/tools.py @@ -1,3 +1,5 @@ +"""Tools for converting data from powerview hub""" + import base64 from aiopvapi.helpers.constants import ATTR_ID @@ -14,6 +16,7 @@ def base64_to_unicode(string): def get_base_path(ip_address, url): + """Convert url and ip to base path""" # Remove scheme if present ip_address = ip_address.split("://")[-1].strip("/") # clean up url (leading or trailing or multiple '/') diff --git a/aiopvapi/hub.py b/aiopvapi/hub.py index 1d029dd..13473f2 100644 --- a/aiopvapi/hub.py +++ b/aiopvapi/hub.py @@ -2,50 +2,38 @@ import logging from aiopvapi.helpers.api_base import ApiBase -from aiopvapi.helpers.tools import join_path, base64_to_unicode - -LOGGER = logging.getLogger(__name__) - - -class UserData(ApiBase): - api_path = "api/userdata" - - def __init__(self, request): - super().__init__(request, self.api_path) - self.hub_name = None - self.ip = None - self.ssid = None - self._raw = None - - def parse(self, raw): - """Convert raw incoming to class attributes.""" - self._raw = raw - self.hub_name = self._parse("userData", "hubName", converter=base64_to_unicode) - self.ip = self._parse("userData", "ip") - self.ssid = self._parse("userData", "ssid") - - def _parse(self, *keys, converter=None): - try: - val = self._raw - for key in keys: - val = val[key] - except KeyError as err: - LOGGER.error(err) - return None - if converter: - return converter(val) - return val - - async def update_user_data(self): - _raw = await self.request.get(self._base_path) - LOGGER.debug("Raw user data: {}".format(_raw)) - self.parse(_raw) +from aiopvapi.helpers.constants import ( + FUNCTION_IDENTIFY, + FUNCTION_REBOOT, + FWVERSION, + HUB_MODEL_MAPPING, + HUB_NAME, + CONFIG, + FIRMWARE, + FIRMWARE_MAINPROCESSOR, + FIRMWARE_NAME, + FIRMWARE_BUILD, + FIRMWARE_REVISION, + FIRMWARE_SUB_REVISION, + NETWORK_STATUS, + SERIAL_NUMBER, + MAC_ADDRESS, + DEFAULT_LEGACY_MAINPROCESSOR, + USER_DATA, +) +from aiopvapi.helpers.tools import ( + get_base_path, + join_path, + base64_to_unicode, +) + +_LOGGER = logging.getLogger(__name__) class Version: """PowerView versioning scheme class.""" - def __init__(self, build, revision, sub_revision, name=None): + def __init__(self, build, revision, sub_revision, name=None) -> None: self._build = build self._revision = revision self._sub_revision = sub_revision @@ -56,53 +44,189 @@ def __repr__(self): self._build, self._revision, self._sub_revision ) + @property + def name(self) -> str: + """Return the name of the device""" + return self._name + + @property + def sw_version(self) -> str | None: + """Version in Home Assistant friendly format""" + return f"{self._revision}.{self._sub_revision}.{self._build}" + def __eq__(self, other): return str(self) == str(other) class Hub(ApiBase): - api_path = "api" + """Powerview Hub Class""" + + def __init__(self, request) -> None: + super().__init__(request, self.api_endpoint) + self._main_processor_version: Version | None = None + self._radio_version: list[Version] | None = None + self.hub_name = None + self.ip = None + self.ssid = None + self.mac_address = None + self.serial_number = None + self.main_processor_info = None + + def is_supported(self, function: str) -> bool: + """Confirm availble features based on api version""" + if self.api_version >= 3: + return function in (FUNCTION_REBOOT, FUNCTION_IDENTIFY) + return False - def __init__(self, request): - super().__init__(request, self.api_path) - self._main_processor_version = None - self._radio_version = None - self.user_data = UserData(request) + @property + def role(self) -> str | None: + """Return the role of the hub in the current system""" + if self.api_version is None or self.api_version <= 2: + return "Primary" + + multi_gateway = self._parse(CONFIG, "mgwStatus", "running") + gateway_primary = self._parse(CONFIG, "mgwConfig", "primary") + + if multi_gateway is False or (multi_gateway and gateway_primary): + return "Primary" + return "Secondary" + + @property + def firmware(self) -> str | None: + """Return the current firmware version""" + return self.main_processor_version.sw_version + + @property + def model(self) -> str | None: + """Return the freindly name for the model of the hub""" + return HUB_MODEL_MAPPING.get( + self.main_processor_version.name, self.main_processor_version.name + ) @property - def main_processor_version(self): + def hub_address(self) -> str | None: + """Return the address of the hub""" + return self.ip + + @property + def main_processor_version(self) -> Version | None: + """Return the main processor version""" return self._main_processor_version @property - def radio_version(self): + def radio_version(self) -> Version | None: + """Return the radio version""" return self._radio_version @property - def name(self): - return self.user_data.hub_name + def name(self) -> str | None: + """The name of the device""" + return self.hub_name @property - def ip(self): - return self.user_data.ip + def url(self) -> str: + """Returns the url of the hub + + Used in Home Assistant as configuration url + """ + if self.api_version >= 3: + return self.base_path + return join_path(self.base_path, "shades") + + async def reboot(self) -> None: + """Reboot the hub""" + if not self.is_supported("reboot"): + _LOGGER.error("Method not supported") + return + + url = get_base_path(self.request.hub_ip, join_path("gateway", "reboot")) + await self.request.post(url) + + async def identify(self, interval: int = 10) -> None: + """Identify the hub""" + if not self.is_supported("identify"): + _LOGGER.error("Method not supported") + return + + url = get_base_path(self.request.hub_ip, join_path("gateway", "identify")) + await self.request.get(url, params={"time": interval}) async def query_firmware(self): - """Query the firmware versions.""" - - _version = await self.request.get(join_path(self._base_path, "/fwversion")) - _fw = _version.get("firmware") - if _fw: - _main = _fw.get("mainProcessor") - if _main: - self._main_processor_version = self._make_version(_main) - _radio = _fw.get("radio") - if _radio: - self._radio_version = self._make_version(_radio) + """ + Query the firmware versions. If API version is not set yet, get the API version first. + """ + if not self.api_version: + await self.request.set_api_version() + if self.api_version >= 3: + await self._query_firmware_g3() + else: + await self._query_firmware_g2() + _LOGGER.debug("Raw hub data: %s", self._raw_data) + + async def _query_firmware_g2(self): + # self._raw_data = await self.request.get(join_path(self._base_path, "userdata")) + self._raw_data = await self.request.get(join_path(self.base_path, "userdata")) + + _main = self._parse(USER_DATA, FIRMWARE, FIRMWARE_MAINPROCESSOR) + if not _main: + # do some checking for legacy v1 failures + _fw = await self.request.get(join_path(self.base_path, FWVERSION)) + # _fw = await self.request.get(join_path(self._base_path, FWVERSION)) + if FIRMWARE in _fw: + _main = self._parse(FIRMWARE, FIRMWARE_MAINPROCESSOR, data=_fw) + else: + _main = DEFAULT_LEGACY_MAINPROCESSOR + # if we are failing here lets drop api version down + # used in HA as v1 do not support stop + self.request.api_version = 1 + _LOGGER.debug("Powerview api version changed to %s", self.api_version) + + if _main: + self._main_processor_version = self._make_version(_main) + self.main_processor_info = _main + + _radio = self._parse(USER_DATA, FIRMWARE, "radio") + if _radio: + # gen 3 has multiple radios, this keeps hub consistent as a list + self._radio_version = [self._make_version(_radio)] + + self.ip = self._parse(USER_DATA, "ip") + self.ssid = self._parse(USER_DATA, "ssid") + self.mac_address = self._parse(USER_DATA, MAC_ADDRESS) + self.serial_number = self._parse(USER_DATA, SERIAL_NUMBER) + + self.hub_name = self._parse(USER_DATA, HUB_NAME, converter=base64_to_unicode) + + async def _query_firmware_g3(self): + gateway = get_base_path(self.request.hub_ip, "gateway") + self._raw_data = await self.request.get(gateway) + + _main = self._parse(CONFIG, FIRMWARE, FIRMWARE_MAINPROCESSOR) + if _main: + self._main_processor_version = self._make_version(_main) + self.main_processor_info = _main + + _radios = self._parse(CONFIG, FIRMWARE, "radios") + if _radios: + self._radio_version = [] + for _radio in _radios: + self._radio_version.append(self._make_version(_radio)) + + self.ip = self._parse(CONFIG, NETWORK_STATUS, "ipAddress") + self.ssid = self._parse(CONFIG, NETWORK_STATUS, "ssid") + self.mac_address = self._parse(CONFIG, NETWORK_STATUS, "primaryMacAddress") + self.serial_number = self._parse(CONFIG, SERIAL_NUMBER) + + self.hub_name = self.mac_address + if HUB_NAME not in self._parse(CONFIG): + # Get gateway name from home API until it is in the gateway API + home = await self.request.get(self.base_path) + self.hub_name = home["gateways"][0]["name"] def _make_version(self, data: dict): - version = Version( - data["build"], data["revision"], data["subRevision"], data.get("name") + return Version( + data[FIRMWARE_BUILD], + data[FIRMWARE_REVISION], + data[FIRMWARE_SUB_REVISION], + data.get(FIRMWARE_NAME), ) - return version - - async def query_user_data(self): - await self.user_data.update_user_data() diff --git a/aiopvapi/resources/automation.py b/aiopvapi/resources/automation.py new file mode 100644 index 0000000..fc538c9 --- /dev/null +++ b/aiopvapi/resources/automation.py @@ -0,0 +1,233 @@ +"""Scene class managing all scenes.""" + +from aiopvapi.helpers.aiorequest import AioRequest, PvApiMaintenance +from aiopvapi.helpers.api_base import ApiResource +from aiopvapi.helpers.tools import get_base_path, join_path +from aiopvapi.helpers.constants import ( + ATTR_SCENE_ID, + ATTR_SCHEDULED_EVENT, + ATTR_ID, + FUNCTION_SCHEDULE, +) +from aiopvapi.resources.scene import Scene + +import logging + +_LOGGER = logging.getLogger(__name__) + + +class Automation(ApiResource): + """Powerview Automation class.""" + + def __init__(self, raw_data: dict, request: AioRequest) -> None: + self.api_endpoint = "scheduledevents" + if request.api_version >= 3: + self.api_endpoint = "automations" + super().__init__(request, self.api_endpoint, raw_data) + self._name = None + self._room_id = None + self._scene: Scene = None + + def is_supported(self, function: str) -> bool: + """Return if api supports this function.""" + if self.api_version >= 3: + return False + else: + if function in FUNCTION_SCHEDULE: + return True + return False + + @property + def enabled(self) -> bool: + """Return the automation state.""" + return self._raw_data.get("enabled") + + @property + def id(self) -> int: + return self._raw_data.get(ATTR_ID) + + @property + def name(self) -> str: + if self._name is not None: + return self._name + return self._raw_data.get(ATTR_SCENE_ID) + + @property + def scene_id(self) -> str: + """Return the scene id of the automation.""" + return self._raw_data.get(ATTR_SCENE_ID) + + @property + def room_id(self) -> int | None: + """Return the room id of the automation.""" + return self._room_id + + def convert_to_12_hour(self, hour: int): + """Convert 24 hour time to 12 hour""" + if hour < 0 or hour > 24: + _LOGGER.error("%s is not a valid 24 hour time", hour) + return 0 + if hour == 0: + return 12 + if 0 < hour <= 12: + return hour + return hour - 12 + + def format_time(self, hour: int, minute: int): + """Convert hour and minute to friendly text format""" + meridiem = "AM" if hour < 12 else "PM" + hour = hour % 12 if hour % 12 != 0 else 12 + + if minute >= 60: + hour += minute // 60 + minute %= 60 + + hour = self.convert_to_12_hour(hour) + return f"{hour}:{str(abs(minute)).zfill(2)} {meridiem}" + + def get_execution_time(self): + """Return a friendly string, in the same format the hub + does incicating when the time the schedule will execute. + """ + # {'id': 437, 'type': 14, 'enabled': False, 'days': 127, 'hour': 1, 'min': 0, 'bleId': 2, 'sceneId': 220, 'errorShd_Ids': []} + # {'enabled': True, 'sceneId': 14067, 'daySunday': False, 'dayMonday': True, 'dayTuesday': True, 'dayWednesday': True, 'dayThursday': True, 'dayFriday': True, 'daySaturday': False, 'eventType': 0, 'hour': 7, 'minute': 0, 'id': 38971} + + if self.api_version >= 3: + # 2 = Before sunrise, 10 = After sunrise + # 6 = Before sunset, 14 = After sunset + sunrise = [2, 10] + valid_events = [2, 6, 10, 14] + else: + # - Sunrise = 1, Sunset = 2 + # before and after are caluclated by hour/minute + sunrise = [1] + valid_events = [1, 2] + + attr_type = "type" if self.api_version >= 3 else "eventType" + attr_hour = "hour" + attr_minute = "min" if self.api_version >= 3 else "minute" + + event_type = self.raw_data.get(attr_type) + hour = self.raw_data.get(attr_hour) + minute = self.raw_data.get(attr_minute) + + # event type 0 represents clock based for all generations + if event_type == 0: + return self.format_time(hour, minute) + + if event_type in valid_events: + when = "Sunrise" if event_type in sunrise else "Sunset" + + if hour == 0 and minute == 0: + return f"At {when}" + + if self.api_version >= 3: + before_after = "Before" if event_type in [2, 6] else "After" + else: + before_after = "Before" if minute < 0 else "After" + hour = abs(minute) // 60 + minute = abs(minute) % 60 + # hour = floor(abs(minute) / 60) + # minute = abs(minute) - (hour * 60) + return f"{hour}h {minute}m {before_after} {when}" + # return f"{abs(hour)}h {abs(minute)}m {before_after} {when}" + + return f"Unknown Event {event_type}" + + def get_execution_days(self): + """Return a friendly string, in the same format the hub + does incicating when the days the schedule will execute. + """ + + if self.api_version >= 3: + day_mapping = { + 0x40: "Sun", + 0x01: "Mon", + 0x02: "Tue", + 0x04: "Wed", + 0x08: "Thu", + 0x10: "Fri", + 0x20: "Sat", + } + + enabled_days = [ + day for bit, day in day_mapping.items() if self.raw_data["days"] & bit + ] + + else: + day_mapping = { + "daySunday": "Sun", + "dayMonday": "Mon", + "dayTuesday": "Tue", + "dayWednesday": "Wed", + "dayThursday": "Thu", + "dayFriday": "Fri", + "daySaturday": "Sat", + } + + enabled_days = [ + day_mapping[key] + for key, value in self.raw_data.items() + if key in day_mapping and value + ] + + if len(enabled_days) == len(day_mapping): + output = "Every Day" + elif set(enabled_days) == {"Sat", "Sun"}: + output = "Weekends" + elif set(enabled_days) == {"Mon", "Tue", "Wed", "Thu", "Fri"}: + output = "Weekdays" + else: + output = ", ".join(enabled_days) + return output + + @property + def details(self) -> dict[str, str]: + """Return the specifics of the automation.""" + details = { + "ID": self.id, + "Time": self.get_execution_time(), + "Days": self.get_execution_days(), + } + + _LOGGER.debug( + "Automation: %s (Enabled: %s), %s, %s", + self.name, + self.enabled, + details.get("Time"), + details.get("Days"), + ) + return details + + async def fetch_associated_scene_data(self) -> None: + """Update the automation with friendly scene info.""" + scene_url = join_path( + get_base_path(self.request.hub_ip, self.api_path), + "scenes", + str(self.scene_id), + ) + self._scene: Scene = Scene( + await self.request.get(scene_url), + self.request, + ) + self._name = self._scene.name + self._room_id = self._scene.room_id + + async def set_state(self, state: bool) -> None: + """Update the automation enabled status.""" + resource_path = join_path(self.base_path, str(self.id)) + data = self.raw_data + data["enabled"] = state + if self.api_version <= 2: + data = {"scheduledEvent": data} + await self.request.put(resource_path, data) + + async def refresh(self): + """Query the hub and for updated automation information.""" + try: + raw_data = await self.request.get(self._resource_path) + # Gen <= 2 API has raw data under shade key. Gen >= 3 API this is flattened. + self._raw_data = raw_data.get(ATTR_SCHEDULED_EVENT, raw_data) + except PvApiMaintenance: + _LOGGER.debug("Hub undergoing maintenance. Please try again") + return diff --git a/aiopvapi/resources/model.py b/aiopvapi/resources/model.py new file mode 100644 index 0000000..0fa76f8 --- /dev/null +++ b/aiopvapi/resources/model.py @@ -0,0 +1,24 @@ +"""Powerview data models""" + +from dataclasses import dataclass +from collections.abc import Iterable +from typing import Any +from aiopvapi.resources.shade import BaseShade +from aiopvapi.hub import Hub +from aiopvapi.resources.scene import Scene +from aiopvapi.resources.automation import Automation +from aiopvapi.resources.room import Room + + +@dataclass +class PowerviewData: + """ + Powerview data in raw and processed form + + :raw - raw json from the hub + + :processed - Class Object grouped by id + """ + + raw: Iterable[dict[str | int, Any]] + processed: dict[str, BaseShade | Hub | Automation | Scene | Room] diff --git a/aiopvapi/resources/room.py b/aiopvapi/resources/room.py index 473adac..79839b5 100644 --- a/aiopvapi/resources/room.py +++ b/aiopvapi/resources/room.py @@ -8,9 +8,11 @@ class Room(ApiResource): - api_path = "api/rooms" + """Powerview Rooms""" - def __init__(self, raw_data: dict, request: AioRequest): + api_endpoint = "rooms" + + def __init__(self, raw_data: dict, request: AioRequest) -> None: if ATTR_ROOM in raw_data: raw_data = raw_data.get(ATTR_ROOM) - super().__init__(request, self.api_path, raw_data) + super().__init__(request, self.api_endpoint, raw_data) diff --git a/aiopvapi/resources/scene.py b/aiopvapi/resources/scene.py index 6f5ac34..6f341da 100644 --- a/aiopvapi/resources/scene.py +++ b/aiopvapi/resources/scene.py @@ -1,22 +1,48 @@ +"""Scene class managing all scenes.""" + from aiopvapi.helpers.aiorequest import AioRequest from aiopvapi.helpers.api_base import ApiResource -from aiopvapi.helpers.constants import ATTR_SCENE, ATTR_ROOM_ID, ATTR_SCENE_ID +from aiopvapi.helpers.tools import join_path +from aiopvapi.helpers.constants import ( + ATTR_SCENE, + ATTR_ROOM_ID, + ATTR_ROOM_IDS, + ATTR_SCENE_ID, + ATTR_SHADE_IDS, +) + +import logging + +_LOGGER = logging.getLogger(__name__) class Scene(ApiResource): - api_path = "api/scenes" + """Powerview Scene class.""" + + api_endpoint = "scenes" - def __init__(self, raw_data: dict, request: AioRequest): + def __init__(self, raw_data: dict, request: AioRequest) -> None: if ATTR_SCENE in raw_data: raw_data = raw_data.get(ATTR_SCENE) - super().__init__(request, self.api_path, raw_data) + super().__init__(request, self.api_endpoint, raw_data) @property def room_id(self): """Return the room id.""" + if self.api_version >= 3: + return self._raw_data.get(ATTR_ROOM_IDS)[0] return self._raw_data.get(ATTR_ROOM_ID) - async def activate(self): + async def activate(self) -> list[int]: """Activate this scene.""" - _val = await self.request.get(self._base_path, params={ATTR_SCENE_ID: self._id}) + if self.request.api_version >= 3: + resource_path = join_path(self.base_path, str(self.id), "activate") + _val = await self.request.put(resource_path) + else: + _val = await self.request.get( + self.base_path, params={ATTR_SCENE_ID: self._id} + ) + # v2 returns format {'sceneIds': ids} so flattening the list to align v3 + _val = _val.get(ATTR_SHADE_IDS) + # should return an array of ID's that belong to the scene return _val diff --git a/aiopvapi/resources/scene_member.py b/aiopvapi/resources/scene_member.py index a05e77d..fbcf9e3 100644 --- a/aiopvapi/resources/scene_member.py +++ b/aiopvapi/resources/scene_member.py @@ -1,30 +1,38 @@ +"""Class for managing scene members.""" + from aiopvapi.helpers.aiorequest import AioRequest from aiopvapi.helpers.api_base import ApiResource -from aiopvapi.helpers.constants import ATTR_SCENE_MEMBER, ATTR_SCENE_ID, ATTR_SHADE_ID +from aiopvapi.helpers.constants import ( + ATTR_SCENE_MEMBER, + ATTR_SCENE_ID, + ATTR_SHADE_ID, +) class SceneMember(ApiResource): """Shades belonging to a scene.""" - api_path = "api/scenemembers" + api_endpoint = "scenemembers" - def __init__(self, raw_data: dict, request: AioRequest): + def __init__(self, raw_data: dict, request: AioRequest) -> None: if ATTR_SCENE_MEMBER in raw_data: raw_data = raw_data.get(ATTR_SCENE_MEMBER) - super().__init__(request, self.api_path, raw_data) + super().__init__(request, self.api_endpoint, raw_data) @property - def scene_id(self): + def scene_id(self) -> str: + """Return scene id of the scene""" return self._raw_data.get(ATTR_SCENE_ID) @property - def shade_id(self): + def shade_id(self) -> str: + """Return shade id of scene members""" return self._raw_data.get(ATTR_SHADE_ID) async def delete(self): """Deletes a scene from a shade""" _val = await self.request.delete( - self._base_path, + self.base_path, params={ ATTR_SCENE_ID: self._raw_data.get(ATTR_SCENE_ID), ATTR_SHADE_ID: self._raw_data.get(ATTR_SHADE_ID), diff --git a/aiopvapi/resources/shade.py b/aiopvapi/resources/shade.py index e387de2..442d318 100644 --- a/aiopvapi/resources/shade.py +++ b/aiopvapi/resources/shade.py @@ -1,13 +1,15 @@ -from audioop import avg +"""Shade class managing all shade types.""" + import logging -from collections import namedtuple from dataclasses import dataclass +from typing import Any -from aiopvapi.helpers.aiorequest import AioRequest +from aiopvapi.helpers.aiorequest import AioRequest, PvApiMaintenance from aiopvapi.helpers.api_base import ApiResource +from aiopvapi.helpers.tools import join_path from aiopvapi.helpers.constants import ( ATTR_CAPABILITIES, - ATTR_POSITION_DATA, + ATTR_POSITIONS, ATTR_SHADE, ATTR_TYPE, ATTR_ID, @@ -16,41 +18,65 @@ ATTR_POSITION1, ATTR_POSITION2, ATTR_POSKIND2, - ATTR_POSITION, - ATTR_COMMAND, - ATTR_MOVE, + ATTR_PRIMARY, + ATTR_SECONDARY, ATTR_TILT, + ATTR_BATTERY_KIND, + ATTR_POWER_TYPE, + FIRMWARE, + FIRMWARE_REVISION, + FIRMWARE_SUB_REVISION, + FIRMWARE_BUILD, MAX_POSITION, MID_POSITION, MIN_POSITION, + MAX_POSITION_V2, + MOTION_STOP, POSKIND_PRIMARY, POSKIND_SECONDARY, - POSKIND_VANE, + POSKIND_TILT, + ATTR_SIGNAL_STRENGTH, + ATTR_SIGNAL_STRENGTH_MAX, + POWERTYPE_BATTERY, + POWERTYPE_HARDWIRED, + POWERTYPE_MAP_V2, + POWERTYPE_MAP_V3, + POWERTYPE_RECHARGABLE, + SHADE_BATTERY_STATUS, + SHADE_BATTERY_STRENGTH, + POSITIONS_V2, + POSITIONS_V3, + BATTERY_KIND_HARDWIRED, + MOTION_VELOCITY, + MOTION_JOG, + MOTION_CALIBRATE, + MOTION_FAVORITE, + FUNCTION_SET_POWER, ) _LOGGER = logging.getLogger(__name__) @dataclass -class ShadeCapabilities: - """Represents the capabilities available for shade.""" +class PowerviewCapabilities: + """Capabilities available from Powerview.""" primary: bool = False secondary: bool = False - tilt90: bool = False - tilt180: bool = False - tiltOnClosed: bool = False - tiltAnywhere: bool = False - tiltOnSecondaryClosed: bool = False - primaryInverted: bool = False - secondaryInverted: bool = False - secondaryOverlapped: bool = False + tilt_90: bool = False + tilt_180: bool = False + tilt_onclosed: bool = False + tilt_anywhere: bool = False + tilt_onsecondaryclosed: bool = False + primary_inverted: bool = False + secondary_inverted: bool = False + secondary_overlapped: bool = False vertical: bool = False @dataclass class ShadeLimits: - """Represents the limits of a shade.""" + """Limits of a shade.""" primary_min: int = MIN_POSITION primary_max: int = MAX_POSITION @@ -60,179 +86,472 @@ class ShadeLimits: tilt_max: int = MAX_POSITION -shade_type = namedtuple("shade_type", ["shade_type", "description"]) -capability = namedtuple("capability", ["type", "capabilities", "description"]) +@dataclass +class ShadePosition: + """Positions for a powerview shade.""" + primary: int | float | None = None + secondary: int | float | None = None + tilt: int | float | None = None + velocity: float | None = None # float only a v3 only property -def factory(raw_data, request): - """Class factory to create different types of shades - depending on shade type.""" - if ATTR_SHADE in raw_data: - raw_data = raw_data.get(ATTR_SHADE) +@dataclass +class ShadeType: + """Shade information based on type and description""" - shade_type = raw_data.get(ATTR_TYPE) + type: int | str + description: str - def find_type(shade): - for tp in shade.shade_types: - if tp.shade_type == shade_type: - return shade(raw_data, tp, request) - return None - shade_capability = raw_data.get(ATTR_CAPABILITIES) +@dataclass +class ShadeCapability: + """Shade capability information""" - def find_capability(shade): - if shade.capability.type == shade_capability: - return shade(raw_data, shade, request) - return None + type: int | str + capabilities: PowerviewCapabilities + description: str - classes = [ - ShadeBottomUp, - ShadeBottomUpTiltOnClosed90, - ShadeBottomUpTiltOnClosed180, #to ensure capability match order here is important - ShadeBottomUpTiltAnywhere, - ShadeVerticalTiltAnywhere, - ShadeVertical, - ShadeTiltOnly, - ShadeTopDown, - ShadeTopDownBottomUp, - ShadeDualOverlapped, - ShadeDualOverlappedTilt90, - ShadeDualOverlappedTilt180, - ] - for cls in classes: - # class check is more concise as we have tested positioning - _shade = find_type(cls) - if _shade: - _LOGGER.debug("Match type : %s - %s", _shade, raw_data) - return _shade +class BaseShade(ApiResource): + """Basic shade class.""" - for cls in classes: - # fallback to a capability check - this should future proof new shades - # type 0 that contain tilt would not be caught here - _shade = find_capability(cls) - if _shade: - _LOGGER.debug("Match capability : %s - %s", _shade, raw_data) - return _shade + api_endpoint = "shades" - _LOGGER.debug("Shade unmatched : %s - %s", BaseShade, raw_data) - return BaseShade(raw_data, BaseShade.shade_types[0], request) + shade_types: tuple[ShadeType] = (ShadeType(0, "undefined type"),) + capability: ShadeCapability = ShadeCapability( + "-1", PowerviewCapabilities(primary=True), "undefined" + ) + _open_position: ShadePosition = ShadePosition(primary=MAX_POSITION) + _close_position: ShadePosition = ShadePosition(primary=MIN_POSITION) + _open_position_tilt: ShadePosition = ShadePosition() + _close_position_tilt: ShadePosition = ShadePosition() + shade_limits: ShadeLimits = ShadeLimits() -class BaseShade(ApiResource): - api_path = "api/shades" - shade_types = (shade_type(0, "undefined type"),) - capability = capability("-1", ShadeCapabilities(primary=True), "undefined") - open_position = {ATTR_POSITION1: MAX_POSITION, ATTR_POSKIND1: 1} - close_position = {ATTR_POSITION1: MIN_POSITION, ATTR_POSKIND1: 1} - open_position_tilt = {} - close_position_tilt = {} - allowed_positions = () - - shade_limits = ShadeLimits() - - def __init__(self, raw_data: dict, shade_type: shade_type, request: AioRequest): + def __init__( + self, raw_data: dict, shade_type: ShadeType, request: AioRequest + ) -> None: self.shade_type = shade_type - super().__init__(request, self.api_path, raw_data) + super().__init__(request, self.api_endpoint, raw_data=raw_data) + + def is_supported(self, function: str) -> bool: + """Return if api supports this function.""" + if self.api_version >= 3: + return function in (MOTION_JOG, MOTION_VELOCITY, MOTION_STOP) + elif self.api_version == 2: + return function in ( + MOTION_JOG, + MOTION_CALIBRATE, + MOTION_FAVORITE, + MOTION_STOP, + FUNCTION_SET_POWER, + ) + else: + return function in ( + MOTION_JOG, + MOTION_CALIBRATE, + MOTION_FAVORITE, + FUNCTION_SET_POWER, + ) + + @property + def current_position(self) -> ShadePosition: + """Return the current position of the shade as a percentage.""" + position = self.raw_to_structured(self._raw_data) + position = self.get_additional_positions(position) + return position + + @property + def room_id(self) -> int: + """Return the room id of the shade.""" + return self._raw_data.get(ATTR_ROOM_ID) + + @property + def type_id(self) -> int: + """Return the type id of the shade.""" + return self._raw_data.get(ATTR_TYPE) + + @property + def type_name(self) -> str: + """Return the type name of the shade.""" + for shade in self.shade_types: + if shade.type == self.type_id: + return shade.description + return self.type_id + + @property + def firmware(self) -> str | None: + """Return firmware string for the shade.""" + if FIRMWARE not in self.raw_data: + return None + firmware = self.raw_data[FIRMWARE] + return f"{firmware[FIRMWARE_REVISION]}.{firmware[FIRMWARE_SUB_REVISION]}.{firmware[FIRMWARE_BUILD]}" + + @property + def url(self) -> str: + """Return url for the shade.""" + return self._resource_path + + @property + def open_position(self) -> ShadePosition: + """Return the shade opened position""" + return self._open_position + + @property + def close_position(self) -> ShadePosition: + """Return the shade closed position""" + return self._close_position + + @property + def open_position_tilt(self) -> ShadePosition: + """Return the tilt opened position""" + return self._open_position_tilt + + @property + def close_position_tilt(self) -> ShadePosition: + """Return the tilt closed position""" + return self._close_position_tilt + + def percent_to_api(self, position: float, position_type: str) -> int | float: + """Convert percentage based position to hunter douglas api position.""" + # get the possible maximum for the shade (some shades only allow 50% position) + max_position_pct_mapping = { + ATTR_PRIMARY: self.shade_limits.primary_max, + ATTR_SECONDARY: self.shade_limits.secondary_max, + ATTR_TILT: self.shade_limits.tilt_max, + } + + max_position_pct = max_position_pct_mapping.get(position_type, 100) + + # ensure the position remains in range 0-100 + position = self.position_limit(position, position_type) + + # gen 3 takes 0.0 -> 1.0 (fractional perentage) - float + if self.api_version >= 3: + max_position_pct = max_position_pct / 100 + return round(position / 100 * max_position_pct, 2) + + # gen 2 requires conversion to 0-65335 - int + max_position_pct = max_position_pct / 100 * MAX_POSITION_V2 + return int(position / 100 * max_position_pct) + + def api_to_percent(self, position: float, position_type: str) -> int: + """Convert hunter douglas api based position to percentage based position.""" + # get the possible maximum for the shade (some shades only allow 50% position) + max_position_pct_mapping = { + ATTR_PRIMARY: self.shade_limits.primary_max, + ATTR_SECONDARY: self.shade_limits.secondary_max, + ATTR_TILT: self.shade_limits.tilt_max, + } + + max_position_pct = max_position_pct_mapping.get(position_type, 100) + + # convert percentage based version of max positioning to api per version + max_position_api = max_position_pct / 100 + if self.api_version < 3: + max_position_api = MAX_POSITION_V2 * max_position_api + + percent = self.position_limit(round((position / max_position_api) * 100)) + return percent + + def structured_to_raw(self, data: ShadePosition) -> dict[str, Any]: + """Convert structured ShadePosition to API relevant dict""" + _LOGGER.debug("Structured Data %s: %s", self.name, data) + + if self.api_version >= 3: + # Gen 3 raw data creation + raw = {ATTR_POSITIONS: {}} + for position_type in POSITIONS_V3: + if getattr(data, position_type) is not None: + raw[ATTR_POSITIONS][position_type] = self.percent_to_api( + getattr(data, position_type), position_type + ) + + else: + # Gen 2 raw data creation + position_data = {} + if data.primary is not None: + # primary is always in position 1 + position_data[ATTR_POSKIND1] = POSKIND_PRIMARY + position_data[ATTR_POSITION1] = self.percent_to_api( + data.primary, ATTR_PRIMARY + ) + if data.secondary is not None: + poskind = ATTR_POSKIND2 + position = ATTR_POSITION2 + if data.primary is None: + # if no primary, secondary should be in position 1 (its a legacy thing) + poskind = ATTR_POSKIND1 + position = ATTR_POSITION1 + position_data[poskind] = POSKIND_SECONDARY + position_data[position] = self.percent_to_api( + data.secondary, ATTR_SECONDARY + ) + if data.tilt is not None: + if data.primary is not None and data.secondary is not None: + # if both primary and secondary exist than tilt cannot be sent + _LOGGER.debug( + "Legacy only accepts 2 positions. Tilt ignored %s", data + ) + elif data.primary is not None or data.secondary is not None: + # if primary or secondary exist move tilt to position 2 (its a legacy thing) + position_data[ATTR_POSKIND2] = POSKIND_TILT + position_data[ATTR_POSITION2] = self.percent_to_api( + data.tilt, ATTR_TILT + ) + else: + position_data[ATTR_POSKIND1] = POSKIND_TILT + position_data[ATTR_POSITION1] = self.percent_to_api( + data.tilt, ATTR_TILT + ) + + raw = {ATTR_SHADE: {ATTR_ID: self.id, ATTR_POSITIONS: position_data}} + + _LOGGER.debug("Raw Conversion %s: %s", self.name, raw) + return raw + + def raw_to_structured(self, shade_data: dict[int | str, Any]) -> ShadePosition: + """Convert API dict info to structured ShadePosition dataclass""" + _LOGGER.debug("Raw Data %s: %s", self.name, shade_data) + + if ATTR_POSITIONS not in shade_data: + return ShadePosition() + + position_data = shade_data[ATTR_POSITIONS] + + position = ShadePosition() + if self.api_version >= 3: + for position_key in POSITIONS_V3: + if position_key in position_data: + setattr( + position, + position_key, + self.api_to_percent(position_data[position_key], position_key), + ) + + else: + position_mapping = { + POSKIND_PRIMARY: ATTR_PRIMARY, + POSKIND_SECONDARY: ATTR_SECONDARY, + POSKIND_TILT: ATTR_TILT, + } + + for position_key, poskind_key in POSITIONS_V2: + if poskind_key in position_data: + target_key = position_mapping.get(position_data[poskind_key]) + setattr( + position, + target_key, + self.api_to_percent(position_data[position_key], target_key), + ) + + _LOGGER.debug("Structured Conversion %s: %s", self.name, position) + return position def _create_shade_data(self, position_data=None, room_id=None): """Create a shade data object to be sent to the hub""" + if self.api_version >= 3: + return {"positions": position_data} + base = {ATTR_SHADE: {ATTR_ID: self.id}} if position_data: - base[ATTR_SHADE][ATTR_POSITION_DATA] = self.clamp(position_data) + base[ATTR_SHADE][ATTR_POSITIONS] = position_data if room_id: base[ATTR_SHADE][ATTR_ROOM_ID] = room_id return base - async def _move(self, position_data): - result = await self.request.put(self._resource_path, data=position_data) - return result - - async def move(self, position_data): + async def move_raw(self, position_data: dict): + """Move the shade to a set position using raw data""" + _LOGGER.debug("Shade %s move to: %s", self.name, position_data) data = self._create_shade_data(position_data=position_data) return await self._move(data) + async def _move(self, position_data: dict): + params = {} + resource_path = self._resource_path + if self.api_version >= 3: + # IDs are required in request params for gen 3. + params = {"ids": self.id} + resource_path = join_path(self.base_path, "positions") + result = await self.request.put( + resource_path, data=position_data, params=params + ) + return result + + async def move(self, position_data: ShadePosition) -> ShadePosition: + """Move the shade to a set position""" + _LOGGER.debug("Shade %s move to: %s", self.name, position_data) + data = self.structured_to_raw(position_data) + await self._move(data) + return self.current_position + + def get_additional_positions(self, positions: ShadePosition) -> ShadePosition: + """Returns additonal positions not reported by the hub""" + return positions + async def open(self): + """Open the shade""" return await self.move(position_data=self.open_position) async def close(self): + """Close the shade""" return await self.move(position_data=self.close_position) - def position_limit(self, value, poskind): - if poskind == POSKIND_PRIMARY: - min = self.shade_limits.primary_min - max = self.shade_limits.primary_max - elif poskind == POSKIND_SECONDARY: - min = self.shade_limits.secondary_min - max = self.shade_limits.secondary_max - elif poskind == POSKIND_VANE: - min = self.shade_limits.tilt_min - max = self.shade_limits.tilt_max - if min <= value <= max: - return value - if value < min: - return min + def position_limit(self, position: int, position_type: str = ""): + """Limit values that can be calculated.""" + # determine the absolute position for the particular shade + limits = { + ATTR_PRIMARY: ( + self.shade_limits.primary_min, + self.shade_limits.primary_max, + ), + ATTR_SECONDARY: ( + self.shade_limits.secondary_min, + self.shade_limits.secondary_max, + ), + ATTR_TILT: (self.shade_limits.tilt_min, self.shade_limits.tilt_max), + } + + min_limit, max_limit = limits.get(position_type, (0, 100)) + + return min(max(min_limit, position), max_limit) + + async def _motion(self, motion): + if self.api_version >= 3: + path = join_path(self._resource_path, "motion") + cmd = {"motion": motion} else: - return max - - def clamp(self, position_data): - """Prevent impossible positions being sent.""" - if (position1 := position_data.get(ATTR_POSITION1)) is not None: - position_data[ATTR_POSITION1] = self.position_limit( - position1, position_data[ATTR_POSKIND1] - ) - if (position2 := position_data.get(ATTR_POSITION2)) is not None: - position_data[ATTR_POSITION2] = self.position_limit( - position2, position_data[ATTR_POSKIND2] - ) - return position_data + path = self._resource_path + cmd = {"shade": {"motion": motion}} + await self.request.put(path, cmd) async def jog(self): """Jog the shade.""" - await self.request.put(self._resource_path, {"shade": {"motion": "jog"}}) + await self._motion(MOTION_JOG) async def calibrate(self): """Calibrate the shade.""" - await self.request.put(self._resource_path, {"shade": {"motion": "calibrate"}}) + await self._motion(MOTION_CALIBRATE) async def favorite(self): """Move the shade to the defined favorite position.""" - await self.request.put(self._resource_path, {"shade": {"motion": "heart"}}) + await self._motion(MOTION_FAVORITE) async def stop(self): """Stop the shade.""" - return await self.request.put(self._resource_path, {"shade": {"motion": "stop"}}) + if not self.is_supported(MOTION_STOP): + _LOGGER.error("Method not supported") + return + + if self.api_version >= 3: + await self.request.put( + join_path(self.base_path, MOTION_STOP), params={"ids": self.id} + ) + else: + await self._motion(MOTION_STOP) async def add_shade_to_room(self, room_id): + """Add shade to room.""" data = self._create_shade_data(room_id=room_id) return await self.request.put(self._resource_path, data) async def refresh(self): """Query the hub and the actual shade to get the most recent shade data. Including current shade position.""" - raw_data = await self.request.get(self._resource_path, {"refresh": "true"}) - if isinstance(raw_data, bool): - _LOGGER.debug("No data available, hub undergoing maintenance. Please try again") - return - self._raw_data = raw_data.get(ATTR_SHADE) + try: + _LOGGER.debug("Refreshing position of: %s", self.name) + raw_data = await self.request.get(self._resource_path, {"refresh": "true"}) + # Gen <= 2 API has raw data under shade key. Gen >= 3 API this is flattened. + self._raw_data = raw_data.get(ATTR_SHADE, raw_data) + except PvApiMaintenance: + _LOGGER.debug("Hub undergoing maintenance. Please try again") + return async def refresh_battery(self): """Query the hub and request the most recent battery state.""" - raw_data = await self.request.get(self._resource_path, {"updateBatteryLevel": "true"}) - if isinstance(raw_data, bool): - _LOGGER.debug("No data available, hub undergoing maintenance. Please try again") - return - self._raw_data = raw_data.get(ATTR_SHADE) + try: + raw_data = await self.request.get( + self._resource_path, {"updateBatteryLevel": "true"} + ) + # Gen <= 2 API has raw data under shade key. Gen >= 3 API this is flattened. + self._raw_data = raw_data.get(ATTR_SHADE, raw_data) + except PvApiMaintenance: + _LOGGER.debug("Hub undergoing maintenance. Please try again") + return - async def set_power_source(self, type): + def has_battery_info(self) -> bool: + """Confirm if the shade has battery info.""" + if self.api_version >= 3: + return bool(SHADE_BATTERY_STATUS in self.raw_data) + return bool(SHADE_BATTERY_STRENGTH in self.raw_data) + + def is_battery_powered(self) -> bool: + """Confirm if the shade is battery or hardwired.""" + attr = ATTR_POWER_TYPE if self.api_version >= 3 else ATTR_BATTERY_KIND + return bool(self.raw_data.get(attr) != BATTERY_KIND_HARDWIRED) + + def supported_power_sources(self) -> list[str]: + """List supported power sources.""" + return [POWERTYPE_HARDWIRED, POWERTYPE_BATTERY, POWERTYPE_RECHARGABLE] + + def get_power_source(self) -> str: + """Get from the hub the type of power source.""" + version_map = POWERTYPE_MAP_V3 if self.api_version >= 3 else POWERTYPE_MAP_V2 + attr = ATTR_POWER_TYPE if self.api_version >= 3 else ATTR_BATTERY_KIND + powertype_map = {v: k for k, v in version_map.items()} + + raw_num = self.raw_data.get(attr) + battery_type = powertype_map.get(raw_num, None) + _LOGGER.debug("Mapping power source %s to %s", raw_num, battery_type) + return battery_type + + async def set_power_source(self, power_source): """Update the hub with the type of power source.""" - if type not in [1, 2, 3]: - _LOGGER.error("Unsupported Power Type. Accepted values are 1, 2 & 3") + if not self.is_supported(FUNCTION_SET_POWER): + _LOGGER.error("Method not supported") return - await self.request.put(self._resource_path, data={"shade": {"batteryKind": type}}) - async def get_current_position(self, refresh=True) -> dict: + if power_source not in (supported := self.supported_power_sources()): + _LOGGER.error("Unsupported Power Source. Accepted values: %s", supported) + return + + version_map = POWERTYPE_MAP_V3 if self.api_version >= 3 else POWERTYPE_MAP_V2 + attr = ATTR_POWER_TYPE if self.api_version >= 3 else ATTR_BATTERY_KIND + await self.request.put( + self._resource_path, + data={"shade": {attr: version_map.get(power_source)}}, + ) + + def get_battery_strength(self) -> int: + """Get battery strength from raw_data and return as a percentage.""" + power_levels = { + 4: 100, # 4 is hardwired + 3: 100, # 3 = 100% to 51% power remaining + 2: 50, # 2 = 50% to 21% power remaining + 1: 20, # 1 = 20% or less power remaining + 0: 0, # 0 = No power remaining + } + battery_status = self.raw_data[SHADE_BATTERY_STATUS] + return power_levels.get(battery_status, 0) + + def has_signal_strength(self) -> bool: + """Confirm if the shade has signal data.""" + return bool(ATTR_SIGNAL_STRENGTH in self.raw_data) + + def get_signal_strength(self) -> int | str: + """Get signal strength from raw_data. + + :v3 is RSSI + :v2 is calculated as a percentage + """ + if self.api_version >= 3: + return self.raw_data[ATTR_SIGNAL_STRENGTH] + return round( + self.raw_data[ATTR_SIGNAL_STRENGTH] / ATTR_SIGNAL_STRENGTH_MAX * 100 + ) + + async def get_current_position_raw(self, refresh=True) -> dict: """Return the current shade position. :param refresh: If True it queries the hub for the latest info. @@ -240,22 +559,47 @@ async def get_current_position(self, refresh=True) -> dict: """ if refresh: await self.refresh() - position = self._raw_data.get(ATTR_POSITION_DATA) + position = self._raw_data.get(ATTR_POSITIONS) return position + async def get_current_position(self, refresh=True) -> ShadePosition: + """Return the current shade position. + + :param refresh: If True it queries the hub for the latest info. + :return: Dictionary with position data. + """ + await self.get_current_position_raw(refresh) + return self.raw_to_structured(self._raw_data) + class BaseShadeTilt(BaseShade): """A shade with move and tilt at bottom capabilities.""" # even for shades that can 180° tilt, this would just result in # two closed positions. 90° will always be the open position - open_position_tilt = {ATTR_POSITION1: MID_POSITION, ATTR_POSKIND1: 3} - close_position_tilt = {ATTR_POSITION1: MIN_POSITION, ATTR_POSKIND1: 3} - async def tilt(self, position_data): + def __init__( + self, raw_data: dict, shade_type: ShadeType, request: AioRequest + ) -> None: + super().__init__(raw_data, shade_type, request) + self._open_position_tilt = ShadePosition(tilt=MAX_POSITION) + self._close_position_tilt = ShadePosition(tilt=MIN_POSITION) + if self.api_version < 3: + self._open_position_tilt = ShadePosition(tilt=MID_POSITION) + + async def tilt_raw(self, position_data): + """Tilt the shade to a set position using raw data""" + _LOGGER.debug("Shade %s tilt to: %s", self.name, position_data) data = self._create_shade_data(position_data=position_data) return await self._move(data) + async def tilt(self, position_data: ShadePosition): + """Tilt the shade to a set position""" + _LOGGER.debug("Shade %s move to: %s", self.name, position_data) + data = self.structured_to_raw(position_data) + await self._move(data) + return self.current_position + async def tilt_open(self): """Tilt to close position.""" return await self.tilt(position_data=self.open_position_tilt) @@ -264,6 +608,14 @@ async def tilt_close(self): """Tilt to close position""" return await self.tilt(position_data=self.close_position_tilt) + def get_additional_positions(self, positions: ShadePosition) -> ShadePosition: + """Returns additonal positions not reported by the hub""" + if positions.primary and positions.tilt is None: + positions.tilt = MIN_POSITION + elif positions.tilt and positions.primary is None: + positions.primary = MIN_POSITION + return positions + class ShadeBottomUp(BaseShade): """Type 0 - Up Down Only. @@ -272,30 +624,33 @@ class ShadeBottomUp(BaseShade): """ shade_types = ( - shade_type(1, "Designer Roller"), - shade_type(4, "Roman"), - shade_type(5, "Bottom Up"), - shade_type(6, "Duette"), - shade_type(10, "Duette and Applause SkyLift"), - shade_type(31, "Vignette"), - shade_type(42, "M25T Roller Blind"), - shade_type(49, "AC Roller"), + ShadeType(1, "Designer Roller"), + ShadeType(4, "Roman"), + ShadeType(5, "Bottom Up"), + ShadeType(6, "Duette"), + ShadeType(10, "Duette and Applause SkyLift"), + ShadeType(31, "Vignette"), + ShadeType(32, "Vignette"), + ShadeType(42, "M25T Roller Blind"), + ShadeType(49, "AC Roller"), + ShadeType(52, "Banded Shades"), + ShadeType(84, "Vignette"), ) - capability = capability( + capability = ShadeCapability( 0, - ShadeCapabilities( + PowerviewCapabilities( primary=True, ), "Bottom Up", ) - open_position = {ATTR_POSITION1: MAX_POSITION, ATTR_POSKIND1: 1} - close_position = {ATTR_POSITION1: MIN_POSITION, ATTR_POSKIND1: 1} - - allowed_positions = ( - {ATTR_POSITION: {ATTR_POSKIND1: 1}, ATTR_COMMAND: ATTR_MOVE}, - ) + def __init__( + self, raw_data: dict, shade_type: ShadeType, request: AioRequest + ) -> None: + super().__init__(raw_data, shade_type, request) + self._open_position = ShadePosition(primary=MAX_POSITION) + self._close_position = ShadePosition(primary=MIN_POSITION) class ShadeBottomUpTiltOnClosed180(BaseShadeTilt): @@ -306,33 +661,31 @@ class ShadeBottomUpTiltOnClosed180(BaseShadeTilt): only model without a distinct capability code. """ - shade_types = ( - shade_type(44, "Twist"), - ) + shade_types = (ShadeType(44, "Twist"),) # via json these have capability 0 # overriding to 1 to trick HA into providing tilt functionality # only difference is these have 180 tilt - capability = capability( + capability = ShadeCapability( 1, - ShadeCapabilities( + PowerviewCapabilities( primary=True, - tiltOnClosed=True, - tilt180=True, + tilt_onclosed=True, + tilt_180=True, ), "Bottom Up Tilt 180°", ) - open_position = {ATTR_POSITION1: MAX_POSITION, ATTR_POSKIND1: 1} - close_position = {ATTR_POSITION1: MIN_POSITION, ATTR_POSKIND1: 1} - - open_position_tilt = {ATTR_POSITION1: MID_POSITION, ATTR_POSKIND1: 3} - close_position_tilt = {ATTR_POSITION1: MIN_POSITION, ATTR_POSKIND1: 3} - - allowed_positions = ( - {ATTR_POSITION: {ATTR_POSKIND1: 1}, ATTR_COMMAND: ATTR_MOVE}, - {ATTR_POSITION: {ATTR_POSKIND1: 3}, ATTR_COMMAND: ATTR_TILT}, - ) + def __init__( + self, raw_data: dict, shade_type: ShadeType, request: AioRequest + ) -> None: + super().__init__(raw_data, shade_type, request) + self._open_position = ShadePosition(primary=MAX_POSITION) + self._close_position = ShadePosition(primary=MIN_POSITION) + self._open_position_tilt = ShadePosition(tilt=MAX_POSITION) + self._close_position_tilt = ShadePosition(tilt=MIN_POSITION) + if self.api_version < 3: + self._open_position_tilt = ShadePosition(tilt=MID_POSITION) class ShadeBottomUpTiltOnClosed90(BaseShadeTilt): @@ -342,33 +695,33 @@ class ShadeBottomUpTiltOnClosed90(BaseShadeTilt): """ shade_types = ( - shade_type(18, "Pirouette"), - shade_type(23, "Silhouette"), - shade_type(43, "Facette"), + ShadeType(18, "Pirouette"), + ShadeType(23, "Silhouette"), + ShadeType(43, "Facette"), ) - capability = capability( + capability = ShadeCapability( 1, - ShadeCapabilities( + PowerviewCapabilities( primary=True, - tiltOnClosed=True, - tilt90=True, + tilt_onclosed=True, + tilt_90=True, ), "Bottom Up Tilt 90°", ) - shade_limits = ShadeLimits(tilt_max=MID_POSITION) - - open_position = {ATTR_POSITION1: MAX_POSITION, ATTR_POSKIND1: 1} - close_position = {ATTR_POSITION1: MIN_POSITION, ATTR_POSKIND1: 1} - - open_position_tilt = {ATTR_POSITION1: MID_POSITION, ATTR_POSKIND1: 3} - close_position_tilt = {ATTR_POSITION1: MIN_POSITION, ATTR_POSKIND1: 3} - - allowed_positions = ( - {ATTR_POSITION: {ATTR_POSKIND1: 1}, ATTR_COMMAND: ATTR_MOVE}, - {ATTR_POSITION: {ATTR_POSKIND1: 3}, ATTR_COMMAND: ATTR_TILT}, - ) + def __init__( + self, raw_data: dict, shade_type: ShadeType, request: AioRequest + ) -> None: + super().__init__(raw_data, shade_type, request) + self.shade_limits = ShadeLimits(tilt_max=MAX_POSITION) + self._open_position = ShadePosition(primary=MAX_POSITION) + self._close_position = ShadePosition(primary=MIN_POSITION) + self._open_position_tilt = ShadePosition(tilt=MAX_POSITION) + self._close_position_tilt = ShadePosition(tilt=MIN_POSITION) + if self.api_version < 3: + self.shade_limits = ShadeLimits(tilt_max=MID_POSITION) + self._open_position_tilt = ShadePosition(tilt=MID_POSITION) class ShadeBottomUpTiltAnywhere(BaseShadeTilt): @@ -378,41 +731,32 @@ class ShadeBottomUpTiltAnywhere(BaseShadeTilt): """ shade_types = ( - shade_type(51, "Venetian, Tilt Anywhere"), - shade_type(62, "Venetian, Tilt Anywhere"), + ShadeType(51, "Venetian, Tilt Anywhere"), + ShadeType(62, "Venetian, Tilt Anywhere"), ) - capability = capability( + capability = ShadeCapability( 2, - ShadeCapabilities( + PowerviewCapabilities( primary=True, - tiltAnywhere=True, - tilt180=True, + tilt_anywhere=True, + tilt_180=True, ), "Bottom Up Tilt 180°", ) - open_position = { - ATTR_POSKIND1: 1, - ATTR_POSITION1: MAX_POSITION, - ATTR_POSKIND2: 3, - ATTR_POSITION2: MID_POSITION, - } - - close_position = { - ATTR_POSKIND1: 1, - ATTR_POSITION1: MIN_POSITION, - ATTR_POSKIND2: 3, - ATTR_POSITION2: MIN_POSITION, - } - - open_position_tilt = {ATTR_POSITION1: MID_POSITION, ATTR_POSKIND1: 3} - close_position_tilt = {ATTR_POSITION1: MIN_POSITION, ATTR_POSKIND1: 3} - - allowed_positions = ( - {ATTR_POSITION: {ATTR_POSKIND1: 1, ATTR_POSKIND2: 3}, ATTR_COMMAND: ATTR_MOVE}, - {ATTR_POSITION: {ATTR_POSKIND1: 3}, ATTR_COMMAND: ATTR_TILT}, - ) + def __init__( + self, raw_data: dict, shade_type: ShadeType, request: AioRequest + ) -> None: + super().__init__(raw_data, shade_type, request) + self._open_position = ShadePosition(primary=MAX_POSITION, tilt=MAX_POSITION) + self._close_position = ShadePosition(primary=MIN_POSITION, tilt=MAX_POSITION) + self._open_position_tilt = ShadePosition(tilt=MAX_POSITION) + self._close_position_tilt = ShadePosition(tilt=MIN_POSITION) + if self.api_version < 3: + self._open_position = ShadePosition(primary=MAX_POSITION, tilt=MID_POSITION) + self._close_position = ShadePosition(primary=MIN_POSITION, tilt=MID_POSITION) + self._open_position_tilt = ShadePosition(tilt=MID_POSITION) class ShadeVertical(ShadeBottomUp): @@ -423,17 +767,17 @@ class ShadeVertical(ShadeBottomUp): """ shade_types = ( - shade_type(26, "Skyline Panel, Left Stack"), - shade_type(27, "Skyline Panel, Right Stack"), - shade_type(28, "Skyline Panel, Split Stack"), - shade_type(69, "Curtain, Left Stack"), - shade_type(70, "Curtain, Right Stack"), - shade_type(71, "Curtain, Split Stack"), + ShadeType(26, "Skyline Panel, Left Stack"), + ShadeType(27, "Skyline Panel, Right Stack"), + ShadeType(28, "Skyline Panel, Split Stack"), + ShadeType(69, "Curtain, Left Stack"), + ShadeType(70, "Curtain, Right Stack"), + ShadeType(71, "Curtain, Split Stack"), ) - capability = capability( + capability = ShadeCapability( 3, - ShadeCapabilities( + PowerviewCapabilities( primary=True, vertical=True, ), @@ -449,17 +793,17 @@ class ShadeVerticalTiltAnywhere(ShadeBottomUpTiltAnywhere): """ shade_types = ( - shade_type(54, "Vertical Slats, Left Stack"), - shade_type(55, "Vertical Slats, Right Stack"), - shade_type(56, "Vertical Slats, Split Stack"), + ShadeType(54, "Vertical Slats, Left Stack"), + ShadeType(55, "Vertical Slats, Right Stack"), + ShadeType(56, "Vertical Slats, Split Stack"), ) - capability = capability( + capability = ShadeCapability( 4, - ShadeCapabilities( + PowerviewCapabilities( primary=True, - tiltAnywhere=True, - tilt180=True, + tilt_anywhere=True, + tilt_180=True, vertical=True, ), "Vertical Tilt Anywhere", @@ -472,33 +816,36 @@ class ShadeTiltOnly(BaseShadeTilt): A shade with tilt anywhere capabilities only. """ - shade_types = ( - shade_type(66, "Palm Beach Shutters"), - ) + shade_types = (ShadeType(66, "Palm Beach Shutters"),) - capability = capability( + capability = ShadeCapability( 5, - ShadeCapabilities( - tiltAnywhere=True, - tilt180=True, + PowerviewCapabilities( + tilt_anywhere=True, + tilt_180=True, ), "Tilt Only 180°", ) - open_position = {} - close_position = {} - - open_position_tilt = {ATTR_POSITION1: MID_POSITION, ATTR_POSKIND1: 3} - close_position_tilt = {ATTR_POSITION1: MIN_POSITION, ATTR_POSKIND1: 3} - - allowed_positions = ( - {ATTR_POSITION: {ATTR_POSKIND1: 3}, ATTR_COMMAND: ATTR_TILT}, - ) - - async def move(self): - _LOGGER.error("Move not supported.") + def __init__( + self, raw_data: dict, shade_type: ShadeType, request: AioRequest + ) -> None: + super().__init__(raw_data, shade_type, request) + self._open_position = ShadePosition() + self._close_position = ShadePosition() + self._open_position_tilt = ShadePosition(tilt=MAX_POSITION) + self._close_position_tilt = ShadePosition(tilt=MIN_POSITION) + if self.api_version < 3: + self._open_position_tilt = ShadePosition(tilt=MID_POSITION) + + async def move(self, position_data=None): + _LOGGER.error("Move unsupported. Position request(%s) ignored", position_data) return + def get_additional_positions(self, positions: ShadePosition) -> ShadePosition: + """Returns additonal positions not reported by the hub""" + return positions + class ShadeTopDown(BaseShade): """Type 6 - Top Down Only @@ -506,25 +853,23 @@ class ShadeTopDown(BaseShade): A shade with top down capabilities only. """ - shade_types = ( - shade_type(7, "Top Down"), - ) + shade_types = (ShadeType(7, "Top Down"),) - capability = capability( + capability = ShadeCapability( 6, - ShadeCapabilities( + PowerviewCapabilities( primary=True, - primaryInverted=True, + primary_inverted=True, ), "Top Down", ) - open_position = {ATTR_POSITION1: MIN_POSITION, ATTR_POSKIND1: 1} - close_position = {ATTR_POSITION1: MAX_POSITION, ATTR_POSKIND1: 1} - - allowed_positions = ( - {ATTR_POSITION: {ATTR_POSKIND1: 1}, ATTR_COMMAND: ATTR_MOVE}, - ) + def __init__( + self, raw_data: dict, shade_type: ShadeType, request: AioRequest + ) -> None: + super().__init__(raw_data, shade_type, request) + self._open_position = ShadePosition(primary=MIN_POSITION) + self._close_position = ShadePosition(primary=MAX_POSITION) class ShadeTopDownBottomUp(BaseShade): @@ -534,38 +879,35 @@ class ShadeTopDownBottomUp(BaseShade): """ shade_types = ( - shade_type(8, "Duette, Top Down Bottom Up"), - shade_type(9, "Duette DuoLite, Top Down Bottom Up"), - shade_type(33, "Duette Architella, Top Down Bottom Up"), - shade_type(47, "Pleated, Top Down Bottom Up"), + ShadeType(8, "Duette, Top Down Bottom Up"), + ShadeType(9, "Duette DuoLite, Top Down Bottom Up"), + ShadeType(33, "Duette Architella, Top Down Bottom Up"), + ShadeType(47, "Pleated, Top Down Bottom Up"), ) - capability = capability( + capability = ShadeCapability( 7, - ShadeCapabilities( + PowerviewCapabilities( primary=True, secondary=True, ), "Top Down Bottom Up", ) - open_position = { - ATTR_POSITION1: MAX_POSITION, - ATTR_POSITION2: MIN_POSITION, - ATTR_POSKIND1: 1, - ATTR_POSKIND2: 2, - } - - close_position = { - ATTR_POSITION1: MIN_POSITION, - ATTR_POSITION2: MIN_POSITION, - ATTR_POSKIND1: 1, - ATTR_POSKIND2: 2, - } - - allowed_positions = ( - {ATTR_POSITION: {ATTR_POSKIND1: 1, ATTR_POSKIND2: 2}, ATTR_COMMAND: ATTR_MOVE}, - ) + def __init__( + self, raw_data: dict, shade_type: ShadeType, request: AioRequest + ) -> None: + super().__init__(raw_data, shade_type, request) + self._open_position = ShadePosition(primary=MAX_POSITION, secondary=MIN_POSITION) + self._close_position = ShadePosition(primary=MIN_POSITION, secondary=MIN_POSITION) + + def get_additional_positions(self, positions: ShadePosition) -> ShadePosition: + """Returns additonal positions not reported by the hub""" + if positions.primary is None: + positions.primary = MIN_POSITION + if positions.secondary is None: + positions.secondary = MIN_POSITION + return positions class ShadeDualOverlapped(BaseShade): @@ -575,27 +917,45 @@ class ShadeDualOverlapped(BaseShade): """ shade_types = ( - shade_type(65, "Vignette Duolite"), - shade_type(79, "Duolite Lift"), + ShadeType(65, "Vignette Duolite"), + ShadeType(79, "Duolite Lift"), ) - capability = capability( + capability = ShadeCapability( 8, - ShadeCapabilities( + PowerviewCapabilities( primary=True, secondary=True, - secondaryOverlapped=True, + secondary_overlapped=True, ), "Dual Shade Overlapped", ) - open_position = {ATTR_POSITION1: MAX_POSITION, ATTR_POSKIND1: 1} - close_position = {ATTR_POSITION1: MIN_POSITION, ATTR_POSKIND1: 2} - - allowed_positions = ( - {ATTR_POSITION: {ATTR_POSKIND1: 1}, ATTR_COMMAND: ATTR_MOVE}, - {ATTR_POSITION: {ATTR_POSKIND1: 2}, ATTR_COMMAND: ATTR_MOVE}, - ) + def __init__( + self, raw_data: dict, shade_type: ShadeType, request: AioRequest + ) -> None: + super().__init__(raw_data, shade_type, request) + self._open_position = ShadePosition(primary=MAX_POSITION) + self._close_position = ShadePosition(secondary=MIN_POSITION) + + def get_additional_positions(self, positions: ShadePosition) -> ShadePosition: + """Returns additonal positions not reported by the hub""" + if positions.primary: + if positions.secondary is None: + positions.secondary = MAX_POSITION + if positions.tilt is None: + positions.tilt = MIN_POSITION + elif positions.secondary: + if positions.primary is None: + positions.primary = MIN_POSITION + if positions.tilt is None: + positions.tilt = MIN_POSITION + elif positions.tilt: + if positions.primary is None: + positions.primary = MIN_POSITION + if positions.secondary is None: + positions.secondary = MAX_POSITION + return positions class ShadeDualOverlappedTilt90(BaseShadeTilt): @@ -605,35 +965,51 @@ class ShadeDualOverlappedTilt90(BaseShadeTilt): Tilt on these is unique in that it requires the rear shade open and front shade closed. """ - shade_types = ( - shade_type(38, "Silhouette Duolite"), - ) + shade_types = (ShadeType(38, "Silhouette Duolite"),) - capability = capability( + capability = ShadeCapability( 9, - ShadeCapabilities( + PowerviewCapabilities( primary=True, secondary=True, - secondaryOverlapped=True, - tilt90=True, - tiltOnClosed=True, + secondary_overlapped=True, + tilt_90=True, + tilt_onclosed=True, ), "Dual Shade Overlapped Tilt 90°", ) - shade_limits = ShadeLimits(tilt_max=MID_POSITION) - - open_position = {ATTR_POSITION1: MAX_POSITION, ATTR_POSKIND1: 1} - close_position = {ATTR_POSITION1: MIN_POSITION, ATTR_POSKIND1: 2} - - open_position_tilt = {ATTR_POSITION2: MID_POSITION, ATTR_POSKIND1: 3} - close_position_tilt = {ATTR_POSITION2: MIN_POSITION, ATTR_POSKIND1: 3} - - allowed_positions = ( - {ATTR_POSITION: {ATTR_POSKIND1: 1}, ATTR_COMMAND: ATTR_MOVE}, - {ATTR_POSITION: {ATTR_POSKIND1: 2}, ATTR_COMMAND: ATTR_MOVE}, - {ATTR_POSITION: {ATTR_POSKIND1: 3}, ATTR_COMMAND: ATTR_TILT}, - ) + def __init__( + self, raw_data: dict, shade_type: ShadeType, request: AioRequest + ) -> None: + super().__init__(raw_data, shade_type, request) + self.shade_limits = ShadeLimits(tilt_max=MAX_POSITION) + self._open_position = ShadePosition(primary=MAX_POSITION) + self._close_position = ShadePosition(secondary=MIN_POSITION) + self._open_position_tilt = ShadePosition(tilt=MAX_POSITION) + self._close_position_tilt = ShadePosition(tilt=MIN_POSITION) + if self.api_version < 3: + self.shade_limits = ShadeLimits(tilt_max=MID_POSITION) + self._open_position_tilt = ShadePosition(tilt=MID_POSITION) + + def get_additional_positions(self, positions: ShadePosition) -> ShadePosition: + """Returns additonal positions not reported by the hub""" + if positions.primary: + if positions.secondary is None: + positions.secondary = MAX_POSITION + if positions.tilt is None: + positions.tilt = MIN_POSITION + elif positions.secondary: + if positions.primary is None: + positions.primary = MIN_POSITION + if positions.tilt is None: + positions.tilt = MIN_POSITION + elif positions.tilt: + if positions.primary is None: + positions.primary = MIN_POSITION + if positions.secondary is None: + positions.secondary = MAX_POSITION + return positions class ShadeDualOverlappedTilt180(ShadeDualOverlappedTilt90): @@ -643,19 +1019,78 @@ class ShadeDualOverlappedTilt180(ShadeDualOverlappedTilt90): Tilt on these is unique in that it requires the rear shade open and front shade closed. """ - shade_types = ( - ) + shade_types = () - capability = capability( + capability = ShadeCapability( 10, - ShadeCapabilities( + PowerviewCapabilities( primary=True, secondary=True, - secondaryOverlapped=True, - tilt180=True, - tiltOnClosed=True, + secondary_overlapped=True, + tilt_180=True, + tilt_onclosed=True, ), "Dual Shade Overlapped Tilt 180°", ) - shade_limits = ShadeLimits(tilt_max=MAX_POSITION) + def __init__( + self, raw_data: dict, shade_type: ShadeType, request: AioRequest + ) -> None: + super().__init__(raw_data, shade_type, request) + if self.api_version < 3: + self.shade_limits = ShadeLimits(tilt_max=MAX_POSITION) + +def factory(raw_data: dict, request: AioRequest): + """Class factory to create different types of shades + depending on shade type.""" + + if ATTR_SHADE in raw_data: + raw_data = raw_data.get(ATTR_SHADE) + + raw_type = raw_data.get(ATTR_TYPE) + + def find_type(shade: BaseShade): + for type_def in shade.shade_types: + if type_def.type == raw_type: + return shade(raw_data, type_def, request) + return None + + shade_capability = raw_data.get(ATTR_CAPABILITIES) + + def find_capability(shade: BaseShade): + if shade.capability.type == shade_capability: + return shade(raw_data, shade, request) + return None + + classes = [ + ShadeBottomUp, + ShadeBottomUpTiltOnClosed90, + ShadeBottomUpTiltOnClosed180, # to ensure capability match order here is important + ShadeBottomUpTiltAnywhere, + ShadeVerticalTiltAnywhere, + ShadeVertical, + ShadeTiltOnly, + ShadeTopDown, + ShadeTopDownBottomUp, + ShadeDualOverlapped, + ShadeDualOverlappedTilt90, + ShadeDualOverlappedTilt180, + ] + + for cls in classes: + # class check is more concise as we have tested positioning + _shade = find_type(cls) + if _shade: + _LOGGER.debug("Match type : %s - %s", _shade, raw_data) + return _shade + + for cls in classes: + # fallback to a capability check - this should future proof new shades + # type 0 that contain tilt would not be caught here + _shade = find_capability(cls) + if _shade: + _LOGGER.debug("Match capability : %s - %s", _shade, raw_data) + return _shade + + _LOGGER.debug("Shade unmatched : %s - %s", BaseShade, raw_data) + return BaseShade(raw_data, BaseShade.shade_types[0], request) diff --git a/aiopvapi/resources/userdata.py b/aiopvapi/resources/userdata.py deleted file mode 100644 index 9402d62..0000000 --- a/aiopvapi/resources/userdata.py +++ /dev/null @@ -1,12 +0,0 @@ -from aiopvapi.helpers.aiorequest import AioRequest -from aiopvapi.helpers.api_base import ApiResource -from aiopvapi.helpers.constants import ATTR_USER_DATA - - -class UserData(ApiResource): - api_path = "api/userdata" - - def __init__(self, raw_data: dict, request: AioRequest): - if ATTR_USER_DATA in raw_data: - raw_data = raw_data.get(ATTR_USER_DATA) - super().__init__(request, self.api_path, raw_data) diff --git a/aiopvapi/rooms.py b/aiopvapi/rooms.py index 71a8322..93d0d04 100644 --- a/aiopvapi/rooms.py +++ b/aiopvapi/rooms.py @@ -3,36 +3,69 @@ import logging from aiopvapi.helpers.api_base import ApiEntryPoint -from aiopvapi.helpers.constants import ATTR_NAME, ATTR_COLOR_ID, ATTR_ICON_ID, ATTR_ROOM +from aiopvapi.helpers.aiorequest import AioRequest +from aiopvapi.helpers.constants import ( + ATTR_ID, + ATTR_NAME, + ATTR_COLOR_ID, + ATTR_ICON_ID, + ATTR_ROOM, + ATTR_ROOM_DATA, +) from aiopvapi.helpers.tools import unicode_to_base64 from aiopvapi.resources.room import Room -_LOGGER = logging.getLogger("__name__") +from aiopvapi.resources.model import PowerviewData -ATTR_ROOM_DATA = "roomData" +_LOGGER = logging.getLogger(__name__) class Rooms(ApiEntryPoint): - api_path = "api/rooms" + """Rooms entry point""" - def __init__(self, request): - super().__init__(request, self.api_path) + api_endpoint = "rooms" + + def __init__(self, request: AioRequest) -> None: + super().__init__(request, self.api_endpoint) async def create_room(self, name, color_id=0, icon_id=0): + """Create a room on the hub""" name = unicode_to_base64(name) data = { - ATTR_ROOM: {ATTR_NAME: name, ATTR_COLOR_ID: color_id, ATTR_ICON_ID: icon_id} + ATTR_ROOM: { + ATTR_NAME: name, + ATTR_COLOR_ID: color_id, + ATTR_ICON_ID: icon_id, + } } - return await self.request.post(self._base_path, data=data) + return await self.request.post(self.base_path, data=data) def _resource_factory(self, raw): return Room(raw, self.request) - @staticmethod - def _loop_raw(raw): - for _raw in raw[ATTR_ROOM_DATA]: + def _loop_raw(self, raw): + if self.api_version < 3: + raw = raw[ATTR_ROOM_DATA] + + for _raw in raw: yield _raw - @staticmethod - def _get_to_actual_data(raw): + def _get_to_actual_data(self, raw): + if self.api_version >= 3: + return raw return raw.get("room") + + async def get_rooms(self) -> PowerviewData: + """Get a list of rooms. + + :returns PowerviewData object + :raises PvApiError when an error occurs. + """ + resources = await self.get_resources() + if self.api_version < 3: + resources = resources[ATTR_ROOM_DATA] + + processed = {entry[ATTR_ID]: Room(entry, self.request) for entry in resources} + + _LOGGER.debug("Raw room data: %s", resources) + return PowerviewData(raw=resources, processed=processed) diff --git a/aiopvapi/scene_members.py b/aiopvapi/scene_members.py index b1bada7..a5e0d65 100644 --- a/aiopvapi/scene_members.py +++ b/aiopvapi/scene_members.py @@ -3,48 +3,57 @@ import logging from aiopvapi.helpers.api_base import ApiEntryPoint -from aiopvapi.helpers.constants import ATTR_SCENE_ID, ATTR_SHADE_ID, ATTR_POSITION_DATA +from aiopvapi.helpers.aiorequest import AioRequest +from aiopvapi.helpers.constants import ( + ATTR_SCENE_ID, + ATTR_SHADE_ID, + ATTR_POSITIONS, + SCENE_MEMBER_DATA, +) from aiopvapi.resources.scene_member import ATTR_SCENE_MEMBER, SceneMember -_LOGGER = logging.getLogger("__name__") +from aiopvapi.resources.model import PowerviewData -SCENE_MEMBER_DATA = "sceneMemberData" +_LOGGER = logging.getLogger("__name__") class SceneMembers(ApiEntryPoint): """A scene member is a device, like a shade, being a member of a specific scene.""" - api_path = "api/scenemembers" + api_endpoint = "scenemembers" - def __init__(self, request): - super().__init__(request, self.api_path) + def __init__(self, request: AioRequest) -> None: + super().__init__(request, self.api_endpoint) async def create_scene_member(self, shade_position, scene_id, shade_id): """Adds a shade to an existing scene""" data = { ATTR_SCENE_MEMBER: { - ATTR_POSITION_DATA: shade_position, + ATTR_POSITIONS: shade_position, ATTR_SCENE_ID: scene_id, ATTR_SHADE_ID: shade_id, } } - return await self.request.post(self._base_path, data=data) + return await self.request.post(self.base_path, data=data) def _resource_factory(self, raw): return SceneMember(raw, self.request) - @staticmethod - def _loop_raw(raw): - for _raw in raw[SCENE_MEMBER_DATA]: + def _loop_raw(self, raw): + if self.api_version < 3: + raw = raw[SCENE_MEMBER_DATA] + + for _raw in raw: yield _raw - @staticmethod - def _get_to_actual_data(raw): - return raw.get("scenemember") + def _get_to_actual_data(self, raw): + if self.api_version >= 3: + return raw + return raw.get(SCENE_MEMBER_DATA) - async def get_scene_members(self, scene_id): + async def get_scene_members_old(self, scene_id): """Return all scene members for a particular Scene ID.""" return await self.get_instances(sceneId=scene_id) @@ -52,5 +61,23 @@ async def get_scene_members(self, scene_id): async def delete_shade_from_scene(self, shade_id, scene_id): """Delete a shade from a scene.""" return await self.request.delete( - self._base_path, params={ATTR_SCENE_ID: scene_id, ATTR_SHADE_ID: shade_id} + self.base_path, + params={ATTR_SCENE_ID: scene_id, ATTR_SHADE_ID: shade_id}, ) + + async def get_scene_members(self) -> PowerviewData: + """Get a list of scene members. + + :raises PvApiError when an error occurs. + """ + resources = await self.get_resources() + if self.api_version < 3: + resources = resources[SCENE_MEMBER_DATA] + + # return array of scenes attached to a shade + processed = { + entry["shadeId"]: SceneMember(entry, self.request) for entry in resources + } + + _LOGGER.debug("Raw scene_member data: %s", resources) + return PowerviewData(raw=resources, processed=processed) diff --git a/aiopvapi/scenes.py b/aiopvapi/scenes.py index 8f6fd11..c33929b 100644 --- a/aiopvapi/scenes.py +++ b/aiopvapi/scenes.py @@ -5,38 +5,70 @@ from aiopvapi.helpers.aiorequest import AioRequest from aiopvapi.helpers.api_base import ApiEntryPoint from aiopvapi.helpers.constants import ( + ATTR_ID, ATTR_NAME, ATTR_ROOM_ID, ATTR_ICON_ID, ATTR_COLOR_ID, + ATTR_SCENE_DATA, ) from aiopvapi.helpers.tools import unicode_to_base64 +from aiopvapi.resources.model import PowerviewData from aiopvapi.resources.scene import Scene -_LOGGER = logging.getLogger("__name__") -ATTR_SCENE_DATA = "sceneData" +_LOGGER = logging.getLogger(__name__) class Scenes(ApiEntryPoint): - api_path = "api/scenes" + """Powerview Scenes""" - def __init__(self, request: AioRequest): - super().__init__(request, self.api_path) + api_endpoint = "scenes" + + def __init__(self, request: AioRequest) -> None: + super().__init__(request, self.api_endpoint) def _resource_factory(self, raw): return Scene(raw, self.request) - @staticmethod - def _loop_raw(raw): - for _raw in raw[ATTR_SCENE_DATA]: + def _loop_raw(self, raw): + if self.api_version < 3: + raw = raw[ATTR_SCENE_DATA] + + for _raw in raw: yield _raw - @staticmethod - def _get_to_actual_data(raw): + def _get_to_actual_data(self, raw): + if self.api_version >= 3: + return raw return raw.get("scene") + async def get_scenes_old(self) -> dict: + """Get a list of scenes. + + :raises PvApiError when an error occurs. + """ + resources = await self.get_resources() + if self.api_version < 3: + return resources[ATTR_SCENE_DATA] + return resources + + async def get_scenes(self) -> PowerviewData: + """Get a list of scenes. + + :raises PvApiError when an error occurs. + """ + resources = await self.get_resources() + if self.api_version < 3: + resources = resources[ATTR_SCENE_DATA] + + # return array of scenes attached to a shade + processed = {entry[ATTR_ID]: Scene(entry, self.request) for entry in resources} + + _LOGGER.debug("Raw scenes data: %s", resources) + return PowerviewData(raw=resources, processed=processed) + async def create_scene(self, room_id, name, color_id=0, icon_id=0): - """Creates am empty scene. + """Creates an empty scene. Scenemembers need to be added after the scene has been created. @@ -51,5 +83,5 @@ async def create_scene(self, room_id, name, color_id=0, icon_id=0): ATTR_ICON_ID: icon_id, } } - _response = await self.request.post(self._base_path, data=_data) + _response = await self.request.post(self.base_path, data=_data) return _response diff --git a/aiopvapi/shades.py b/aiopvapi/shades.py index 44b2028..5c8355f 100644 --- a/aiopvapi/shades.py +++ b/aiopvapi/shades.py @@ -4,53 +4,84 @@ from aiopvapi.helpers.aiorequest import AioRequest from aiopvapi.helpers.api_base import ApiEntryPoint -from aiopvapi.helpers.constants import ATTR_NAME, ATTR_NAME_UNICODE +from aiopvapi.helpers.constants import ( + ATTR_ID, + ATTR_NAME, + ATTR_NAME_UNICODE, + ATTR_SHADE_DATA, +) from aiopvapi.helpers.tools import base64_to_unicode from aiopvapi.resources import shade -LOGGER = logging.getLogger("__name__") +from aiopvapi.resources.model import PowerviewData -ATTR_SHADE_DATA = "shadeData" + +_LOGGER = logging.getLogger(__name__) class Shades(ApiEntryPoint): """Shades entry point""" - api_path = "api/shades" + api_endpoint = "shades" - def __init__(self, request: AioRequest): - super().__init__(request, self.api_path) + def __init__(self, request: AioRequest) -> None: + super().__init__(request, self.api_endpoint) - @staticmethod - def sanitize_resources(resource: dict): - """Cleans up incoming scene data + def _sanitize_resources(self, resources: dict) -> dict | None: + """Cleans up incoming shade data - :param resource: The dict with scene data to be sanitized. - :returns: Cleaned up scene dict. + :param resources: The dict with shade data to be sanitized. + :returns: Cleaned up shade dict. """ + if self.api_version < 3: + resources = resources[ATTR_SHADE_DATA] + try: - for shade in resource[ATTR_SHADE_DATA]: - _name = shade.get(ATTR_NAME) + for _shade in resources: + _name = _shade.get(ATTR_NAME) if _name: - shade[ATTR_NAME_UNICODE] = base64_to_unicode(_name) - return resource + _shade[ATTR_NAME_UNICODE] = base64_to_unicode(_name) + return resources except (KeyError, TypeError): - LOGGER.debug("no shade data available") + _LOGGER.debug("No shade data available") return None def _resource_factory(self, raw): return shade.factory(raw, self.request) - @staticmethod - def _loop_raw(raw): - for _raw in raw[ATTR_SHADE_DATA]: + def _loop_raw(self, raw): + if self.api_version < 3: + raw = raw[ATTR_SHADE_DATA] + + for _raw in raw: yield _raw - @staticmethod - def _get_to_actual_data(raw): + def _get_to_actual_data(self, raw): + if self.api_version >= 3: + return raw return raw.get("shade") - # async def get_shade(self, shade_id: int): + async def get_shades(self) -> PowerviewData: + """Get a list of shades. + + :returns PowerviewData object + :raises PvApiError when an error occurs. + """ + resources = await self.get_resources() + if self.api_version < 3: + resources = resources[ATTR_SHADE_DATA] + + _LOGGER.debug("Raw shades data: %s", resources) + + processed = { + entry[ATTR_ID]: shade.factory(entry, self.request) for entry in resources + } + + _LOGGER.debug("Raw shades data: %s", resources) + return PowerviewData(raw=resources, processed=processed) + + # async def get_shade(self, shade_id: int): + # _url = '{}/{}'.format(self.api_path, shade_id) # _raw = await self.request.get(_url) # return shade.factory(_raw, self.request) diff --git a/aiopvapi/userdata.py b/aiopvapi/userdata.py deleted file mode 100644 index 8e0f9fd..0000000 --- a/aiopvapi/userdata.py +++ /dev/null @@ -1,33 +0,0 @@ -"""Scenes class managing all scene data.""" - -import logging - -from aiopvapi.helpers.aiorequest import AioRequest -from aiopvapi.helpers.api_base import ApiEntryPoint -from aiopvapi.helpers.tools import base64_to_unicode - -LOGGER = logging.getLogger("__name__") - -ATTR_HUB_NAME = "hubName" -ATTR_HUB_NAME_UNICODE = "hubNameUnicode" - - -class UserData(ApiEntryPoint): - api_path = "api/userdata" - - def __init__(self, request: AioRequest): - super().__init__(request, self.api_path) - - @staticmethod - def sanitize_resources(resource): - """Cleans up incoming scene data - - :param resource: The dict with scene data to be sanitized. - :returns: Cleaned up dict. - """ - try: - resource[ATTR_HUB_NAME_UNICODE] = base64_to_unicode(resource[ATTR_HUB_NAME]) - return resource - except (KeyError, TypeError): - LOGGER.debug("no data available") - return None diff --git a/readme.md b/readme.md index edcf488..45da58c 100644 --- a/readme.md +++ b/readme.md @@ -56,6 +56,14 @@ Have a look at the examples folder for some guidance how to use it. - Handle calls to update shade position during maintenance - Raise error directly on hub calls instead of logger +### v3.0.0 + +- Major overhaul to incorporate gateway version 3 API. Version can be automatically detected or manually specified. +- UserData class is deprecated and replaced with Hub. +- ShadePosition class now replaces the raw json management of shades in support of cross generational management. +- Schedules / Automations are now supported by the API +- New get_*objecttype* methods available to returned structured data objects for consistent management + ## Links --- diff --git a/requirements-dev.txt b/requirements-dev.txt index f22e2bd..26b527b 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -5,3 +5,4 @@ nox pytest-cov python-coveralls wheel +json diff --git a/tests/fake_server.py b/tests/fake_server.py index 7af8e1a..77f3d04 100644 --- a/tests/fake_server.py +++ b/tests/fake_server.py @@ -20,6 +20,9 @@ ROOM_VALUE = """ {"room":{"id":46688,"name":"TGl2aW5nIHJvb20=","order":0,"colorId":7,"iconId":0,"type":0}} """ +ROOM_VALUE_V3 = """ +{"id":46688,"name":"TGl2aW5nIHJvb20=","order":0,"colorId":7,"iconId":0,"type":0} +""" SHADES_VALUE = """ {"shadeIds":[29889,56112],"shadeData":[ @@ -33,6 +36,9 @@ SHADE_VALUE = """ {"shade":{"id":11155,"name":"U2hhZGUgMQ==","roomId":46688,"groupId":38058,"order":0,"type":8,"batteryStrength":146,"batteryStatus":3,"positions":{"position1":8404,"posKind1":1,"position2":42209,"posKind2":2},"firmware":{"revision":1,"subRevision":4,"build":1643},"signalStrength":4}} """ +SHADE_VALUE_V3 = """ +{"id":11155,"name":"U2hhZGUgMQ==","roomId":46688,"groupId":38058,"order":0,"type":8,"batteryStrength":146,"batteryStatus":3,"positions":{"position1":8404,"posKind1":1,"position2":42209,"posKind2":2},"firmware":{"revision":1,"subRevision":4,"build":1643},"signalStrength":4} +""" SCENES_VALUE = """ {"sceneIds":[37217,64533],"sceneData":[ @@ -43,12 +49,92 @@ SCENE_VALUE = """ {"scene":{"roomId":46688,"name":"VGVzdA==","colorId":7,"iconId":0,"id":43436,"order":0}} """ +SCENE_VALUE_V3 = """ +{"roomId":46688,"name":"VGVzdA==","colorId":7,"iconId":0,"id":43436,"order":0} +""" + +USER_DATA_VALUE = """ +{ + "userData" : { + "firmware" : { + "mainProcessor" : { + "build" : 395, + "name" : "PV Hub2.0", + "revision" : 2, + "subRevision" : 0 + }, + "radio" : { + "build" : 1307, + "revision" : 2, + "subRevision" : 0 + } + }, + "gateway" : "192.168.1.1", + "hubName" : "SHViYnk=", + "ip" : "192.168.1.100", + "macAddress" : "00:26:74:af:fd:ae", + "mask" : "255.255.255.0", + "serialNumber" : "927FD402C11CE424", + "ssid" : "cisco789" + } +} +""" + +FWVERSION_VALUE = """ +{ + "firmware" : { + "mainProcessor" : { + "build" : 395, + "name" : "PV Hub2.0", + "revision" : 2, + "subRevision" : 0 + }, + "radio" : { + "build" : 1307, + "revision" : 2, + "subRevision" : 0 + } + } +} +""" + +GATEWAY_VALUE = """ +{ + "config":{ + "firmware":{ + "mainProcessor":{ + "build":395, + "name":"PV Hub2.0", + "revision":2, + "subRevision":0 + }, + "radios":[ + { + "build":1307, + "revision":2, + "subRevision":0 + } + ] + }, + "networkStatus":{ + "gateway":"192.168.1.1", + "ipAddress":"192.168.1.100", + "primaryMacAddress":"00:26:74:af:fd:ae", + "mask":"255.255.255.0", + "ssid":"cisco789" + }, + "serialNumber":"927FD402C11CE424" + } +} +""" + +HOME_VALUE = """ +{"gateways":[{"name": "Hubby"}]} +""" class FakeResolver: - _LOCAL_HOST = {0: '127.0.0.1', - socket.AF_INET: '127.0.0.1', - socket.AF_INET6: '::1'} + _LOCAL_HOST = {0: "127.0.0.1", socket.AF_INET: "127.0.0.1", socket.AF_INET6: "::1"} def __init__(self, fakes, *, loop): """fakes -- dns -> port dict""" @@ -58,47 +144,80 @@ def __init__(self, fakes, *, loop): async def resolve(self, host, port=0, family=socket.AF_INET): fake_port = self._fakes.get(host) if fake_port is not None: - return [{'hostname': host, - 'host': self._LOCAL_HOST[family], 'port': fake_port, - 'family': family, 'proto': 0, - 'flags': socket.AI_NUMERICHOST}] + return [ + { + "hostname": host, + "host": self._LOCAL_HOST[family], + "port": fake_port, + "family": family, + "proto": 0, + "flags": socket.AI_NUMERICHOST, + } + ] else: return await self._resolver.resolve(host, port, family) class FakePowerViewHub: - - def __init__(self, *, loop): + def __init__(self, *, loop, api_version=2): self.loop = loop self.app = web.Application() - self.app.router.add_routes( - [ - web.get('/test_get_status_200', self.test_get_status_200), - web.get('/wrong_status', self.wrong_status), - web.get('/invalid_json', self.invalid_json), - web.get('/get_timeout', self.get_timeout), - web.post('/post_status_200', self.post_status_200), - web.post('/post_status_201', self.post_status_201), - web.get('/api/rooms', self.get_rooms), - web.get('/api/rooms/1234', self.get_room), - web.post('/api/rooms', self.new_room), - web.delete('/api/rooms/{room_id}', self.delete_room), - web.get('/api/scenes', self.handle_scene), - web.get('/api/scenes/43436', self.get_scene), - web.post('/api/scenes', self.create_scene), - web.get('/api/shades', self.get_shades), - web.get('/api/shades/11155', self.get_shade), - web.put('/api/shades/{shade_id}', self.add_shade_to_room), - web.delete('/api/sceneMembers', self.remove_shade_from_scene), - - ]) + self.api_version = api_version + if api_version >= 3: + self.app.router.add_routes( + [ + web.get("/test_get_status_200", self.test_get_status_200), + web.get("/wrong_status", self.wrong_status), + web.get("/invalid_json", self.invalid_json), + web.get("/get_timeout", self.get_timeout), + web.post("/post_status_200", self.post_status_200), + web.post("/post_status_201", self.post_status_201), + web.get("/home/rooms", self.get_rooms), + web.get("/home/rooms/1234", self.get_room), + web.post("/home/rooms", self.new_room), + web.delete("/home/rooms/{room_id}", self.delete_room), + web.get("/home/scenes", self.handle_scene), + web.get("/home/scenes/43436", self.get_scene), + web.post("/home/scenes", self.create_scene), + web.get("/home/shades", self.get_shades), + web.get("/home/shades/11155", self.get_shade), + web.put("/home/shades/{shade_id}", self.add_shade_to_room), + web.delete("/home/sceneMembers", self.remove_shade_from_scene), + web.get("/gateway", self.get_gateway), + web.get("/home", self.get_home), + ] + ) + else: + self.app.router.add_routes( + [ + web.get("/test_get_status_200", self.test_get_status_200), + web.get("/wrong_status", self.wrong_status), + web.get("/invalid_json", self.invalid_json), + web.get("/get_timeout", self.get_timeout), + web.post("/post_status_200", self.post_status_200), + web.post("/post_status_201", self.post_status_201), + web.get("/api/rooms", self.get_rooms), + web.get("/api/rooms/1234", self.get_room), + web.post("/api/rooms", self.new_room), + web.delete("/api/rooms/{room_id}", self.delete_room), + web.get("/api/scenes", self.handle_scene), + web.get("/api/scenes/43436", self.get_scene), + web.post("/api/scenes", self.create_scene), + web.get("/api/shades", self.get_shades), + web.get("/api/shades/11155", self.get_shade), + web.put("/api/shades/{shade_id}", self.add_shade_to_room), + web.delete("/api/sceneMembers", self.remove_shade_from_scene), + web.get("/api/fwversion", self.get_fwversion), + web.get("/userdata", self.get_user_data), + ] + ) self.runner = None async def start(self): port = unused_port() self.runner = web.AppRunner(self.app) await self.runner.setup() - site = web.TCPSite(self.runner, '127.0.0.1', port) + site = web.TCPSite(self.runner, "127.0.0.1", port) await site.start() return {FAKE_BASE_URL: port} @@ -107,9 +226,7 @@ async def stop(self): await self.runner.cleanup() async def test_get_status_200(self, request): - return web.json_response( - {"title": "test"} - , status=200) + return web.json_response({"title": "test"}, status=200) async def wrong_status(self, request): return web.json_response({}, status=201) @@ -118,79 +235,74 @@ async def invalid_json(self, request): return web.Response( body='{"title": "test}', status=200, - headers={'content-type': 'application/json'}) + headers={"content-type": "application/json"}, + ) async def get_timeout(self, request): await asyncio.sleep(2) return web.json_response({}) async def post_status_200(self, request): - return web.json_response({ - 'a': 'b', 'c': 'd' - }) + return web.json_response({"a": "b", "c": "d"}) async def post_status_201(self, request): - return web.json_response({ - 'a': 'b', 'c': 'd' - }) + return web.json_response({"a": "b", "c": "d"}) async def get_rooms(self, request): return web.Response( - body=ROOMS_VALUE, - headers={'content-type': 'application/json'}) + body=ROOMS_VALUE, headers={"content-type": "application/json"} + ) async def get_room(self, request): + room_value = ROOM_VALUE_V3 if self.api_version >= 3 else ROOM_VALUE return web.Response( - body=ROOM_VALUE, - headers={'content-type': 'application/json'} + body=room_value, headers={"content-type": "application/json"} ) async def new_room(self, request): _js = await request.json() - return web.json_response({ - "id": 1, "name": _js['room']["name"] - }, status=201) + return web.json_response({"id": 1, "name": _js["room"]["name"]}, status=201) async def delete_room(self, request): _id = request.match_info["room_id"] - if _id == '26756': + if _id == "26756": return web.Response(status=204) else: return web.Response(status=404) async def handle_scene(self, request): - _id = request.query.get('sceneId') + _id = request.query.get("sceneId") if _id is None: return web.Response( - body=SCENES_VALUE, - headers={'content-type': 'application/json'}) - elif _id is not None and _id == '10': - return web.json_response({'id': 10}) + body=SCENES_VALUE, headers={"content-type": "application/json"} + ) + elif _id is not None and _id == "10": + return web.json_response({"id": 10}) else: return web.Response(status=404) async def get_scene(self, request): + scene_value = SCENE_VALUE_V3 if self.api_version >= 3 else SCENE_VALUE return web.Response( - body=SCENE_VALUE, - headers={'content-type': 'application/json'} + body=scene_value, headers={"content-type": "application/json"} ) async def get_shades(self, request): return web.Response( - body=SHADES_VALUE, - headers={'content-type': 'application/json'}) + body=SHADES_VALUE, headers={"content-type": "application/json"} + ) async def get_shade(self, request): + shade_value = SHADE_VALUE_V3 if self.api_version >= 3 else SHADE_VALUE return web.Response( - body=SHADE_VALUE, - headers={'content-type': 'application/json'} + body=shade_value, headers={"content-type": "application/json"} ) async def add_shade_to_room(self, request): # todo: finish this. - _shade_id = request.match_info['shade_id'] + _shade_id = request.match_info["shade_id"] _js = await request.json() - if _js['shade']['roomId'] == 123: + if _js["shade"]["roomId"] == 123: return web.json_response({"shade": True}) async def create_scene(self, request): @@ -198,21 +310,41 @@ async def create_scene(self, request): return web.json_response(_js) async def remove_shade_from_scene(self, request): - shade_id = request.query.get['shadeId'] - scene_id = request.query.get['sceneId'] - return web.json_response({'scene_id': scene_id, 'shade_id': shade_id}) + shade_id = request.query.get["shadeId"] + scene_id = request.query.get["sceneId"] + return web.json_response({"scene_id": scene_id, "shade_id": shade_id}) + + async def get_user_data(self, request): + return web.Response( + body=USER_DATA_VALUE, headers={"content-type": "application/json"} + ) + async def get_fwversion(self, request): + return web.Response( + body=FWVERSION_VALUE, headers={"content-type": "application/json"} + ) -async def main(loop): - fake_powerview_hub = FakePowerViewHub(loop=loop) + async def get_gateway(self, request): + return web.Response( + body=GATEWAY_VALUE, headers={"content-type": "application/json"} + ) + + async def get_home(self, request): + return web.Response( + body=HOME_VALUE, headers={"content-type": "application/json"} + ) + + +async def main(loop, api_version=2): + fake_powerview_hub = FakePowerViewHub(loop=loop, api_version=api_version) info = await fake_powerview_hub.start() resolver = FakeResolver(info, loop=loop) connector = aiohttp.TCPConnector(loop=loop, resolver=resolver) - async with aiohttp.ClientSession(connector=connector, - loop=loop) as session: - async with session.get('http://{}/v2.7/me'.format(FAKE_BASE_URL), - ) as resp: + async with aiohttp.ClientSession(connector=connector, loop=loop) as session: + async with session.get( + "http://{}/v2.7/me".format(FAKE_BASE_URL), + ) as resp: print(await resp.json()) # async with session.get('https://graph.facebook.com/v2.7/me/friends', @@ -228,9 +360,13 @@ async def main(loop): class TestFakeServer(unittest.TestCase): + def __init__(self, methodName: str = "runTest") -> None: + super().__init__(methodName) + self.api_version = 2 + def setUp(self): self.loop = asyncio.get_event_loop() - self.server = FakePowerViewHub(loop=self.loop) + self.server = FakePowerViewHub(loop=self.loop, api_version=self.api_version) self.request = None def tearDown(self): @@ -238,14 +374,16 @@ def tearDown(self): self.loop.run_until_complete(self.request.websession.close()) self.loop.run_until_complete(self.server.stop()) - async def start_fake_server(self): + async def start_fake_server(self, api_version=2): info = await self.server.start() resolver = FakeResolver(info, loop=self.loop) connector = aiohttp.TCPConnector(loop=self.loop, resolver=resolver) _session = aiohttp.ClientSession(connector=connector, loop=self.loop) - self.request = AioRequest(FAKE_BASE_URL, loop=self.loop, - websession=_session, timeout=1) + self.request = AioRequest( + FAKE_BASE_URL, loop=self.loop, websession=_session, timeout=1 + ) + self.request.api_version = api_version def make_url(end_point): - return 'http://{}/{}'.format(FAKE_BASE_URL, end_point) + return "http://{}/{}".format(FAKE_BASE_URL, end_point) diff --git a/tests/test_apiresource.py b/tests/test_apiresource.py index 3510b81..ba3ae91 100644 --- a/tests/test_apiresource.py +++ b/tests/test_apiresource.py @@ -15,6 +15,7 @@ def get_resource(self): """Get the resource being tested.""" _request = Mock() _request.hub_ip = FAKE_BASE_URL + _request.api_version = 2 return ApiResource(_request, "base", self.get_resource_raw_data()) def setUp(self): @@ -30,7 +31,7 @@ def test_id_property(self): def test_full_path(self): self.assertEqual( - self.resource._base_path, "http://{}/base".format(FAKE_BASE_URL) + self.resource._base_path, "http://{}/api/base".format(FAKE_BASE_URL) ) def test_name_property(self): @@ -52,6 +53,20 @@ def test_raw_data_property(self): data["name"] = "no name" self.assertEqual({"name": "name"}, self.resource.raw_data) + +class TestApiResource_V3(TestApiResource): + def get_resource(self): + """Get the resource being tested.""" + _request = Mock() + _request.hub_ip = FAKE_BASE_URL + _request.api_version = 3 + return ApiResource(_request, "base", self.get_resource_raw_data()) + + def test_full_path(self): + self.assertEqual( + self.resource._base_path, "http://{}/home/base".format(FAKE_BASE_URL) + ) + # def test_delete_200(self, mocked): # """Test delete resources with status 200.""" # mocked.delete(self.get_resource_uri(), @@ -168,5 +183,9 @@ def test_raw_data_property(self): def test_clean_names(): req = Mock() req.hub_ip = "123.123.123" + req.api_version = 2 api = ApiEntryPoint(req, "abc") - clean = api._sanitize_resources(test_data1) + try: + api._sanitize_resources(test_data1) + except NotImplementedError: + pass diff --git a/tests/test_hub.py b/tests/test_hub.py index 46b2848..2c9d6cc 100644 --- a/tests/test_hub.py +++ b/tests/test_hub.py @@ -1,16 +1,18 @@ -import asyncio from unittest.mock import MagicMock +import json import pytest from aiopvapi.helpers.aiorequest import AioRequest from aiopvapi.hub import Version, Hub from tests.test_scene_members import AsyncMock +from tests.fake_server import TestFakeServer, FAKE_BASE_URL, USER_DATA_VALUE @pytest.fixture def fake_aiorequest(): - request = AioRequest('127.0.0.1', websession=MagicMock()) + request = AioRequest("127.0.0.1", websession=MagicMock()) + request.api_version = 2 request.get = AsyncMock() request.put = AsyncMock() return request @@ -30,19 +32,53 @@ def test_version(): assert not version1 == version3 - version1 = Version('abc', 'def', 'ghi') - version2 = Version('abc', 'def', 'ghi') + version1 = Version("abc", "def", "ghi") + version2 = Version("abc", "def", "ghi") assert version1 == version2 -def test_hub(fake_aiorequest): - hub = Hub(fake_aiorequest) - assert hub._base_path == 'http://127.0.0.1/api' - loop = asyncio.get_event_loop() +class TestHub_v2(TestFakeServer): + def test_hub_init(self): + async def go(): + await self.start_fake_server() + hub = Hub(self.request) + await hub.query_firmware() + return hub - fake_aiorequest.get = AsyncMock(return_value={}) + hub = self.loop.run_until_complete(go()) - loop.run_until_complete(hub.query_firmware()) + assert hub._base_path == "http://" + FAKE_BASE_URL + "/api" - hub.request.get.mock.assert_called_once_with( - 'http://127.0.0.1/api/fwversion') + # self.request.get.mock.assert_called_once_with(FAKE_BASE_URL + "/userdata") + data = json.loads(USER_DATA_VALUE) + + assert hub.main_processor_info == data["userData"]["firmware"]["mainProcessor"] + assert hub.main_processor_version == "BUILD: 395 REVISION: 2 SUB_REVISION: 0" + assert hub.radio_version == "BUILD: 1307 REVISION: 2 SUB_REVISION: 0" + assert hub.ssid == "cisco789" + assert hub.name == "Hubby" + + +class TestHub_v3(TestFakeServer): + def __init__(self, methodName: str = "runTest") -> None: + super().__init__(methodName) + self.api_version = 3 + + def test_hub_init(self): + async def go(): + await self.start_fake_server(api_version=3) + hub = Hub(self.request) + await hub.query_firmware() + return hub + + hub = self.loop.run_until_complete(go()) + + assert hub._base_path == "http://" + FAKE_BASE_URL + "/gateway" + + # self.request.get.mock.assert_called_once_with(FAKE_BASE_URL + "/userdata") + data = json.loads(USER_DATA_VALUE) + + assert hub.main_processor_info == data["userData"]["firmware"]["mainProcessor"] + assert hub.main_processor_version == "BUILD: 395 REVISION: 2 SUB_REVISION: 0" + assert hub.radio_version == ["BUILD: 1307 REVISION: 2 SUB_REVISION: 0"] + assert hub.ssid == "cisco789" diff --git a/tests/test_room.py b/tests/test_room.py index 2bbca8f..718b4ab 100644 --- a/tests/test_room.py +++ b/tests/test_room.py @@ -5,30 +5,37 @@ from tests.fake_server import FAKE_BASE_URL from tests.test_apiresource import TestApiResource -ROOM_RAW_DATA = {"order": 2, "name": "RGluaW5nIFJvb20=", - "colorId": 0, "iconId": 0, "id": 26756, "type": 0} +ROOM_RAW_DATA = { + "order": 2, + "name": "RGluaW5nIFJvb20=", + "colorId": 0, + "iconId": 0, + "id": 26756, + "type": 0, +} class TestRoom(TestApiResource): - def get_resource_raw_data(self): return ROOM_RAW_DATA def get_resource_uri(self): - return 'http://{}/api/rooms/26756'.format(FAKE_BASE_URL) + return "http://{}/api/rooms/26756".format(FAKE_BASE_URL) def get_resource(self): _request = Mock() _request.hub_ip = FAKE_BASE_URL + _request.api_version = 2 return Room(ROOM_RAW_DATA, _request) def test_full_path(self): - self.assertEqual(self.resource._base_path, - 'http://{}/api/rooms'.format(FAKE_BASE_URL)) + self.assertEqual( + self.resource._base_path, "http://{}/api/rooms".format(FAKE_BASE_URL) + ) def test_name_property(self): # No name_unicode, so base64 encoded is returned - self.assertEqual('RGluaW5nIFJvb20=', self.resource.name) + self.assertEqual("RGluaW5nIFJvb20=", self.resource.name) def test_delete_room_success(self): """Tests deleting a room""" @@ -48,7 +55,7 @@ def test_delete_room_fail(self): async def go(): await self.start_fake_server() room = Room(self.get_resource_raw_data(), self.request) - room._resource_path += '1' + room._resource_path += "1" resp = await room.delete() return resp diff --git a/tests/test_scene.py b/tests/test_scene.py index 3bbbfb2..4ee0248 100644 --- a/tests/test_scene.py +++ b/tests/test_scene.py @@ -26,6 +26,7 @@ def get_resource_uri(self): def get_resource(self): _request = Mock() _request.hub_ip = FAKE_BASE_URL + _request.api_version = 2 return Scene(SCENE_RAW_DATA, _request) def test_name_property(self): diff --git a/tests/test_shade.py b/tests/test_shade.py index ca3b494..c7147d6 100644 --- a/tests/test_shade.py +++ b/tests/test_shade.py @@ -2,48 +2,120 @@ from aiopvapi.helpers.aiorequest import AioRequest from aiopvapi.resources.shade import BaseShade, shade_type +from aiopvapi.helpers.constants import ( + ATTR_POSKIND1, + ATTR_POSITION1, + ATTR_POSITION2, + ATTR_POSKIND2, + ATTR_PRIMARY, + ATTR_TILT, + MAX_POSITION, + MID_POSITION, + MID_POSITION_V2, + MAX_POSITION_V2, +) + from tests.fake_server import FAKE_BASE_URL from tests.test_apiresource import TestApiResource -SHADE_RAW_DATA = {"id": 29889, "type": 6, "batteryStatus": 0, - "batteryStrength": 0, - "name": "UmlnaHQ=", "roomId": 12372, - "groupId": 18480, - "positions": {"posKind1": 1, "position1": 0}, - "firmware": {"revision": 1, "subRevision": 8, "build": 1944}} + +SHADE_RAW_DATA = { + "id": 29889, + "type": 6, + "batteryStatus": 0, + "batteryStrength": 0, + "name": "UmlnaHQ=", + "roomId": 12372, + "groupId": 18480, + "positions": {"posKind1": 1, "position1": 0}, + "firmware": {"revision": 1, "subRevision": 8, "build": 1944}, +} class TestShade(TestApiResource): + def get_resource_raw_data(self): + return SHADE_RAW_DATA + + def get_resource_uri(self): + return "http://{}/api/shades/29889".format(FAKE_BASE_URL) + + def get_resource(self): + _request = Mock(spec=AioRequest) + _request.hub_ip = FAKE_BASE_URL + _request.api_version = 2 + return BaseShade(SHADE_RAW_DATA, shade_type(0, ""), _request) + + def test_full_path(self): + self.assertEqual( + self.resource._base_path, "http://{}/api/shades".format(FAKE_BASE_URL) + ) + + def test_name_property(self): + # No name_unicode, so base64 encoded is returned + self.assertEqual("UmlnaHQ=", self.resource.name) + + def test_add_shade_to_room(self): + async def go(): + await self.start_fake_server() + shade = BaseShade({"id": 111}, shade_type(0, ""), self.request) + res = await shade.add_shade_to_room(123) + return res + + resource = self.loop.run_until_complete(go()) + self.assertTrue(resource["shade"]) + + def test_convert_g2(self): + shade = self.get_resource() + self.assertEqual( + shade.convert_to_v2({ATTR_PRIMARY: MAX_POSITION}), + {ATTR_POSITION1: MAX_POSITION_V2, ATTR_POSKIND1: 1}, + ) + self.assertEqual( + shade.convert_to_v2({ATTR_TILT: MID_POSITION}), + {ATTR_POSITION1: MID_POSITION_V2, ATTR_POSKIND1: 3}, + ) + self.assertEqual( + shade.convert_to_v2({ATTR_PRIMARY: MAX_POSITION, ATTR_TILT: MID_POSITION}), + { + ATTR_POSKIND1: 1, + ATTR_POSITION1: MAX_POSITION_V2, + ATTR_POSKIND2: 3, + ATTR_POSITION2: MID_POSITION_V2, + }, + ) + +class TestShade_V3(TestApiResource): def get_resource_raw_data(self): return SHADE_RAW_DATA def get_resource_uri(self): - return 'http://{}/api/shades/29889'.format(FAKE_BASE_URL) + return "http://{}/home/shades/29889".format(FAKE_BASE_URL) def get_resource(self): _request = Mock(spec=AioRequest) _request.hub_ip = FAKE_BASE_URL - return BaseShade(SHADE_RAW_DATA, shade_type(0, ''), _request) + _request.api_version = 3 + return BaseShade(SHADE_RAW_DATA, shade_type(0, ""), _request) def test_full_path(self): self.assertEqual( - self.resource._base_path, - 'http://{}/api/shades'.format(FAKE_BASE_URL)) + self.resource._base_path, "http://{}/home/shades".format(FAKE_BASE_URL) + ) def test_name_property(self): # No name_unicode, so base64 encoded is returned - self.assertEqual('UmlnaHQ=', self.resource.name) + self.assertEqual("UmlnaHQ=", self.resource.name) def test_add_shade_to_room(self): async def go(): await self.start_fake_server() - shade = BaseShade({'id': 111}, shade_type(0, ''), self.request) + shade = BaseShade({"id": 111}, shade_type(0, ""), self.request) res = await shade.add_shade_to_room(123) return res resource = self.loop.run_until_complete(go()) - self.assertTrue(resource['shade']) + self.assertTrue(resource["shade"]) # def test_open(self): # mocked.put('http://127.0.0.1/api/shades/29889',