diff --git a/custom_components/tech/__init__.py b/custom_components/tech/__init__.py index 08dea6c..e22e40c 100644 --- a/custom_components/tech/__init__.py +++ b/custom_components/tech/__init__.py @@ -9,7 +9,7 @@ from homeassistant.core import HomeAssistant from homeassistant.helpers import aiohttp_client from homeassistant.helpers.typing import ConfigType -from custom_components.tech.tech_update_coordinator import TechUpdateCoordinator +from .tech_update_coordinator import TechUpdateCoordinator from .const import DOMAIN from .tech import Tech diff --git a/custom_components/tech/climate.py b/custom_components/tech/climate.py index fe6d387..56c94b6 100644 --- a/custom_components/tech/climate.py +++ b/custom_components/tech/climate.py @@ -4,7 +4,8 @@ import logging from typing import Any, Final -from custom_components.tech.tech_update_coordinator import TechUpdateCoordinator +from .models.module import ZoneElement +from .tech_update_coordinator import TechUpdateCoordinator from homeassistant.components.climate import ( ClimateEntity, ClimateEntityFeature, @@ -63,18 +64,18 @@ class TechThermostat(CoordinatorEntity, ClimateEntity): _attr_supported_features = ClimateEntityFeature.TARGET_TEMPERATURE | ClimateEntityFeature.PRESET_MODE _attr_preset_modes = DEFAULT_PRESETS - def __init__(self, device: dict[str, Any], coordinator, api: Tech) -> None: + def __init__(self, device: ZoneElement, coordinator, api: Tech) -> None: """Initialize the Tech device.""" self._api = api - self._id: int = device["zone"]["id"] - self._zone_mode_id = device["mode"]["id"] + self._id: int = device.zone.id + self._zone_mode_id = device.mode.id self._udid = coordinator.udid # Set unique_id first as it's required for entity registry self._attr_unique_id = f"{self._udid}_{self._id}" self._attr_device_info = { "identifiers": {(DOMAIN, self._attr_unique_id)}, - "name": device["description"]["name"], + "name": device.description.name, "manufacturer": "Tech", } @@ -91,21 +92,21 @@ def __init__(self, device: dict[str, Any], coordinator, api: Tech) -> None: self.update_properties(coordinator.get_zones()[self._id]) - def update_properties(self, device: dict[str, Any]) -> None: + def update_properties(self, device: ZoneElement) -> None: """Update the properties from device data.""" - self._attr_name = device["description"]["name"] + self._attr_name = device.description.name - zone = device["zone"] - if zone["setTemperature"] is not None: - self._attr_target_temperature = zone["setTemperature"] / 10 + zone = device.zone + if zone.setTemperature is not None: + self._attr_target_temperature = zone.setTemperature / 10 - if zone["currentTemperature"] is not None: - self._attr_current_temperature = zone["currentTemperature"] / 10 + if zone.currentTemperature is not None: + self._attr_current_temperature = zone.currentTemperature / 10 - if zone["humidity"] is not None: - self._attr_current_humidity = zone["humidity"] + if zone.humidity is not None: + self._attr_current_humidity = zone.humidity - state = zone["flags"]["relayState"] + state = zone.flags.relayState if state == "on": self._attr_hvac_action = HVACAction.HEATING elif state == "off": @@ -113,7 +114,7 @@ def update_properties(self, device: dict[str, Any]) -> None: else: self._attr_hvac_action = HVACAction.OFF - mode = zone["zoneState"] + mode = zone.zoneState self._attr_hvac_mode = HVACMode.HEAT if mode in ["zoneOn", "noAlarm"] else HVACMode.OFF @callback @@ -137,32 +138,6 @@ async def async_set_temperature(self, **kwargs: Any) -> None: temperature, ex ) - - async def async_set_preset_mode(self, preset_mode: str) -> None: - try: - if self._attr_preset_mode == CHANGE_PRESET: - _LOGGER.debug("Preset mode change already in progress for %s", self._attr_name) - return - - preset_mode_id = DEFAULT_PRESETS.index(preset_mode) - await self._api.set_module_menu( - self._udid, - "mu", - 1000, - preset_mode_id - ) - - self._attr_preset_modes = [CHANGE_PRESET] - self._attr_preset_mode = CHANGE_PRESET - - await self.coordinator.async_request_refresh() - except Exception as ex: - _LOGGER.error( - "Failed to set preset mode for %s to %s: %s", - self._attr_name, - preset_mode, - ex - ) async def async_set_hvac_mode(self, hvac_mode: str) -> None: """Set new target hvac mode.""" diff --git a/custom_components/tech/config_flow.py b/custom_components/tech/config_flow.py index 9cd28e0..75cef72 100644 --- a/custom_components/tech/config_flow.py +++ b/custom_components/tech/config_flow.py @@ -1,4 +1,5 @@ """Config flow for Tech Sterowniki integration.""" +from typing import Any import logging, uuid import voluptuous as vol from homeassistant import config_entries, core, exceptions @@ -7,6 +8,7 @@ from .const import DOMAIN # pylint:disable=unused-import from .tech import Tech from types import MappingProxyType +from .models.module import Module, UserModule _LOGGER = logging.getLogger(__name__) @@ -28,7 +30,7 @@ async def validate_input(hass: core.HomeAssistant, data): if not await api.authenticate(data["username"], data["password"]): raise InvalidAuth modules = await api.list_modules() - + # If you cannot connect: # throw CannotConnect # If the authentication is wrong: @@ -42,6 +44,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): """Handle a config flow for Tech Sterowniki.""" VERSION = 1 + MINOR_VERSION = 1 # Pick one of the available connection classes in homeassistant/config_entries.py CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL @@ -50,10 +53,10 @@ async def async_step_user(self, user_input=None): errors = {} if user_input is not None: try: - _LOGGER.debug("Context: " + str(self.context)) + _LOGGER.debug("Context: " + str(self.context)) validated_input = await validate_input(self.hass, user_input) - modules = self._create_modules_array(validated_input=validated_input) + modules: list[UserModule] = self._create_modules_array(validated_input=validated_input) if len(modules) == 0: return self.async_abort("no_modules") @@ -61,8 +64,8 @@ async def async_step_user(self, user_input=None): if len(modules) > 1: for module in modules[1:len(modules)]: await self.hass.config_entries.async_add(self._create_config_entry(module=module)) - - return self.async_create_entry(title=modules[0]["version"], data=modules[0]) + + return self.async_create_entry(title=modules[0].module_title, data=modules[0]) except CannotConnect: errors["base"] = "cannot_connect" except InvalidAuth: @@ -85,10 +88,10 @@ async def async_step_reauth(self, user_input=None): return await self.async_step_user() - def _create_config_entry(self, module: dict) -> ConfigEntry: + def _create_config_entry(self, module: UserModule) -> ConfigEntry: return ConfigEntry( data=module, - title=module["version"], + title=module.module_title, entry_id=uuid.uuid4().hex, discovery_keys=MappingProxyType({}), domain=DOMAIN, @@ -99,19 +102,19 @@ def _create_config_entry(self, module: dict) -> ConfigEntry: unique_id=None, subentries_data=[]) - def _create_modules_array(self, validated_input: dict) -> [dict]: + def _create_modules_array(self, validated_input: dict) -> list[UserModule]: return [ self._create_module_dict(validated_input, module_dict) for module_dict in validated_input["modules"] ] - def _create_module_dict(self, validated_input: dict, module_dict: dict) -> dict: - return { - "user_id": validated_input["user_id"], - "token": validated_input["token"], - "module": module_dict, - "version": module_dict["version"] + ": " + module_dict["name"] - } + def _create_module_dict(self, validated_input: dict, module: Module) -> UserModule: + return UserModule( + user_id=validated_input["user_id"], + token=validated_input["token"], + module=module, + module_title=module.version + ": " + module.name + ) class CannotConnect(exceptions.HomeAssistantError): @@ -119,4 +122,4 @@ class CannotConnect(exceptions.HomeAssistantError): class InvalidAuth(exceptions.HomeAssistantError): - """Error to indicate there is invalid auth.""" + """Error to indicate there is invalid auth.""" \ No newline at end of file diff --git a/custom_components/tech/models/__init__.py b/custom_components/tech/models/__init__.py new file mode 100644 index 0000000..5f10151 --- /dev/null +++ b/custom_components/tech/models/__init__.py @@ -0,0 +1,50 @@ +"""Models package for Tech API responses.""" +from .module_menu import ( + MenuElementOption, + MenuElementParams, + MenuElement, + ModuleMenuData, + ModuleMenuResponse, +) +from .module import ( + Module, + ModuleData, + Zones, + ZoneElement, + Zone, + ZoneDescription, + ZoneMode, + ZoneSchedule, + ZoneFlags, + ScheduleInterval, + GlobalSchedules, + GlobalSchedule, + ControllerParameters, + ControllerMode, + Tile, + TileParams, +) + +__all__ = [ + "MenuElementOption", + "MenuElementParams", + "MenuElement", + "ModuleMenuData", + "ModuleMenuResponse", + "Module", + "ModuleData", + "Zones", + "ZoneElement", + "Zone", + "ZoneDescription", + "ZoneMode", + "ZoneSchedule", + "ZoneFlags", + "ScheduleInterval", + "GlobalSchedules", + "GlobalSchedule", + "ControllerParameters", + "ControllerMode", + "Tile", + "TileParams", +] diff --git a/custom_components/tech/models/module.py b/custom_components/tech/models/module.py new file mode 100644 index 0000000..60ae07f --- /dev/null +++ b/custom_components/tech/models/module.py @@ -0,0 +1,190 @@ +"""Models for module API responses.""" +from typing import Any, Optional +from pydantic import BaseModel + +class Module(BaseModel): + """Represents a single module.""" + id: int + default: bool + name: str + email: str + type: str + controllerStatus: str + moduleStatus: str + additionalInformation: str + phoneNumber: Optional[str] = None + zipCode: str + tag: Optional[str] = None + country: Optional[str] = None + gmtId: int + gmtTime: str + postcodePolicyAccepted: bool + style: str + version: str + company: str + udid: str + +class UserModule(BaseModel): + """Represents a user module.""" + user_id: str + token: str + module: Module + module_title: str + +# Models for get_module_data response + +class ZoneFlags(BaseModel): + """Represents zone flags.""" + relayState: str + minOneWindowOpen: bool + algorithm: str + floorSensor: int + humidityAlgorytm: int + zoneExcluded: int + + +class Zone(BaseModel): + """Represents zone information.""" + id: int + parentId: int + time: str + duringChange: bool + index: int + currentTemperature: int + setTemperature: int + flags: ZoneFlags + zoneState: str + signalStrength: Optional[int] = None + batteryLevel: Optional[int] = None + actuatorsOpen: int + humidity: int + visibility: bool + + +class ZoneDescription(BaseModel): + """Represents zone description.""" + id: int + parentId: int + name: str + styleId: int + styleIcon: str + duringChange: bool + + +class ZoneMode(BaseModel): + """Represents zone mode.""" + id: int + parentId: int + mode: str + constTempTime: int + setTemperature: int + scheduleIndex: int + + +class ScheduleInterval(BaseModel): + """Represents a schedule interval.""" + start: int + stop: int + temp: int + + +class ZoneSchedule(BaseModel): + """Represents zone schedule.""" + id: int + parentId: int + index: int + p0Days: list[str] + p0Intervals: list[ScheduleInterval] + p0SetbackTemp: int + p1Days: list[str] + p1Intervals: list[ScheduleInterval] + p1SetbackTemp: int + + +class ZoneElement(BaseModel): + """Represents a zone element.""" + zone: Zone + description: ZoneDescription + mode: ZoneMode + schedule: ZoneSchedule + actuators: list[Any] + underfloor: dict[str, Any] + windowsSensors: list[Any] + additionalContacts: list[Any] + + +class GlobalSchedule(BaseModel): + """Represents a global schedule.""" + id: int + parentId: int + index: int + name: str + p0Days: list[str] + p0SetbackTemp: int + p0Intervals: list[ScheduleInterval] + p1Days: list[str] + p1SetbackTemp: int + p1Intervals: list[ScheduleInterval] + + +class GlobalSchedules(BaseModel): + """Represents global schedules container.""" + time: str + duringChange: bool + elements: list[GlobalSchedule] + + +class ControllerMode(BaseModel): + """Represents controller mode.""" + id: int + parentId: int + type: int + txtId: int + iconId: int + value: int + menuId: int + + +class ControllerParameters(BaseModel): + """Represents controller parameters.""" + controllerMode: ControllerMode + globalSchedulesNumber: dict[str, Any] + + +class Zones(BaseModel): + """Represents zones container.""" + transaction_time: str + elements: list[ZoneElement] + globalSchedules: GlobalSchedules + controllerParameters: ControllerParameters + + +class TileParams(BaseModel): + """Represents tile parameters.""" + description: str + txtId: int + iconId: int + version: Optional[str] = None + companyId: Optional[int] = None + controllerName: Optional[str] = None + mainControllerId: Optional[int] = None + workingStatus: Optional[bool] = None + + +class Tile(BaseModel): + """Represents a tile.""" + id: int + parentId: int + type: int + menuId: int + orderId: Optional[int] = None + visibility: bool + params: TileParams + + +class ModuleData(BaseModel): + """Represents the full response from get_module_data API.""" + zones: Zones + tiles: list[Tile] + tilesOrder: Optional[Any] = None + tilesLastUpdate: str diff --git a/custom_components/tech/models/module_menu.py b/custom_components/tech/models/module_menu.py new file mode 100644 index 0000000..e96f6a1 --- /dev/null +++ b/custom_components/tech/models/module_menu.py @@ -0,0 +1,46 @@ +"""Models for module menu API responses.""" +from typing import Optional +from pydantic import BaseModel + + +class MenuElementOption(BaseModel): + """Represents a single option in a radio button control.""" + txtId: int + value: int + + +class MenuElementParams(BaseModel): + """Represents parameters of a menu element.""" + description: str + value: Optional[int] = None + default: Optional[int] = None + options: Optional[list[MenuElementOption]] = None + txtId: Optional[int] = None + type: Optional[int] = None + blockHide: Optional[int] = None + + +class MenuElement(BaseModel): + """Represents a single menu element.""" + menuType: str + type: int + id: int + parentId: int + access: bool + txtId: int + wikiTxtId: int + iconId: int + params: MenuElementParams + duringChange: Optional[str] + + +class ModuleMenuData(BaseModel): + """Represents the data section of module menu response.""" + elements: list[MenuElement] + transaction_time: str + + +class ModuleMenuResponse(BaseModel): + """Represents the full response from get_module_menu API.""" + status: str + data: ModuleMenuData diff --git a/custom_components/tech/select.py b/custom_components/tech/select.py index 048beff..204b690 100644 --- a/custom_components/tech/select.py +++ b/custom_components/tech/select.py @@ -4,7 +4,9 @@ import logging from typing import Any -from custom_components.tech.tech_update_coordinator import TechUpdateCoordinator +from .models.module import Module, UserModule +from .models.module_menu import MenuElement, MenuElement, ModuleMenuData +from .tech_update_coordinator import TechUpdateCoordinator from homeassistant.components.select import SelectEntity @@ -32,25 +34,26 @@ async def async_setup_entry( entry: ConfigEntry, async_add_entities: AddEntitiesCallback, ) -> bool: - """Set up Tech climate based on config_entry.""" + """Set up Tech select based on config_entry.""" api: Tech = hass.data[DOMAIN][entry.entry_id]["api"] coordinator: TechUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]["coordinator"] + module_data = Module(**entry.data["module"]) try: async_add_entities( - [TechHub(entry.data["module"], coordinator, api)] + [TechHub(module_data, coordinator, api)] ) return True except Exception as ex: - _LOGGER.error("Failed to set up Tech climate: %s", ex) + _LOGGER.error("Failed to set up Tech select: %s", ex) return False class TechHub(CoordinatorEntity, SelectEntity): _attr_options: list[str] = list(DEFAULT_PRESETS.values()) _attr_current_option: str | None = None - def __init__(self, hub, coordinator, api: Tech) -> None: + def __init__(self, module: Module, coordinator: TechUpdateCoordinator, api: Tech) -> None: """Initialize the Tech Hub device.""" self._api = api self._udid = coordinator.udid @@ -60,14 +63,14 @@ def __init__(self, hub, coordinator, api: Tech) -> None: self._attr_unique_id = self._udid self._attr_device_info = { "identifiers": {(DOMAIN, self._attr_unique_id)}, - "name": hub["name"], + "name": module.name, "manufacturer": "Tech", } super().__init__(coordinator, context=self._udid) # Initialize attributes that will be updated - self._attr_name: str | None = hub["name"] + self._attr_name: str | None = module.name self.update_properties(coordinator.get_menu()) @@ -78,19 +81,19 @@ def _handle_coordinator_update(self) -> None: self.update_properties(self.coordinator.get_menu()) self.async_write_ha_state() - def update_properties(self, device_menu_config: dict[str, Any] | None) -> None: - heating_mode = self.get_heating_mode_from_menu_config(device_menu_config) if device_menu_config else None + def update_properties(self, module_menu_data: ModuleMenuData | None) -> None: + heating_mode = self.get_heating_mode_from_menu_config(module_menu_data) if module_menu_data else None _LOGGER.debug("Updating heating mode for hub %s: %s", self._attr_name, heating_mode) if heating_mode is not None: - if heating_mode["duringChange"] == "t": + if heating_mode.duringChange == "t": _LOGGER.debug("Preset mode change in progress for %s", self._attr_name) self._attr_options = [CHANGE_PRESET] self._attr_current_option = CHANGE_PRESET _LOGGER.debug("Current preset mode for %s: %s", self._attr_name, self._attr_current_option) else: self._attr_options = list(DEFAULT_PRESETS.values()) - heating_mode_id = heating_mode["params"]["value"] + heating_mode_id = heating_mode.params.value self._attr_current_option = self.map_heating_mode_id_to_name(heating_mode_id) _LOGGER.debug("Current preset mode for %s: %s", self._attr_name, self._attr_current_option) else: @@ -122,15 +125,14 @@ async def async_select_option(self, option: str) -> None: ex ) - def get_heating_mode_from_menu_config(self, menu_config: dict[str, Any]) -> dict[str, Any] | None: + def get_heating_mode_from_menu_config(self, menu_config: ModuleMenuData) -> MenuElement | None: """Get current preset mode from menu config.""" - element = None + heating_mode_menu_id = 1000 - for e in menu_config["elements"]: - if e["id"] == heating_mode_menu_id: - element = e - break - return element + for e in menu_config.elements: + if e.id == heating_mode_menu_id: + return e + return None def map_heating_mode_id_to_name(self, heating_mode_id) -> str: """Map heating mode id to preset mode name.""" diff --git a/custom_components/tech/tech.py b/custom_components/tech/tech.py index 419884a..c0a2ce8 100644 --- a/custom_components/tech/tech.py +++ b/custom_components/tech/tech.py @@ -6,7 +6,13 @@ import json import time import asyncio +from typing import Type, TypeVar, overload from aiocache import Cache, cached +from pydantic import BaseModel, TypeAdapter + +from .models import Module, ModuleData, ModuleMenuResponse, ZoneElement + +T = TypeVar("T", bound=BaseModel) logging.basicConfig(level=logging.DEBUG) _LOGGER = logging.getLogger(__name__) @@ -33,7 +39,13 @@ def __init__(self, session: aiohttp.ClientSession, user_id = None, token = None, self.authenticated = False self.zones = {} - async def get(self, request_path): + @overload + async def get(self, request_path: str) -> dict: ... + + @overload + async def get(self, request_path: str, response_type: Type[T]) -> T: ... + + async def get(self, request_path: str, response_type: Type[T] | None = None) -> dict | T: url = self.base_url + request_path _LOGGER.debug("Sending GET request: " + url) async with self.session.get(url, headers=self.headers) as response: @@ -43,6 +55,9 @@ async def get(self, request_path): data = await response.json() _LOGGER.debug(data) + + if response_type is not None: + return response_type.model_validate(data) return data async def post(self, request_path, post_data): @@ -72,25 +87,24 @@ async def authenticate(self, username, password): } return result["authenticated"] - async def list_modules(self): + async def list_modules(self) -> list[Module]: if self.authenticated: path = "users/" + self.user_id + "/modules" result = await self.get(path) + return TypeAdapter(list[Module]).validate_python(result) else: raise TechError(401, "Unauthorized") - return result - async def get_module_data(self, module_udid): + async def get_module_data(self, module_udid) -> ModuleData: _LOGGER.debug("Getting module data..." + module_udid + ", " + self.user_id) if self.authenticated: path = "users/" + self.user_id + "/modules/" + module_udid - result = await self.get(path) + return await self.get(path, ModuleData) else: raise TechError(401, "Unauthorized") - return result @cached(ttl=10, cache=Cache.MEMORY) - async def get_module_zones(self, module_udid): + async def get_module_zones(self, module_udid) -> dict[int, ZoneElement]: """Returns Tech module zones either from cache or it will update all the cached values for Tech module assuming no update has occurred for at least the [update_interval]. @@ -103,11 +117,11 @@ async def get_module_zones(self, module_udid): Dictionary of zones indexed by zone ID. """ result = await self.get_module_data(module_udid) - zones = result["zones"]["elements"] - zones = list(filter(lambda e: e['zone']['zoneState'] != "zoneUnregistered", zones)) - return { zone["zone"]["id"]: zone for zone in zones } + zones = result.zones.elements + zones = list(filter(lambda e: e.zone.zoneState != "zoneUnregistered", zones)) + return { zone.zone.id: zone for zone in zones } - async def get_zone(self, module_udid, zone_id): + async def get_zone(self, module_udid, zone_id) -> ZoneElement: """Returns zone from Tech API cache. Parameters: @@ -115,7 +129,7 @@ async def get_zone(self, module_udid, zone_id): zone_id (int): The Tech module zone ID. Returns: - Dictionary of zone. + ZoneElement object. """ zones = await self.get_module_zones(module_udid) return zones[zone_id] @@ -178,7 +192,7 @@ async def set_zone(self, module_udid, zone_id, on = True): return result @cached(ttl=10, cache=Cache.MEMORY) - async def get_module_menu(self, module_udid, menu_type): + async def get_module_menu(self, module_udid, menu_type) -> ModuleMenuResponse: """ Gets module menu options Parameters: @@ -186,16 +200,15 @@ async def get_module_menu(self, module_udid, menu_type): menu_type (string): Menu type, one of the following: "MU", "MI", "MS", "MP" Return: - JSON object with results + ModuleMenuResponse object with results """ _LOGGER.debug("Getting module menu: %s", menu_type) if self.authenticated: path = f"users/{self.user_id}/modules/{module_udid}/menu/{menu_type}" - result = await self.get(path) + return await self.get(path, ModuleMenuResponse) else: raise TechError(401, "Unauthorized") - return result async def set_module_menu(self, module_udid, menu_type, menu_id, menu_value): """ Sets module menu value diff --git a/custom_components/tech/tech_update_coordinator.py b/custom_components/tech/tech_update_coordinator.py index cb9b421..7704994 100644 --- a/custom_components/tech/tech_update_coordinator.py +++ b/custom_components/tech/tech_update_coordinator.py @@ -6,6 +6,8 @@ import async_timeout +from custom_components.tech.models.module import ZoneElement +from custom_components.tech.models.module_menu import ModuleMenuData from custom_components.tech.tech import (Tech, TechError) from homeassistant.exceptions import ConfigEntryAuthFailed from homeassistant.helpers.update_coordinator import ( @@ -37,11 +39,11 @@ def get_data(self) -> dict[str, Any]: """Return the latest data.""" return self.data - def get_zones(self) -> dict[str, Any]: + def get_zones(self) -> dict[int, ZoneElement]: """Return the latest zones data.""" return self.data["zones"] - def get_menu(self) -> dict[str, Any] | None: + def get_menu(self) -> ModuleMenuData | None: """Return the latest menu data.""" return self.data["menu"] @@ -59,11 +61,11 @@ async def _async_update_data(self): zones = await self.tech_api.get_module_zones(self.udid) menu = await self.tech_api.get_module_menu(self.udid, "mu") - if menu["status"] != "success": + if menu.status != "success": _LOGGER.warning("Failed to get menu config for Tech module %s, response: %s", self.udid, menu) menu = None - self.data = {"zones": zones, "menu": menu["data"] if menu else None} + self.data = {"zones": zones, "menu": menu.data if menu else None} return self.data except TechError as err: raise UpdateFailed(f"Error communicating with API: {err}")