Add Motionblinds BLE integration (#109497)

* initial fork

* intial tests

* Initial test coverage

* extra coverage

* complete config flow tests

* fix generated

* Update CODEOWNERS

* Move logic to PyPi library and update to pass config_flow test and pre-commit

* Remove Button, Select and Sensor platform for initial PR

* Update manifest.json

* Change info logs to debug in cover

* Use _abort_if_unique_id_configured instead of custom loop checking existing entries

* Change platforms list to PLATFORMS global

Co-authored-by: Josef Zweck <24647999+zweckj@users.noreply.github.com>

* Remove VERSION from ConfigFlow

Co-authored-by: Josef Zweck <24647999+zweckj@users.noreply.github.com>

* Replace all info logs by debug

* Use instance attributes in ConfigFlow

Co-authored-by: Josef Zweck <24647999+zweckj@users.noreply.github.com>

* Add return type and docstring to init in ConfigFlow

* Add recovery to tests containing errors

* Make NoBluetoothAdapter and NoDevicesFound abort instead of show error

* Change info logs to debug

* Add and change integration type from hub to device

* Use CONF_ADDRESS from homeassistant.const

* Move cover attributes initialization out of constructor

* Change CONF_ADDRESS in tests from const to homeassistant.const

* Remove unused part of tests

* Change 'not motion_device' to 'motion_device is None'

* Change _attr_connection_type to _connection_type

* Add connections to DeviceInfo

* Add model to DeviceInfo and change MotionBlindType values

* Remove identifiers from DeviceInfo

* Move constants from const to library

* Move calibration and running to library, re-add all platforms

* Remove platforms from init

* Remove button platform

* Remove select platform

* Remove sensor platform

* Bump motionblindsble to 0.0.4

* Remove closed, opening and closing attribute default values

Co-authored-by: Josef Zweck <24647999+zweckj@users.noreply.github.com>

* Remove CONFIG_SCHEMA from init

Co-authored-by: Josef Zweck <24647999+zweckj@users.noreply.github.com>

* Remove unused platform attributes and icons

* Re-add _attr_is_closed to GenericBlind to fix error

* Use entry.async_create_background_task for library instead of entry.async_create_task

* Move updating of position on disconnect to library

* Remove type hints, keep for _attr_is_closed

* Use DISPLAY_NAME constant from library for display name

* Add TYPE_CHECKING condition to assert in config_flow

* Re-add CONFIG_SCHEMA to __init__ to pass hassfest

* Change FlowResult type to ConfigFlowResult

* Fix import in tests

* Fix ruff import

* Fix tests by using value of enum

* Use lowercase name of MotionBlindType enum for data schema selector and translation

* Fix using name instead of value for MotionBlindType

* Improve position None handling

Co-authored-by: starkillerOG <starkiller.og@gmail.com>

* Improve tilt None handling

Co-authored-by: starkillerOG <starkiller.og@gmail.com>

* Change BLIND_TO_ENTITY_TYPE name

* Set entity name of cover to None and use DeviceInfo name

* Add base entity

* Move async_update to base entity

* Move unique ID with suffix to base class

* Add entity.py to .coveragerc

* Remove extra state attribute connection type

* Remove separate line hass.data.setdefault(DOMAIN, {})

* Remove use of field for key and translation_key in MotionCoverEntityDescription

* Remove entity translation with extra state_attributes from strings.json

* Use super().__init__(device, entry)

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* Change if block in async_update_running

* Use if blocks in async_update_position

* Add additional scanner check before show_form

* Remove default value of device_class in MotionCoverEntityDescription

* Fix entry.data[CONF_BLIND_TYPE] uppercase

* Fix device info model name

* Bump motionblindsble to 0.0.5

* Fix tests

* Move entity_description to MotionblindsBLEEntity

* Change double roller blind name

* Bump motionblindsble to 0.0.6

* Fix ruff

* Use status_query for async_update

* Bump motionblindsble to 0.0.7

* Change bluetooth local name

* Set kw_only=True for dataclass

* Change name of GenericBlind

* Change scanner_count conditional

* Wrap async_register_callback in entry.async_on_unload

* Bump motionblindsble to 0.0.8

* Use set_create_task_factory and set_call_later_factory

* Update bluetooth.py generated

* Simplify COVER_TYPES dictionary

* Move registering callbacks to async_added_to_hass

* Remove check for ATTR_POSITION and ATTR_TILT_POSITION in kwargs

* Add naming consistency for device and entry

* Use if block instead of ternary for _attr_unique_id

* Improve errors ternary in config_flow

* Use set instead of list for running_type

* Improve errors ternary in config_flow

* Remove init from MotionblindsBLECoverEntity and move debug log to async_added_to_hass

* Update debug log create cover

* Fix ruff

* Use identity check instead of equals

* Use identity check instead of equals

* Change MotionblindsBLECoverEntityDescription name

* Change debug log text

* Remove ATTR_CONNECTION from const

* Add types for variables in async_setup_entry

* Add types for variables in async_setup_entry

* Change PositionBlind class name to PositionCover etc

* Improve docstrings

* Improve docstrings

---------

Co-authored-by: starkillerOG <starkiller.og@gmail.com>
Co-authored-by: Josef Zweck <24647999+zweckj@users.noreply.github.com>
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

.coveragerc

@@ -806,6 +806,9 @@ omit =
     homeassistant/components/motion_blinds/cover.py
     homeassistant/components/motion_blinds/entity.py
     homeassistant/components/motion_blinds/sensor.py
+    homeassistant/components/motionblinds_ble/__init__.py
+    homeassistant/components/motionblinds_ble/cover.py
+    homeassistant/components/motionblinds_ble/entity.py
     homeassistant/components/motionmount/__init__.py
     homeassistant/components/motionmount/binary_sensor.py
     homeassistant/components/motionmount/entity.py

CODEOWNERS

@@ -841,6 +841,8 @@ build.json @home-assistant/supervisor
 /tests/components/mopeka/ @bdraco
 /homeassistant/components/motion_blinds/ @starkillerOG
 /tests/components/motion_blinds/ @starkillerOG
+/homeassistant/components/motionblinds_ble/ @LennP @jerrybboy
+/tests/components/motionblinds_ble/ @LennP @jerrybboy
 /homeassistant/components/motioneye/ @dermotduffy
 /tests/components/motioneye/ @dermotduffy
 /homeassistant/components/motionmount/ @RJPoelstra

homeassistant/components/motionblinds_ble/__init__.py

@@ -0,0 +1,104 @@
+"""Motionblinds BLE integration."""
+
+from __future__ import annotations
+
+from functools import partial
+import logging
+
+from motionblindsble.const import MotionBlindType
+from motionblindsble.crypt import MotionCrypt
+from motionblindsble.device import MotionDevice
+
+from homeassistant.components.bluetooth import (
+    BluetoothCallbackMatcher,
+    BluetoothChange,
+    BluetoothScanningMode,
+    BluetoothServiceInfoBleak,
+    async_ble_device_from_address,
+    async_register_callback,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_ADDRESS, Platform
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers import config_validation as cv
+from homeassistant.helpers.event import async_call_later
+from homeassistant.helpers.typing import ConfigType
+
+from .const import CONF_BLIND_TYPE, CONF_MAC_CODE, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+PLATFORMS: list[Platform] = [
+    Platform.COVER,
+]
+
+CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
+
+
+async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
+    """Set up Motionblinds BLE integration."""
+
+    _LOGGER.debug("Setting up Motionblinds BLE integration")
+
+    # The correct time is needed for encryption
+    _LOGGER.debug("Setting timezone for encryption: %s", hass.config.time_zone)
+    MotionCrypt.set_timezone(hass.config.time_zone)
+
+    return True
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Motionblinds BLE device from a config entry."""
+
+    _LOGGER.debug("(%s) Setting up device", entry.data[CONF_MAC_CODE])
+
+    ble_device = async_ble_device_from_address(hass, entry.data[CONF_ADDRESS])
+    device = MotionDevice(
+        ble_device if ble_device is not None else entry.data[CONF_ADDRESS],
+        blind_type=MotionBlindType[entry.data[CONF_BLIND_TYPE].upper()],
+    )
+
+    # Register Home Assistant functions to use in the library
+    device.set_create_task_factory(
+        partial(
+            entry.async_create_background_task,
+            hass=hass,
+            name=device.ble_device.address,
+        )
+    )
+    device.set_call_later_factory(partial(async_call_later, hass=hass))
+
+    # Register a callback that updates the BLEDevice in the library
+    @callback
+    def async_update_ble_device(
+        service_info: BluetoothServiceInfoBleak, change: BluetoothChange
+    ) -> None:
+        """Update the BLEDevice."""
+        _LOGGER.debug("(%s) New BLE device found", service_info.address)
+        device.set_ble_device(service_info.device, rssi=service_info.advertisement.rssi)
+
+    entry.async_on_unload(
+        async_register_callback(
+            hass,
+            async_update_ble_device,
+            BluetoothCallbackMatcher(address=entry.data[CONF_ADDRESS]),
+            BluetoothScanningMode.ACTIVE,
+        )
+    )
+
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = device
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    _LOGGER.debug("(%s) Finished setting up device", entry.data[CONF_MAC_CODE])
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload Motionblinds BLE device from a config entry."""
+
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

homeassistant/components/motionblinds_ble/config_flow.py

@@ -0,0 +1,214 @@
+"""Config flow for Motionblinds BLE integration."""
+
+from __future__ import annotations
+
+import logging
+import re
+from typing import TYPE_CHECKING, Any
+
+from bleak.backends.device import BLEDevice
+from motionblindsble.const import DISPLAY_NAME, MotionBlindType
+import voluptuous as vol
+
+from homeassistant.components import bluetooth
+from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_ADDRESS
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.selector import (
+    SelectSelector,
+    SelectSelectorConfig,
+    SelectSelectorMode,
+)
+
+from .const import (
+    CONF_BLIND_TYPE,
+    CONF_LOCAL_NAME,
+    CONF_MAC_CODE,
+    DOMAIN,
+    ERROR_COULD_NOT_FIND_MOTOR,
+    ERROR_INVALID_MAC_CODE,
+    ERROR_NO_BLUETOOTH_ADAPTER,
+    ERROR_NO_DEVICES_FOUND,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+CONFIG_SCHEMA = vol.Schema({vol.Required(CONF_MAC_CODE): str})
+
+
+class FlowHandler(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Motionblinds BLE."""
+
+    def __init__(self) -> None:
+        """Initialize a ConfigFlow."""
+        self._discovery_info: BluetoothServiceInfoBleak | BLEDevice | None = None
+        self._mac_code: str | None = None
+        self._display_name: str | None = None
+        self._blind_type: MotionBlindType | None = None
+
+    async def async_step_bluetooth(
+        self, discovery_info: BluetoothServiceInfoBleak
+    ) -> ConfigFlowResult:
+        """Handle the bluetooth discovery step."""
+        _LOGGER.debug(
+            "Discovered Motionblinds bluetooth device: %s", discovery_info.as_dict()
+        )
+        await self.async_set_unique_id(discovery_info.address)
+        self._abort_if_unique_id_configured()
+
+        self._discovery_info = discovery_info
+        self._mac_code = get_mac_from_local_name(discovery_info.name)
+        self._display_name = DISPLAY_NAME.format(mac_code=self._mac_code)
+        self.context["local_name"] = discovery_info.name
+        self.context["title_placeholders"] = {"name": self._display_name}
+
+        return await self.async_step_confirm()
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle a flow initialized by the user."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            mac_code = user_input[CONF_MAC_CODE]
+            # Discover with BLE
+            try:
+                await self.async_discover_motionblind(mac_code)
+            except NoBluetoothAdapter:
+                return self.async_abort(reason=EXCEPTION_MAP[NoBluetoothAdapter])
+            except NoDevicesFound:
+                return self.async_abort(reason=EXCEPTION_MAP[NoDevicesFound])
+            except tuple(EXCEPTION_MAP.keys()) as e:
+                errors = {"base": EXCEPTION_MAP.get(type(e), str(type(e)))}
+                return self.async_show_form(
+                    step_id="user", data_schema=CONFIG_SCHEMA, errors=errors
+                )
+            return await self.async_step_confirm()
+
+        scanner_count = bluetooth.async_scanner_count(self.hass, connectable=True)
+        if not scanner_count:
+            _LOGGER.error("No bluetooth adapter found")
+            return self.async_abort(reason=EXCEPTION_MAP[NoBluetoothAdapter])
+
+        return self.async_show_form(
+            step_id="user", data_schema=CONFIG_SCHEMA, errors=errors
+        )
+
+    async def async_step_confirm(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Confirm a single device."""
+        if user_input is not None:
+            self._blind_type = user_input[CONF_BLIND_TYPE]
+
+            if TYPE_CHECKING:
+                assert self._discovery_info is not None
+
+            return self.async_create_entry(
+                title=str(self._display_name),
+                data={
+                    CONF_ADDRESS: self._discovery_info.address,
+                    CONF_LOCAL_NAME: self._discovery_info.name,
+                    CONF_MAC_CODE: self._mac_code,
+                    CONF_BLIND_TYPE: self._blind_type,
+                },
+            )
+
+        return self.async_show_form(
+            step_id="confirm",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_BLIND_TYPE): SelectSelector(
+                        SelectSelectorConfig(
+                            options=[
+                                blind_type.name.lower()
+                                for blind_type in MotionBlindType
+                            ],
+                            translation_key=CONF_BLIND_TYPE,
+                            mode=SelectSelectorMode.DROPDOWN,
+                        )
+                    )
+                }
+            ),
+            description_placeholders={"display_name": self._display_name},
+        )
+
+    async def async_discover_motionblind(self, mac_code: str) -> None:
+        """Discover Motionblinds initialized by the user."""
+        if not is_valid_mac(mac_code):
+            _LOGGER.error("Invalid MAC code: %s", mac_code.upper())
+            raise InvalidMACCode
+
+        scanner_count = bluetooth.async_scanner_count(self.hass, connectable=True)
+        if not scanner_count:
+            _LOGGER.error("No bluetooth adapter found")
+            raise NoBluetoothAdapter
+
+        bleak_scanner = bluetooth.async_get_scanner(self.hass)
+        devices = await bleak_scanner.discover()
+
+        if len(devices) == 0:
+            _LOGGER.error("Could not find any bluetooth devices")
+            raise NoDevicesFound
+
+        motion_device: BLEDevice | None = next(
+            (
+                device
+                for device in devices
+                if device
+                and device.name
+                and f"MOTION_{mac_code.upper()}" in device.name
+            ),
+            None,
+        )
+
+        if motion_device is None:
+            _LOGGER.error("Could not find a motor with MAC code: %s", mac_code.upper())
+            raise CouldNotFindMotor
+
+        await self.async_set_unique_id(motion_device.address, raise_on_progress=False)
+        self._abort_if_unique_id_configured()
+
+        self._discovery_info = motion_device
+        self._mac_code = mac_code.upper()
+        self._display_name = DISPLAY_NAME.format(mac_code=self._mac_code)
+
+
+def is_valid_mac(data: str) -> bool:
+    """Validate the provided MAC address."""
+
+    mac_regex = r"^[0-9A-Fa-f]{4}$"
+    return bool(re.match(mac_regex, data))
+
+
+def get_mac_from_local_name(data: str) -> str | None:
+    """Get the MAC address from the bluetooth local name."""
+
+    mac_regex = r"^MOTION_([0-9A-Fa-f]{4})$"
+    match = re.search(mac_regex, data)
+    return str(match.group(1)) if match else None
+
+
+class CouldNotFindMotor(HomeAssistantError):
+    """Error to indicate no motor with that MAC code could be found."""
+
+
+class InvalidMACCode(HomeAssistantError):
+    """Error to indicate the MAC code is invalid."""
+
+
+class NoBluetoothAdapter(HomeAssistantError):
+    """Error to indicate no bluetooth adapter could be found."""
+
+
+class NoDevicesFound(HomeAssistantError):
+    """Error to indicate no bluetooth devices could be found."""
+
+
+EXCEPTION_MAP = {
+    NoBluetoothAdapter: ERROR_NO_BLUETOOTH_ADAPTER,
+    NoDevicesFound: ERROR_NO_DEVICES_FOUND,
+    CouldNotFindMotor: ERROR_COULD_NOT_FIND_MOTOR,
+    InvalidMACCode: ERROR_INVALID_MAC_CODE,
+}

homeassistant/components/motionblinds_ble/const.py

@@ -0,0 +1,16 @@
+"""Constants for the Motionblinds BLE integration."""
+
+CONF_LOCAL_NAME = "local_name"
+CONF_MAC_CODE = "mac_code"
+CONF_BLIND_TYPE = "blind_type"
+
+DOMAIN = "motionblinds_ble"
+
+ERROR_COULD_NOT_FIND_MOTOR = "could_not_find_motor"
+ERROR_INVALID_MAC_CODE = "invalid_mac_code"
+ERROR_NO_BLUETOOTH_ADAPTER = "no_bluetooth_adapter"
+ERROR_NO_DEVICES_FOUND = "no_devices_found"
+
+ICON_VERTICAL_BLIND = "mdi:blinds-vertical-closed"
+
+MANUFACTURER = "Motionblinds"

homeassistant/components/motionblinds_ble/cover.py

@@ -0,0 +1,230 @@
+"""Cover entities for the Motionblinds BLE integration."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import logging
+from typing import Any
+
+from motionblindsble.const import MotionBlindType, MotionRunningType
+from motionblindsble.device import MotionDevice
+
+from homeassistant.components.cover import (
+    ATTR_POSITION,
+    ATTR_TILT_POSITION,
+    CoverDeviceClass,
+    CoverEntity,
+    CoverEntityDescription,
+    CoverEntityFeature,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .const import CONF_BLIND_TYPE, CONF_MAC_CODE, DOMAIN, ICON_VERTICAL_BLIND
+from .entity import MotionblindsBLEEntity
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True, kw_only=True)
+class MotionblindsBLECoverEntityDescription(CoverEntityDescription):
+    """Entity description of a cover entity with default values."""
+
+    key: str = CoverDeviceClass.BLIND.value
+    translation_key: str = CoverDeviceClass.BLIND.value
+
+
+SHADE_ENTITY_DESCRIPTION = MotionblindsBLECoverEntityDescription(
+    device_class=CoverDeviceClass.SHADE
+)
+BLIND_ENTITY_DESCRIPTION = MotionblindsBLECoverEntityDescription(
+    device_class=CoverDeviceClass.BLIND
+)
+CURTAIN_ENTITY_DESCRIPTION = MotionblindsBLECoverEntityDescription(
+    device_class=CoverDeviceClass.CURTAIN
+)
+VERTICAL_ENTITY_DESCRIPTION = MotionblindsBLECoverEntityDescription(
+    device_class=CoverDeviceClass.CURTAIN, icon=ICON_VERTICAL_BLIND
+)
+
+BLIND_TYPE_TO_ENTITY_DESCRIPTION: dict[str, MotionblindsBLECoverEntityDescription] = {
+    MotionBlindType.HONEYCOMB.name: SHADE_ENTITY_DESCRIPTION,
+    MotionBlindType.ROMAN.name: SHADE_ENTITY_DESCRIPTION,
+    MotionBlindType.ROLLER.name: SHADE_ENTITY_DESCRIPTION,
+    MotionBlindType.DOUBLE_ROLLER.name: SHADE_ENTITY_DESCRIPTION,
+    MotionBlindType.VENETIAN.name: BLIND_ENTITY_DESCRIPTION,
+    MotionBlindType.VENETIAN_TILT_ONLY.name: BLIND_ENTITY_DESCRIPTION,
+    MotionBlindType.CURTAIN.name: CURTAIN_ENTITY_DESCRIPTION,
+    MotionBlindType.VERTICAL.name: VERTICAL_ENTITY_DESCRIPTION,
+}
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Set up cover entity based on a config entry."""
+
+    cover_class: type[MotionblindsBLECoverEntity] = BLIND_TYPE_TO_CLASS[
+        entry.data[CONF_BLIND_TYPE].upper()
+    ]
+    device: MotionDevice = hass.data[DOMAIN][entry.entry_id]
+    entity_description: MotionblindsBLECoverEntityDescription = (
+        BLIND_TYPE_TO_ENTITY_DESCRIPTION[entry.data[CONF_BLIND_TYPE].upper()]
+    )
+    entity: MotionblindsBLECoverEntity = cover_class(device, entry, entity_description)
+
+    async_add_entities([entity])
+
+
+class MotionblindsBLECoverEntity(MotionblindsBLEEntity, CoverEntity):
+    """Representation of a cover entity."""
+
+    _attr_is_closed: bool | None = None
+    _attr_name = None
+
+    async def async_added_to_hass(self) -> None:
+        """Register device callbacks."""
+        _LOGGER.debug(
+            "(%s) Added %s cover entity (%s)",
+            self.entry.data[CONF_MAC_CODE],
+            MotionBlindType[self.entry.data[CONF_BLIND_TYPE].upper()].value.lower(),
+            BLIND_TYPE_TO_CLASS[self.entry.data[CONF_BLIND_TYPE].upper()].__name__,
+        )
+        self.device.register_running_callback(self.async_update_running)
+        self.device.register_position_callback(self.async_update_position)
+
+    async def async_stop_cover(self, **kwargs: Any) -> None:
+        """Stop moving the cover entity."""
+        _LOGGER.debug("(%s) Stopping", self.entry.data[CONF_MAC_CODE])
+        await self.device.stop()
+
+    @callback
+    def async_update_running(
+        self, running_type: MotionRunningType | None, write_state: bool = True
+    ) -> None:
+        """Update the running type (e.g. opening/closing) of the cover entity."""
+        if running_type in {None, MotionRunningType.STILL, MotionRunningType.UNKNOWN}:
+            self._attr_is_opening = False
+            self._attr_is_closing = False
+        else:
+            self._attr_is_opening = running_type is MotionRunningType.OPENING
+            self._attr_is_closing = running_type is not MotionRunningType.OPENING
+        if running_type is not MotionRunningType.STILL:
+            self._attr_is_closed = None
+        if write_state:
+            self.async_write_ha_state()
+
+    @callback
+    def async_update_position(
+        self,
+        position: int | None,
+        tilt: int | None,
+    ) -> None:
+        """Update the position of the cover entity."""
+        if position is None:
+            self._attr_current_cover_position = None
+            self._attr_is_closed = None
+        else:
+            self._attr_current_cover_position = 100 - position
+            self._attr_is_closed = self._attr_current_cover_position == 0
+        if tilt is None:
+            self._attr_current_cover_tilt_position = None
+        else:
+            self._attr_current_cover_tilt_position = 100 - round(100 * tilt / 180)
+        self.async_write_ha_state()
+
+
+class PositionCover(MotionblindsBLECoverEntity):
+    """Representation of a cover entity with position capability."""
+
+    _attr_supported_features = (
+        CoverEntityFeature.OPEN
+        | CoverEntityFeature.CLOSE
+        | CoverEntityFeature.STOP
+        | CoverEntityFeature.SET_POSITION
+    )
+
+    async def async_open_cover(self, **kwargs: Any) -> None:
+        """Open the cover entity."""
+        _LOGGER.debug("(%s) Opening", self.entry.data[CONF_MAC_CODE])
+        await self.device.open()
+
+    async def async_close_cover(self, **kwargs: Any) -> None:
+        """Close the cover entity."""
+        _LOGGER.debug("(%s) Closing", self.entry.data[CONF_MAC_CODE])
+        await self.device.close()
+
+    async def async_set_cover_position(self, **kwargs: Any) -> None:
+        """Move the cover entity to a specific position."""
+        new_position: int = 100 - int(kwargs[ATTR_POSITION])
+
+        _LOGGER.debug(
+            "(%s) Setting position to %i",
+            self.entry.data[CONF_MAC_CODE],
+            new_position,
+        )
+        await self.device.position(new_position)
+
+
+class TiltCover(MotionblindsBLECoverEntity):
+    """Representation of a cover entity with tilt capability."""
+
+    _attr_supported_features = (
+        CoverEntityFeature.OPEN_TILT
+        | CoverEntityFeature.CLOSE_TILT
+        | CoverEntityFeature.STOP_TILT
+        | CoverEntityFeature.SET_TILT_POSITION
+    )
+
+    async def async_open_cover_tilt(self, **kwargs: Any) -> None:
+        """Tilt the cover entity open."""
+        _LOGGER.debug("(%s) Tilt opening", self.entry.data[CONF_MAC_CODE])
+        await self.device.open_tilt()
+
+    async def async_close_cover_tilt(self, **kwargs: Any) -> None:
+        """Tilt the cover entity closed."""
+        _LOGGER.debug("(%s) Tilt closing", self.entry.data[CONF_MAC_CODE])
+        await self.device.close_tilt()
+
+    async def async_stop_cover_tilt(self, **kwargs: Any) -> None:
+        """Stop tilting the cover entity."""
+        await self.async_stop_cover(**kwargs)
+
+    async def async_set_cover_tilt_position(self, **kwargs: Any) -> None:
+        """Tilt the cover entity to a specific position."""
+        new_tilt: int = 100 - int(kwargs[ATTR_TILT_POSITION])
+
+        _LOGGER.debug(
+            "(%s) Setting tilt position to %i",
+            self.entry.data[CONF_MAC_CODE],
+            new_tilt,
+        )
+        await self.device.tilt(round(180 * new_tilt / 100))
+
+
+class PositionTiltCover(PositionCover, TiltCover):
+    """Representation of a cover entity with position & tilt capabilities."""
+
+    _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
+    )
+
+
+BLIND_TYPE_TO_CLASS: dict[str, type[MotionblindsBLECoverEntity]] = {
+    MotionBlindType.ROLLER.name: PositionCover,
+    MotionBlindType.HONEYCOMB.name: PositionCover,
+    MotionBlindType.ROMAN.name: PositionCover,
+    MotionBlindType.VENETIAN.name: PositionTiltCover,
+    MotionBlindType.VENETIAN_TILT_ONLY.name: TiltCover,
+    MotionBlindType.DOUBLE_ROLLER.name: PositionTiltCover,
+    MotionBlindType.CURTAIN.name: PositionCover,
+    MotionBlindType.VERTICAL.name: PositionTiltCover,
+}

homeassistant/components/motionblinds_ble/entity.py

@@ -0,0 +1,52 @@
+"""Base entities for the Motionblinds BLE integration."""
+
+import logging
+
+from motionblindsble.const import MotionBlindType
+from motionblindsble.device import MotionDevice
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_ADDRESS
+from homeassistant.helpers.device_registry import CONNECTION_BLUETOOTH, DeviceInfo
+from homeassistant.helpers.entity import Entity, EntityDescription
+
+from .const import CONF_BLIND_TYPE, CONF_MAC_CODE, MANUFACTURER
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class MotionblindsBLEEntity(Entity):
+    """Base class for Motionblinds BLE entities."""
+
+    _attr_has_entity_name = True
+    _attr_should_poll = False
+
+    device: MotionDevice
+    entry: ConfigEntry
+
+    def __init__(
+        self,
+        device: MotionDevice,
+        entry: ConfigEntry,
+        entity_description: EntityDescription,
+        unique_id_suffix: str | None = None,
+    ) -> None:
+        """Initialize the entity."""
+        if unique_id_suffix is None:
+            self._attr_unique_id = entry.data[CONF_ADDRESS]
+        else:
+            self._attr_unique_id = f"{entry.data[CONF_ADDRESS]}_{unique_id_suffix}"
+        self.device = device
+        self.entry = entry
+        self.entity_description = entity_description
+        self._attr_device_info = DeviceInfo(
+            connections={(CONNECTION_BLUETOOTH, entry.data[CONF_ADDRESS])},
+            manufacturer=MANUFACTURER,
+            model=MotionBlindType[entry.data[CONF_BLIND_TYPE].upper()].value,
+            name=device.display_name,
+        )
+
+    async def async_update(self) -> None:
+        """Update state, called by HA if there is a poll interval and by the service homeassistant.update_entity."""
+        _LOGGER.debug("(%s) Updating entity", self.entry.data[CONF_MAC_CODE])
+        await self.device.status_query()

homeassistant/components/motionblinds_ble/manifest.json

@@ -0,0 +1,18 @@
+{
+  "domain": "motionblinds_ble",
+  "name": "Motionblinds BLE",
+  "bluetooth": [
+    {
+      "local_name": "MOTION_*",
+      "connectable": true
+    }
+  ],
+  "codeowners": ["@LennP", "@jerrybboy"],
+  "config_flow": true,
+  "dependencies": ["bluetooth_adapters"],
+  "documentation": "https://www.home-assistant.io/integrations/motionblinds_ble",
+  "integration_type": "device",
+  "iot_class": "assumed_state",
+  "loggers": ["motionblindsble"],
+  "requirements": ["motionblindsble==0.0.8"]
+}

homeassistant/components/motionblinds_ble/strings.json

@@ -0,0 +1,37 @@
+{
+  "config": {
+    "abort": {
+      "no_bluetooth_adapter": "No bluetooth adapter found",
+      "no_devices_found": "Could not find any bluetooth devices"
+    },
+    "error": {
+      "could_not_find_motor": "Could not find a motor with that MAC code",
+      "invalid_mac_code": "Invalid MAC code"
+    },
+    "step": {
+      "user": {
+        "description": "Fill in the 4-character MAC code of your motor, for example F3ED or E3A6",
+        "data": {
+          "mac_code": "MAC code"
+        }
+      },
+      "confirm": {
+        "description": "What kind of blind is {display_name}?"
+      }
+    }
+  },
+  "selector": {
+    "blind_type": {
+      "options": {
+        "roller": "Roller blind",
+        "honeycomb": "Honeycomb blind",
+        "roman": "Roman blind",
+        "venetian": "Venetian blind",
+        "venetian_tilt_only": "Venetian blind (tilt-only)",
+        "double_roller": "Double roller blind",
+        "curtain": "Curtain blind",
+        "vertical": "Vertical blind"
+      }
+    }
+  }
+}

homeassistant/generated/bluetooth.py

@@ -428,6 +428,11 @@ BLUETOOTH: list[dict[str, bool | str | int | list[int]]] = [
         "manufacturer_id": 89,
         "service_uuid": "0000fee5-0000-1000-8000-00805f9b34fb",
     },
+    {
+        "connectable": True,
+        "domain": "motionblinds_ble",
+        "local_name": "MOTION_*",
+    },
     {
         "domain": "oralb",
         "manufacturer_id": 220,

homeassistant/generated/config_flows.py

@@ -326,6 +326,7 @@ FLOWS = {
         "moon",
         "mopeka",
         "motion_blinds",
+        "motionblinds_ble",
         "motioneye",
         "motionmount",
         "mqtt",

homeassistant/generated/integrations.json

@@ -3734,6 +3734,12 @@
       "config_flow": true,
       "iot_class": "local_push"
     },
+    "motionblinds_ble": {
+      "name": "Motionblinds BLE",
+      "integration_type": "device",
+      "config_flow": true,
+      "iot_class": "assumed_state"
+    },
     "motioneye": {
       "name": "motionEye",
       "integration_type": "hub",

requirements_all.txt

@@ -1327,6 +1327,9 @@ mopeka-iot-ble==0.7.0
 # homeassistant.components.motion_blinds
 motionblinds==0.6.23
 
+# homeassistant.components.motionblinds_ble
+motionblindsble==0.0.8
+
 # homeassistant.components.motioneye
 motioneye-client==0.3.14
 

requirements_test_all.txt

@@ -1066,6 +1066,9 @@ mopeka-iot-ble==0.7.0
 # homeassistant.components.motion_blinds
 motionblinds==0.6.23
 
+# homeassistant.components.motionblinds_ble
+motionblindsble==0.0.8
+
 # homeassistant.components.motioneye
 motioneye-client==0.3.14
 

tests/components/motionblinds_ble/__init__.py

@@ -0,0 +1 @@
+"""Tests for the Motionblinds BLE integration."""

tests/components/motionblinds_ble/conftest.py

@@ -0,0 +1,31 @@
+"""Setup the MotionBlinds BLE tests."""
+
+from unittest.mock import AsyncMock, Mock, patch
+
+import pytest
+
+TEST_MAC = "abcd"
+TEST_NAME = f"MOTION_{TEST_MAC.upper()}"
+TEST_ADDRESS = "test_adress"
+
+
+@pytest.fixture(name="motionblinds_ble_connect", autouse=True)
+def motion_blinds_connect_fixture(enable_bluetooth):
+    """Mock motion blinds ble connection and entry setup."""
+    device = Mock()
+    device.name = TEST_NAME
+    device.address = TEST_ADDRESS
+
+    bleak_scanner = AsyncMock()
+    bleak_scanner.discover.return_value = [device]
+
+    with patch(
+        "homeassistant.components.motionblinds_ble.config_flow.bluetooth.async_scanner_count",
+        return_value=1,
+    ), patch(
+        "homeassistant.components.motionblinds_ble.config_flow.bluetooth.async_get_scanner",
+        return_value=bleak_scanner,
+    ), patch(
+        "homeassistant.components.motionblinds_ble.async_setup_entry", return_value=True
+    ):
+        yield bleak_scanner, device

tests/components/motionblinds_ble/test_config_flow.py

@@ -0,0 +1,256 @@
+"""Test the MotionBlinds BLE config flow."""
+
+from unittest.mock import patch
+
+from motionblindsble.const import MotionBlindType
+
+from homeassistant import config_entries, data_entry_flow
+from homeassistant.components.bluetooth.models import BluetoothServiceInfoBleak
+from homeassistant.components.motionblinds_ble import const
+from homeassistant.const import CONF_ADDRESS
+from homeassistant.core import HomeAssistant
+
+from .conftest import TEST_ADDRESS, TEST_MAC, TEST_NAME
+
+from tests.components.bluetooth import generate_advertisement_data, generate_ble_device
+
+TEST_BLIND_TYPE = MotionBlindType.ROLLER.name.lower()
+
+BLIND_SERVICE_INFO = BluetoothServiceInfoBleak(
+    name=TEST_NAME,
+    address=TEST_ADDRESS,
+    device=generate_ble_device(
+        address="cc:cc:cc:cc:cc:cc",
+        name=TEST_NAME,
+    ),
+    rssi=-61,
+    manufacturer_data={000: b"test"},
+    service_data={
+        "test": bytearray(b"0000"),
+    },
+    service_uuids=[
+        "test",
+    ],
+    source="local",
+    advertisement=generate_advertisement_data(
+        manufacturer_data={000: b"test"},
+        service_uuids=["test"],
+    ),
+    connectable=True,
+    time=0,
+)
+
+
+async def test_config_flow_manual_success(
+    hass: HomeAssistant, motionblinds_ble_connect
+) -> None:
+    """Successful flow manually initialized by the user."""
+    result = await hass.config_entries.flow.async_init(
+        const.DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] is data_entry_flow.FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {}
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {const.CONF_MAC_CODE: TEST_MAC},
+    )
+    assert result["type"] is data_entry_flow.FlowResultType.FORM
+    assert result["step_id"] == "confirm"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {const.CONF_BLIND_TYPE: MotionBlindType.ROLLER.name.lower()},
+    )
+    assert result["type"] is data_entry_flow.FlowResultType.CREATE_ENTRY
+    assert result["title"] == f"Motionblind {TEST_MAC.upper()}"
+    assert result["data"] == {
+        CONF_ADDRESS: TEST_ADDRESS,
+        const.CONF_LOCAL_NAME: TEST_NAME,
+        const.CONF_MAC_CODE: TEST_MAC.upper(),
+        const.CONF_BLIND_TYPE: TEST_BLIND_TYPE,
+    }
+    assert result["options"] == {}
+
+
+async def test_config_flow_manual_error_invalid_mac(
+    hass: HomeAssistant, motionblinds_ble_connect
+) -> None:
+    """Invalid MAC code error flow manually initialized by the user."""
+
+    # Initialize
+    result = await hass.config_entries.flow.async_init(
+        const.DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] is data_entry_flow.FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {}
+
+    # Try invalid MAC code
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {const.CONF_MAC_CODE: "AABBCC"},  # A MAC code should be 4 characters
+    )
+    assert result["type"] is data_entry_flow.FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {"base": const.ERROR_INVALID_MAC_CODE}
+
+    # Recover
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {const.CONF_MAC_CODE: TEST_MAC},
+    )
+    assert result["type"] is data_entry_flow.FlowResultType.FORM
+    assert result["step_id"] == "confirm"
+
+    # Finish flow
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {const.CONF_BLIND_TYPE: MotionBlindType.ROLLER.name.lower()},
+    )
+    assert result["type"] is data_entry_flow.FlowResultType.CREATE_ENTRY
+    assert result["title"] == f"Motionblind {TEST_MAC.upper()}"
+    assert result["data"] == {
+        CONF_ADDRESS: TEST_ADDRESS,
+        const.CONF_LOCAL_NAME: TEST_NAME,
+        const.CONF_MAC_CODE: TEST_MAC.upper(),
+        const.CONF_BLIND_TYPE: TEST_BLIND_TYPE,
+    }
+    assert result["options"] == {}
+
+
+async def test_config_flow_manual_error_no_bluetooth_adapter(
+    hass: HomeAssistant, motionblinds_ble_connect
+) -> None:
+    """No Bluetooth adapter error flow manually initialized by the user."""
+
+    # Try step_user with zero Bluetooth adapters
+    with patch(
+        "homeassistant.components.motionblinds_ble.config_flow.bluetooth.async_scanner_count",
+        return_value=0,
+    ):
+        result = await hass.config_entries.flow.async_init(
+            const.DOMAIN, context={"source": config_entries.SOURCE_USER}
+        )
+    assert result["type"] is data_entry_flow.FlowResultType.ABORT
+    assert result["reason"] == const.ERROR_NO_BLUETOOTH_ADAPTER
+
+    # Try discovery with zero Bluetooth adapters
+    result = await hass.config_entries.flow.async_init(
+        const.DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] is data_entry_flow.FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {}
+
+    with patch(
+        "homeassistant.components.motionblinds_ble.config_flow.bluetooth.async_scanner_count",
+        return_value=0,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {const.CONF_MAC_CODE: TEST_MAC},
+        )
+    assert result["type"] is data_entry_flow.FlowResultType.ABORT
+    assert result["reason"] == const.ERROR_NO_BLUETOOTH_ADAPTER
+
+
+async def test_config_flow_manual_error_could_not_find_motor(
+    hass: HomeAssistant, motionblinds_ble_connect
+) -> None:
+    """Could not find motor error flow manually initialized by the user."""
+
+    # Initialize
+    result = await hass.config_entries.flow.async_init(
+        const.DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] is data_entry_flow.FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {}
+
+    # Try with MAC code that cannot be found
+    motionblinds_ble_connect[1].name = "WRONG_NAME"
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {const.CONF_MAC_CODE: TEST_MAC},
+    )
+    assert result["type"] is data_entry_flow.FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {"base": const.ERROR_COULD_NOT_FIND_MOTOR}
+
+    # Recover
+    motionblinds_ble_connect[1].name = TEST_NAME
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {const.CONF_MAC_CODE: TEST_MAC},
+    )
+    assert result["type"] is data_entry_flow.FlowResultType.FORM
+    assert result["step_id"] == "confirm"
+
+    # Finish flow
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {const.CONF_BLIND_TYPE: MotionBlindType.ROLLER.name.lower()},
+    )
+    assert result["type"] is data_entry_flow.FlowResultType.CREATE_ENTRY
+    assert result["title"] == f"Motionblind {TEST_MAC.upper()}"
+    assert result["data"] == {
+        CONF_ADDRESS: TEST_ADDRESS,
+        const.CONF_LOCAL_NAME: TEST_NAME,
+        const.CONF_MAC_CODE: TEST_MAC.upper(),
+        const.CONF_BLIND_TYPE: TEST_BLIND_TYPE,
+    }
+    assert result["options"] == {}
+
+
+async def test_config_flow_manual_error_no_devices_found(
+    hass: HomeAssistant, motionblinds_ble_connect
+) -> None:
+    """No devices found error flow manually initialized by the user."""
+
+    # Initialize
+    result = await hass.config_entries.flow.async_init(
+        const.DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] is data_entry_flow.FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {}
+
+    # Try with zero found bluetooth devices
+    motionblinds_ble_connect[0].discover.return_value = []
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {const.CONF_MAC_CODE: TEST_MAC},
+    )
+    assert result["type"] is data_entry_flow.FlowResultType.ABORT
+    assert result["reason"] == const.ERROR_NO_DEVICES_FOUND
+
+
+async def test_config_flow_bluetooth_success(
+    hass: HomeAssistant, motionblinds_ble_connect
+) -> None:
+    """Successful bluetooth discovery flow."""
+    result = await hass.config_entries.flow.async_init(
+        const.DOMAIN,
+        context={"source": config_entries.SOURCE_BLUETOOTH},
+        data=BLIND_SERVICE_INFO,
+    )
+
+    assert result["type"] is data_entry_flow.FlowResultType.FORM
+    assert result["step_id"] == "confirm"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {const.CONF_BLIND_TYPE: MotionBlindType.ROLLER.name.lower()},
+    )
+
+    assert result["type"] is data_entry_flow.FlowResultType.CREATE_ENTRY
+    assert result["title"] == f"Motionblind {TEST_MAC.upper()}"
+    assert result["data"] == {
+        CONF_ADDRESS: TEST_ADDRESS,
+        const.CONF_LOCAL_NAME: TEST_NAME,
+        const.CONF_MAC_CODE: TEST_MAC.upper(),
+        const.CONF_BLIND_TYPE: TEST_BLIND_TYPE,
+    }
+    assert result["options"] == {}