Home Assistant / Programowanie

Ustatnia modyfikacja 3 grudnia 2025 przez Olek

Jest gotowa integracja, która umożliwia podłączenie urządzeń w standardzie TUYA do Home assistanta. Niestety, niektóre urządzenia, szczególnie te tańsze, nie są wspierane przez tę integrację. Nie chcę się rozwodzić dlaczego, po prostu nie widać encji dla danego urządzenia.

Z pozycji aplikacji mobilnej wszystko ładnie widać.

Bardzo często ten problem rozwiązuje się inną integracją, TUYA Local, jeżeli urządzenie jest w sieci lokalnej z HA.
Jednak co jeżeli urządzenie jest poza siecią i chcemy skorzystać z informacji zawartej w chmurze TUYA. Wiedziałem, że jest jakiś sposób, bo przecież aplikacja mobilna skąd te dane uzyskuje. Szukałem w sieci i nie znalazłem gotowego rozwiązania.
Dlatego sam zacząłem rozpracowywać temat. Korzystając z Pythona zacząłem rozgryzać Rest API z chmurą TUYA. Jaki komponent komunikacyjna posłużyła mi biblioteka TinyTuya do tego języka. Zaimplementowane procedury w tej bibliotece nie zwracały mi żądanych informacji, ale można było za jej pomocą tworzyć nowe zapytania. Zagłębiając się Cloud Services API Reference znalazłem wreszcie rozwiązanie. Czyli
GET: /v2.0/cloud/thing/*****/shadow/properties.
Skąd mogłem uzyskać wszystkie informacje.

Poniżej „czysty” (bez HA) kod do komunikacji z chmurą TUYA.
W dokumentacji TinyTuya są instrukcje, w jaki sposób uzyskać ***** dane.

# TinyTuya 
# -*- coding: utf-8 -*-
"""
 TinyTuya - Tuya Cloud Functions
 Author: Jason A. Cox
 For more information see https://github.com/jasonacox/tinytuya
""" 
import tinytuya
import json
import time

ACCESS_ID = " ***** "
ACCESS_KEY = " **** "
DEVICE_ID = " **** " #WIFI dual meter
device_id = DEVICE_ID

print("Start")
tinytuya.set_debug(False)
# Connect to Tuya Cloud
c = tinytuya.Cloud(
        apiRegion="eu", 
        apiKey=ACCESS_ID, 
        apiSecret=ACCESS_KEY, 
        apiDeviceID=DEVICE_ID)

result = c.cloudrequest(f'/v2.0/cloud/thing/{device_id}/shadow/properties')
print("properties")
print(json.dumps(result, indent=3))

while True:
    def ValueCode(propertis, code):
        for item in propertis:
            if item['code'] == code:
                return item['value']
                break
        return None
    result = c.cloudrequest(f'/v2.0/cloud/thing/{device_id}/shadow/properties')
    if result and result['result'] and result['result']['properties']:
        propertis = result['result']['properties']
        energy_forword_a = ValueCode(propertis, 'energy_forword_a')/100
        energy_forword_b = ValueCode(propertis, 'energy_forword_b')/100
        print(f'{energy_forword_a=} {energy_forword_b=}')
    else:
        print(f'błąd')
    time.sleep(10)

Zgodnie z przykładem budowania integracji do HA i przykładem example_sensor zbudowałem dwa komponenty. Dla każdego urządzenia osobny.

# Platform for sensor integration
# sensor.py dla stacji pogody
from __future__ import annotations

from homeassistant.components.sensor import (
    SensorDeviceClass,
    SensorEntity,
    SensorStateClass,
)
from homeassistant.const import UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType

import tinytuya
import json

ACCESS_ID = " **** "
ACCESS_KEY = " **** "
device_id =  " **** " #Pogoda

propertis = None

def setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None
) -> None:
    add_entities([RivaTemperatureInSensor(), 
                  RivaHumidityInSensor(),
                  RivaTemperatureOutSensor(),
                  RivaHumidityOutSensor()
                  ])


def ValueCode(propertis, code):
    for item in propertis:
        if item['code'] == code:
            return item['value']
            break
    return None


class RivaTemperatureInSensor(SensorEntity):
    _attr_name = "Riva temperatura wewnętrzna"
    _attr_unique_id = "riva_temperature_in"
    _attr_native_unit_of_measurement = "°C"
    tinytuya.set_debug(False)
    # Connect to Tuya Cloud
    c = tinytuya.Cloud(
        apiRegion="eu", 
        apiKey=ACCESS_ID, 
        apiSecret=ACCESS_KEY, 
        apiDeviceID=device_id)       
    def update(self) -> None:
        global propertis
        result = self.c.cloudrequest(f'/v2.0/cloud/thing/{device_id}/shadow/properties')
        propertis = result['result']['properties']
        v = ValueCode(propertis, 'intemp')        
        self._attr_native_value = v/10
            
class RivaHumidityInSensor(SensorEntity):
    _attr_name = "Riva wilgotność wewnętrzna"
    _attr_unique_id = "riva_humidity_in"
    _attr_native_unit_of_measurement = "%"
    def update(self):
        if propertis:
            v = ValueCode(propertis, 'inhum')        
            self._attr_native_value = v/10        


class RivaTemperatureOutSensor(SensorEntity):
    _attr_name = "Riva temperatura zewnętrzna"
    _attr_unique_id = "riva_temperature_out"
    _attr_native_unit_of_measurement = "°C"
    def update(self):
        if propertis:
            v = ValueCode(propertis, 'ch1temp')        
            self._attr_native_value = v/10        

class RivaHumidityOutSensor(SensorEntity):
    _attr_name = "Riva wilgotność zewnętrzna"
    _attr_unique_id = "riva_humidity_out"
    _attr_native_unit_of_measurement = "%"
    def update(self):
        if propertis:
            v = ValueCode(propertis, 'ch1hum')        
            self._attr_native_value = v/10        

# Platform for sensor integration
# sensor.py dla miernika energi
from __future__ import annotations

from homeassistant.components.sensor import (
    SensorDeviceClass,
    SensorEntity,
    SensorStateClass,
)
from homeassistant.const import UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType

import tinytuya
import json

ACCESS_ID = " **** "
ACCESS_KEY = " **** "
device_id =  " **** " #WIFI dual meter

propertis = None

def setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None
) -> None:
    add_entities([RivaEnergiaAllSensor(), 
                  RivaEnergiaKlimaSensor()
                  ])


def ValueCode(propertis, code):
    for item in propertis:
        if item['code'] == code:
            return item['value']
            break
    return None


class RivaEnergiaAllSensor(SensorEntity):
    _attr_name = "Riva enargia całość"
    _attr_unique_id = "riva_energy_all"
    _attr_native_unit_of_measurement = "kWh"
    tinytuya.set_debug(False)
    # Connect to Tuya Cloud
    c = tinytuya.Cloud(
        apiRegion="eu", 
        apiKey=ACCESS_ID, 
        apiSecret=ACCESS_KEY, 
        apiDeviceID=device_id)       
    def update(self) -> None:
        global propertis
        result = self.c.cloudrequest(f'/v2.0/cloud/thing/{device_id}/shadow/properties')
        propertis = result['result']['properties']
        v = ValueCode(propertis, 'energy_forword_a')        
        self._attr_native_value = v/100
            
class RivaEnergiaKlimaSensor(SensorEntity):
    _attr_name = "Riva enargia klimatyzacja"
    _attr_unique_id = "riva_energy_clim"
    _attr_native_unit_of_measurement = "kWh"
    def update(self):
        if propertis:
            v = ValueCode(propertis, 'energy_forword_b')        
            self._attr_native_value = v/100

i zapis w pliku configuration.yaml

sensor:
  - platform: riva_pogoda
    scan_interval: 300
  - platform: riva_energia
    scan_interval: 300

Powyższa wersja kodu przedstawia stare rozwiązanie kodowania komponentów. Teraz zalecana jest budowanie kodu w trybie asynchronicznym (wątkowym). Da tak prostych komponentów, stare rozwiązanie też spełnia w 100% swoje zadanie.

Dodaj komentarz