"""Support for 1-Wire environment switches.""" from __future__ import annotations from datetime import timedelta import os from typing import Any from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription from homeassistant.const import EntityCategory from homeassistant.core import HomeAssistant from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from .const import DEVICE_KEYS_0_3, DEVICE_KEYS_0_7, DEVICE_KEYS_A_B from .entity import OneWireEntity from .onewirehub import ( SIGNAL_NEW_DEVICE_CONNECTED, OneWireConfigEntry, OneWireHub, OWDeviceDescription, ) # the library uses non-persistent connections # and concurrent access to the bus is managed by the server PARALLEL_UPDATES = 0 SCAN_INTERVAL = timedelta(seconds=30) DEVICE_SWITCHES: dict[str, tuple[SwitchEntityDescription, ...]] = { "05": ( SwitchEntityDescription( key="PIO", entity_registry_enabled_default=False, translation_key="pio", ), ), "12": tuple( [ SwitchEntityDescription( key=f"PIO.{device_key}", entity_registry_enabled_default=False, translation_key="pio_id", translation_placeholders={"id": str(device_key)}, ) for device_key in DEVICE_KEYS_A_B ] + [ SwitchEntityDescription( key=f"latch.{device_key}", entity_registry_enabled_default=False, translation_key="latch_id", translation_placeholders={"id": str(device_key)}, ) for device_key in DEVICE_KEYS_A_B ] ), "26": ( SwitchEntityDescription( key="IAD", entity_registry_enabled_default=False, entity_category=EntityCategory.CONFIG, translation_key="iad", ), ), "29": tuple( [ SwitchEntityDescription( key=f"PIO.{device_key}", entity_registry_enabled_default=False, translation_key="pio_id", translation_placeholders={"id": str(device_key)}, ) for device_key in DEVICE_KEYS_0_7 ] + [ SwitchEntityDescription( key=f"latch.{device_key}", entity_registry_enabled_default=False, translation_key="latch_id", translation_placeholders={"id": str(device_key)}, ) for device_key in DEVICE_KEYS_0_7 ] ), "3A": tuple( SwitchEntityDescription( key=f"PIO.{device_key}", entity_registry_enabled_default=False, translation_key="pio_id", translation_placeholders={"id": str(device_key)}, ) for device_key in DEVICE_KEYS_A_B ), "EF": (), # "HobbyBoard": special } # EF sensors are usually hobbyboards specialized sensors. HOBBYBOARD_EF: dict[str, tuple[SwitchEntityDescription, ...]] = { "HB_HUB": tuple( SwitchEntityDescription( key=f"hub/branch.{device_key}", entity_registry_enabled_default=False, entity_category=EntityCategory.CONFIG, translation_key="hub_branch_id", translation_placeholders={"id": str(device_key)}, ) for device_key in DEVICE_KEYS_0_3 ), "HB_MOISTURE_METER": tuple( [ SwitchEntityDescription( key=f"moisture/is_leaf.{device_key}", entity_registry_enabled_default=False, entity_category=EntityCategory.CONFIG, translation_key="leaf_sensor_id", translation_placeholders={"id": str(device_key)}, ) for device_key in DEVICE_KEYS_0_3 ] + [ SwitchEntityDescription( key=f"moisture/is_moisture.{device_key}", entity_registry_enabled_default=False, entity_category=EntityCategory.CONFIG, translation_key="moisture_sensor_id", translation_placeholders={"id": str(device_key)}, ) for device_key in DEVICE_KEYS_0_3 ] ), } def get_sensor_types( device_sub_type: str, ) -> dict[str, tuple[SwitchEntityDescription, ...]]: """Return the proper info array for the device type.""" if "HobbyBoard" in device_sub_type: return HOBBYBOARD_EF return DEVICE_SWITCHES async def async_setup_entry( hass: HomeAssistant, config_entry: OneWireConfigEntry, async_add_entities: AddConfigEntryEntitiesCallback, ) -> None: """Set up 1-Wire platform.""" async def _add_entities( hub: OneWireHub, devices: list[OWDeviceDescription] ) -> None: """Add 1-Wire entities for all devices.""" if not devices: return async_add_entities(get_entities(hub, devices), True) hub = config_entry.runtime_data await _add_entities(hub, hub.devices) config_entry.async_on_unload( async_dispatcher_connect(hass, SIGNAL_NEW_DEVICE_CONNECTED, _add_entities) ) def get_entities( onewire_hub: OneWireHub, devices: list[OWDeviceDescription] ) -> list[OneWireSwitchEntity]: """Get a list of entities.""" entities: list[OneWireSwitchEntity] = [] for device in devices: family = device.family device_type = device.type device_id = device.id device_info = device.device_info device_sub_type = "std" if device_type and "EF" in family: device_sub_type = "HobbyBoard" family = device_type elif "A6" in family: # A6 is a secondary family code for DS2438 family = "26" if family not in get_sensor_types(device_sub_type): continue for description in get_sensor_types(device_sub_type)[family]: device_file = os.path.join(os.path.split(device.path)[0], description.key) entities.append( OneWireSwitchEntity( description=description, device_id=device_id, device_file=device_file, device_info=device_info, owproxy=onewire_hub.owproxy, ) ) return entities class OneWireSwitchEntity(OneWireEntity, SwitchEntity): """Implementation of a 1-Wire switch.""" @property def is_on(self) -> bool | None: """Return true if switch is on.""" if (state := self._state) is None: return None return state == "1" async def async_turn_on(self, **kwargs: Any) -> None: """Turn the entity on.""" await self._write_value(b"1") async def async_turn_off(self, **kwargs: Any) -> None: """Turn the entity off.""" await self._write_value(b"0")