"""Shade class managing all shade types.""" import asyncio from dataclasses import dataclass import logging from typing import Any from aiopvapi.helpers.aiorequest import AioRequest, PvApiMaintenance from aiopvapi.helpers.api_base import ApiResource from aiopvapi.helpers.constants import ( ATTR_BATTERY_KIND, ATTR_CAPABILITIES, ATTR_ID, ATTR_POSITION1, ATTR_POSITION2, ATTR_POSITIONS, ATTR_POSKIND1, ATTR_POSKIND2, ATTR_POWER_TYPE, ATTR_PRIMARY, ATTR_ROOM_ID, ATTR_SECONDARY, ATTR_SHADE, ATTR_SIGNAL_STRENGTH, ATTR_SIGNAL_STRENGTH_MAX, ATTR_TILT, ATTR_TYPE, CLOSED_POSITION, CLOSED_POSITION_V2, FIRMWARE, FIRMWARE_BUILD, FIRMWARE_REVISION, FIRMWARE_SUB_REVISION, FUNCTION_SET_POWER, MAX_POSITION, MAX_POSITION_V2, MID_POSITION, MIN_POSITION, MOTION_CALIBRATE, MOTION_FAVORITE, MOTION_JOG, MOTION_STOP, MOTION_VELOCITY, POSITIONS_V2, POSITIONS_V3, POSKIND_PRIMARY, POSKIND_SECONDARY, POSKIND_TILT, POWER_SOURCE_HARDWIRED, POWERTYPE_BATTERY, POWERTYPE_HARDWIRED, POWERTYPE_MAP_V2, POWERTYPE_MAP_V3, POWERTYPE_RECHARGABLE, SHADE_BATTERY_STATUS, SHADE_BATTERY_STRENGTH, ) from aiopvapi.helpers.tools import deep_update_dict, join_path _LOGGER = logging.getLogger(__name__) @dataclass class PowerviewCapabilities: """Capabilities available from Powerview.""" primary: bool = False secondary: 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 light: bool = False @dataclass class ShadeLimits: """Limits of a shade.""" primary_min: int = MIN_POSITION primary_max: int = MAX_POSITION secondary_min: int = MIN_POSITION secondary_max: int = MAX_POSITION tilt_min: int = MIN_POSITION tilt_max: int = MAX_POSITION @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 @dataclass class ShadeType: """Shade information based on type and description.""" type: int | str description: str @dataclass class ShadeCapability: """Shade capability information.""" type: int | str capabilities: PowerviewCapabilities description: str class BaseShade(ApiResource): """Basic shade class.""" api_endpoint = "shades" 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() def __init__( self, raw_data: dict, shade_type: ShadeType, request: AioRequest ) -> None: """Initialize Base shade.""" self.shade_type = shade_type 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, ) if self.api_version == 2: return function in ( MOTION_JOG, MOTION_CALIBRATE, MOTION_FAVORITE, MOTION_STOP, FUNCTION_SET_POWER, ) 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) return self.get_additional_positions(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] revision = firmware[FIRMWARE_REVISION] sub_revision = firmware[FIRMWARE_SUB_REVISION] build = firmware[FIRMWARE_BUILD] return f"{revision}.{sub_revision}.{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.""" _LOGGER.warning( "Percent to API %s (%s): %s", self.name, position_type, 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 # this may not be needed, causing issues with MID_POSITION and working fine win 0/100 for all other positions # position = self.position_limit(position, position_type) position = self.position_limit(position) _LOGGER.warning( "Percent to position_limit %s (%s): %s", self.name, position_type, position ) # 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((position / max_position_api) * 100) return round(percent) def structured_to_raw(self, data: ShadePosition) -> dict[str, Any]: """Convert structured ShadePosition to API relevant dict.""" _LOGGER.warning("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.warning("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( float(position_data[position_key] or 0), 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( float(position_data[position_key] or 0), 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_POSITIONS] = position_data if room_id: base[ATTR_SHADE][ATTR_ROOM_ID] = room_id return base def _update_position_from_dict(self, updates: dict) -> None: updates = updates.get(ATTR_SHADE, updates) # Gen 2 position dict is embedded self._raw_data = deep_update_dict(self._raw_data, updates) 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") # store the requested position in the shade data response = await self.request.put(resource_path, data=position_data, params=params) self._update_position_from_dict(position_data) return response 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) # _LOGGER.warning("before call: %s", self._raw_data.get("positions")) data = self.structured_to_raw(position_data) # _LOGGER.warning("Shade %s move to: %s", self.name, data) await self._move(data) # _LOGGER.warning("after call: %s", self._raw_data.get("positions")) return self.current_position # async def move_and_wait(self, position_data: ShadePosition) -> ShadePosition: # """Move the shade to a set position, wait for the move to complete.""" # _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: """Return additional 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, 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)) if self.api_version < 3 and position != 0 and position < CLOSED_POSITION_V2: _LOGGER.debug( "%s: Assuming shade is closed as %s is less than %s", self.name, position, CLOSED_POSITION, ) position = CLOSED_POSITION 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: path = self._resource_path cmd = {"shade": {"motion": motion}} await self.request.put(path, cmd) async def jog(self): """Jog the shade.""" await self._motion(MOTION_JOG) async def calibrate(self): """Calibrate the shade.""" await self._motion(MOTION_CALIBRATE) async def favorite(self): """Move the shade to the defined favorite position.""" await self._motion(MOTION_FAVORITE) async def stop(self): """Stop the shade.""" 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, suppress_timeout: bool = False, **kwargs): """Query the hub and refresh the most recent position state. :param kwargs: Keyword arguments to be passed to the get request. For example, timeout can be passed as kwargs. """ try: _LOGGER.debug("Refreshing position of: %s", self.name) raw_data = await self.request.get( self._resource_path, {"refresh": "true"}, suppress_timeout=suppress_timeout, **kwargs, ) if raw_data is None: _LOGGER.debug("No update received for: %s", self.name) return # 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 # Encountering hub issues with failures (gen 2) # even the retry logic is not working, and just fails every time # TODO: Fix this, for now we're just using the normal refresh # async def refresh_with_retry( # self, suppress_timeout: bool = False, retries: int = 3, **kwargs # ): # """Query the hub and refresh the most recent position state, with retry logic. # :param suppress_timeout: Whether to suppress timeouts. # :param retries: Number of retry attempts for transient errors. # :param kwargs: Additional arguments for the request. # """ # for attempt in range(retries): # try: # _LOGGER.debug( # "Refreshing position of: %s (attempt %s/%s)", # self.name, # attempt + 1, # retries, # ) # if attempt >= 1: # _LOGGER.warning( # "Refreshing position of: %s (attempt %s/%s)", # self.name, # attempt + 1, # retries, # ) # raw_data = await self.request.get( # self._resource_path, # {"refresh": "true"}, # suppress_timeout=suppress_timeout, # **kwargs, # ) # if raw_data is None: # _LOGGER.warning("No update received for: %s", self.name) # return # # 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) # return # ✅ Success, exit after first successful call # except PvApiMaintenance: # _LOGGER.debug("Hub undergoing maintenance. Please try again") # return # except PvApiConnectionError as ex: # if attempt < retries - 1: # delay = 2**attempt # _LOGGER.warning( # "Hub connection issue for %s (attempt %s/%s). Retrying in %s seconds...", # self.name, # attempt + 1, # retries, # delay, # ) # await asyncio.sleep(delay) # else: # raise PvApiConnectionError( # f"PowerView hub unreachable after {retries} attempts: {ex}" # ) from ex async def refresh_battery(self, suppress_timeout: bool = False, **kwargs): """Query the hub and request the most recent battery state. :param kwargs: Keyword arguments to be passed to the get request. For example, timeout can be passed as kwargs. """ if not self.is_battery_powered: _LOGGER.debug("Shade %s is not battery powered", self.name) return try: # the refresh can sometimes first wake the shade, resulting in a timeout # retry to try and get a true value _LOGGER.debug("Refreshing battery of: %s", self.name) retries = 3 for attempt in range(retries): raw_data = await self.request.get( self._resource_path, {"updateBatteryLevel": "true"}, suppress_timeout=suppress_timeout, **kwargs, ) if raw_data is None: _LOGGER.debug("No update received for: %s", self.name) return # 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) _LOGGER.debug( "Shade battery %s %d: %s", self.name, attempt, self._raw_data ) if not self._raw_data.get("timedOut", False): _LOGGER.debug("Shade battery %s %d: Refreshed", self.name, attempt) break # timeout is false, so we're done if attempt < retries - 1: _LOGGER.debug( "Shade %s timed out, retrying in 2 minutes (attempt %d/%d)", self.name, attempt + 1, retries, ) await asyncio.sleep(120) else: _LOGGER.warning( "Shade battery refresh %s timed out after %d attempts", self.name, retries, ) return except PvApiMaintenance: _LOGGER.debug("Hub undergoing maintenance. Please try again") return 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 get_battery_info(self) -> int: """Return the battery powerType.""" attr = ATTR_POWER_TYPE if self.api_version >= 3 else ATTR_BATTERY_KIND return self.raw_data.get(attr) def is_battery_powered(self) -> bool: """Confirm if the shade is battery or hardwired.""" return bool(self.get_battery_info() not in POWER_SOURCE_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) _LOGGER.debug("%s: Mapping %s %s to %s", self.name, attr, 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 not self.is_supported(FUNCTION_SET_POWER): _LOGGER.error("Method not supported") return 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.""" if self.api_version < 3: # SHADE_BATTERY_STRENGTH is in tenths of a volt (e.g., 146 = 14.6V), max is 18.0V (180) # use min to ensure we don't exceed 100% when more than 18.0V is supplied return min(100, round((self.raw_data[SHADE_BATTERY_STRENGTH] / 180) * 100)) # gen 3 dont return the same information for batteries and # while gen 2 do support the below, it is less accurate than the above 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. :return: Dictionary with position data. """ if refresh: await self.refresh() return self._raw_data.get(ATTR_POSITIONS) 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 def __init__( self, raw_data: dict, shade_type: ShadeType, request: AioRequest ) -> None: """Initialize shade with tilt.""" 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 open position.""" return await self.tilt(position_data=self.open_position_tilt) 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: """Return additional positions not reported by the hub.""" if positions.primary is not None and positions.tilt is None: positions.tilt = MIN_POSITION elif positions.tilt is not None and positions.primary is None: positions.primary = MIN_POSITION return positions class ShadeBottomUp(BaseShade): """Type 0 - Up Down Only. A simple open/close shade. """ shade_types = ( ShadeType(1, "Designer Roller"), ShadeType(4, "Roman"), ShadeType(5, "Bottom Up"), ShadeType(6, "Duette"), ShadeType(10, "Duette and Applause SkyLift"), ShadeType(19, "Provenance Woven Wood"), ShadeType(31, "Vignette"), ShadeType(32, "Vignette"), ShadeType(42, "M25T Roller Blind"), ShadeType(49, "AC Roller"), ShadeType(52, "Banded Shades"), ShadeType(53, "Sonnette"), ShadeType(57, "Carole Roman Shades"), ShadeType(84, "Vignette"), ) capability = ShadeCapability( 0, PowerviewCapabilities( primary=True, ), "Bottom Up", ) def __init__( self, raw_data: dict, shade_type: ShadeType, request: AioRequest ) -> None: """Initialize Standard Bottom Up shade.""" super().__init__(raw_data, shade_type, request) self._open_position = ShadePosition(primary=MAX_POSITION) self._close_position = ShadePosition(primary=MIN_POSITION) # self.shade_limits = ShadeLimits(primary_max=MID_POSITION) class ShadeBottomUpTiltOnClosed180(BaseShadeTilt): """Type 0 - Up Down tiltOnClosed 180°. A shade with move and tilt at when closed capabilities. These are believed to be an oversight by the HD Powerview team and the only model without a distinct capability code. """ 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 = ShadeCapability( 1, PowerviewCapabilities( primary=True, tilt_onclosed=True, tilt_180=True, ), "Bottom Up TiltOnClosed 180°", ) def __init__( self, raw_data: dict, shade_type: ShadeType, request: AioRequest ) -> None: """Initialize shade with tilt on closed functions.""" 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): """Type 1 - Up Down tiltOnClosed 90°. A shade with move and tilt at bottom capabilities with only a 90° tilt. """ shade_types = ( ShadeType(18, "Pirouette"), ShadeType(23, "Silhouette"), ShadeType(43, "Facette"), ShadeType(72, "Silhouette"), ) capability = ShadeCapability( 1, PowerviewCapabilities( primary=True, tilt_onclosed=True, tilt_90=True, ), "Bottom Up TiltOnClosed 90°", ) def __init__( self, raw_data: dict, shade_type: ShadeType, request: AioRequest ) -> None: """Initialize shade with tilt on closed functions.""" 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): """Type 2 - Up Down tiltAnywhere 180°. A shade with move and tilt anywhere capabilities. """ shade_types = ( ShadeType(51, "Venetian, Tilt Anywhere"), ShadeType(62, "Venetian, Tilt Anywhere"), ) capability = ShadeCapability( 2, PowerviewCapabilities( primary=True, tilt_anywhere=True, tilt_180=True, ), "Bottom Up TiltAnywhere 180°", ) def __init__( self, raw_data: dict, shade_type: ShadeType, request: AioRequest ) -> None: """Initialize shade with tilt anywhere.""" 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=MIN_POSITION ) self._open_position_tilt = ShadePosition(tilt=MID_POSITION) class ShadeVertical(ShadeBottomUp): """Type 3 - Vertical Open Close. A vertical shade with open/close only Same capabilities as type 0 (no tilt) but vertical. """ shade_types = ( 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 = ShadeCapability( 3, PowerviewCapabilities( primary=True, vertical=True, ), "Vertical", ) class ShadeVerticalTiltAnywhere(ShadeBottomUpTiltAnywhere): """Type 4 - Vertical tiltAnywhere 180°. A vertical shade with open/close and tilt anywhere Same capabilities as type 2 but vertical. """ shade_types = ( ShadeType(54, "Vertical Slats, Left Stack"), ShadeType(55, "Vertical Slats, Right Stack"), ShadeType(56, "Vertical Slats, Split Stack"), ) capability = ShadeCapability( 4, PowerviewCapabilities( primary=True, tilt_anywhere=True, tilt_180=True, vertical=True, ), "Vertical Tilt Anywhere", ) def get_additional_positions(self, positions: ShadePosition) -> ShadePosition: """Return additional positions not reported by the hub.""" if positions.primary is None: positions.primary = MIN_POSITION if positions.tilt is None: positions.tilt = MIN_POSITION return positions class ShadeTiltOnly(BaseShadeTilt): """Type 5 - Tilt Only 180°. A shade with tilt anywhere capabilities only. """ shade_types = ( ShadeType(40, "Everwood Alternative Wood Blinds"), ShadeType(66, "Palm Beach Shutters"), ) capability = ShadeCapability( 5, PowerviewCapabilities( tilt_anywhere=True, tilt_180=True, ), "Tilt Only 180°", ) def __init__( self, raw_data: dict, shade_type: ShadeType, request: AioRequest ) -> None: """Initialize shade with tilt only.""" super().__init__(raw_data, shade_type, request) self._open_position = ShadePosition() self._close_position = ShadePosition() self._open_position_tilt = ShadePosition(tilt=MID_POSITION) self._close_position_tilt = ShadePosition(tilt=MIN_POSITION) def get_additional_positions(self, positions: ShadePosition) -> ShadePosition: """Return additional positions not reported by the hub.""" # bug where tilt only return posKind1=1 # https://github.com/home-assistant/core/issues/115257 if positions.primary is not None: positions.tilt = positions.primary positions.primary = None return positions class ShadeTopDown(BaseShade): """Type 6 - Top Down Only. A shade with top down capabilities only. """ shade_types = (ShadeType(7, "Top Down"),) capability = ShadeCapability( 6, PowerviewCapabilities( primary=True, primary_inverted=True, ), "Top Down", ) def __init__( self, raw_data: dict, shade_type: ShadeType, request: AioRequest ) -> None: """Initialize shade with top down only.""" super().__init__(raw_data, shade_type, request) self._open_position = ShadePosition(primary=MIN_POSITION) self._close_position = ShadePosition(primary=MAX_POSITION) class ShadeTopDownBottomUp(BaseShade): """Type 7 - Top Down Bottom Up. A shade with top down bottom up capabilities. """ shade_types = ( 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 = ShadeCapability( 7, PowerviewCapabilities( primary=True, secondary=True, ), "Top Down Bottom Up", ) def __init__( self, raw_data: dict, shade_type: ShadeType, request: AioRequest ) -> None: """Initialize shade with top down bottom up.""" 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: """Return additional 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): """Type 8 - Dual Shade Overlapped. A shade with a front sheer and rear blackout shade. """ shade_types = ( ShadeType(65, "Vignette Duolite"), ShadeType(79, "Duolite Lift"), ) capability = ShadeCapability( 8, PowerviewCapabilities( primary=True, secondary=True, secondary_overlapped=True, ), "Dual Shade Overlapped", ) def __init__( self, raw_data: dict, shade_type: ShadeType, request: AioRequest ) -> None: """Initialize shade with sheer front and rear blockout.""" 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: """Return additional positions not reported by the hub.""" if positions.primary is not None: if positions.secondary is None: positions.secondary = MAX_POSITION if positions.tilt is None: positions.tilt = MIN_POSITION elif positions.secondary is not None: if positions.primary is None: positions.primary = MIN_POSITION if positions.tilt is None: positions.tilt = MIN_POSITION elif positions.tilt is not None: if positions.primary is None: positions.primary = MIN_POSITION if positions.secondary is None: positions.secondary = MAX_POSITION return positions class ShadeDualOverlappedTilt90(BaseShadeTilt): """Type 9 - Dual Shade Overlapped with tiltOnClosed. A shade with a front sheer and rear blackout shade. Tilt on these is unique in that it requires the rear shade open and front shade closed. """ shade_types = (ShadeType(38, "Silhouette Duolite"),) capability = ShadeCapability( 9, PowerviewCapabilities( primary=True, secondary=True, secondary_overlapped=True, tilt_90=True, tilt_onclosed=True, ), "Dual Shade Overlapped Tilt 90°", ) def __init__( self, raw_data: dict, shade_type: ShadeType, request: AioRequest ) -> None: """Initialize shade with sheer front and rear blockout + tilt.""" 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: """Return additional positions not reported by the hub.""" if positions.primary is not None: if positions.secondary is None: positions.secondary = MAX_POSITION if positions.tilt is None: positions.tilt = MIN_POSITION elif positions.secondary is not None: if positions.primary is None: positions.primary = MIN_POSITION if positions.tilt is None: positions.tilt = MIN_POSITION elif positions.tilt is not None: if positions.primary is None: positions.primary = MIN_POSITION if positions.secondary is None: positions.secondary = MAX_POSITION return positions class ShadeDualOverlappedTilt180(ShadeDualOverlappedTilt90): """Type 10 - Dual Shade Overlapped with tiltOnClosed. A shade with a front sheer and rear blackout shade. Tilt on these is unique in that it requires the rear shade open and front shade closed. """ shade_types = () capability = ShadeCapability( 10, PowerviewCapabilities( primary=True, secondary=True, secondary_overlapped=True, tilt_180=True, tilt_onclosed=True, ), "Dual Shade Overlapped Tilt 180°", ) def __init__( self, raw_data: dict, shade_type: ShadeType, request: AioRequest ) -> None: """Initialize shade with sheer front and rear blockout + tilt.""" super().__init__(raw_data, shade_type, request) if self.api_version < 3: self.shade_limits = ShadeLimits(tilt_max=MAX_POSITION) class ShadeDualOverlappedIlluminated(ShadeDualOverlapped): """Type 11 - Illuminated Shades. A shade with a front sheer and rear blackout shade, plus an embedded light. """ shade_types = ( ShadeType(95, "Aura Illuminated, Roller"), # TODO: Capabilites 11 (light) ) capability = ShadeCapability( 11, PowerviewCapabilities( primary=True, secondary=True, secondary_overlapped=True, light=True, ), "Illuminated Shades", ) def factory(raw_data: dict, request: AioRequest): """Class factory to create different shade types.""" 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, ShadeDualOverlappedIlluminated, ] for cls in classes: # class check is more concise as we have tested positioning _shade = find_type(cls) if _shade: _LOGGER.debug("Shade match on 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("Shade match on capability: %s - %s", _shade, raw_data) return _shade _LOGGER.warning( "Shade type not found. Falling back to basic bottom up capabilities: %s - %s", BaseShade, raw_data, ) return BaseShade(raw_data, BaseShade.shade_types[0], request)