Initial support for Parkland/TypeID 39 shades (#12)

* disabled home_id filter

* stronger typing

* Update .gitignore

* extend set_position() for all values

* support setting tilt

* completed tilt control

* add tilt functions

* Initial support for Parkland/TypeID 39 shades

* Update cover.py

* Update README.md

* simplify set_position()

* fix spelling issues

* Update .gitignore

* remove unverified shade types

* clean code

* fix ruff

* update linting

* Update lint.yml

---------

Co-authored-by: patman15 <14628713+patman15@users.noreply.github.com>
This commit is contained in:
Dustin Brewer
2025-12-30 01:00:39 -08:00
committed by GitHub
parent b558083b50
commit 5d498b8753
8 changed files with 200 additions and 30 deletions

View File

@@ -8,8 +8,8 @@ on:
- cron: '0 5 * * 6' - cron: '0 5 * * 6'
jobs: jobs:
ruff: lint:
name: "Ruff" name: "Lint the code"
runs-on: "ubuntu-latest" runs-on: "ubuntu-latest"
steps: steps:
- name: "Checkout the repository" - name: "Checkout the repository"
@@ -24,5 +24,11 @@ jobs:
- name: "Install requirements" - name: "Install requirements"
run: python3 -m pip install -r requirements.txt run: python3 -m pip install -r requirements.txt
- name: "Run Ruff" - name: "Run ruff"
run: python3 -m ruff check . run: ruff check .
- name: "Run mypy"
run: mypy .
- name: "Run codespell"
run: codespell -L hass

View File

@@ -25,6 +25,7 @@ Type* | Description
10 | Duette and Applause SkyLift 10 | Duette and Applause SkyLift
19 | Provenance Woven Wood 19 | Provenance Woven Wood
31, 32, 84 | Vignette 31, 32, 84 | Vignette
39 | Parkland
42 | M25T Roller Blind 42 | M25T Roller Blind
49 | AC Roller 49 | AC Roller
52 | Banded Shades 52 | Banded Shades
@@ -85,6 +86,9 @@ In case you have severe troubles,
- disable the log (Home Assistant will prompt you to download the log), and finally - disable the log (Home Assistant will prompt you to download the log), and finally
- [open an issue](https://github.com/patman15/hdpv_ble/issues/new?assignees=&labels=Bug&projects=&template=bug.yml) with a good description of what happened and attach the log. - [open an issue](https://github.com/patman15/hdpv_ble/issues/new?assignees=&labels=Bug&projects=&template=bug.yml) with a good description of what happened and attach the log.
# Thanks To
[@mannkind](https://github.com/mannkind)
[license-shield]: https://img.shields.io/github/license/patman15/hdpv_ble.svg?style=for-the-badge [license-shield]: https://img.shields.io/github/license/patman15/hdpv_ble.svg?style=for-the-badge
[releases-shield]: https://img.shields.io/github/release/patman15/hdpv_ble.svg?style=for-the-badge [releases-shield]: https://img.shields.io/github/release/patman15/hdpv_ble.svg?style=for-the-badge
[releases]: https://github.com//patman15/hdpv_ble/releases [releases]: https://github.com//patman15/hdpv_ble/releases

View File

@@ -13,10 +13,15 @@ from homeassistant.const import Platform
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
from .const import DOMAIN, LOGGER from .const import LOGGER
from .coordinator import PVCoordinator from .coordinator import PVCoordinator
PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.COVER, Platform.SENSOR, Platform.BUTTON] PLATFORMS: list[Platform] = [
Platform.BINARY_SENSOR,
Platform.COVER,
Platform.SENSOR,
Platform.BUTTON,
]
type ConfigEntryType = ConfigEntry[PVCoordinator] type ConfigEntryType = ConfigEntry[PVCoordinator]
@@ -43,8 +48,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntryType) -> bool
except BleakError as err: except BleakError as err:
raise ConfigEntryNotReady("Unable to query device info.") from err raise ConfigEntryNotReady("Unable to query device info.") from err
# Insert the coordinator in the global registry
hass.data.setdefault(DOMAIN, {})
entry.runtime_data = coordinator entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(coordinator.async_start()) entry.async_on_unload(coordinator.async_start())

View File

@@ -17,7 +17,10 @@ from cryptography.hazmat.primitives.ciphers.base import (
AEADEncryptionContext, AEADEncryptionContext,
) )
from homeassistant.components.cover import ATTR_CURRENT_POSITION from homeassistant.components.cover import (
ATTR_CURRENT_POSITION,
ATTR_CURRENT_TILT_POSITION,
)
from .const import LOGGER, TIMEOUT from .const import LOGGER, TIMEOUT
@@ -48,7 +51,11 @@ SHADE_TYPE: Final[dict[int, str]] = {
8: "Duette, Top Down Bottom Up", 8: "Duette, Top Down Bottom Up",
9: "Duette DuoLite, Top Down Bottom Up", 9: "Duette DuoLite, Top Down Bottom Up",
33: "Duette Architella, Top Down Bottom Up", 33: "Duette Architella, Top Down Bottom Up",
39: "Parkland",
47: "Pleated, Top Down Bottom Up", 47: "Pleated, Top Down Bottom Up",
# top down, tilt anywhere
51: "Venetian, Tilt Anywhere",
62: "Venetian, Tilt Anywhere",
} }
OPEN_POSITION: Final[int] = 100 OPEN_POSITION: Final[int] = 100
@@ -138,9 +145,7 @@ class PowerViewBLE:
return self._client.is_connected return self._client.is_connected
# general cmd: uint16_t cmd, uint8_t seqID, uint8_t data_len # general cmd: uint16_t cmd, uint8_t seqID, uint8_t data_len
async def _cmd( async def _cmd(self, cmd: tuple[ShadeCmd, bytes], disconnect: bool = True) -> None:
self, cmd: tuple[ShadeCmd, bytes], disconnect: bool = True
) -> None:
self._cmd_next = cmd self._cmd_next = cmd
if self._cmd_lock.locked(): if self._cmd_lock.locked():
LOGGER.debug("%s: device busy, queuing %s command", self.name, cmd[0]) LOGGER.debug("%s: device busy, queuing %s command", self.name, cmd[0])
@@ -182,13 +187,13 @@ class PowerViewBLE:
if len(data) != 9: if len(data) != 9:
LOGGER.debug("not a V2 record!") LOGGER.debug("not a V2 record!")
return [] return []
pos: int = int.from_bytes(data[3:5], byteorder="little") pos: Final[int] = int.from_bytes(data[3:5], byteorder="little")
pos2: int = (int(data[5]) << 4) + (int(data[4]) >> 4) pos2: Final[int] = (int(data[5]) << 4) + (int(data[4]) >> 4)
return [ return [
(ATTR_CURRENT_POSITION, ((pos >> 2) / 10)), (ATTR_CURRENT_POSITION, ((pos >> 2) / 10)),
("position2", pos2 >> 2), ("position2", pos2 >> 2),
("position3", int(data[6])), ("position3", int(data[6])),
("tilt", int(data[7])), (ATTR_CURRENT_TILT_POSITION, int(data[7])),
("home_id", int.from_bytes(data[0:2], byteorder="little")), ("home_id", int.from_bytes(data[0:2], byteorder="little")),
("type_id", int(data[2])), ("type_id", int(data[2])),
("is_opening", bool(pos & 0x3 == 0x2)), ("is_opening", bool(pos & 0x3 == 0x2)),
@@ -200,16 +205,33 @@ class PowerViewBLE:
] ]
# position cmd: uint16_t pos1, uint16_t pos2, uint16_t pos3, uint16_t tilt, uint8_t velocity # position cmd: uint16_t pos1, uint16_t pos2, uint16_t pos3, uint16_t tilt, uint8_t velocity
async def set_position(self, value: int, disconnect: bool = True) -> None: async def set_position(
self,
pos1: int,
pos2: int = 0x8000,
pos3: int = 0x8000,
tilt: int = 0x8000,
velocity: int = 0x0,
disconnect: bool = True,
) -> None:
"""Set position of device.""" """Set position of device."""
LOGGER.debug("%s setting position to %i", self.name, value) LOGGER.debug(
"%s setting position to %i/%i/%i, tilt %i, velocity %s",
self.name,
pos1,
pos2,
pos3,
tilt,
velocity,
)
await self._cmd( await self._cmd(
( (
ShadeCmd.SET_POSITION, ShadeCmd.SET_POSITION,
bytes( int.to_bytes(pos1, 2, byteorder="little")
int.to_bytes(value * 100, 2, byteorder="little") + int.to_bytes(pos2, 2, byteorder="little")
+ bytes([0x00, 0x80, 0x00, 0x80, 0x00, 0x80, 0x0]) + int.to_bytes(pos3, 2, byteorder="little")
), + int.to_bytes(tilt, 2, byteorder="little")
+ int.to_bytes(velocity, 1),
), ),
disconnect, disconnect,
) )
@@ -291,8 +313,8 @@ class PowerViewBLE:
.copy() .copy()
.decode("UTF-8") .decode("UTF-8")
) )
except Exception as ex: except BleakError as ex:
LOGGER.error("Error: %s - %s", type(ex).__name__, ex) LOGGER.debug("%s: querying failed: %s", self.name, ex)
raise raise
finally: finally:
await self.disconnect() await self.disconnect()
@@ -310,7 +332,11 @@ class PowerViewBLE:
if self._cipher is not None and self._is_encrypted: if self._cipher is not None and self._is_encrypted:
dec: AEADDecryptionContext = self._cipher.decryptor() dec: AEADDecryptionContext = self._cipher.decryptor()
self._data = bytes(dec.update(bytes(data)) + dec.finalize()) self._data = bytes(dec.update(bytes(data)) + dec.finalize())
LOGGER.debug("%s %s", "decoded data: ".rjust(19+len(self.name)), self._data.hex(" ")) LOGGER.debug(
"%s %s",
"decoded data: ".rjust(19 + len(self.name)),
self._data.hex(" "),
)
self._data_event.set() self._data_event.set()

View File

@@ -9,7 +9,9 @@ from homeassistant.components.bluetooth.passive_update_coordinator import (
) )
from homeassistant.components.cover import ( from homeassistant.components.cover import (
ATTR_CURRENT_POSITION, ATTR_CURRENT_POSITION,
ATTR_CURRENT_TILT_POSITION,
ATTR_POSITION, ATTR_POSITION,
ATTR_TILT_POSITION,
CoverDeviceClass, CoverDeviceClass,
CoverEntity, CoverEntity,
CoverEntityFeature, CoverEntityFeature,
@@ -32,11 +34,18 @@ async def async_setup_entry(
"""Set up the demo cover platform.""" """Set up the demo cover platform."""
coordinator: PVCoordinator = config_entry.runtime_data coordinator: PVCoordinator = config_entry.runtime_data
async_add_entities([PowerViewCover(coordinator)]) model: Final[str|None] = coordinator.dev_details.get("model")
entities: list[PowerViewCover] = []
if model in ["39"]:
entities.append(PowerViewCoverTiltOnly(coordinator))
else:
entities.append(PowerViewCover(coordinator))
async_add_entities(entities)
class PowerViewCover(PassiveBluetoothCoordinatorEntity[PVCoordinator], CoverEntity): # type: ignore[reportIncompatibleVariableOverride] class PowerViewCover(PassiveBluetoothCoordinatorEntity[PVCoordinator], CoverEntity): # type: ignore[reportIncompatibleVariableOverride]
"""Representation of a powerview shade.""" """Representation of a PowerView shade with Up/Down functionality only."""
_attr_has_entity_name = True _attr_has_entity_name = True
_attr_device_class = CoverDeviceClass.SHADE _attr_device_class = CoverDeviceClass.SHADE
@@ -52,8 +61,9 @@ class PowerViewCover(PassiveBluetoothCoordinatorEntity[PVCoordinator], CoverEnti
coordinator: PVCoordinator, coordinator: PVCoordinator,
) -> None: ) -> None:
"""Initialize the shade.""" """Initialize the shade."""
LOGGER.debug("%s: init() PowerViewCover", coordinator.name)
self._attr_name = CoverDeviceClass.SHADE self._attr_name = CoverDeviceClass.SHADE
self._coord = coordinator self._coord: PVCoordinator = coordinator
self._attr_device_info = self._coord.device_info self._attr_device_info = self._coord.device_info
self._target_position: int | None = round( self._target_position: int | None = round(
self._coord.data.get(ATTR_CURRENT_POSITION, OPEN_POSITION) self._coord.data.get(ATTR_CURRENT_POSITION, OPEN_POSITION)
@@ -171,3 +181,117 @@ class PowerViewCover(PassiveBluetoothCoordinatorEntity[PVCoordinator], CoverEnti
self.async_write_ha_state() self.async_write_ha_state()
except BleakError as err: except BleakError as err:
LOGGER.error("Failed to stop cover '%s': %s", self.name, err) LOGGER.error("Failed to stop cover '%s': %s", self.name, err)
class PowerViewCoverTilt(PowerViewCover):
"""Representation of a PowerView shade with additional tilt functionality."""
_attr_supported_features = (
CoverEntityFeature.OPEN
| CoverEntityFeature.CLOSE
| CoverEntityFeature.STOP
| CoverEntityFeature.SET_POSITION
| CoverEntityFeature.OPEN_TILT
| CoverEntityFeature.CLOSE_TILT
| CoverEntityFeature.STOP_TILT
| CoverEntityFeature.SET_TILT_POSITION
)
def __init__(
self,
coordinator: PVCoordinator,
) -> None:
"""Initialize the shade with tilt."""
LOGGER.debug("%s: init() PowerViewCoverTilt", coordinator.name)
super().__init__(coordinator)
@property
def current_cover_tilt_position(self) -> int | None: # type: ignore[reportIncompatibleVariableOverride]
"""Return current tilt of cover.
None is unknown
"""
pos: Final = self._coord.data.get(ATTR_CURRENT_TILT_POSITION)
return round(pos) if pos is not None else None
async def async_set_cover_tilt_position(self, **kwargs: Any) -> None:
"""Move the tilt to a specific position."""
if isinstance(target_position := kwargs.get(ATTR_TILT_POSITION), int):
LOGGER.debug("set cover tilt to position %i", target_position)
if (
self.current_cover_tilt_position == round(target_position)
or self.current_cover_position is None
):
return
try:
await self._coord.api.set_position(
self.current_cover_position, tilt=target_position
)
self.async_write_ha_state()
except BleakError as err:
LOGGER.error(
"Failed to tilt cover '%s' to %f%%: %s",
self.name,
target_position,
err,
)
async def async_stop_cover_tilt(self, **kwargs: Any) -> None:
"""Stop the cover."""
await self.async_stop_cover(kwargs=kwargs)
async def async_open_cover_tilt(self, **kwargs: Any) -> None:
"""Open the cover tilt."""
LOGGER.debug("open cover tilt")
_kwargs = {**kwargs, ATTR_TILT_POSITION: OPEN_POSITION}
await self.async_set_cover_tilt_position(**_kwargs)
async def async_close_cover_tilt(self, **kwargs: Any) -> None:
"""Close the cover tilt."""
LOGGER.debug("close cover tilt")
_kwargs = {**kwargs, ATTR_TILT_POSITION: CLOSED_POSITION}
await self.async_set_cover_tilt_position(**_kwargs)
class PowerViewCoverTiltOnly(PowerViewCoverTilt):
"""Representation of a PowerView shade with additional tilt functionality."""
OPENCLOSED_THRESHOLD = 5
_attr_device_class = CoverDeviceClass.BLIND
_attr_supported_features = (
CoverEntityFeature.OPEN_TILT
| CoverEntityFeature.CLOSE_TILT
| CoverEntityFeature.STOP_TILT
| CoverEntityFeature.SET_TILT_POSITION
)
def __init__(
self,
coordinator: PVCoordinator,
) -> None:
"""Initialize the shade with tilt only."""
LOGGER.debug("%s: init() PowerViewCoverTiltOnly", coordinator.name)
super().__init__(coordinator)
@property
def is_opening(self) -> bool | None: # type: ignore[reportIncompatibleVariableOverride]
"""Return if the cover is opening or not."""
return False
@property
def is_closing(self) -> bool | None: # type: ignore[reportIncompatibleVariableOverride]
"""Return if the cover is closing or not."""
return False
@property
def is_closed(self) -> bool: # type: ignore[reportIncompatibleVariableOverride]
"""Return if the cover is closed."""
return isinstance(self.current_cover_tilt_position, int) and (
self.current_cover_tilt_position
>= OPEN_POSITION - PowerViewCoverTiltOnly.OPENCLOSED_THRESHOLD
or self.current_cover_tilt_position
<= CLOSED_POSITION + PowerViewCoverTiltOnly.OPENCLOSED_THRESHOLD
)

View File

@@ -4,8 +4,7 @@
"bluetooth": [ "bluetooth": [
{ {
"service_uuid": "0000fdc1-0000-1000-8000-00805f9b34fb", "service_uuid": "0000fdc1-0000-1000-8000-00805f9b34fb",
"manufacturer_id": 2073, "manufacturer_id": 2073
"manufacturer_data_start": [0,0]
} }
], ],
"codeowners": ["@patman15"], "codeowners": ["@patman15"],

View File

@@ -2,6 +2,12 @@
* Emulate a Hunter Douglas PowerView cover device using ESP32 * Emulate a Hunter Douglas PowerView cover device using ESP32
* used e.g. to gain the home_key from an existing installation via BLE * used e.g. to gain the home_key from an existing installation via BLE
* *
* REQUIRES:
* - ESP32 Board Definitions 3.0.x (tested on 3.0.7)
* - WolfSSL 5.7.x (tested on 5.7.6)
* - Phone Region: Potentially an alternative region depending on the app response
* - e.g. To add Parkland shades in the US, phone region set to the UK temporarily
*
* TODO: * TODO:
* - cleanup code * - cleanup code
* - think about emulating a remote * - think about emulating a remote

View File

@@ -1,4 +1,6 @@
homeassistant==2025.11.0 homeassistant==2025.11.0
pip>=21.3.1 pip>=21.3.1
ruff>=0.9.1,<=0.15.0 ruff>=0.9.1,<=0.15.0
types-requests
mypy~=1.19.1
codespell