diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 0000000..07448bf Binary files /dev/null and b/.DS_Store differ diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml old mode 100644 new mode 100755 index de14144..d367e89 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -1,7 +1,8 @@ +--- repos: - repo: https://github.com/astral-sh/ruff-pre-commit # Ruff version. - rev: v0.11.13 + rev: v0.13.0 hooks: # Run the linter. - id: ruff diff --git a/custom_components/.DS_Store b/custom_components/.DS_Store new file mode 100644 index 0000000..f495017 Binary files /dev/null and b/custom_components/.DS_Store differ diff --git a/custom_components/tech/__init__.py b/custom_components/tech/__init__.py index 373ee44..2c929ed 100644 --- a/custom_components/tech/__init__.py +++ b/custom_components/tech/__init__.py @@ -43,10 +43,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: websession = async_get_clientsession(hass) coordinator = TechCoordinator(hass, websession, user_id, token) + coordinator.config_entry = entry hass.data[DOMAIN][entry.entry_id] = coordinator await coordinator.async_config_entry_first_refresh() + # Load filter reset date from storage + await coordinator.async_load_filter_reset_date() + await assets.load_subtitles(language_code, Tech(websession, user_id, token)) await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) diff --git a/custom_components/tech/binary_sensor.py b/custom_components/tech/binary_sensor.py index 005962d..1d9f980 100644 --- a/custom_components/tech/binary_sensor.py +++ b/custom_components/tech/binary_sensor.py @@ -5,6 +5,10 @@ from homeassistant.components import binary_sensor from homeassistant.config_entries import ConfigEntry from homeassistant.const import ( + ATTR_IDENTIFIERS, + ATTR_MANUFACTURER, + CONF_MODEL, + CONF_NAME, CONF_PARAMS, CONF_TYPE, STATE_OFF, @@ -12,17 +16,27 @@ EntityCategory, ) from homeassistant.core import HomeAssistant +from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.typing import StateType, UndefinedType +from homeassistant.helpers.update_coordinator import CoordinatorEntity from . import TechCoordinator, assets from .const import ( CONTROLLER, DOMAIN, + FILTER_ALARM_MAX_DAYS, + INCLUDE_HUB_IN_NAME, + MANUFACTURER, + RECUPERATION_EXHAUST_FLOW, + RECUPERATION_SUPPLY_FLOW, + RECUPERATION_SUPPLY_FLOW_ALT, TYPE_ADDITIONAL_PUMP, TYPE_FIRE_SENSOR, TYPE_RELAY, + TYPE_TEMPERATURE_CH, UDID, + VER, VISIBILITY, ) from .entity import TileEntity @@ -63,6 +77,28 @@ async def async_setup_entry( if tile[CONF_TYPE] == TYPE_ADDITIONAL_PUMP: entities.append(RelaySensor(tile, coordinator, config_entry)) + # Check if we have recuperation system for filter replacement sensor + has_recuperation_flow = False + for t in tiles: + tile = tiles[t] + if tile.get("type") == TYPE_TEMPERATURE_CH: + widget1_txt_id = tile.get("params", {}).get("widget1", {}).get("txtId", 0) + widget2_txt_id = tile.get("params", {}).get("widget2", {}).get("txtId", 0) + for flow_sensor in [RECUPERATION_EXHAUST_FLOW, RECUPERATION_SUPPLY_FLOW, RECUPERATION_SUPPLY_FLOW_ALT]: + if flow_sensor["txt_id"] in [widget1_txt_id, widget2_txt_id]: + has_recuperation_flow = True + break + if has_recuperation_flow: + break + + # Add recuperation binary sensors if system detected + if has_recuperation_flow: + _LOGGER.debug("Creating recuperation binary sensors") + entities.extend([ + FilterReplacementSensor(coordinator, config_entry), + RecuperationSystemStatusSensor(coordinator, config_entry), + ]) + async_add_entities(entities, True) @@ -109,3 +145,143 @@ def __init__( def get_state(self, device): """Get device state.""" return device[CONF_PARAMS]["workingStatus"] + + +class FilterReplacementSensor(CoordinatorEntity, binary_sensor.BinarySensorEntity): + """Binary sensor to indicate when filter replacement is needed.""" + + _attr_has_entity_name = True + _attr_device_class = binary_sensor.BinarySensorDeviceClass.PROBLEM + _attr_icon = "mdi:air-filter-outline" + + def __init__( + self, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the filter replacement sensor.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_unique_id = f"{self._udid}_filter_replacement_needed" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Filter Replacement Needed" + + @property + def name(self) -> str: + """Return the name of the sensor.""" + return self._name + + @property + def is_on(self) -> bool | None: + """Return True if filter replacement is needed.""" + # Calculate if filter replacement is needed based on usage days + if hasattr(self._coordinator, '_filter_reset_date') and self._coordinator._filter_reset_date: + from datetime import datetime + reset_date = datetime.fromisoformat(self._coordinator._filter_reset_date) + current_date = datetime.now() + days_since_reset = (current_date - reset_date).days + return days_since_reset >= FILTER_ALARM_MAX_DAYS + else: + # No reset date stored, check if we're over the default max days + # Assume filter is new if no reset date + return False + + @property + def entity_category(self): + """Return the entity category for diagnostic entities.""" + return EntityCategory.DIAGNOSTIC + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, + CONF_NAME: f"{self._config_entry.title} Recuperation", + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), + ATTR_MANUFACTURER: MANUFACTURER, + } + + +class RecuperationSystemStatusSensor(CoordinatorEntity, binary_sensor.BinarySensorEntity): + """Binary sensor to indicate recuperation system operational status.""" + + _attr_has_entity_name = True + _attr_device_class = binary_sensor.BinarySensorDeviceClass.RUNNING + _attr_icon = "mdi:air-filter" + + def __init__( + self, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the system status sensor.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_unique_id = f"{self._udid}_recuperation_running" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Recuperation Running" + + @property + def name(self) -> str: + """Return the name of the sensor.""" + return self._name + + @property + def is_on(self) -> bool | None: + """Return True if recuperation system is running.""" + # Check if any flow sensors show activity + if self._coordinator.data and "tiles" in self._coordinator.data: + for tile_id, tile_data in self._coordinator.data["tiles"].items(): + if tile_data.get("type") == TYPE_TEMPERATURE_CH: + # Check flow sensors for activity + widget1_data = tile_data.get("params", {}).get("widget1", {}) + widget2_data = tile_data.get("params", {}).get("widget2", {}) + + # Check if any flow sensor shows activity (> 0) + for widget_data in [widget1_data, widget2_data]: + widget_txt_id = widget_data.get("txtId", 0) + for flow_sensor in [RECUPERATION_EXHAUST_FLOW, RECUPERATION_SUPPLY_FLOW, RECUPERATION_SUPPLY_FLOW_ALT]: + if flow_sensor["txt_id"] == widget_txt_id: + flow_value = widget_data.get("value", 0) + if flow_value and flow_value > 0: + return True + return False + + @property + def entity_category(self): + """Return the entity category for diagnostic entities.""" + return EntityCategory.DIAGNOSTIC + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, + CONF_NAME: f"{self._config_entry.title} Recuperation", + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), + ATTR_MANUFACTURER: MANUFACTURER, + } diff --git a/custom_components/tech/button.py b/custom_components/tech/button.py new file mode 100644 index 0000000..b67b60c --- /dev/null +++ b/custom_components/tech/button.py @@ -0,0 +1,274 @@ +"""Support for Tech HVAC button controls.""" + +import logging +from typing import Any + +from homeassistant.components.button import ButtonEntity +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import ( + ATTR_IDENTIFIERS, + ATTR_MANUFACTURER, + CONF_MODEL, + CONF_NAME, +) +from homeassistant.core import HomeAssistant +from homeassistant.helpers.device_registry import DeviceInfo +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.update_coordinator import CoordinatorEntity + +from .const import ( + CONTROLLER, + DOMAIN, + INCLUDE_HUB_IN_NAME, + MANUFACTURER, + RECUPERATION_EXHAUST_FLOW, + RECUPERATION_SUPPLY_FLOW, + RECUPERATION_SUPPLY_FLOW_ALT, + TYPE_TEMPERATURE_CH, + UDID, + VER, +) +from .coordinator import TechCoordinator + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up entry.""" + _LOGGER.debug( + "Setting up button entry, controller udid: %s", + config_entry.data[CONTROLLER][UDID], + ) + coordinator = hass.data[DOMAIN][config_entry.entry_id] + controller_udid = config_entry.data[CONTROLLER][UDID] + + tiles = await coordinator.api.get_module_tiles(controller_udid) + + entities = [] + + # Check if we have recuperation system (detected by flow sensors) + has_recuperation_flow = False + for t in tiles: + tile = tiles[t] + if tile.get("type") == TYPE_TEMPERATURE_CH: + widget1_txt_id = tile.get("params", {}).get("widget1", {}).get("txtId", 0) + widget2_txt_id = tile.get("params", {}).get("widget2", {}).get("txtId", 0) + for flow_sensor in [RECUPERATION_EXHAUST_FLOW, RECUPERATION_SUPPLY_FLOW, RECUPERATION_SUPPLY_FLOW_ALT]: + if flow_sensor["txt_id"] in [widget1_txt_id, widget2_txt_id]: + has_recuperation_flow = True + break + if has_recuperation_flow: + break + + # Create recuperation buttons if we have recuperation + if has_recuperation_flow: + _LOGGER.debug("Creating recuperation buttons: filter reset and party mode") + entities.append(FilterResetButton(coordinator, config_entry)) + entities.append(PartyModeButton(coordinator, config_entry)) + entities.append(QuickBoostButton(coordinator, config_entry)) + else: + _LOGGER.debug("No recuperation flow detected, skipping recuperation buttons") + + async_add_entities(entities, True) + + +class FilterResetButton(CoordinatorEntity, ButtonEntity): + """Representation of filter reset button.""" + + _attr_has_entity_name = True + _attr_icon = "mdi:air-filter" + + def __init__( + self, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the filter reset button.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_unique_id = f"{self._udid}_filter_reset" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Filter Reset" + + @property + def name(self) -> str: + """Return the name of the button.""" + return self._name + + async def async_press(self) -> None: + """Press the button to reset filter timer.""" + _LOGGER.debug("Resetting filter timer") + + # Store the current date as filter reset date + from datetime import datetime + current_date = datetime.now().isoformat() + + # Store in coordinator for immediate use by filter usage sensor + self._coordinator._filter_reset_date = current_date + + # Call API to reset filter data + await self._coordinator.api.update_filter_data(self._udid) + + # Store the reset date in Home Assistant storage + from homeassistant.helpers import storage + store = storage.Store( + self._coordinator.hass, + version=1, + key=f"{DOMAIN}_{self._udid}_filter_data" + ) + await store.async_save({ + "filter_reset_date": current_date + }) + + # Refresh coordinator data + await self._coordinator.async_request_refresh() + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, # Unique identifiers for the device + CONF_NAME: f"{self._config_entry.title} Recuperation", # Name of the device + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), # Model of the device + ATTR_MANUFACTURER: MANUFACTURER, # Manufacturer of the device + } + + +class PartyModeButton(CoordinatorEntity, ButtonEntity): + """Button to activate party mode with default duration.""" + + _attr_has_entity_name = True + _attr_icon = "mdi:party-popper" + + def __init__( + self, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the party mode button.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_unique_id = f"{self._udid}_party_mode_button" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Party Mode (60min)" + + @property + def name(self) -> str: + """Return the name of the button.""" + return self._name + + async def async_press(self) -> None: + """Press the button to activate party mode for 60 minutes.""" + _LOGGER.debug("Activating party mode for 60 minutes") + + try: + # Activate party mode for 60 minutes (reasonable default) + await self._coordinator.api.set_party_mode(self._udid, 60) + + # Refresh coordinator data to reflect the change + await self._coordinator.async_request_refresh() + + _LOGGER.info("Party mode activated successfully for 60 minutes") + + except Exception as err: + _LOGGER.error("Failed to activate party mode: %s", err) + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, + CONF_NAME: f"{self._config_entry.title} Recuperation", + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), + ATTR_MANUFACTURER: MANUFACTURER, + } + + +class QuickBoostButton(CoordinatorEntity, ButtonEntity): + """Button to activate quick boost mode for 30 minutes.""" + + _attr_has_entity_name = True + _attr_icon = "mdi:rocket-launch" + + def __init__( + self, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the quick boost button.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_unique_id = f"{self._udid}_quick_boost_button" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Quick Boost (30min)" + + @property + def name(self) -> str: + """Return the name of the button.""" + return self._name + + async def async_press(self) -> None: + """Press the button to activate quick boost mode for 30 minutes.""" + _LOGGER.debug("Activating quick boost mode for 30 minutes") + + try: + # Activate party mode for 30 minutes (short boost) + await self._coordinator.api.set_party_mode(self._udid, 30) + + # Refresh coordinator data to reflect the change + await self._coordinator.async_request_refresh() + + _LOGGER.info("Quick boost mode activated successfully for 30 minutes") + + except Exception as err: + _LOGGER.error("Failed to activate quick boost mode: %s", err) + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, + CONF_NAME: f"{self._config_entry.title} Recuperation", + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), + ATTR_MANUFACTURER: MANUFACTURER, + } \ No newline at end of file diff --git a/custom_components/tech/config_flow.py b/custom_components/tech/config_flow.py index 73bb3f0..4b9d3c9 100644 --- a/custom_components/tech/config_flow.py +++ b/custom_components/tech/config_flow.py @@ -1,33 +1,22 @@ """Config flow for Tech Sterowniki integration.""" import logging +import uuid from types import MappingProxyType from typing import Any -import uuid import voluptuous as vol - from homeassistant import config_entries, core, exceptions -from homeassistant.config_entries import SOURCE_USER, ConfigEntry, ConfigFlowResult -from homeassistant.const import ( - ATTR_ID, - CONF_NAME, - CONF_PASSWORD, - CONF_TOKEN, - CONF_USERNAME, -) -from homeassistant.helpers import aiohttp_client, config_validation as cv +from homeassistant.config_entries import (SOURCE_USER, ConfigEntry, + ConfigFlowResult) +from homeassistant.const import (ATTR_ID, CONF_NAME, CONF_PASSWORD, CONF_TOKEN, + CONF_USERNAME) +from homeassistant.helpers import aiohttp_client +from homeassistant.helpers import config_validation as cv from . import assets -from .const import ( - CONTROLLER, - CONTROLLERS, - DOMAIN, - INCLUDE_HUB_IN_NAME, - UDID, - USER_ID, - VER, -) +from .const import (CONTROLLER, CONTROLLERS, DOMAIN, INCLUDE_HUB_IN_NAME, UDID, + USER_ID, VER) from .tech import Tech, TechError, TechLoginError _LOGGER = logging.getLogger(__name__) @@ -226,6 +215,7 @@ def _create_config_entry(self, controller: dict) -> ConfigEntry: source=SOURCE_USER, options={}, unique_id=None, + subentries_data=[], ) def _create_controllers_array(self, validated_input: dict[str, Any]) -> list[dict]: diff --git a/custom_components/tech/const.py b/custom_components/tech/const.py index 59fcd49..06bb005 100644 --- a/custom_components/tech/const.py +++ b/custom_components/tech/const.py @@ -27,7 +27,7 @@ DEFAULT_ICON = "mdi:eye" -PLATFORMS = [Platform.BINARY_SENSOR, Platform.CLIMATE, Platform.SENSOR] +PLATFORMS = [Platform.BINARY_SENSOR, Platform.CLIMATE, Platform.SENSOR, Platform.FAN, Platform.NUMBER, Platform.SELECT, Platform.BUTTON, Platform.SWITCH] SCAN_INTERVAL: Final = timedelta(seconds=60) API_TIMEOUT: Final = 60 @@ -39,6 +39,7 @@ TYPE_RELAY = 11 TYPE_ADDITIONAL_PUMP = 21 TYPE_FAN = 22 +TYPE_RECUPERATION = 122 # Custom type for recuperation units TYPE_VALVE = 23 TYPE_MIXING_VALVE = 24 TYPE_FUEL_SUPPLY = 31 @@ -59,6 +60,7 @@ TYPE_FIRE_SENSOR: "mdi:fire", TYPE_ADDITIONAL_PUMP: "mdi:arrow-right-drop-circle-outline", TYPE_FAN: "mdi:fan", + TYPE_RECUPERATION: "mdi:air-filter", TYPE_VALVE: "mdi:valve", TYPE_MIXING_VALVE: "mdi:valve", # TODO: find a better icon TYPE_OPEN_THERM: "mdi:home-thermometer", @@ -68,6 +70,7 @@ TXT_ID_BY_TYPE = { TYPE_FIRE_SENSOR: 205, TYPE_FAN: 4135, + TYPE_RECUPERATION: 4135, # Using same txt_id as fan for now TYPE_VALVE: 991, TYPE_MIXING_VALVE: 5731, TYPE_FUEL_SUPPLY: 961, @@ -85,6 +88,113 @@ OPENTHERM_SET_TEMP = {"txt_id": 1058, "state_key": "setCurrentTemp"} OPENTHERM_SET_TEMP_DHW = {"txt_id": 1059, "state_key": "setTempDHW"} +# Recuperation flow sensor measured values +RECUPERATION_EXHAUST_FLOW = {"txt_id": 6131, "widget": "widget2", "name": "exhaust_flow"} +RECUPERATION_SUPPLY_FLOW = {"txt_id": 6132, "widget": "widget1", "name": "supply_flow"} +RECUPERATION_SUPPLY_FLOW_ALT = {"txt_id": 5994, "widget": "widget2", "name": "supply_flow"} + +# Recuperation temperature sensors (common txtIds) +RECUPERATION_TEMP_SENSORS = [ + {"txt_id": 119, "name": "Supply Air Temperature", "device_class": "temperature"}, + {"txt_id": 120, "name": "Exhaust Air Temperature", "device_class": "temperature"}, + {"txt_id": 121, "name": "External Air Temperature", "device_class": "temperature"}, + {"txt_id": 122, "name": "Discharge Air Temperature", "device_class": "temperature"}, + {"txt_id": 126, "name": "Supply Air Temperature", "device_class": "temperature"}, + {"txt_id": 127, "name": "Exhaust Air Temperature", "device_class": "temperature"}, + {"txt_id": 5995, "name": "Fresh Air Temperature", "device_class": "temperature"}, + {"txt_id": 5996, "name": "Extract Air Temperature", "device_class": "temperature"}, + {"txt_id": 5997, "name": "Supply Air Temperature", "device_class": "temperature"}, + {"txt_id": 5998, "name": "Exhaust Air Temperature", "device_class": "temperature"}, + {"txt_id": 5999, "name": "Heat Exchanger Temperature", "device_class": "temperature"}, + {"txt_id": 6000, "name": "Preheater Temperature", "device_class": "temperature"}, +] + +# Recuperation speed control endpoints (ido_id mapping) +RECUPERATION_SPEED_ENDPOINTS = { + 1: {"ido_id": 1737}, # Low speed + 2: {"ido_id": 1748}, # Medium speed + 3: {"ido_id": 1739}, # High speed +} + +# Default values for speed configuration (can be overridden by user) +DEFAULT_SPEED_VALUES = { + 1: 120, # Low speed default - 120 m³/h + 2: 280, # Medium speed default - 280 m³/h + 3: 390, # High speed default - 390 m³/h +} + +# Speed configuration ranges for each gear +SPEED_RANGES = { + 1: {"min": 60, "max": 280, "step": 10}, # Speed 1: 60-280 m³/h + 2: {"min": 120, "max": 400, "step": 10}, # Speed 2: 120-400 m³/h + 3: {"min": 280, "max": 400, "step": 10}, # Speed 3: 280-400 m³/h +} + +# Configuration keys for storing user values +SPEED_CONFIG_KEYS = { + 1: "recuperation_speed_1_flow", + 2: "recuperation_speed_2_flow", + 3: "recuperation_speed_3_flow", +} + +# Humidity sensor txtId values (unit: 8, type: 2) +# 2024: Pokój czujnik bezprzewodowy 1 +# 2025: Pokój czujnik bezprzewodowy 2 +# 2027: Łazienka czujnik bezprzewodowy 4 +# 2658: Łazienka 2 czujnik bezprzewodowy 5 +# TODO: Add txtId for "łazienka 3 czujnik bezprzewodowy 6" when available +HUMIDITY_SENSOR_TXT_IDS = [2024, 2025, 2027, 2658] + +# Party mode settings +PARTY_MODE_IDO_ID = 1447 +PARTY_MODE_MIN_MINUTES = 15 +PARTY_MODE_MAX_MINUTES = 720 + +# Direct gear control settings +GEAR_CONTROL_IDO_ID = 1833 +GEAR_OPTIONS = { + "stop": 0, # Zatrzymaj wentylator + "speed_1": 1, # 1 bieg + "speed_2": 2, # 2 bieg + "speed_3": 3, # 3 bieg +} +GEAR_OPTIONS_REVERSE = {v: k for k, v in GEAR_OPTIONS.items()} + +# Fan mode select settings (timed mode changes) +FAN_MODE_IDO_ID = 1966 +FAN_MODE_OPTIONS = { + "stop": 0, # Zatrzymaj wentylator + "speed_1": 1, # 1 bieg + "speed_2": 2, # 2 bieg + "speed_3": 3, # 3 bieg +} +FAN_MODE_OPTIONS_REVERSE = {v: k for k, v in FAN_MODE_OPTIONS.items()} + +# Filter management settings +FILTER_ALARM_IDO_ID = 2080 +FILTER_ALARM_MIN_DAYS = 30 +FILTER_ALARM_MAX_DAYS = 120 + +# Filter usage tracking +FILTER_USAGE_IDO_ID = 2081 + +# Ventilation parameters settings +VENTILATION_ROOM_IDO_ID = 2170 +VENTILATION_BATHROOM_IDO_ID = 2171 +CO2_THRESHOLD_IDO_ID = 2115 +HYSTERESIS_IDO_ID = 2239 + +# Flow balancing setting +FLOW_BALANCING_IDO_ID = 1733 + +# Ventilation parameter ranges +VENTILATION_MIN_PERCENT = 10 +VENTILATION_MAX_PERCENT = 90 +CO2_MIN_PPM = 400 +CO2_MAX_PPM = 2000 +HYSTERESIS_MIN_PERCENT = 5 +HYSTERESIS_MAX_PERCENT = 10 + TECH_SUPPORTED_LANGUAGES = [ "en", "fr", diff --git a/custom_components/tech/coordinator.py b/custom_components/tech/coordinator.py index 8c96d5d..94d279a 100644 --- a/custom_components/tech/coordinator.py +++ b/custom_components/tech/coordinator.py @@ -35,6 +35,24 @@ def __init__( update_interval=SCAN_INTERVAL, ) self.api = Tech(session, user_id, token) + self._filter_reset_date = None + self.hass = hass + + async def async_load_filter_reset_date(self) -> None: + """Load filter reset date from storage.""" + if self.config_entry: + from homeassistant.helpers import storage + + udid = self.config_entry.data[CONTROLLER][UDID] + store = storage.Store( + self.hass, + version=1, + key=f"{DOMAIN}_{udid}_filter_data" + ) + data = await store.async_load() + if data and "filter_reset_date" in data: + self._filter_reset_date = data["filter_reset_date"] + _LOGGER.debug("Loaded filter reset date: %s", self._filter_reset_date) async def _async_update_data(self) -> dict: """Fetch data from TECH API endpoint(s).""" diff --git a/custom_components/tech/fan.py b/custom_components/tech/fan.py new file mode 100644 index 0000000..c1d161a --- /dev/null +++ b/custom_components/tech/fan.py @@ -0,0 +1,503 @@ +"""Support for Tech HVAC fans.""" + +import logging +from typing import Any + +from homeassistant.components.fan import FanEntity, FanEntityFeature +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import ( + ATTR_IDENTIFIERS, + ATTR_MANUFACTURER, + CONF_DESCRIPTION, + CONF_MODEL, + CONF_NAME, + CONF_PARAMS, + CONF_TYPE, +) +from homeassistant.core import HomeAssistant +from homeassistant.helpers.device_registry import DeviceInfo +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.update_coordinator import CoordinatorEntity +from homeassistant.util.percentage import ( + int_states_in_range, + percentage_to_ranged_value, + ranged_value_to_percentage, +) + +from . import assets +from .const import ( + CONTROLLER, + DOMAIN, + INCLUDE_HUB_IN_NAME, + MANUFACTURER, + TYPE_FAN, + TYPE_RECUPERATION, + TYPE_TEMPERATURE_CH, + UDID, + VER, +) +from .coordinator import TechCoordinator +from .entity import TileEntity + +_LOGGER = logging.getLogger(__name__) + +SPEED_RANGE = (1, 3) # Recuperation typically has 3 speeds + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up entry.""" + _LOGGER.debug( + "Setting up fan entry, controller udid: %s", + config_entry.data[CONTROLLER][UDID], + ) + coordinator = hass.data[DOMAIN][config_entry.entry_id] + controller_udid = config_entry.data[CONTROLLER][UDID] + + tiles = await coordinator.api.get_module_tiles(controller_udid) + + entities = [] + for t in tiles: + tile = tiles[t] + if not tile.get("visibility", False): + continue + + # Check if this is a fan tile that could be a recuperation unit + if tile[CONF_TYPE] == TYPE_FAN: + # Check if it has recuperation-like characteristics + description = tile[CONF_PARAMS].get(CONF_DESCRIPTION, "").lower() + if any(keyword in description for keyword in ["recuperation", "rekuperacja", "ventilation", "wentylacja"]): + entities.append(TileRecuperationFanEntity(tile, coordinator, config_entry)) + else: + entities.append(TileFanEntity(tile, coordinator, config_entry)) + elif tile[CONF_TYPE] == TYPE_RECUPERATION: + entities.append(TileRecuperationFanEntity(tile, coordinator, config_entry)) + + # Check if we found any recuperation flow sensors and create a virtual fan control + has_recuperation_flow = False + for t in tiles: + tile = tiles[t] + if tile[CONF_TYPE] == TYPE_TEMPERATURE_CH: + widget1_txt_id = tile[CONF_PARAMS].get("widget1", {}).get("txtId", 0) + widget2_txt_id = tile[CONF_PARAMS].get("widget2", {}).get("txtId", 0) + from .const import ( + RECUPERATION_EXHAUST_FLOW, + RECUPERATION_SUPPLY_FLOW, + RECUPERATION_SUPPLY_FLOW_ALT, + ) + for flow_sensor in [RECUPERATION_EXHAUST_FLOW, RECUPERATION_SUPPLY_FLOW, RECUPERATION_SUPPLY_FLOW_ALT]: + if flow_sensor["txt_id"] in [widget1_txt_id, widget2_txt_id]: + has_recuperation_flow = True + break + if has_recuperation_flow: + break + + # Create recuperation fan control if we detected flow sensors + if has_recuperation_flow: + # Create a virtual tile for recuperation control + virtual_tile = { + "id": 9999, # Virtual ID + "type": TYPE_RECUPERATION, + "params": {"description": "Recuperation Control"} + } + entities.append(VirtualRecuperationFanEntity(virtual_tile, coordinator, config_entry)) + + async_add_entities(entities, True) + + +class TileFanEntity(TileEntity, CoordinatorEntity, FanEntity): + """Representation of a Tech Fan.""" + + _attr_supported_features = FanEntityFeature.SET_SPEED | FanEntityFeature.PRESET_MODE + _attr_preset_modes = ["Stop", "Speed 1", "Speed 2", "Speed 3", "Party Mode"] + + def __init__( + self, + device, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the fan.""" + TileEntity.__init__(self, device, coordinator, config_entry) + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_icon = assets.get_icon_by_type(device[CONF_TYPE]) + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + assets.get_text_by_type(device[CONF_TYPE]) + + @property + def unique_id(self) -> str: + """Return a unique ID.""" + return f"{self._unique_id}_tile_fan" + + @property + def name(self) -> str: + """Return the name of the fan.""" + return self._name + + @property + def is_on(self) -> bool: + """Return true if the fan is on.""" + if self._coordinator.data and "tiles" in self._coordinator.data: + tile_data = self._coordinator.data["tiles"].get(self._id) + if tile_data: + gear = tile_data[CONF_PARAMS].get("gear", 0) + return gear > 0 + return False + + @property + def percentage(self) -> int | None: + """Return the current speed percentage.""" + if self._coordinator.data and "tiles" in self._coordinator.data: + tile_data = self._coordinator.data["tiles"].get(self._id) + if tile_data: + gear = tile_data[CONF_PARAMS].get("gear", 0) + if gear == 0: + return 0 + return ranged_value_to_percentage(SPEED_RANGE, gear) + return 0 + + @property + def speed_count(self) -> int: + """Return the number of speeds the fan supports.""" + return int_states_in_range(SPEED_RANGE) + + async def async_set_percentage(self, percentage: int) -> None: + """Set the speed percentage of the fan.""" + if percentage == 0: + await self.async_turn_off() + else: + speed_level = percentage_to_ranged_value(SPEED_RANGE, percentage) + # For now, use None for configured_values (will use defaults) + await self._coordinator.api.set_recuperation_speed(self._udid, speed_level, None) + await self._coordinator.async_request_refresh() + + async def async_turn_on( + self, + percentage: int | None = None, + preset_mode: str | None = None, + **kwargs: Any, + ) -> None: + """Turn on the fan.""" + if percentage is None: + percentage = 33 # Default to speed 1 (33%) + await self.async_set_percentage(percentage) + + async def async_turn_off(self, **kwargs: Any) -> None: + """Turn off the fan.""" + await self._coordinator.api.set_recuperation_speed(self._udid, 0, None) + await self._coordinator.async_request_refresh() + + @property + def preset_mode(self) -> str | None: + """Return the current preset mode.""" + # Check if party mode is active + if self._is_party_mode_active(): + return "Party Mode" + + if not self.is_on: + return "Stop" + + percentage = self.percentage + if percentage == 0: + return "Stop" + elif percentage <= 33: + return "Speed 1" + elif percentage <= 66: + return "Speed 2" + else: + return "Speed 3" + + async def async_set_preset_mode(self, preset_mode: str) -> None: + """Set the preset mode.""" + _LOGGER.debug("Setting fan preset mode to %s", preset_mode) + + if preset_mode == "Party Mode": + # Activate party mode with configured duration + party_duration = self._get_party_mode_duration() + await self._coordinator.api.set_party_mode(self._udid, party_duration) + await self._coordinator.async_request_refresh() + return + + preset_to_mode = { + "Stop": 0, + "Speed 1": 1, + "Speed 2": 2, + "Speed 3": 3 + } + + mode_value = preset_to_mode.get(preset_mode, 0) + + if mode_value == 0: + await self.async_turn_off() + else: + # Use configured speed values + configured_values = self._get_configured_speed_values() + await self._coordinator.api.set_recuperation_speed(self._udid, mode_value, configured_values) + await self._coordinator.async_request_refresh() + + def _get_configured_speed_values(self) -> dict: + """Get configured speed values from number entities or use defaults.""" + from .const import DEFAULT_SPEED_VALUES + + configured_values = DEFAULT_SPEED_VALUES.copy() + + # Try to get current values from number entities via Home Assistant + if hasattr(self._coordinator, 'hass'): + hass = self._coordinator.hass + for speed_level in [1, 2, 3]: + entity_id = f"number.{self._udid}_recuperation_speed_{speed_level}_config" + state = hass.states.get(entity_id) + if state and state.state not in ["unavailable", "unknown"]: + try: + configured_values[speed_level] = int(float(state.state)) + _LOGGER.debug("Got configured speed %s: %s m³/h", speed_level, configured_values[speed_level]) + except (ValueError, TypeError): + _LOGGER.debug("Could not parse speed %s value: %s", speed_level, state.state) + + return configured_values + + def _is_party_mode_active(self) -> bool: + """Check if party mode is currently active.""" + # This would need to be implemented based on how party mode state is tracked + # For now, we can't easily detect if party mode is active without additional API calls + # Could be enhanced by checking a stored party mode end time + return False + + def _get_party_mode_duration(self) -> int: + """Get configured party mode duration from number entity or use default.""" + + default_duration = 60 # Default 1 hour + + # Try to get current value from party mode number entity + if hasattr(self._coordinator, 'hass'): + hass = self._coordinator.hass + entity_id = f"number.{self._udid}_recuperation_party_mode" + state = hass.states.get(entity_id) + if state and state.state not in ["unavailable", "unknown"]: + try: + duration = int(float(state.state)) + _LOGGER.debug("Got configured party mode duration: %s minutes", duration) + return duration + except (ValueError, TypeError): + _LOGGER.debug("Could not parse party mode duration: %s", state.state) + + return default_duration + + def get_state(self, device): + """Get device state.""" + gear = device[CONF_PARAMS].get("gear", 0) + return gear > 0 + + def update_properties(self, device): + """Update the properties of the device.""" + self._state = self.is_on + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, self._unique_id) + }, # Unique identifiers for the device + CONF_NAME: self._name, # Name of the device + CONF_MODEL: self._config_entry.data[CONTROLLER][CONF_NAME] + ": " + self._config_entry.data[CONTROLLER][VER], # Model of the device + ATTR_MANUFACTURER: MANUFACTURER, # Manufacturer of the device + } + + +class TileRecuperationFanEntity(TileFanEntity): + """Representation of a Tech Recuperation Unit.""" + + _attr_supported_features = FanEntityFeature.SET_SPEED | FanEntityFeature.PRESET_MODE + _attr_preset_modes = ["Stop", "Speed 1", "Speed 2", "Speed 3", "Party Mode"] + + def __init__( + self, + device, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the recuperation unit.""" + super().__init__(device, coordinator, config_entry) + self._attr_icon = "mdi:air-filter" + + # Update name to indicate it's a recuperation unit + base_name = assets.get_text_by_type(device[CONF_TYPE]) + if "recuperation" not in base_name.lower() and "rekuperacja" not in base_name.lower(): + base_name = f"Recuperation {base_name}" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + base_name + + @property + def unique_id(self) -> str: + """Return a unique ID.""" + return f"{self._unique_id}_tile_recuperation" + + +class VirtualRecuperationFanEntity(TileFanEntity): + """Virtual recuperation fan entity for systems without direct fan tiles.""" + + _attr_supported_features = FanEntityFeature.SET_SPEED | FanEntityFeature.PRESET_MODE + _attr_preset_modes = ["Stop", "Speed 1", "Speed 2", "Speed 3", "Party Mode"] + + def __init__( + self, + device, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the virtual recuperation fan.""" + super().__init__(device, coordinator, config_entry) + self._attr_icon = "mdi:air-filter" + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Recuperation" + + @property + def unique_id(self) -> str: + """Return a unique ID.""" + return f"{self._udid}_virtual_recuperation" + + @property + def is_on(self) -> bool: + """Return true if the recuperation is on.""" + # Check flow values to determine if recuperation is running + if self._coordinator.data and "tiles" in self._coordinator.data: + from .const import ( + RECUPERATION_EXHAUST_FLOW, + RECUPERATION_SUPPLY_FLOW, + RECUPERATION_SUPPLY_FLOW_ALT, + ) + + for tile_id, tile_data in self._coordinator.data["tiles"].items(): + if tile_data.get("type") == TYPE_TEMPERATURE_CH: + widget1_txt_id = tile_data.get("params", {}).get("widget1", {}).get("txtId", 0) + widget2_txt_id = tile_data.get("params", {}).get("widget2", {}).get("txtId", 0) + + # Check for flow values + for flow_sensor in [RECUPERATION_EXHAUST_FLOW, RECUPERATION_SUPPLY_FLOW, RECUPERATION_SUPPLY_FLOW_ALT]: + if flow_sensor["txt_id"] in [widget1_txt_id, widget2_txt_id]: + widget_key = flow_sensor["widget"] + flow_value = tile_data.get("params", {}).get(widget_key, {}).get("value", 0) + if flow_value > 0: + return True + return False + + @property + def percentage(self) -> int | None: + """Return the current speed percentage based on flow values.""" + if not self.is_on: + return 0 + + # Try to determine speed based on flow values + if self._coordinator.data and "tiles" in self._coordinator.data: + from .const import ( + RECUPERATION_EXHAUST_FLOW, + RECUPERATION_SUPPLY_FLOW, + RECUPERATION_SUPPLY_FLOW_ALT, + ) + + for tile_id, tile_data in self._coordinator.data["tiles"].items(): + if tile_data.get("type") == TYPE_TEMPERATURE_CH: + widget1_txt_id = tile_data.get("params", {}).get("widget1", {}).get("txtId", 0) + widget2_txt_id = tile_data.get("params", {}).get("widget2", {}).get("txtId", 0) + + for flow_sensor in [RECUPERATION_EXHAUST_FLOW, RECUPERATION_SUPPLY_FLOW, RECUPERATION_SUPPLY_FLOW_ALT]: + if flow_sensor["txt_id"] in [widget1_txt_id, widget2_txt_id]: + widget_key = flow_sensor["widget"] + flow_value = tile_data.get("params", {}).get(widget_key, {}).get("value", 0) + + # Map flow values to speed levels based on defaults + # Use more flexible ranges since users can configure speeds + if flow_value >= 300: # High speed range + return ranged_value_to_percentage(SPEED_RANGE, 3) # High + elif flow_value >= 200: # Medium speed range + return ranged_value_to_percentage(SPEED_RANGE, 2) # Medium + elif flow_value >= 50: # Low speed range + return ranged_value_to_percentage(SPEED_RANGE, 1) # Low + + return 33 # Default to low speed if we can't determine + + @property + def preset_mode(self) -> str | None: + """Return the current preset mode.""" + # Check if party mode is active + if self._is_party_mode_active(): + return "Party Mode" + + # Determine current mode based on flow values or stored state + if not self.is_on: + return "Stop" + + percentage = self.percentage + if percentage == 0: + return "Stop" + elif percentage <= 33: + return "Speed 1" + elif percentage <= 66: + return "Speed 2" + else: + return "Speed 3" + + async def async_set_preset_mode(self, preset_mode: str) -> None: + """Set the preset mode.""" + _LOGGER.debug("Setting recuperation preset mode to %s", preset_mode) + + if preset_mode == "Party Mode": + # Activate party mode with configured duration + party_duration = self._get_party_mode_duration() + await self._coordinator.api.set_party_mode(self._udid, party_duration) + await self._coordinator.async_request_refresh() + return + + preset_to_mode = { + "Stop": 0, + "Speed 1": 1, + "Speed 2": 2, + "Speed 3": 3 + } + + mode_value = preset_to_mode.get(preset_mode, 0) + + if mode_value == 0: + # Stop the fan + await self._coordinator.api.set_fan_mode(self._udid, 0) + else: + # Use configured speed values from number entities + configured_values = self._get_configured_speed_values() + await self._coordinator.api.set_recuperation_speed(self._udid, mode_value, configured_values) + + await self._coordinator.async_request_refresh() + + def _get_configured_speed_values(self) -> dict: + """Get configured speed values from number entities or use defaults.""" + from .const import DEFAULT_SPEED_VALUES + + configured_values = DEFAULT_SPEED_VALUES.copy() + + # Try to get current values from number entities via Home Assistant + if hasattr(self._coordinator, 'hass'): + hass = self._coordinator.hass + for speed_level in [1, 2, 3]: + entity_id = f"number.{self._udid}_recuperation_speed_{speed_level}_config" + state = hass.states.get(entity_id) + if state and state.state not in ["unavailable", "unknown"]: + try: + configured_values[speed_level] = int(float(state.state)) + _LOGGER.debug("Got configured speed %s: %s m³/h", speed_level, configured_values[speed_level]) + except (ValueError, TypeError): + _LOGGER.debug("Could not parse speed %s value: %s", speed_level, state.state) + + return configured_values diff --git a/custom_components/tech/number.py b/custom_components/tech/number.py new file mode 100644 index 0000000..ea3927a --- /dev/null +++ b/custom_components/tech/number.py @@ -0,0 +1,606 @@ +"""Support for Tech HVAC number controls.""" + +import logging +from typing import Any + +from homeassistant.components.number import NumberEntity, NumberMode +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import ( + ATTR_IDENTIFIERS, + ATTR_MANUFACTURER, + CONF_MODEL, + CONF_NAME, + UnitOfTime, +) +from homeassistant.core import HomeAssistant +from homeassistant.helpers.device_registry import DeviceInfo +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.update_coordinator import CoordinatorEntity + +from . import assets +from .const import ( + CONTROLLER, + DEFAULT_SPEED_VALUES, + DOMAIN, + FILTER_ALARM_IDO_ID, + FILTER_ALARM_MAX_DAYS, + FILTER_ALARM_MIN_DAYS, + HUMIDITY_SENSOR_TXT_IDS, + INCLUDE_HUB_IN_NAME, + MANUFACTURER, + PARTY_MODE_IDO_ID, + PARTY_MODE_MAX_MINUTES, + PARTY_MODE_MIN_MINUTES, + RECUPERATION_EXHAUST_FLOW, + RECUPERATION_SUPPLY_FLOW, + RECUPERATION_SUPPLY_FLOW_ALT, + SPEED_CONFIG_KEYS, + SPEED_RANGES, + TYPE_TEMPERATURE_CH, + UDID, + VER, +) +from .coordinator import TechCoordinator + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up entry.""" + _LOGGER.debug( + "Setting up number entry, controller udid: %s", + config_entry.data[CONTROLLER][UDID], + ) + coordinator = hass.data[DOMAIN][config_entry.entry_id] + controller_udid = config_entry.data[CONTROLLER][UDID] + + tiles = await coordinator.api.get_module_tiles(controller_udid) + + entities = [] + + # Check if we have recuperation system (detected by flow sensors) + has_recuperation_flow = False + for t in tiles: + tile = tiles[t] + if tile.get("type") == TYPE_TEMPERATURE_CH: + widget1_txt_id = tile.get("params", {}).get("widget1", {}).get("txtId", 0) + widget2_txt_id = tile.get("params", {}).get("widget2", {}).get("txtId", 0) + for flow_sensor in [RECUPERATION_EXHAUST_FLOW, RECUPERATION_SUPPLY_FLOW, RECUPERATION_SUPPLY_FLOW_ALT]: + if flow_sensor["txt_id"] in [widget1_txt_id, widget2_txt_id]: + has_recuperation_flow = True + break + if has_recuperation_flow: + break + + # Create controls if we have recuperation + if has_recuperation_flow: + _LOGGER.debug("Creating recuperation number controls - party mode and speed configs") + # Add party mode control + entities.append(RecuperationPartyModeNumber(coordinator, config_entry)) + + # Add speed configuration controls + for speed_level in [1, 2, 3]: + entities.append(RecuperationSpeedConfigNumber(coordinator, config_entry, speed_level)) + + # Add filter alarm control for recuperation systems + entities.append(FilterAlarmNumber(coordinator, config_entry)) + + # Add ventilation parameters + entities.append(VentilationRoomParameterNumber(coordinator, config_entry)) + entities.append(VentilationBathroomParameterNumber(coordinator, config_entry)) + entities.append(Co2ThresholdNumber(coordinator, config_entry)) + entities.append(HysteresisNumber(coordinator, config_entry)) + else: + _LOGGER.debug("No recuperation flow detected, skipping number controls") + + async_add_entities(entities, True) + + +class RecuperationPartyModeNumber(CoordinatorEntity, NumberEntity): + """Representation of recuperation party mode number control.""" + + _attr_has_entity_name = True + _attr_native_unit_of_measurement = UnitOfTime.MINUTES + _attr_native_min_value = PARTY_MODE_MIN_MINUTES + _attr_native_max_value = PARTY_MODE_MAX_MINUTES # Full recuperator range 15-720 minutes + _attr_native_step = 15 # 15-minute increments for easier selection + _attr_mode = NumberMode.BOX + _attr_icon = "mdi:party-popper" + + def __init__( + self, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the party mode number.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_unique_id = f"{self._udid}_recuperation_party_mode" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Party Mode Duration" + + @property + def name(self) -> str: + """Return the name of the number.""" + return self._name + + @property + def native_value(self) -> float | None: + """Return the current value.""" + # For party mode, always return the minimum as default + # The actual value would need to be tracked separately in HA storage + return float(PARTY_MODE_MIN_MINUTES) + + async def async_set_native_value(self, value: float) -> None: + """Set the party mode duration.""" + duration_minutes = int(value) + _LOGGER.debug("Setting party mode to %s minutes", duration_minutes) + + await self._coordinator.api.set_party_mode(self._udid, duration_minutes) + await self._coordinator.async_request_refresh() + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, # Unique identifiers for the device + CONF_NAME: f"{self._config_entry.title} Recuperation", # Name of the device + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), # Model of the device + ATTR_MANUFACTURER: MANUFACTURER, # Manufacturer of the device + } + + +class RecuperationSpeedConfigNumber(CoordinatorEntity, NumberEntity): + """Representation of recuperation speed flow configuration.""" + + _attr_has_entity_name = True + _attr_native_unit_of_measurement = "m³/h" + _attr_mode = NumberMode.BOX + _attr_icon = "mdi:speedometer" + + def __init__( + self, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + speed_level: int, + ) -> None: + """Initialize the speed config number.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._speed_level = speed_level + self._attr_unique_id = f"{self._udid}_recuperation_speed_{speed_level}_config" + + # Set min/max values based on speed level + speed_range = SPEED_RANGES.get(speed_level, {"min": 50, "max": 500, "step": 10}) + self._attr_native_min_value = speed_range["min"] + self._attr_native_max_value = speed_range["max"] + self._attr_native_step = speed_range["step"] + + speed_names = { + 1: f"Speed 1 (Low) {speed_range['min']}-{speed_range['max']} m³/h", + 2: f"Speed 2 (Medium) {speed_range['min']}-{speed_range['max']} m³/h", + 3: f"Speed 3 (High) {speed_range['min']}-{speed_range['max']} m³/h" + } + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + f"Recuperation {speed_names[speed_level]}" + + @property + def name(self) -> str: + """Return the name of the number.""" + return self._name + + @property + def native_value(self) -> float | None: + """Return the current configured flow value.""" + # Get from Home Assistant configuration storage + # For now, use default values + return float(DEFAULT_SPEED_VALUES[self._speed_level]) + + async def async_set_native_value(self, value: float) -> None: + """Set the configured flow value.""" + flow_value = int(value) + _LOGGER.debug("Setting speed %s flow config to %s m³/h", self._speed_level, flow_value) + + # Store the value in Home Assistant's data storage + # This would need to be implemented with hass.data or config entries + # For now, just log the change + _LOGGER.info("Speed %s flow configured to %s m³/h", self._speed_level, flow_value) + + # Note: The actual storage implementation would go here + # We'd need to update the coordinator or config entry data + + @property + def entity_category(self): + """Return the entity category for configuration entities.""" + from homeassistant.helpers.entity import EntityCategory + return EntityCategory.CONFIG + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, # Unique identifiers for the device + CONF_NAME: f"{self._config_entry.title} Recuperation", # Name of the device + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), # Model of the device + ATTR_MANUFACTURER: MANUFACTURER, # Manufacturer of the device + } + + +class FilterAlarmNumber(CoordinatorEntity, NumberEntity): + """Representation of filter alarm interval number control.""" + + _attr_has_entity_name = True + _attr_native_unit_of_measurement = "days" + _attr_native_min_value = FILTER_ALARM_MIN_DAYS + _attr_native_max_value = FILTER_ALARM_MAX_DAYS + _attr_native_step = 1 + _attr_mode = NumberMode.BOX + _attr_icon = "mdi:air-filter" + + def __init__( + self, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the filter alarm number.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_unique_id = f"{self._udid}_filter_alarm_interval" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Filter Alarm Interval" + + @property + def name(self) -> str: + """Return the name of the number.""" + return self._name + + @property + def native_value(self) -> float | None: + """Return the current configured alarm interval.""" + # Default to 30 days - the actual value would be stored in HA data + return float(FILTER_ALARM_MIN_DAYS) + + async def async_set_native_value(self, value: float) -> None: + """Set the filter alarm interval.""" + days = int(value) + _LOGGER.debug("Setting filter alarm interval to %s days", days) + + await self._coordinator.api.set_filter_alarm(self._udid, days) + await self._coordinator.async_request_refresh() + + @property + def entity_category(self): + """Return the entity category for configuration entities.""" + from homeassistant.helpers.entity import EntityCategory + return EntityCategory.CONFIG + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, # Unique identifiers for the device + CONF_NAME: f"{self._config_entry.title} Recuperation", # Name of the device + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), # Model of the device + ATTR_MANUFACTURER: MANUFACTURER, # Manufacturer of the device + } + + + +class VentilationRoomParameterNumber(CoordinatorEntity, NumberEntity): + """Representation of room ventilation parameter number control.""" + + _attr_has_entity_name = True + _attr_native_unit_of_measurement = "%" + _attr_native_min_value = 10 + _attr_native_max_value = 90 + _attr_native_step = 1 + _attr_mode = NumberMode.SLIDER + _attr_icon = "mdi:air-filter" + + def __init__( + self, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the room ventilation parameter number.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_unique_id = f"{self._udid}_ventilation_room_parameter" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Room Ventilation Parameter" + + @property + def name(self) -> str: + """Return the name of the number.""" + return self._name + + @property + def native_value(self) -> float | None: + """Return the current value.""" + return 45.0 + + async def async_set_native_value(self, value: float) -> None: + """Set the room ventilation parameter.""" + percent = int(value) + _LOGGER.debug("Setting room ventilation parameter to %s%%", percent) + + await self._coordinator.api.set_ventilation_room_parameter(self._udid, percent) + await self._coordinator.async_request_refresh() + + @property + def entity_category(self): + """Return the entity category for configuration entities.""" + from homeassistant.helpers.entity import EntityCategory + return EntityCategory.CONFIG + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, + CONF_NAME: f"{self._config_entry.title} Recuperation", + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), + ATTR_MANUFACTURER: MANUFACTURER, + } + + +class VentilationBathroomParameterNumber(CoordinatorEntity, NumberEntity): + """Representation of bathroom ventilation parameter number control.""" + + _attr_has_entity_name = True + _attr_native_unit_of_measurement = "%" + _attr_native_min_value = 10 + _attr_native_max_value = 90 + _attr_native_step = 1 + _attr_mode = NumberMode.SLIDER + _attr_icon = "mdi:shower" + + def __init__( + self, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the bathroom ventilation parameter number.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_unique_id = f"{self._udid}_ventilation_bathroom_parameter" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Bathroom Ventilation Parameter" + + @property + def name(self) -> str: + """Return the name of the number.""" + return self._name + + @property + def native_value(self) -> float | None: + """Return the current value.""" + return 55.0 + + async def async_set_native_value(self, value: float) -> None: + """Set the bathroom ventilation parameter.""" + percent = int(value) + _LOGGER.debug("Setting bathroom ventilation parameter to %s%%", percent) + + await self._coordinator.api.set_ventilation_bathroom_parameter(self._udid, percent) + await self._coordinator.async_request_refresh() + + @property + def entity_category(self): + """Return the entity category for configuration entities.""" + from homeassistant.helpers.entity import EntityCategory + return EntityCategory.CONFIG + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, + CONF_NAME: f"{self._config_entry.title} Recuperation", + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), + ATTR_MANUFACTURER: MANUFACTURER, + } + + +class Co2ThresholdNumber(CoordinatorEntity, NumberEntity): + """Representation of CO2 threshold number control.""" + + _attr_has_entity_name = True + _attr_native_unit_of_measurement = "ppm" + _attr_native_min_value = 400 + _attr_native_max_value = 2000 + _attr_native_step = 50 + _attr_mode = NumberMode.SLIDER + _attr_icon = "mdi:molecule-co2" + + def __init__( + self, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the CO2 threshold number.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_unique_id = f"{self._udid}_co2_threshold" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "CO2 Threshold" + + @property + def name(self) -> str: + """Return the name of the number.""" + return self._name + + @property + def native_value(self) -> float | None: + """Return the current value.""" + return 1000.0 + + async def async_set_native_value(self, value: float) -> None: + """Set the CO2 threshold.""" + ppm = int(value) + _LOGGER.debug("Setting CO2 threshold to %s ppm", ppm) + + await self._coordinator.api.set_co2_threshold(self._udid, ppm) + await self._coordinator.async_request_refresh() + + @property + def entity_category(self): + """Return the entity category for configuration entities.""" + from homeassistant.helpers.entity import EntityCategory + return EntityCategory.CONFIG + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, + CONF_NAME: f"{self._config_entry.title} Recuperation", + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), + ATTR_MANUFACTURER: MANUFACTURER, + } + + +class HysteresisNumber(CoordinatorEntity, NumberEntity): + """Representation of hysteresis number control.""" + + _attr_has_entity_name = True + _attr_native_unit_of_measurement = "%" + _attr_native_min_value = 5 + _attr_native_max_value = 10 + _attr_native_step = 1 + _attr_mode = NumberMode.SLIDER + _attr_icon = "mdi:tune-vertical" + + def __init__( + self, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the hysteresis number.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_unique_id = f"{self._udid}_hysteresis" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Hysteresis" + + @property + def name(self) -> str: + """Return the name of the number.""" + return self._name + + @property + def native_value(self) -> float | None: + """Return the current value.""" + return 10.0 + + async def async_set_native_value(self, value: float) -> None: + """Set the hysteresis.""" + percent = int(value) + _LOGGER.debug("Setting hysteresis to %s%%", percent) + + await self._coordinator.api.set_hysteresis(self._udid, percent) + await self._coordinator.async_request_refresh() + + @property + def entity_category(self): + """Return the entity category for configuration entities.""" + from homeassistant.helpers.entity import EntityCategory + return EntityCategory.CONFIG + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, + CONF_NAME: f"{self._config_entry.title} Recuperation", + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), + ATTR_MANUFACTURER: MANUFACTURER, + } diff --git a/custom_components/tech/select.py b/custom_components/tech/select.py new file mode 100644 index 0000000..76ee484 --- /dev/null +++ b/custom_components/tech/select.py @@ -0,0 +1,283 @@ +"""Support for Tech HVAC select controls.""" + +import logging +from typing import Any + +from homeassistant.components.select import SelectEntity +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import ( + ATTR_IDENTIFIERS, + ATTR_MANUFACTURER, + CONF_MODEL, + CONF_NAME, +) +from homeassistant.core import HomeAssistant +from homeassistant.helpers.device_registry import DeviceInfo +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.update_coordinator import CoordinatorEntity +from homeassistant.core import callback + +from . import assets +from .const import ( + CONTROLLER, + DOMAIN, + FAN_MODE_IDO_ID, + FAN_MODE_OPTIONS, + FAN_MODE_OPTIONS_REVERSE, + GEAR_CONTROL_IDO_ID, + GEAR_OPTIONS, + GEAR_OPTIONS_REVERSE, + INCLUDE_HUB_IN_NAME, + MANUFACTURER, + RECUPERATION_EXHAUST_FLOW, + RECUPERATION_SUPPLY_FLOW, + RECUPERATION_SUPPLY_FLOW_ALT, + TYPE_TEMPERATURE_CH, + UDID, + VER, +) +from .coordinator import TechCoordinator + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up entry.""" + _LOGGER.debug( + "Setting up select entry, controller udid: %s", + config_entry.data[CONTROLLER][UDID], + ) + coordinator = hass.data[DOMAIN][config_entry.entry_id] + controller_udid = config_entry.data[CONTROLLER][UDID] + + tiles = await coordinator.api.get_module_tiles(controller_udid) + + entities = [] + + # Check if we have recuperation system (detected by flow sensors) + has_recuperation_flow = False + for t in tiles: + tile = tiles[t] + if tile.get("type") == TYPE_TEMPERATURE_CH: + widget1_txt_id = tile.get("params", {}).get("widget1", {}).get("txtId", 0) + widget2_txt_id = tile.get("params", {}).get("widget2", {}).get("txtId", 0) + for flow_sensor in [RECUPERATION_EXHAUST_FLOW, RECUPERATION_SUPPLY_FLOW, RECUPERATION_SUPPLY_FLOW_ALT]: + if flow_sensor["txt_id"] in [widget1_txt_id, widget2_txt_id]: + has_recuperation_flow = True + break + if has_recuperation_flow: + break + + # Create select controls if we have recuperation + if has_recuperation_flow: + _LOGGER.debug("Creating recuperation select controls") + entities.append(RecuperationGearSelect(coordinator, config_entry)) # Direct gear control + entities.append(RecuperationModeSelect(coordinator, config_entry)) # Timed fan mode + else: + _LOGGER.debug("No recuperation flow detected, skipping select controls") + + async_add_entities(entities, True) + + +class RecuperationModeSelect(CoordinatorEntity, SelectEntity): + """Representation of recuperation fan mode select control (timed mode changes).""" + + _attr_has_entity_name = True + _attr_icon = "mdi:fan-speed" + + def __init__( + self, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the fan mode select.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_unique_id = f"{self._udid}_recuperation_fan_mode" + + # Define options with Polish names + self._attr_options = [ + "Zatrzymaj wentylator", + "1 bieg", + "2 bieg", + "3 bieg" + ] + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Fan Mode (Timed)" + + @property + def name(self) -> str: + """Return the name of the select.""" + return self._name + + @property + def current_option(self) -> str | None: + """Return the current selected option.""" + # For now, always return default - the actual state tracking + # would need more complex implementation with HA storage + return "Zatrzymaj wentylator" + + async def async_select_option(self, option: str) -> None: + """Select the fan mode option.""" + _LOGGER.debug("Selecting fan mode option: %s", option) + + # Map option name to value + option_value_map = { + "Zatrzymaj wentylator": 0, + "1 bieg": 1, + "2 bieg": 2, + "3 bieg": 3 + } + + mode_value = option_value_map.get(option) + if mode_value is None: + _LOGGER.error("Unknown fan mode option: %s", option) + return + + await self._coordinator.api.set_fan_mode(self._udid, mode_value) + await self._coordinator.async_request_refresh() + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, # Unique identifiers for the device + CONF_NAME: f"{self._config_entry.title} Recuperation", # Name of the device + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), # Model of the device + ATTR_MANUFACTURER: MANUFACTURER, # Manufacturer of the device + } + + +class RecuperationGearSelect(CoordinatorEntity, SelectEntity): + """Representation of direct recuperation gear control.""" + + _attr_has_entity_name = True + _attr_icon = "mdi:speedometer" + + def __init__( + self, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the gear select.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_unique_id = f"{self._udid}_recuperation_gear" + self._attr_options = ["Zatrzymaj", "1 bieg", "2 bieg", "3 bieg"] + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Recuperation Gear" + + # Initialize with default, will be updated by coordinator data + self._last_selected_gear = "Zatrzymaj" + + @callback + def _handle_coordinator_update(self) -> None: + """Handle updated data from the coordinator.""" + # Update current gear from tiles data when coordinator updates + if self._coordinator.data and "tiles" in self._coordinator.data: + for tile_data in self._coordinator.data["tiles"].values(): + if tile_data.get("type") in [22, 122]: # TYPE_FAN or TYPE_RECUPERATION + current_gear = tile_data.get("params", {}).get("gear", 0) + gear_mapping = { + 0: "Zatrzymaj", + 1: "1 bieg", + 2: "2 bieg", + 3: "3 bieg" + } + new_gear = gear_mapping.get(current_gear, "Zatrzymaj") + if new_gear != self._last_selected_gear: + self._last_selected_gear = new_gear + _LOGGER.debug("Updated gear from coordinator: %s", new_gear) + break + + super()._handle_coordinator_update() + + @property + def name(self) -> str: + """Return the name of the select.""" + return self._name + + @property + def current_option(self) -> str | None: + """Return the current selected gear.""" + # First try to get from tiles data (faster) + if self._coordinator.data and "tiles" in self._coordinator.data: + for tile_id, tile_data in self._coordinator.data["tiles"].items(): + # Look for recuperation fan tile or similar + if tile_data.get("type") in [22, 122]: # TYPE_FAN or TYPE_RECUPERATION + current_gear = tile_data.get("params", {}).get("gear", 0) + gear_mapping = { + 0: "Zatrzymaj", + 1: "1 bieg", + 2: "2 bieg", + 3: "3 bieg" + } + return gear_mapping.get(current_gear, "Zatrzymaj") + + # Fallback: check stored state from last API call + return getattr(self, '_last_selected_gear', "Zatrzymaj") + + async def async_select_option(self, option: str) -> None: + """Change the selected gear.""" + gear_mapping = { + "Zatrzymaj": 0, + "1 bieg": 1, + "2 bieg": 2, + "3 bieg": 3, + } + + gear_value = gear_mapping.get(option, 0) + _LOGGER.debug("Setting recuperation gear to: %s (value: %d)", option, gear_value) + + try: + # Use the direct gear control API + await self._coordinator.api.set_gear_direct(self._udid, gear_value) + + # Store the selected option for immediate display + self._last_selected_gear = option + + # Refresh coordinator data + await self._coordinator.async_request_refresh() + + _LOGGER.info("Successfully set recuperation gear to: %s", option) + + except Exception as err: + _LOGGER.error("Failed to set recuperation gear to %s: %s", option, err) + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, + CONF_NAME: f"{self._config_entry.title} Recuperation", + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), + ATTR_MANUFACTURER: MANUFACTURER, + } \ No newline at end of file diff --git a/custom_components/tech/sensor.py b/custom_components/tech/sensor.py index 7c36bdc..fd5a97a 100644 --- a/custom_components/tech/sensor.py +++ b/custom_components/tech/sensor.py @@ -24,6 +24,8 @@ PERCENTAGE, STATE_OFF, STATE_ON, + STATE_UNAVAILABLE, + STATE_UNKNOWN, EntityCategory, UnitOfTemperature, ) @@ -41,17 +43,23 @@ BATTERY_LEVEL, CONTROLLER, DOMAIN, + HUMIDITY_SENSOR_TXT_IDS, INCLUDE_HUB_IN_NAME, MANUFACTURER, OPENTHERM_CURRENT_TEMP, OPENTHERM_CURRENT_TEMP_DHW, OPENTHERM_SET_TEMP, OPENTHERM_SET_TEMP_DHW, + RECUPERATION_EXHAUST_FLOW, + RECUPERATION_SUPPLY_FLOW, + RECUPERATION_SUPPLY_FLOW_ALT, + RECUPERATION_TEMP_SENSORS, SIGNAL_STRENGTH, TYPE_FAN, TYPE_FUEL_SUPPLY, TYPE_MIXING_VALVE, TYPE_OPEN_THERM, + TYPE_RECUPERATION, TYPE_TEMPERATURE, TYPE_TEMPERATURE_CH, TYPE_TEXT, @@ -96,30 +104,114 @@ async def async_setup_entry( if tile[VISIBILITY] is False or tile.get(WORKING_STATUS, True) is False: continue if tile[CONF_TYPE] == TYPE_TEMPERATURE: - signal_strength = tile[CONF_PARAMS][SIGNAL_STRENGTH] - battery_level = tile[CONF_PARAMS][BATTERY_LEVEL] - create_devices = False - if signal_strength not in (None, "null"): - create_devices = True - entities.append( - TileTemperatureSignalSensor( - tile, coordinator, config_entry, create_devices + _LOGGER.debug("Creating temperature sensor: id=%s, txtId=%s, value=%s", + tile[CONF_ID], tile[CONF_PARAMS].get("txtId"), tile[CONF_PARAMS].get("value")) + + # Check if this is a recuperation temperature sensor by txtId + tile_txt_id = tile[CONF_PARAMS].get("txtId", 0) + is_recuperation_temp = False + for temp_sensor in RECUPERATION_TEMP_SENSORS: + if temp_sensor["txt_id"] == tile_txt_id: + is_recuperation_temp = True + _LOGGER.debug("Creating recuperation temperature sensor from TYPE_TEMPERATURE tile: %s (txtId: %s)", temp_sensor["name"], temp_sensor["txt_id"]) + entities.append( + SimpleRecuperationTemperatureSensor( + tile, coordinator, config_entry, temp_sensor + ) ) - ) - if battery_level not in (None, "null"): - create_devices = True - entities.append( - TileTemperatureBatterySensor( - tile, coordinator, config_entry, create_devices + break + + if not is_recuperation_temp: + # Regular temperature sensor processing + signal_strength = tile[CONF_PARAMS].get(SIGNAL_STRENGTH) + battery_level = tile[CONF_PARAMS].get(BATTERY_LEVEL) + create_devices = False + if signal_strength not in (None, "null"): + create_devices = True + entities.append( + TileTemperatureSignalSensor( + tile, coordinator, config_entry, create_devices + ) + ) + if battery_level not in (None, "null"): + create_devices = True + entities.append( + TileTemperatureBatterySensor( + tile, coordinator, config_entry, create_devices + ) ) + entities.append( + TileTemperatureSensor(tile, coordinator, config_entry, create_devices) ) - entities.append( - TileTemperatureSensor(tile, coordinator, config_entry, create_devices) - ) if tile[CONF_TYPE] == TYPE_TEMPERATURE_CH: - entities.append(TileWidgetSensor(tile, coordinator, config_entry)) + # Check if this tile contains recuperation flow data + widget1_txt_id = tile[CONF_PARAMS].get("widget1", {}).get("txtId", 0) + widget2_txt_id = tile[CONF_PARAMS].get("widget2", {}).get("txtId", 0) + + # Check for recuperation flow sensors + is_recuperation_flow = False + for flow_sensor in [RECUPERATION_EXHAUST_FLOW, RECUPERATION_SUPPLY_FLOW, RECUPERATION_SUPPLY_FLOW_ALT]: + if flow_sensor["txt_id"] in [widget1_txt_id, widget2_txt_id]: + is_recuperation_flow = True + # Create flow sensor if widget has data + widget_key = flow_sensor["widget"] + if tile[CONF_PARAMS].get(widget_key, {}).get("txtId") == flow_sensor["txt_id"]: + entities.append( + TileRecuperationFlowSensor( + tile, coordinator, config_entry, flow_sensor + ) + ) + + # Check for recuperation temperature sensors + is_recuperation_temp = False + for widget_key in ["widget1", "widget2"]: + widget_data = tile[CONF_PARAMS].get(widget_key, {}) + widget_txt_id = widget_data.get("txtId", 0) + widget_unit = widget_data.get("unit", -1) + widget_type = widget_data.get("type", 0) + + # Check if this is a recuperation temperature sensor + for temp_sensor in RECUPERATION_TEMP_SENSORS: + if temp_sensor["txt_id"] == widget_txt_id: + is_recuperation_temp = True + # Always create sensor if txtId matches, even if current value is None/0 + entities.append( + TileRecuperationTemperatureSensor( + tile, coordinator, config_entry, widget_key, temp_sensor + ) + ) + _LOGGER.debug("Created temperature sensor: %s (txtId: %s)", temp_sensor["name"], temp_sensor["txt_id"]) + break + + # Check for humidity sensors in widgets + is_humidity_sensor = False + for widget_key in ["widget1", "widget2"]: + widget_data = tile[CONF_PARAMS].get(widget_key, {}) + widget_txt_id = widget_data.get("txtId", 0) + widget_unit = widget_data.get("unit", -1) + widget_type = widget_data.get("type", 0) + + # Check if this is a humidity sensor (unit: 8, type: 2) + if widget_txt_id in HUMIDITY_SENSOR_TXT_IDS or (widget_unit == 8 and widget_type == 2): + is_humidity_sensor = True + if widget_data.get("value", 0) > 0 or widget_txt_id > 0: # Has data + entities.append( + TileHumiditySensorWidget( + tile, coordinator, config_entry, widget_key, widget_txt_id + ) + ) + + # If not recuperation flow data, humidity sensor, or temperature sensor, create regular widget sensor + if not is_recuperation_flow and not is_humidity_sensor and not is_recuperation_temp and widget1_txt_id > 0: + entities.append(TileWidgetSensor(tile, coordinator, config_entry)) if tile[CONF_TYPE] == TYPE_FAN: - entities.append(TileFanSensor(tile, coordinator, config_entry)) + # Check if this fan is handled by the fan platform (recuperation) + description = tile[CONF_PARAMS].get(CONF_DESCRIPTION, "").lower() + if not any(keyword in description for keyword in ["recuperation", "rekuperacja", "ventilation", "wentylacja"]): + entities.append(TileFanSensor(tile, coordinator, config_entry)) + if tile[CONF_TYPE] == TYPE_RECUPERATION: + # TYPE_RECUPERATION fans are handled by fan platform, create sensor for monitoring only + entities.append(TileRecuperationSensor(tile, coordinator, config_entry)) if tile[CONF_TYPE] == TYPE_VALVE: entities.append(TileValveSensor(tile, coordinator, config_entry)) for valve_sensor in [ @@ -153,6 +245,27 @@ async def async_setup_entry( ) ) + # Check if we have recuperation system (detected by flow sensors) for filter usage sensor + has_recuperation_flow = False + for t in tiles: + tile = tiles[t] + if tile.get("type") == TYPE_TEMPERATURE_CH: + widget1_txt_id = tile.get("params", {}).get("widget1", {}).get("txtId", 0) + widget2_txt_id = tile.get("params", {}).get("widget2", {}).get("txtId", 0) + for flow_sensor in [RECUPERATION_EXHAUST_FLOW, RECUPERATION_SUPPLY_FLOW, RECUPERATION_SUPPLY_FLOW_ALT]: + if flow_sensor["txt_id"] in [widget1_txt_id, widget2_txt_id]: + has_recuperation_flow = True + break + if has_recuperation_flow: + break + + # Create filter usage sensor, efficiency sensor, and outdoor temperature if we have recuperation + if has_recuperation_flow: + _LOGGER.debug("Creating filter usage, efficiency, and outdoor temperature sensors") + entities.append(FilterUsageSensor(coordinator, config_entry)) + entities.append(RecuperationEfficiencySensor(coordinator, config_entry)) + entities.append(OutdoorTemperatureSensor(coordinator, config_entry)) + async_add_entities(entities, True) # async_add_entities( @@ -1759,3 +1872,664 @@ def device_info(self) -> DeviceInfo | None: CONF_MODEL: self.model, # Model of the device ATTR_MANUFACTURER: self.manufacturer, # Manufacturer of the device } + + +class TileRecuperationSensor(TileSensor, SensorEntity): + """Representation of a Tile Recuperation Sensor for monitoring.""" + + def __init__(self, device, coordinator, config_entry) -> None: + """Initialize the sensor.""" + TileSensor.__init__(self, device, coordinator, config_entry) + self.state_class = SensorStateClass.MEASUREMENT + self._attr_icon = assets.get_icon_by_type(TYPE_RECUPERATION) + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Recuperation Status" + + @property + def unique_id(self) -> str: + """Return a unique ID.""" + return f"{self._unique_id}_tile_recuperation_sensor" + + @property + def name(self) -> str | UndefinedType | None: + """Return the name of the sensor.""" + return self._name + + def get_state(self, device) -> Any: + """Get the state of the device.""" + gear = device[CONF_PARAMS].get("gear", 0) + if gear == 0: + return "off" + elif gear == 1: + return "low" + elif gear == 2: + return "medium" + elif gear == 3: + return "high" + else: + return f"speed_{gear}" + + +class TileRecuperationFlowSensor(TileSensor, SensorEntity): + """Representation of a Tile Recuperation Flow Sensor.""" + + _attr_has_entity_name = True + _attr_native_unit_of_measurement = "m³/h" + _attr_state_class = SensorStateClass.MEASUREMENT + _attr_icon = "mdi:air-filter" + + def __init__( + self, + device, + coordinator: TechCoordinator, + config_entry, + flow_sensor_config, + ) -> None: + """Initialize the sensor.""" + self._flow_config = flow_sensor_config + self._widget_key = flow_sensor_config["widget"] + self._flow_name = flow_sensor_config["name"] + + TileSensor.__init__(self, device, coordinator, config_entry) + + # Set the appropriate device class and icon based on flow type + if self._flow_name == "exhaust_flow": + self._attr_icon = "mdi:arrow-up-bold" + self._flow_display_name = "Exhaust Flow" + elif self._flow_name == "supply_flow": + self._attr_icon = "mdi:arrow-down-bold" + self._flow_display_name = "Supply Flow" + else: + self._flow_display_name = "Air Flow" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + f"Recuperation {self._flow_display_name}" + + @property + def name(self) -> str | UndefinedType | None: + """Return the name of the device.""" + return self._name + + @property + def unique_id(self) -> str: + """Return a unique ID.""" + return f"{self._unique_id}_tile_recuperation_{self._flow_name}" + + def get_state(self, device) -> Any: + """Get the state of the device.""" + widget_data = device[CONF_PARAMS].get(self._widget_key, {}) + flow_value = widget_data.get("value", 0) + # Flow values are in m³/h units, no conversion needed + return flow_value if flow_value else 0 + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._unique_id}_recuperation") + }, # Unique identifiers for the device + CONF_NAME: f"{self._config_entry.title} Recuperation", # Name of the device + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), # Model of the device + ATTR_MANUFACTURER: MANUFACTURER, # Manufacturer of the device + } + + +class TileHumiditySensorWidget(TileSensor, SensorEntity): + """Representation of a Tile Humidity Sensor Widget.""" + + _attr_has_entity_name = True + _attr_native_unit_of_measurement = "%" + _attr_device_class = SensorDeviceClass.HUMIDITY + _attr_state_class = SensorStateClass.MEASUREMENT + _attr_icon = "mdi:water-percent" + + def __init__( + self, + device, + coordinator: TechCoordinator, + config_entry, + widget_key: str, + txt_id: int, + ) -> None: + """Initialize the humidity sensor.""" + self._widget_key = widget_key + self._txt_id = txt_id + + TileSensor.__init__(self, device, coordinator, config_entry) + + # Get the proper name from assets using txtId + sensor_name = assets.get_text(txt_id) if txt_id > 0 else f"Humidity Sensor {txt_id}" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + sensor_name + + @property + def name(self) -> str | UndefinedType | None: + """Return the name of the device.""" + return self._name + + @property + def unique_id(self) -> str: + """Return a unique ID.""" + return f"{self._unique_id}_tile_humidity_{self._txt_id}" + + def get_state(self, device) -> Any: + """Get the state of the device.""" + widget_data = device[CONF_PARAMS].get(self._widget_key, {}) + humidity_value = widget_data.get("value", 0) + # Humidity values are already in percent, no conversion needed + return humidity_value if humidity_value else 0 + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._unique_id}_humidity_sensors") + }, # Unique identifiers for the device + CONF_NAME: f"{self._config_entry.title} Humidity Sensors", # Name of the device + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), # Model of the device + ATTR_MANUFACTURER: MANUFACTURER, # Manufacturer of the device + } + + +class FilterUsageSensor(CoordinatorEntity, SensorEntity): + """Representation of filter usage tracking sensor.""" + + _attr_has_entity_name = True + _attr_native_unit_of_measurement = "days" + _attr_icon = "mdi:air-filter" + _attr_device_class = SensorDeviceClass.DURATION + _attr_state_class = SensorStateClass.TOTAL_INCREASING + + def __init__( + self, + coordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the filter usage sensor.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_unique_id = f"{self._udid}_filter_usage_days" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Filter Usage Days" + + @property + def name(self) -> str: + """Return the name of the sensor.""" + return self._name + + @property + def native_value(self) -> int | None: + """Return the current filter usage in days.""" + # Calculate days since last filter reset from stored date + if hasattr(self._coordinator, '_filter_reset_date') and self._coordinator._filter_reset_date: + from datetime import datetime + reset_date = datetime.fromisoformat(self._coordinator._filter_reset_date) + current_date = datetime.now() + days_diff = (current_date - reset_date).days + return days_diff + else: + # No reset date stored, assume filter is new (0 days) + return 0 + + @property + def entity_category(self): + """Return the entity category for diagnostic entities.""" + return EntityCategory.DIAGNOSTIC + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, # Unique identifiers for the device + CONF_NAME: f"{self._config_entry.title} Recuperation", # Name of the device + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), # Model of the device + ATTR_MANUFACTURER: MANUFACTURER, # Manufacturer of the device + } + + +class TileRecuperationTemperatureSensor(TileSensor, SensorEntity): + """Representation of a recuperation temperature sensor.""" + + def __init__( + self, + device, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + widget_key: str, + temp_sensor: dict, + ) -> None: + """Initialize the recuperation temperature sensor.""" + TileSensor.__init__(self, device, coordinator, config_entry) + self._widget_key = widget_key + self._temp_sensor = temp_sensor + self._attr_unique_id = f"{self._unique_id}_{widget_key}_{temp_sensor['txt_id']}_temp" + self._attr_device_class = SensorDeviceClass.TEMPERATURE + self._attr_native_unit_of_measurement = UnitOfTemperature.CELSIUS + self._attr_state_class = SensorStateClass.MEASUREMENT + self._attr_icon = "mdi:thermometer" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + temp_sensor["name"] + + @property + def name(self) -> str: + """Return the name of the sensor.""" + return self._name + + @property + def unique_id(self) -> str: + """Return a unique ID.""" + return self._attr_unique_id + + @property + def native_value(self) -> float | None: + """Return the temperature value.""" + if self._coordinator.data and "tiles" in self._coordinator.data: + tile_data = self._coordinator.data["tiles"].get(self._id) + if tile_data: + widget_data = tile_data.get("params", {}).get(self._widget_key, {}) + temp_value = widget_data.get("value") + # Temperature values might need conversion from tenths of degrees + if temp_value is not None: + return float(temp_value) / 10.0 if temp_value > 100 else float(temp_value) + return None + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, + CONF_NAME: f"{self._config_entry.title} Recuperation", + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), + ATTR_MANUFACTURER: MANUFACTURER, + } + + +class SimpleRecuperationTemperatureSensor(TileSensor, SensorEntity): + """Representation of a simple recuperation temperature sensor following TYPE_TEMPERATURE pattern.""" + + _attr_has_entity_name = True + _attr_native_unit_of_measurement = UnitOfTemperature.CELSIUS + _attr_device_class = SensorDeviceClass.TEMPERATURE + _attr_state_class = SensorStateClass.MEASUREMENT + + def __init__( + self, + device, + coordinator: TechCoordinator, + config_entry, + temp_sensor: dict, + ) -> None: + """Initialize the sensor.""" + TileSensor.__init__(self, device, coordinator, config_entry) + self._coordinator = coordinator + self._temp_sensor = temp_sensor + self._attr_unique_id = f"{self._unique_id}_recuperation_temp_{temp_sensor['txt_id']}" + self._attr_icon = "mdi:thermometer" + + # Use the temperature sensor name from the config + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + temp_sensor["name"] + + @property + def name(self) -> str: + """Return the name of the sensor.""" + return self._name + + @property + def unique_id(self) -> str: + """Return a unique ID.""" + return self._attr_unique_id + + def get_state(self, device) -> Any: + """Get the state of the device using TYPE_TEMPERATURE pattern.""" + temp_value = device[CONF_PARAMS].get(VALUE) + if temp_value is not None: + # Follow the same pattern as other temperature sensors: divide by 10 + return temp_value / 10 + return None + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._config_entry.data[CONTROLLER][UDID]}_recuperation") + }, + CONF_NAME: f"{self._config_entry.title} Recuperation", + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), + ATTR_MANUFACTURER: MANUFACTURER, + } +class RecuperationEfficiencySensor(CoordinatorEntity, SensorEntity): + """Sensor for calculating recuperation heat recovery efficiency.""" + + _attr_has_entity_name = True + _attr_native_unit_of_measurement = PERCENTAGE + _attr_icon = "mdi:percent" + _attr_state_class = SensorStateClass.MEASUREMENT + + def __init__( + self, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the efficiency sensor.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_unique_id = f"{self._udid}_recuperation_efficiency" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Recuperation Efficiency" + + @property + def name(self) -> str: + """Return the name of the sensor.""" + return self._name + + @property + def native_value(self) -> float | None: + """Calculate heat recovery efficiency based on temperature sensors.""" + if self._coordinator.data and "tiles" in self._coordinator.data: + tiles_data = self._coordinator.data["tiles"] + if not tiles_data: + _LOGGER.debug("No tiles data available for efficiency calculation") + return None + + _LOGGER.debug("Starting efficiency calculation, searching for temperature sensors...") + _LOGGER.debug("Found %d tiles to examine", len(tiles_data)) + + # Try to find temperature sensors for efficiency calculation + # Standard efficiency formula: (Supply - Outdoor) / (Exhaust - Outdoor) * 100 + supply_temp = None + exhaust_temp = None + outdoor_temp = None + + # First, log all available tiles for debugging + _LOGGER.debug("Available tiles for efficiency calculation:") + for tile_id, tile_data in tiles_data.items(): + tile_type = tile_data.get("type", "unknown") + tile_params = tile_data.get("params", {}) + # Safe logging - only log key parameters to avoid issues + txtId = tile_params.get("txtId", "none") + value = tile_params.get("value", "none") + _LOGGER.debug("Tile %s: type=%s, txtId=%s, value=%s", tile_id, tile_type, txtId, value) + + # Look for temperature sensors in tiles + for tile_id, tile_data in tiles_data.items(): + # Check TYPE_TEMPERATURE tiles for recuperation sensors + if tile_data.get("type") == TYPE_TEMPERATURE: + tile_txt_id = tile_data.get("params", {}).get("txtId", 0) + temp_value = tile_data.get("params", {}).get("value") + + _LOGGER.debug("Found TYPE_TEMPERATURE tile: txtId=%s, value=%s", tile_txt_id, temp_value) + + if temp_value is not None: + try: + # Temperature processing - handle both formats (tenths vs direct) + if isinstance(temp_value, (int, float)) and temp_value > 100: + temp_celsius = temp_value / 10 # Likely in tenths (e.g., 215 = 21.5°C) + else: + temp_celsius = float(temp_value) # Already in correct format + + _LOGGER.debug("Temperature processed: txtId=%s, raw_value=%s, temp_celsius=%.1f", tile_txt_id, temp_value, temp_celsius) + except (ValueError, TypeError) as e: + _LOGGER.debug("Could not process temperature value: txtId=%s, value=%s, error=%s", tile_txt_id, temp_value, e) + continue + + # Comprehensive txtId mapping - trying all possible recuperation temperature sensors + # Supply Air Temperature (expanded list from various implementations) + if tile_txt_id in [119, 126, 127, 5997, 2010, 6001, 6002, 6003]: + supply_temp = temp_celsius + _LOGGER.debug("Found Supply Air Temperature: %.1f°C (txtId: %s)", temp_celsius, tile_txt_id) + # Exhaust Air Temperature (expanded list) + elif tile_txt_id in [120, 127, 128, 5998, 5996, 2011, 6004, 6005, 6006]: + exhaust_temp = temp_celsius + _LOGGER.debug("Found Exhaust Air Temperature: %.1f°C (txtId: %s)", temp_celsius, tile_txt_id) + # External/Fresh Air Temperature (expanded list) + elif tile_txt_id in [121, 122, 129, 5995, 2012, 6007, 6008, 6009]: + outdoor_temp = temp_celsius + _LOGGER.debug("Found External Air Temperature: %.1f°C (txtId: %s)", temp_celsius, tile_txt_id) + # Log any reasonable temperature for manual identification + elif -30 <= temp_celsius <= 70: # Reasonable temperature range + _LOGGER.warning("UNRECOGNIZED TEMPERATURE SENSOR: txtId=%s, temp=%.1f°C - could be supply/exhaust/outdoor air", tile_txt_id, temp_celsius) + else: + _LOGGER.debug("Found non-temperature sensor: txtId=%s, value=%s", tile_txt_id, temp_value) + + # Check all other tile types too in case temperature is elsewhere + elif tile_data.get("type") not in [TYPE_TEMPERATURE_CH]: # Skip TYPE_TEMPERATURE_CH as we handle it separately + tile_params = tile_data.get("params", {}) + if "value" in tile_params and isinstance(tile_params.get("value"), (int, float)): + potential_temp = tile_params["value"] + txtId = tile_params.get("txtId", 0) + if txtId > 0: # Valid txtId + _LOGGER.debug("Found other tile type %s with numeric value: txtId=%s, value=%s", tile_data.get("type"), txtId, potential_temp) + + # Also check TYPE_TEMPERATURE_CH widgets + elif tile_data.get("type") == TYPE_TEMPERATURE_CH: + for widget_key in ["widget1", "widget2"]: + widget_data = tile_data.get("params", {}).get(widget_key, {}) + widget_txt_id = widget_data.get("txtId", 0) + temp_value = widget_data.get("value") + + if widget_txt_id > 0: # Only log if we have a valid txtId + _LOGGER.debug("Found TYPE_TEMPERATURE_CH widget: %s, txtId=%s, value=%s", widget_key, widget_txt_id, temp_value) + + if temp_value is not None: + try: + # Same temperature processing for widgets + if isinstance(temp_value, (int, float)) and temp_value > 100: + temp_celsius = temp_value / 10 + else: + temp_celsius = float(temp_value) + + _LOGGER.debug("Widget temperature processed: txtId=%s, raw_value=%s, temp_celsius=%.1f", widget_txt_id, temp_value, temp_celsius) + except (ValueError, TypeError) as e: + _LOGGER.debug("Could not process widget temperature value: txtId=%s, value=%s, error=%s", widget_txt_id, temp_value, e) + continue + + # Same comprehensive txtId mapping for widgets + if widget_txt_id in [119, 126, 127, 5997, 2010, 6001, 6002, 6003]: # Supply Air + supply_temp = temp_celsius + _LOGGER.debug("Found Supply Air Temperature (widget): %.1f°C (txtId: %s)", temp_celsius, widget_txt_id) + elif widget_txt_id in [120, 127, 128, 5998, 5996, 2011, 6004, 6005, 6006]: # Exhaust Air + exhaust_temp = temp_celsius + _LOGGER.debug("Found Exhaust Air Temperature (widget): %.1f°C (txtId: %s)", temp_celsius, widget_txt_id) + elif widget_txt_id in [121, 122, 129, 5995, 2012, 6007, 6008, 6009]: # External Air + outdoor_temp = temp_celsius + _LOGGER.debug("Found External Air Temperature (widget): %.1f°C (txtId: %s)", temp_celsius, widget_txt_id) + elif -30 <= temp_celsius <= 70: # Reasonable temperature range + _LOGGER.warning("UNRECOGNIZED TEMPERATURE WIDGET: txtId=%s, temp=%.1f°C - could be supply/exhaust/outdoor air", widget_txt_id, temp_celsius) + else: + _LOGGER.debug("Found non-temperature widget: txtId=%s, value=%s", widget_txt_id, temp_value) + + # Try to get outdoor temperature from existing sensor if not found in tiles + if outdoor_temp is None: + try: + external_sensor_state = self._coordinator.hass.states.get("sensor.external_air_temperature") + if external_sensor_state and external_sensor_state.state not in [STATE_UNKNOWN, STATE_UNAVAILABLE]: + outdoor_temp = float(external_sensor_state.state) + _LOGGER.debug("Using external_air_temperature sensor: %.1f°C", outdoor_temp) + except (ValueError, AttributeError) as e: + _LOGGER.debug("Could not get outdoor temperature from external sensor: %s", e) + + # Final temperature summary and calculation + _LOGGER.debug("Temperature summary: Supply=%.1f, Exhaust=%.1f, Outdoor=%.1f", + supply_temp if supply_temp is not None else float('nan'), + exhaust_temp if exhaust_temp is not None else float('nan'), + outdoor_temp if outdoor_temp is not None else float('nan')) + + # Calculate efficiency if we have all required temperatures + if supply_temp is not None and exhaust_temp is not None and outdoor_temp is not None: + # Avoid division by zero + temperature_diff = exhaust_temp - outdoor_temp + _LOGGER.debug("Temperature difference (Exhaust - Outdoor): %.1f°C", temperature_diff) + + if abs(temperature_diff) > 0.1: # Minimum difference threshold + efficiency = ((supply_temp - outdoor_temp) / temperature_diff) * 100 + _LOGGER.debug("Calculated efficiency: %.1f%% (formula: (%.1f - %.1f) / %.1f * 100)", + efficiency, supply_temp, outdoor_temp, temperature_diff) + + # Clamp efficiency between 0 and 100% + final_efficiency = max(0, min(100, round(efficiency, 1))) + _LOGGER.debug("Final clamped efficiency: %.1f%%", final_efficiency) + return final_efficiency + else: + _LOGGER.debug("Temperature difference too small: %.1f°C", temperature_diff) + else: + missing = [] + if supply_temp is None: missing.append("Supply") + if exhaust_temp is None: missing.append("Exhaust") + if outdoor_temp is None: missing.append("Outdoor") + _LOGGER.debug("Missing temperature sensors for efficiency calculation: %s", ", ".join(missing)) + + return None + + @property + def entity_category(self): + """Return the entity category for diagnostic entities.""" + return EntityCategory.DIAGNOSTIC + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, + CONF_NAME: f"{self._config_entry.title} Recuperation", + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), + ATTR_MANUFACTURER: MANUFACTURER, + } + + +class OutdoorTemperatureSensor(CoordinatorEntity, SensorEntity): + """Sensor for outdoor temperature from recuperation system - HomeKit friendly.""" + + _attr_has_entity_name = True + _attr_native_unit_of_measurement = UnitOfTemperature.CELSIUS + _attr_device_class = SensorDeviceClass.TEMPERATURE + _attr_state_class = SensorStateClass.MEASUREMENT + _attr_icon = "mdi:thermometer" + + def __init__( + self, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the outdoor temperature sensor.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_unique_id = f"{self._udid}_outdoor_temperature" + + # Simple, HomeKit-friendly name + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Outdoor Temperature" + + @property + def name(self) -> str: + """Return the name of the sensor.""" + return self._name + + @property + def native_value(self) -> float | None: + """Return the outdoor temperature value.""" + if self._coordinator.data and "tiles" in self._coordinator.data: + # Search for outdoor temperature in both TYPE_TEMPERATURE and TYPE_TEMPERATURE_CH tiles + for tile_data in self._coordinator.data["tiles"].values(): + # Check TYPE_TEMPERATURE tiles + if tile_data.get("type") == TYPE_TEMPERATURE: + tile_txt_id = tile_data.get("params", {}).get("txtId", 0) + temp_value = tile_data.get("params", {}).get("value") + + if temp_value is not None and tile_txt_id in [121, 5995]: # External/Fresh Air + return temp_value / 10 # Convert from tenths + + # Check TYPE_TEMPERATURE_CH widgets + elif tile_data.get("type") == TYPE_TEMPERATURE_CH: + for widget_key in ["widget1", "widget2"]: + widget_data = tile_data.get("params", {}).get(widget_key, {}) + widget_txt_id = widget_data.get("txtId", 0) + temp_value = widget_data.get("value") + + if temp_value is not None and widget_txt_id in [121, 5995]: # External/Fresh Air + return temp_value / 10 # Convert from tenths + + return None + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format - separate device for HomeKit.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_weather_station") + }, + CONF_NAME: f"{self._config_entry.title} Weather Station", + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), + ATTR_MANUFACTURER: MANUFACTURER, + } diff --git a/custom_components/tech/switch.py b/custom_components/tech/switch.py new file mode 100644 index 0000000..2164504 --- /dev/null +++ b/custom_components/tech/switch.py @@ -0,0 +1,144 @@ +"""Support for Tech HVAC switch controls.""" + +import logging +from typing import Any + +from homeassistant.components.switch import SwitchEntity +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import ( + ATTR_IDENTIFIERS, + ATTR_MANUFACTURER, + CONF_MODEL, + CONF_NAME, +) +from homeassistant.core import HomeAssistant +from homeassistant.helpers.device_registry import DeviceInfo +from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.helpers.update_coordinator import CoordinatorEntity + +from .const import ( + CONTROLLER, + DOMAIN, + INCLUDE_HUB_IN_NAME, + MANUFACTURER, + RECUPERATION_EXHAUST_FLOW, + RECUPERATION_SUPPLY_FLOW, + RECUPERATION_SUPPLY_FLOW_ALT, + TYPE_TEMPERATURE_CH, + UDID, + VER, +) +from .coordinator import TechCoordinator + +_LOGGER = logging.getLogger(__name__) + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + """Set up entry.""" + _LOGGER.debug( + "Setting up switch entry, controller udid: %s", + config_entry.data[CONTROLLER][UDID], + ) + coordinator = hass.data[DOMAIN][config_entry.entry_id] + controller_udid = config_entry.data[CONTROLLER][UDID] + + tiles = await coordinator.api.get_module_tiles(controller_udid) + + entities = [] + + # Check if we have recuperation system (detected by flow sensors) + has_recuperation_flow = False + for t in tiles: + tile = tiles[t] + if tile.get("type") == TYPE_TEMPERATURE_CH: + widget1_txt_id = tile.get("params", {}).get("widget1", {}).get("txtId", 0) + widget2_txt_id = tile.get("params", {}).get("widget2", {}).get("txtId", 0) + for flow_sensor in [RECUPERATION_EXHAUST_FLOW, RECUPERATION_SUPPLY_FLOW, RECUPERATION_SUPPLY_FLOW_ALT]: + if flow_sensor["txt_id"] in [widget1_txt_id, widget2_txt_id]: + has_recuperation_flow = True + break + if has_recuperation_flow: + break + + # Create flow balancing switch if we have recuperation + if has_recuperation_flow: + _LOGGER.debug("Creating flow balancing switch") + entities.append(FlowBalancingSwitch(coordinator, config_entry)) + else: + _LOGGER.debug("No recuperation flow detected, skipping flow balancing switch") + + async_add_entities(entities, True) + + +class FlowBalancingSwitch(CoordinatorEntity, SwitchEntity): + """Representation of flow balancing switch.""" + + _attr_has_entity_name = True + _attr_icon = "mdi:scale-balance" + + def __init__( + self, + coordinator: TechCoordinator, + config_entry: ConfigEntry, + ) -> None: + """Initialize the flow balancing switch.""" + super().__init__(coordinator) + self._coordinator = coordinator + self._config_entry = config_entry + self._udid = config_entry.data[CONTROLLER][UDID] + self._attr_unique_id = f"{self._udid}_flow_balancing" + + self._name = ( + self._config_entry.title + " " + if self._config_entry.data[INCLUDE_HUB_IN_NAME] + else "" + ) + "Flow Balancing" + + @property + def name(self) -> str: + """Return the name of the switch.""" + return self._name + + @property + def is_on(self) -> bool: + """Return true if the switch is on.""" + # Default to True (enabled) + return True + + async def async_turn_on(self, **kwargs: Any) -> None: + """Turn the switch on.""" + _LOGGER.debug("Turning on flow balancing") + await self._coordinator.api.set_flow_balancing(self._udid, True) + await self._coordinator.async_request_refresh() + + async def async_turn_off(self, **kwargs: Any) -> None: + """Turn the switch off.""" + _LOGGER.debug("Turning off flow balancing") + await self._coordinator.api.set_flow_balancing(self._udid, False) + await self._coordinator.async_request_refresh() + + @property + def entity_category(self): + """Return the entity category for configuration entities.""" + from homeassistant.helpers.entity import EntityCategory + return EntityCategory.CONFIG + + @property + def device_info(self) -> DeviceInfo | None: + """Returns device information in a dictionary format.""" + return { + ATTR_IDENTIFIERS: { + (DOMAIN, f"{self._udid}_recuperation") + }, # Unique identifiers for the device + CONF_NAME: f"{self._config_entry.title} Recuperation", # Name of the device + CONF_MODEL: ( + self._config_entry.data[CONTROLLER][CONF_NAME] + + ": " + + self._config_entry.data[CONTROLLER][VER] + ), # Model of the device + ATTR_MANUFACTURER: MANUFACTURER, # Manufacturer of the device + } \ No newline at end of file diff --git a/custom_components/tech/tech.py b/custom_components/tech/tech.py index 92e69e4..40f5cc9 100644 --- a/custom_components/tech/tech.py +++ b/custom_components/tech/tech.py @@ -37,7 +37,11 @@ def __init__( """ _LOGGER.debug("Init Tech") - self.headers = {"Accept": "application/json", "Accept-Encoding": "gzip"} + self.headers = { + "Accept": "application/json", + "Accept-Encoding": "gzip", + "User-Agent": "TechController/1.0 (Home Assistant Integration)" + } self.base_url = base_url self.session = session if user_id and token: @@ -120,6 +124,7 @@ async def authenticate(self, username, password): self.headers = { "Accept": "application/json", "Accept-Encoding": "gzip", + "User-Agent": "TechController/1.0 (Home Assistant Integration)", "Authorization": f"Bearer {self.token}", } except TechError as err: @@ -395,6 +400,383 @@ async def set_zone(self, module_udid, zone_id, on=True): raise TechError(401, "Unauthorized") return result + async def set_fan_gear(self, module_udid, tile_id, gear): + """Set the gear (speed) of a fan/recuperation unit. + + Args: + module_udid (string): The Tech module udid. + tile_id (int): The tile ID. + gear (int): The gear/speed level (0 = off, 1-3 = speeds). + + Returns: + JSON object with the result. + + """ + _LOGGER.debug("Setting fan gear: tile_id=%s, gear=%s", tile_id, gear) + if self.authenticated: + path = f"users/{self.user_id}/modules/{module_udid}/menu/MI/ido/{tile_id}" + data = {"gear": gear} + _LOGGER.debug("POST data: %s", data) + result = await self.post(path, json.dumps(data)) + _LOGGER.debug("Response: %s", result) + else: + raise TechError(401, "Unauthorized") + return result + + async def set_recuperation_speed(self, module_udid, speed_level, configured_values=None): + """Set the speed of recuperation unit. + + Args: + module_udid (string): The Tech module udid. + speed_level (int): The speed level (1-3 = speeds, 0 = off). + configured_values (dict): Optional configured flow values for each speed. + + Returns: + JSON object with the result. + + """ + from .const import RECUPERATION_SPEED_ENDPOINTS, DEFAULT_SPEED_VALUES + + _LOGGER.debug("Setting recuperation speed: speed_level=%s", speed_level) + if self.authenticated: + if speed_level == 0: + # Turn off - set speed 1 to 0 + ido_id = RECUPERATION_SPEED_ENDPOINTS[1]["ido_id"] + path = f"users/{self.user_id}/modules/{module_udid}/menu/MI/ido/{ido_id}" + data = {"value": 0} + elif speed_level in RECUPERATION_SPEED_ENDPOINTS: + endpoint_config = RECUPERATION_SPEED_ENDPOINTS[speed_level] + ido_id = endpoint_config["ido_id"] + + # Use configured value if available, otherwise default + if configured_values and speed_level in configured_values: + flow_value = configured_values[speed_level] + else: + flow_value = DEFAULT_SPEED_VALUES[speed_level] + + path = f"users/{self.user_id}/modules/{module_udid}/menu/MI/ido/{ido_id}" + data = {"value": flow_value} + else: + raise TechError(400, f"Invalid speed level: {speed_level}") + + _LOGGER.debug("POST to: %s with data: %s", path, data) + result = await self.post(path, json.dumps(data)) + _LOGGER.debug("Response: %s", result) + else: + raise TechError(401, "Unauthorized") + return result + + async def set_party_mode(self, module_udid, duration_minutes): + """Set party mode duration for recuperation. + + Args: + module_udid (string): The Tech module udid. + duration_minutes (int): Duration in minutes (15-720). + + Returns: + JSON object with the result. + + """ + from .const import PARTY_MODE_IDO_ID, PARTY_MODE_MIN_MINUTES, PARTY_MODE_MAX_MINUTES + + _LOGGER.debug("Setting party mode: duration=%s minutes", duration_minutes) + if self.authenticated: + # Validate duration + if not (PARTY_MODE_MIN_MINUTES <= duration_minutes <= PARTY_MODE_MAX_MINUTES): + raise TechError(400, f"Invalid duration: {duration_minutes}. Must be between {PARTY_MODE_MIN_MINUTES} and {PARTY_MODE_MAX_MINUTES} minutes") + + path = f"users/{self.user_id}/modules/{module_udid}/menu/MU/ido/{PARTY_MODE_IDO_ID}" + data = {"value": duration_minutes} + _LOGGER.debug("POST to: %s with data: %s", path, data) + result = await self.post(path, json.dumps(data)) + _LOGGER.debug("Response: %s", result) + else: + raise TechError(401, "Unauthorized") + return result + + async def set_fan_mode(self, module_udid, mode_value): + """Set fan mode (stop, speed 1-3). + + Args: + module_udid (string): The Tech module udid. + mode_value (int): Mode value (0=stop, 1-3=speeds). + + Returns: + JSON object with the result. + + """ + from .const import FAN_MODE_IDO_ID + + _LOGGER.debug("Setting fan mode: mode_value=%s", mode_value) + if self.authenticated: + # Validate mode value + if not (0 <= mode_value <= 3): + raise TechError(400, f"Invalid mode value: {mode_value}. Must be between 0 and 3") + + path = f"users/{self.user_id}/modules/{module_udid}/menu/MU/ido/{FAN_MODE_IDO_ID}" + data = {"value": mode_value} + _LOGGER.debug("POST to: %s with data: %s", path, data) + result = await self.post(path, json.dumps(data)) + _LOGGER.debug("Response: %s", result) + else: + raise TechError(401, "Unauthorized") + return result + + async def set_filter_alarm(self, module_udid, days): + """Set filter alarm interval in days. + + Args: + module_udid (string): The Tech module udid. + days (int): Days until filter alarm (30-120). + + Returns: + JSON object with the result. + + """ + from .const import FILTER_ALARM_IDO_ID, FILTER_ALARM_MIN_DAYS, FILTER_ALARM_MAX_DAYS + + _LOGGER.debug("Setting filter alarm: days=%s", days) + if self.authenticated: + # Validate days + if not (FILTER_ALARM_MIN_DAYS <= days <= FILTER_ALARM_MAX_DAYS): + raise TechError(400, f"Invalid days: {days}. Must be between {FILTER_ALARM_MIN_DAYS} and {FILTER_ALARM_MAX_DAYS} days") + + path = f"users/{self.user_id}/modules/{module_udid}/menu/MI/ido/{FILTER_ALARM_IDO_ID}" + data = {"value": days} + _LOGGER.debug("POST to: %s with data: %s", path, data) + result = await self.post(path, json.dumps(data)) + _LOGGER.debug("Response: %s", result) + else: + raise TechError(401, "Unauthorized") + return result + + async def get_filter_usage(self, module_udid): + """Get filter usage counter. + + Args: + module_udid (string): The Tech module udid. + + Returns: + JSON object with the result. + + """ + from .const import FILTER_USAGE_IDO_ID + + _LOGGER.debug("Getting filter usage") + if self.authenticated: + path = f"users/{self.user_id}/modules/{module_udid}/menu/MI/ido/{FILTER_USAGE_IDO_ID}" + _LOGGER.debug("GET from: %s", path) + result = await self.get(path) + _LOGGER.debug("Response: %s", result) + else: + raise TechError(401, "Unauthorized") + return result + + async def update_filter_data(self, module_udid): + """Update filter data/reset filter timer. + + Args: + module_udid (string): The Tech module udid. + + Returns: + JSON object with the result. + + """ + _LOGGER.debug("Updating filter data") + if self.authenticated: + # Get current timestamp + from datetime import datetime + current_time = datetime.now().isoformat() + + path = f"users/{self.user_id}/modules/{module_udid}/update/data/parents/[]/alarm_ids/[]/last_update/{current_time}" + result = await self.post(path, "{}") + _LOGGER.debug("Filter update response: %s", result) + else: + raise TechError(401, "Unauthorized") + return result + + async def set_ventilation_room_parameter(self, module_udid, percent): + """Set room ventilation parameter. + + Args: + module_udid (string): The Tech module udid. + percent (int): Ventilation percentage (10-90). + + Returns: + JSON object with the result. + + """ + from .const import VENTILATION_ROOM_IDO_ID, VENTILATION_MIN_PERCENT, VENTILATION_MAX_PERCENT + + _LOGGER.debug("Setting room ventilation parameter: percent=%s", percent) + if self.authenticated: + # Validate percent + if not (VENTILATION_MIN_PERCENT <= percent <= VENTILATION_MAX_PERCENT): + raise TechError(400, f"Invalid percent: {percent}. Must be between {VENTILATION_MIN_PERCENT} and {VENTILATION_MAX_PERCENT}%") + + path = f"users/{self.user_id}/modules/{module_udid}/menu/MI/ido/{VENTILATION_ROOM_IDO_ID}" + data = {"value": percent} + _LOGGER.debug("POST to: %s with data: %s", path, data) + result = await self.post(path, json.dumps(data)) + _LOGGER.debug("Response: %s", result) + else: + raise TechError(401, "Unauthorized") + return result + + async def set_ventilation_bathroom_parameter(self, module_udid, percent): + """Set bathroom ventilation parameter. + + Args: + module_udid (string): The Tech module udid. + percent (int): Ventilation percentage (10-90). + + Returns: + JSON object with the result. + + """ + from .const import VENTILATION_BATHROOM_IDO_ID, VENTILATION_MIN_PERCENT, VENTILATION_MAX_PERCENT + + _LOGGER.debug("Setting bathroom ventilation parameter: percent=%s", percent) + if self.authenticated: + # Validate percent + if not (VENTILATION_MIN_PERCENT <= percent <= VENTILATION_MAX_PERCENT): + raise TechError(400, f"Invalid percent: {percent}. Must be between {VENTILATION_MIN_PERCENT} and {VENTILATION_MAX_PERCENT}%") + + path = f"users/{self.user_id}/modules/{module_udid}/menu/MI/ido/{VENTILATION_BATHROOM_IDO_ID}" + data = {"value": percent} + _LOGGER.debug("POST to: %s with data: %s", path, data) + result = await self.post(path, json.dumps(data)) + _LOGGER.debug("Response: %s", result) + else: + raise TechError(401, "Unauthorized") + return result + + async def set_co2_threshold(self, module_udid, ppm): + """Set CO2 threshold parameter. + + Args: + module_udid (string): The Tech module udid. + ppm (int): CO2 threshold in PPM (400-2000). + + Returns: + JSON object with the result. + + """ + from .const import CO2_THRESHOLD_IDO_ID, CO2_MIN_PPM, CO2_MAX_PPM + + _LOGGER.debug("Setting CO2 threshold: ppm=%s", ppm) + if self.authenticated: + # Validate ppm + if not (CO2_MIN_PPM <= ppm <= CO2_MAX_PPM): + raise TechError(400, f"Invalid ppm: {ppm}. Must be between {CO2_MIN_PPM} and {CO2_MAX_PPM} PPM") + + path = f"users/{self.user_id}/modules/{module_udid}/menu/MI/ido/{CO2_THRESHOLD_IDO_ID}" + data = {"value": ppm} + _LOGGER.debug("POST to: %s with data: %s", path, data) + result = await self.post(path, json.dumps(data)) + _LOGGER.debug("Response: %s", result) + else: + raise TechError(401, "Unauthorized") + return result + + async def set_hysteresis(self, module_udid, percent): + """Set hysteresis parameter. + + Args: + module_udid (string): The Tech module udid. + percent (int): Hysteresis percentage (5-10). + + Returns: + JSON object with the result. + + """ + from .const import HYSTERESIS_IDO_ID, HYSTERESIS_MIN_PERCENT, HYSTERESIS_MAX_PERCENT + + _LOGGER.debug("Setting hysteresis: percent=%s", percent) + if self.authenticated: + # Validate percent + if not (HYSTERESIS_MIN_PERCENT <= percent <= HYSTERESIS_MAX_PERCENT): + raise TechError(400, f"Invalid percent: {percent}. Must be between {HYSTERESIS_MIN_PERCENT} and {HYSTERESIS_MAX_PERCENT}%") + + path = f"users/{self.user_id}/modules/{module_udid}/menu/MI/ido/{HYSTERESIS_IDO_ID}" + data = {"value": percent} + _LOGGER.debug("POST to: %s with data: %s", path, data) + result = await self.post(path, json.dumps(data)) + _LOGGER.debug("Response: %s", result) + else: + raise TechError(401, "Unauthorized") + return result + + async def set_flow_balancing(self, module_udid, enabled): + """Set flow balancing setting. + + Args: + module_udid (string): The Tech module udid. + enabled (bool): Flow balancing enabled (True/False). + + Returns: + JSON object with the result. + + """ + from .const import FLOW_BALANCING_IDO_ID + + _LOGGER.debug("Setting flow balancing: enabled=%s", enabled) + if self.authenticated: + value = 1 if enabled else 0 + path = f"users/{self.user_id}/modules/{module_udid}/menu/MI/ido/{FLOW_BALANCING_IDO_ID}" + data = {"value": value} + _LOGGER.debug("POST to: %s with data: %s", path, data) + result = await self.post(path, json.dumps(data)) + _LOGGER.debug("Response: %s", result) + else: + raise TechError(401, "Unauthorized") + return result + + async def get_current_gear(self, module_udid): + """Get current recuperation gear value. + + Args: + module_udid (string): The Tech module udid. + + Returns: + Current gear value (0-3). + + """ + from .const import GEAR_CONTROL_IDO_ID + + if self.authenticated: + path = f"users/{self.user_id}/modules/{module_udid}/menu/MU/ido/{GEAR_CONTROL_IDO_ID}" + _LOGGER.debug("GET from: %s", path) + result = await self.get(path) + _LOGGER.debug("Current gear response: %s", result) + return result.get("value", 0) if result else 0 + else: + raise TechError(401, "Unauthorized") + + async def set_gear_direct(self, module_udid, gear_value): + """Set recuperation gear directly. + + Args: + module_udid (string): The Tech module udid. + gear_value (int): Gear value (0=stop, 1=speed1, 2=speed2, 3=speed3). + + Returns: + JSON object with the result. + + """ + from .const import GEAR_CONTROL_IDO_ID + + _LOGGER.debug("Setting recuperation gear to: %d", gear_value) + if self.authenticated: + path = f"users/{self.user_id}/modules/{module_udid}/menu/MU/ido/{GEAR_CONTROL_IDO_ID}" + data = {"value": gear_value} + _LOGGER.debug("POST to: %s with data: %s", path, data) + result = await self.post(path, json.dumps(data)) + _LOGGER.debug("Response: %s", result) + else: + raise TechError(401, "Unauthorized") + return result + class TechError(Exception): """Raised when Tech API request ended in error. diff --git a/devenv.lock b/devenv.lock old mode 100644 new mode 100755 index 4ea8a98..9dd09d1 --- a/devenv.lock +++ b/devenv.lock @@ -3,10 +3,10 @@ "devenv": { "locked": { "dir": "src/modules", - "lastModified": 1750097069, + "lastModified": 1758122374, "owner": "cachix", "repo": "devenv", - "rev": "ab8a1563714393d366b2db7d93f021afb45e7a6f", + "rev": "ed1d02598f4cd58643fd8b168c8341fe23d815a3", "type": "github" }, "original": { @@ -55,10 +55,10 @@ ] }, "locked": { - "lastModified": 1749636823, + "lastModified": 1758108966, "owner": "cachix", "repo": "git-hooks.nix", - "rev": "623c56286de5a3193aa38891a6991b28f9bab056", + "rev": "54df955a695a84cd47d4a43e08e1feaf90b1fd9b", "type": "github" }, "original": { @@ -89,10 +89,10 @@ }, "nixpkgs": { "locked": { - "lastModified": 1746807397, + "lastModified": 1755783167, "owner": "cachix", "repo": "devenv-nixpkgs", - "rev": "c5208b594838ea8e6cca5997fbf784b7cca1ca90", + "rev": "4a880fb247d24fbca57269af672e8f78935b0328", "type": "github" }, "original": { @@ -110,10 +110,10 @@ ] }, "locked": { - "lastModified": 1749760516, + "lastModified": 1755249745, "owner": "cachix", "repo": "nixpkgs-python", - "rev": "908dbb466af5955ea479ac95953333fd64387216", + "rev": "b6632af2db9f47c79dac8f4466388c7b1b6c3071", "type": "github" }, "original": { diff --git a/devenv.nix b/devenv.nix old mode 100644 new mode 100755 index 288a164..6b54090 --- a/devenv.nix +++ b/devenv.nix @@ -16,6 +16,8 @@ pkgs.git pkgs.ruff pkgs.ffmpeg + pkgs.libpcap + pkgs.libjpeg ]; # https://devenv.sh/languages/ @@ -23,15 +25,15 @@ enable = true; version = "3.13"; uv.enable = true; - venv.enable = true; - venv.requirements = builtins.readFile ./requirements.txt + "\n" + builtins.readFile ./requirements_test_api.txt; + uv.sync.enable = true; }; # https://devenv.sh/scripts/ scripts.setup = { exec = '' echo '🛠️ Running setup' - uv pip sync requirements.txt requirements_test_api.txt + # Sync with pyproject.toml via pip + uv sync --group test_api ''; description = "Install dependencies"; }; @@ -84,6 +86,7 @@ echo echo 🦾 Available scripts: echo 🦾 + . .devenv/state/venv/bin/activate ${pkgs.gnused}/bin/sed -e 's| |••|g' -e 's|=| |' <=61.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "tech-controllers" +version = "1.0.0" +description = "Home Assistant integration for TECH Sterowniki heating controllers" +readme = "README.md" +license = { file = "LICENSE" } +authors = [ + { name = "Mariusz Ostaja-Swierczynski", email = "mariusz.ostaja.swierczynski@gmail.com" }, + { name = "anarion80", email = "anarion80@users.noreply.github.com" }, +] +maintainers = [ + { name = "anarion80", email = "anarion80@users.noreply.github.com" }, +] +classifiers = [ + "Development Status :: 5 - Production/Stable", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.13", +] +requires-python = ">=3.13.2" +dependencies = [ + "aiodns==3.5.0", + "colorlog==6.9.0", + "homeassistant==2025.9.3", + "pip==25.2", + "pre_commit==4.3.0", + "ruff==0.13.0", + "zlib-ng==1.0.0", +] + +[dependency-groups] +test_api = [ + "acme==5.0.0", + "aiodns==3.5.0", + "aiofiles==24.1.0", + "aiohappyeyeballs==2.6.1", + "aiohasupervisor==0.3.2", + "aiohttp-asyncmdnsresolver==0.1.1", + "aiohttp-fast-zlib==0.3.0", + "aiohttp==3.12.15", + "aiohttp_cors==0.8.1", + "aiooui==0.1.9", + "aiosignal==1.4.0", + "aiozoneinfo==0.2.3", + "annotatedyaml==0.4.5", + "astral==2.2", + "async-interrupt==1.2.2", + "async_timeout==5.0.1", + "atomicwrites-homeassistant==1.4.1", + "attrs==25.3.0", + "audioop-lts==0.2.1", + "awesomeversion==25.5.0", + "bcrypt==4.3.0", + "bleak-retry-connector==4.4.3", + "bleak==1.1.1", + "bluetooth-auto-recovery==1.5.3", + "bluetooth-data-tools==1.28.2", + "bluetooth_adapters==2.1.1", + "boto3==1.40.32", + "botocore==1.40.32", + "btsocket==0.3.0", + "certifi>=2021.5.30", + "cffi==2.0.0", + "ciso8601==2.3.2", + "coverage==7.10.0", + "cronsim==2.6", + "cryptography==45.0.3", + "dbus_fast==2.44.3", + "envs==1.4", + "execnet==2.1.1", + "fnv-hash-fast==1.5.0", + "fnvhash>=0.1.0,<0.2.0", + "frozenlist==1.7.0", + "habluetooth==5.6.4", + "hass-nabucasa==1.1.1", + "home-assistant-bluetooth==1.13.1", + "httpx==0.28.1", + "idna==3.10", + "ifaddr==0.2.0", + "iniconfig==2.1.0", + "Jinja2==3.1.6", + "jmespath==1.0.1", + "josepy==2.1.0", + "libpcap>=1.11.0b25", + "lru-dict==1.3.0", + "markupsafe==3.0.2", + "mashumaro==3.16", + "multidict==6.6.4", + "orjson==3.11.3", + "packaging>=23.1", + "Pillow==11.3.0", + "pluggy==1.6.0", + "propcache==0.3.2", + "psutil-home-assistant==0.0.1", + "pycognito==2024.5.1", + "Pygments==2.19.2", + "PyJWT==2.10.1", + "pyOpenSSL==25.1.0", + "pyrfc3339==2.1.0", + "pytest-aiohttp==1.1.0", + "pytest-asyncio==1.1.0", + "pytest-cov==6.2.1", + "pytest-emoji==0.2.0", + "pytest-md==0.2.0", + "pytest-sugar==1.0.0", + "pytest-test-groups==1.2.1", + "pytest-timeout==2.4.0", + "pytest-xdist==3.8.0", + "pytest==8.4.1", + "python-dateutil>=2.1,<3.0.0", + "python-slugify==8.0.4", + "PyTurboJPEG==1.8.2", + "pytz==2025.2", + "PyYAML==6.0.2", + "regex==2024.11.6", + "requests==2.32.4", + "securetar==2025.2.1", + "sentence-stream==1.2.0", + "six==1.17.0", + "snitun==0.44.0", + "SQLAlchemy==2.0.41", + "standard-aifc==3.13.0", + "standard-telnetlib==3.13.0", + "termcolor==3.1.0", + "text_unidecode==1.3", + "typing-extensions>=4.15.0,<5.0", + "ulid-transform==1.4.0", + "urllib3>=2.0", + "uv==0.8.9", + "voluptuous-openapi==0.1.0", + "voluptuous-serialize==2.7.0", + "voluptuous==0.15.2", + "webrtc-models==0.3.0", + "yarl==1.20.1", + "zeroconf==0.147.0", +] + +test = [ + "acme==5.0.0", + "aiodns==3.5.0", + "aiofiles==24.1.0", + "aiohappyeyeballs==2.6.1", + "aiohasupervisor==0.3.2", + "aiohttp-asyncmdnsresolver==0.1.1", + "aiohttp-fast-zlib==0.3.0", + "aiohttp==3.12.15", + "aiohttp_cors==0.8.1", + "aiooui==0.1.9", + "aiosignal==1.4.0", + "aiozoneinfo==0.2.3", + "annotatedyaml==0.4.5", + "astral==2.2", + "async-interrupt==1.2.2", + "async_timeout==5.0.1", + "atomicwrites-homeassistant==1.4.1", + "attrs==25.3.0", + "audioop-lts==0.2.1", + "awesomeversion==25.5.0", + "bcrypt==4.3.0", + "bleak-retry-connector==4.4.3", + "bleak==1.1.1", + "bluetooth-auto-recovery==1.5.3", + "bluetooth-data-tools==1.28.2", + "bluetooth_adapters==2.1.1", + "boto3==1.40.32", + "botocore==1.40.32", + "btsocket==0.3.0", + "certifi>=2021.5.30", + "cffi==2.0.0", + "ciso8601==2.3.2", + "coverage==7.10.0", + "cronsim==2.6", + "cryptography==45.0.3", + "dbus_fast==2.44.3", + "envs==1.4", + "execnet==2.1.1", + "fnv-hash-fast==1.5.0", + "fnvhash>=0.1.0,<0.2.0", + "frozenlist==1.7.0", + "habluetooth==5.6.4", + "hass-nabucasa==1.1.1", + "home-assistant-bluetooth==1.13.1", + "httpx==0.28.1", + "idna==3.10", + "ifaddr==0.2.0", + "iniconfig==2.1.0", + "Jinja2==3.1.6", + "jmespath==1.0.1", + "josepy==2.1.0", + "libpcap>=1.11.0b25", + "lru-dict==1.3.0", + "markupsafe==3.0.2", + "mashumaro==3.16", + "multidict==6.6.4", + "orjson==3.11.3", + "packaging>=23.1", + "Pillow==11.3.0", + "pluggy==1.6.0", + "propcache==0.3.2", + "psutil-home-assistant==0.0.1", + "pycognito==2024.5.1", + "Pygments==2.19.2", + "PyJWT==2.10.1", + "pyOpenSSL==25.1.0", + "pyrfc3339==2.1.0", + "pytest-aiohttp==1.1.0", + "pytest-asyncio==1.1.0", + "pytest-cov==6.2.1", + "pytest-emoji==0.2.0", + "pytest-homeassistant-custom-component==0.13.280", + "pytest-md==0.2.0", + "pytest-sugar==1.0.0", + "pytest-test-groups==1.2.1", + "pytest-timeout==2.4.0", + "pytest-xdist==3.8.0", + "pytest==8.4.1", + "python-dateutil>=2.1,<3.0.0", + "python-slugify==8.0.4", + "PyTurboJPEG==1.8.2", + "pytz==2025.2", + "PyYAML==6.0.2", + "regex==2024.11.6", + "requests==2.32.4", + "securetar==2025.2.1", + "sentence-stream==1.2.0", + "six==1.17.0", + "snitun==0.44.0", + "SQLAlchemy==2.0.41", + "standard-aifc==3.13.0", + "standard-telnetlib==3.13.0", + "termcolor==3.1.0", + "text_unidecode==1.3", + "typing-extensions>=4.15.0,<5.0", + "ulid-transform==1.4.0", + "urllib3>=2.0", + "uv==0.8.9", + "voluptuous-openapi==0.1.0", + "voluptuous-serialize==2.7.0", + "voluptuous==0.15.2", + "webrtc-models==0.3.0", + "yarl==1.20.1", + "zeroconf==0.147.0", +] + +[tool.setuptools.packages.find] +where = ["custom_components"] + +[tool.setuptools.package-data] +"*" = ["*.json", "*.yaml", "*.yml"] + +[project.urls] +Homepage = "https://github.com/mariusz-ostoja-swierczynski/tech-controllers" +Repository = "https://github.com/mariusz-ostoja-swierczynski/tech-controllers.git" +Issues = "https://github.com/mariusz-ostoja-swierczynski/tech-controllers/issues" +Changelog = "https://github.com/mariusz-ostoja-swierczynski/tech-controllers/releases"