Initial release

CHANGELOG.md

@@ -0,0 +1,10 @@
+# Changelog
+
+All notable changes to this project will be documented in this file.
+
+The format is based on [Keep a Changelog],
+and this project adheres to [Semantic Versioning].
+
+## [0.1.0] - 2023-04-22
+
+- Initial release

README.md

@@ -0,0 +1,33 @@
+# Home Assistant support for Tuya BLE devices
+
+## Overview
+
+This integration supports various Mobile-Alerts sensors. The integration acts as proxy server between Mobile-Alerts gateway and cloud. 
+
+_Inspired by [@redphx] code (https://github.com/redphx/poc-tuya-ble-fingerbot)
+
+## Installation
+
+Place the `custom_components` folder in your configuration directory (or add its contents to an existing `custom_components` folder). Alternatively install via [HACS](https://hacs.xyz/).
+
+[![Open your Home Assistant instance and open a repository inside the Home Assistant Community Store.](https://my.home-assistant.io/badges/hacs_repository.svg)](https://my.home-assistant.io/redirect/hacs_repository/?owner=PlusPlus-ua&repository=ha_tuya_ble&category=integration)
+
+## Usage
+
+After adding to Home Assistan integration should discover all supported Bluetooth devices, or you can add discoverable devices manually.
+
+The integration works locally, but connection to Tuya BLE device requires device ID and encryption key from Tuya IOT cloud. It could be obtained using the same credentials as in official Tuya integreation. To obtain the credentials please refer to official Tuya integreation [documentation](https://www.home-assistant.io/integrations/tuya/)
+
+## Supported devices list
+
+* Fingerbots (category_id 'szjqr')
+  + Fingerbot (product_id 'yrnk7mnn'), original device fists in category, powered by CR2 battery.
+  + Fingerbot Plus (product_id 'yiihr7zh'), almost same as original, has sensor button for manual control.
+  + CubeTouch II (product_id 'xhf790if'), bult-in battery with USB type C charging.
+  All features available in Home Assistant, except programming (series of actions) - it's not documented and looks useless becouse it could be implemented by Home Assistant scripts or automations.
+
+* Temperature and humidity sensors (category_id 'wsdcg')
+  + Soil moisture sensor (product_id 'ojzlzzsw').
+
+* CO2 sensors (category_id 'co2bj')
+  + CO2 Detector (product_id '59s19z5m').

custom_components/tuya_ble/__init__.py

@@ -0,0 +1,107 @@
+"""The Tuya BLE integration."""
+from __future__ import annotations
+
+import logging
+
+from bleak_retry_connector import BLEAK_RETRY_EXCEPTIONS as BLEAK_EXCEPTIONS, get_device
+
+from homeassistant.components import bluetooth
+from homeassistant.components.bluetooth.match import ADDRESS, BluetoothCallbackMatcher
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_ADDRESS, EVENT_HOMEASSISTANT_STOP, Platform
+from homeassistant.core import Event, HomeAssistant, callback
+from homeassistant.exceptions import ConfigEntryNotReady
+
+from .tuya_ble import TuyaBLEDevice
+
+from .cloud import HASSTuyaBLEDeviceManager
+from .const import DOMAIN
+from .devices import TuyaBLECoordinator, TuyaBLEData, get_device_product_info
+
+PLATFORMS: list[Platform] = [
+    Platform.BUTTON,
+    Platform.NUMBER,
+    Platform.SENSOR, 
+    Platform.SELECT,
+    Platform.SWITCH,
+]
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Tuya BLE from a config entry."""
+    address: str = entry.data[CONF_ADDRESS]
+    ble_device = bluetooth.async_ble_device_from_address(
+        hass, address.upper(), True
+    ) or await get_device(address)
+    if not ble_device:
+        raise ConfigEntryNotReady(
+            f"Could not find Tuya BLE device with address {address}"
+        )   
+    manager = HASSTuyaBLEDeviceManager(hass, entry.options.copy())
+    device = TuyaBLEDevice(manager, ble_device)
+    await device.initialize()
+    product_info = get_device_product_info(device)
+
+    coordinator = TuyaBLECoordinator(hass, device)
+    try:
+        await device.update()
+    except BLEAK_EXCEPTIONS as ex:
+        raise ConfigEntryNotReady(
+            f"Could not communicate with Tuya BLE device with address {address}"
+        ) from ex
+
+    @callback
+    def _async_update_ble(
+        service_info: bluetooth.BluetoothServiceInfoBleak,
+        change: bluetooth.BluetoothChange,
+    ) -> None:
+        """Update from a ble callback."""
+        device.set_ble_device_and_advertisement_data(
+            service_info.device, service_info.advertisement
+        )
+
+    entry.async_on_unload(
+        bluetooth.async_register_callback(
+            hass,
+            _async_update_ble,
+            BluetoothCallbackMatcher({ADDRESS: address}),
+            bluetooth.BluetoothScanningMode.ACTIVE,
+        )
+    )
+
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = TuyaBLEData(
+        entry.title,
+        device,
+        product_info,
+        manager,
+        coordinator,
+    )
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+    entry.async_on_unload(entry.add_update_listener(_async_update_listener))
+
+    async def _async_stop(event: Event) -> None:
+        """Close the connection."""
+        await device.stop()
+
+    entry.async_on_unload(
+        hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _async_stop)
+    )
+    return True
+
+async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
+    """Handle options update."""
+    data: TuyaBLEData = hass.data[DOMAIN][entry.entry_id]
+    if entry.title != data.title:
+        await hass.config_entries.async_reload(entry.entry_id)
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        data: TuyaBLEData = hass.data[DOMAIN].pop(entry.entry_id)
+        await data.device.stop()
+
+    return unload_ok

custom_components/tuya_ble/button.py

@@ -0,0 +1,163 @@
+"""The Tuya BLE integration."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import logging
+from typing import Callable
+
+from homeassistant.components.button import (
+    ButtonEntityDescription,
+    ButtonEntity,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import DOMAIN
+from .devices import TuyaBLEData, TuyaBLEEntity, TuyaBLEProductInfo
+from .tuya_ble import TuyaBLEDataPointType, TuyaBLEDevice
+
+_LOGGER = logging.getLogger(__name__)
+
+
+TuyaBLEButtonIsAvailable = Callable[
+    ['TuyaBLEButton', TuyaBLEProductInfo], bool
+] | None
+
+
+@dataclass
+class TuyaBLEButtonMapping:
+    dp_id: int
+    description: ButtonEntityDescription
+    force_add: bool = True
+    dp_type: TuyaBLEDataPointType | None = None
+    is_available: TuyaBLEButtonIsAvailable = None
+
+
+def is_fingerbot_in_push_mode(
+    self: TuyaBLEButton,
+    product: TuyaBLEProductInfo
+) -> bool:
+    result: bool = False
+    if product.fingerbot:
+        datapoint = self._device.datapoints[product.fingerbot.mode]
+        if datapoint:
+            result = datapoint.value == 0
+    return result
+
+
+@dataclass
+class TuyaBLEFingerbotModeMapping(TuyaBLEButtonMapping):
+    description: ButtonEntityDescription = ButtonEntityDescription(
+        key="push",
+    )
+    is_available: TuyaBLEButtonIsAvailable = is_fingerbot_in_push_mode
+
+
+@dataclass
+class TuyaBLECategoryButtonMapping:
+    products: dict[str, list[TuyaBLEButtonMapping]] | None = None
+    mapping: list[TuyaBLEButtonMapping] | None = None
+
+
+mapping: dict[str, TuyaBLECategoryButtonMapping] = {
+    "szjqr": TuyaBLECategoryButtonMapping(
+        products={
+            "xhf790if":  # CubeTouch II
+            [
+                TuyaBLEFingerbotModeMapping(dp_id=1),
+            ],
+            "yiihr7zh":  # Fingerbot Plus
+            [
+                TuyaBLEFingerbotModeMapping(dp_id=2),
+            ],
+            "yrnk7mnn":  # Fingerbot
+            [
+                TuyaBLEFingerbotModeMapping(dp_id=2),
+            ],
+        },
+    ),
+}
+
+
+def get_mapping_by_device(
+    device: TuyaBLEDevice
+) -> list[TuyaBLECategoryButtonMapping]:
+    category = mapping.get(device.category)
+    if category is not None and category.products is not None:
+        product_mapping = category.products.get(device.product_id)
+        if product_mapping is not None:
+            return product_mapping
+        if category.mapping is not None:
+            return category.mapping
+        else:
+            return []
+    else:
+        return []
+
+
+class TuyaBLEButton(TuyaBLEEntity, ButtonEntity):
+    """Representation of a Tuya BLE Button."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        coordinator: DataUpdateCoordinator,
+        device: TuyaBLEDevice,
+        product: TuyaBLEProductInfo,
+        mapping: TuyaBLEButtonMapping,
+    ) -> None:
+        super().__init__(
+            hass,
+            coordinator,
+            device,
+            product,
+            mapping.description
+        )
+        self._mapping = mapping
+
+    def press(self) -> None:
+        """Press the button."""
+        datapoint = self._device.datapoints.get_or_create(
+            self._mapping.dp_id,
+            TuyaBLEDataPointType.DT_BOOL,
+            False,
+        )
+        if datapoint:
+            self._hass.create_task(
+                datapoint.set_value(not bool(datapoint.value))
+            )
+
+    @property
+    def available(self) -> bool:
+        """Return if entity is available."""
+        result = super().available
+        if result and self._mapping.is_available:
+            result = self._mapping.is_available(self, self._product)
+        return result
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up the Tuya BLE sensors."""
+    data: TuyaBLEData = hass.data[DOMAIN][entry.entry_id]
+    mappings = get_mapping_by_device(data.device)
+    entities: list[TuyaBLEButton] = []
+    for mapping in mappings:
+        if (
+            mapping.force_add or
+            data.device.datapoints.has_id(mapping.dp_id, mapping.dp_type)
+        ):
+            entities.append(TuyaBLEButton(
+                hass,
+                data.coordinator,
+                data.device,
+                data.product,
+                mapping,
+            ))
+    async_add_entities(entities)