diff --git a/homeassistant/components/shelly/__init__.py b/homeassistant/components/shelly/__init__.py index 6cd2a101a25..34a09338c81 100644 --- a/homeassistant/components/shelly/__init__.py +++ b/homeassistant/components/shelly/__init__.py @@ -59,7 +59,7 @@ from .utils import ( BLOCK_PLATFORMS: Final = ["binary_sensor", "cover", "light", "sensor", "switch"] BLOCK_SLEEPING_PLATFORMS: Final = ["binary_sensor", "sensor"] -RPC_PLATFORMS: Final = ["light", "switch"] +RPC_PLATFORMS: Final = ["binary_sensor", "light", "sensor", "switch"] _LOGGER: Final = logging.getLogger(__name__) COAP_SCHEMA: Final = vol.Schema( @@ -410,7 +410,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: entry, RPC_PLATFORMS ) if unload_ok: - hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id][RPC].shutdown() + await hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id][RPC].shutdown() hass.data[DOMAIN][DATA_CONFIG_ENTRY].pop(entry.entry_id) return unload_ok diff --git a/homeassistant/components/shelly/binary_sensor.py b/homeassistant/components/shelly/binary_sensor.py index 02183c3628e..16ffe8b4ee5 100644 --- a/homeassistant/components/shelly/binary_sensor.py +++ b/homeassistant/components/shelly/binary_sensor.py @@ -24,13 +24,20 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback from .entity import ( BlockAttributeDescription, RestAttributeDescription, + RpcAttributeDescription, ShellyBlockAttributeEntity, ShellyRestAttributeEntity, + ShellyRpcAttributeEntity, ShellySleepingBlockAttributeEntity, async_setup_entry_attribute_entities, async_setup_entry_rest, + async_setup_entry_rpc, +) +from .utils import ( + get_device_entry_gen, + is_block_momentary_input, + is_rpc_momentary_input, ) -from .utils import is_momentary_input SENSORS: Final = { ("device", "overtemp"): BlockAttributeDescription( @@ -69,19 +76,19 @@ SENSORS: Final = { name="Input", device_class=DEVICE_CLASS_POWER, default_enabled=False, - removal_condition=is_momentary_input, + removal_condition=is_block_momentary_input, ), ("relay", "input"): BlockAttributeDescription( name="Input", device_class=DEVICE_CLASS_POWER, default_enabled=False, - removal_condition=is_momentary_input, + removal_condition=is_block_momentary_input, ), ("device", "input"): BlockAttributeDescription( name="Input", device_class=DEVICE_CLASS_POWER, default_enabled=False, - removal_condition=is_momentary_input, + removal_condition=is_block_momentary_input, ), ("sensor", "extInput"): BlockAttributeDescription( name="External Input", @@ -112,6 +119,41 @@ REST_SENSORS: Final = { ), } +RPC_SENSORS: Final = { + "input": RpcAttributeDescription( + key="input", + name="Input", + value=lambda status, _: status["state"], + device_class=DEVICE_CLASS_POWER, + default_enabled=False, + removal_condition=is_rpc_momentary_input, + ), + "cloud": RpcAttributeDescription( + key="cloud", + name="Cloud", + value=lambda status, _: status["connected"], + device_class=DEVICE_CLASS_CONNECTIVITY, + default_enabled=False, + ), + "fwupdate": RpcAttributeDescription( + key="sys", + name="Firmware Update", + device_class=DEVICE_CLASS_UPDATE, + value=lambda status, _: status["available_updates"], + default_enabled=False, + extra_state_attributes=lambda status: { + "latest_stable_version": status["available_updates"].get( + "stable", + {"version": ""}, + )["version"], + "beta_version": status["available_updates"].get( + "beta", + {"version": ""}, + )["version"], + }, + ), +} + async def async_setup_entry( hass: HomeAssistant, @@ -119,29 +161,34 @@ async def async_setup_entry( async_add_entities: AddEntitiesCallback, ) -> None: """Set up sensors for device.""" + if get_device_entry_gen(config_entry) == 2: + return await async_setup_entry_rpc( + hass, config_entry, async_add_entities, RPC_SENSORS, RpcBinarySensor + ) + if config_entry.data["sleep_period"]: await async_setup_entry_attribute_entities( hass, config_entry, async_add_entities, SENSORS, - ShellySleepingBinarySensor, + BlockSleepingBinarySensor, ) else: await async_setup_entry_attribute_entities( - hass, config_entry, async_add_entities, SENSORS, ShellyBinarySensor + hass, config_entry, async_add_entities, SENSORS, BlockBinarySensor ) await async_setup_entry_rest( hass, config_entry, async_add_entities, REST_SENSORS, - ShellyRestBinarySensor, + RestBinarySensor, ) -class ShellyBinarySensor(ShellyBlockAttributeEntity, BinarySensorEntity): - """Shelly binary sensor entity.""" +class BlockBinarySensor(ShellyBlockAttributeEntity, BinarySensorEntity): + """Represent a block binary sensor entity.""" @property def is_on(self) -> bool: @@ -149,8 +196,8 @@ class ShellyBinarySensor(ShellyBlockAttributeEntity, BinarySensorEntity): return bool(self.attribute_value) -class ShellyRestBinarySensor(ShellyRestAttributeEntity, BinarySensorEntity): - """Shelly REST binary sensor entity.""" +class RestBinarySensor(ShellyRestAttributeEntity, BinarySensorEntity): + """Represent a REST binary sensor entity.""" @property def is_on(self) -> bool: @@ -158,10 +205,17 @@ class ShellyRestBinarySensor(ShellyRestAttributeEntity, BinarySensorEntity): return bool(self.attribute_value) -class ShellySleepingBinarySensor( - ShellySleepingBlockAttributeEntity, BinarySensorEntity -): - """Represent a shelly sleeping binary sensor.""" +class RpcBinarySensor(ShellyRpcAttributeEntity, BinarySensorEntity): + """Represent a RPC binary sensor entity.""" + + @property + def is_on(self) -> bool: + """Return true if RPC sensor state is on.""" + return bool(self.attribute_value) + + +class BlockSleepingBinarySensor(ShellySleepingBlockAttributeEntity, BinarySensorEntity): + """Represent a block sleeping binary sensor.""" @property def is_on(self) -> bool: diff --git a/homeassistant/components/shelly/const.py b/homeassistant/components/shelly/const.py index 14b56d2c584..917c10ff57c 100644 --- a/homeassistant/components/shelly/const.py +++ b/homeassistant/components/shelly/const.py @@ -113,3 +113,6 @@ KELVIN_MIN_VALUE_WHITE: Final = 2700 KELVIN_MIN_VALUE_COLOR: Final = 3000 UPTIME_DEVIATION: Final = 5 + +# Max RPC switch/input key instances +MAX_RPC_KEY_INSTANCES = 4 diff --git a/homeassistant/components/shelly/entity.py b/homeassistant/components/shelly/entity.py index fd8dfe281ff..13fd3aade3b 100644 --- a/homeassistant/components/shelly/entity.py +++ b/homeassistant/components/shelly/entity.py @@ -24,11 +24,19 @@ from homeassistant.helpers.restore_state import RestoreEntity from homeassistant.helpers.typing import StateType from . import BlockDeviceWrapper, RpcDeviceWrapper, ShellyDeviceRestWrapper -from .const import AIOSHELLY_DEVICE_TIMEOUT_SEC, BLOCK, DATA_CONFIG_ENTRY, DOMAIN, REST +from .const import ( + AIOSHELLY_DEVICE_TIMEOUT_SEC, + BLOCK, + DATA_CONFIG_ENTRY, + DOMAIN, + REST, + RPC, +) from .utils import ( async_remove_shelly_entity, get_block_entity_name, get_rpc_entity_name, + get_rpc_key_instances, ) _LOGGER: Final = logging.getLogger(__name__) @@ -139,6 +147,45 @@ async def async_restore_block_attribute_entities( async_add_entities(entities) +async def async_setup_entry_rpc( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, + sensors: dict[str, RpcAttributeDescription], + sensor_class: Callable, +) -> None: + """Set up entities for REST sensors.""" + wrapper: RpcDeviceWrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][ + config_entry.entry_id + ][RPC] + + entities = [] + for sensor_id in sensors: + description = sensors[sensor_id] + key_instances = get_rpc_key_instances(wrapper.device.status, description.key) + + for key in key_instances: + # Filter and remove entities that according to settings should not create an entity + if description.removal_condition and description.removal_condition( + wrapper.device.config, key + ): + domain = sensor_class.__module__.split(".")[-1] + unique_id = f"{wrapper.mac}-{key}-{sensor_id}" + await async_remove_shelly_entity(hass, domain, unique_id) + else: + entities.append((key, sensor_id, description)) + + if not entities: + return + + async_add_entities( + [ + sensor_class(wrapper, key, sensor_id, description) + for key, sensor_id, description in entities + ] + ) + + async def async_setup_entry_rest( hass: HomeAssistant, config_entry: ConfigEntry, @@ -187,6 +234,23 @@ class BlockAttributeDescription: extra_state_attributes: Callable[[Block], dict | None] | None = None +@dataclass +class RpcAttributeDescription: + """Class to describe a RPC sensor.""" + + key: str + name: str + icon: str | None = None + unit: str | None = None + value: Callable[[dict, Any], Any] | None = None + device_class: str | None = None + state_class: str | None = None + default_enabled: bool = True + available: Callable[[dict], bool] | None = None + removal_condition: Callable[[dict, str], bool] | None = None + extra_state_attributes: Callable[[dict], dict | None] | None = None + + @dataclass class RestAttributeDescription: """Class to describe a REST sensor.""" @@ -472,6 +536,58 @@ class ShellyRestAttributeEntity(update_coordinator.CoordinatorEntity): return self.description.extra_state_attributes(self.wrapper.device.status) +class ShellyRpcAttributeEntity(ShellyRpcEntity, entity.Entity): + """Helper class to represent a rpc attribute.""" + + def __init__( + self, + wrapper: RpcDeviceWrapper, + key: str, + attribute: str, + description: RpcAttributeDescription, + ) -> None: + """Initialize sensor.""" + super().__init__(wrapper, key) + self.attribute = attribute + self.description = description + + self._attr_unique_id = f"{super().unique_id}-{attribute}" + self._attr_name = get_rpc_entity_name(wrapper.device, key, description.name) + self._attr_entity_registry_enabled_default = description.default_enabled + self._attr_device_class = description.device_class + self._attr_icon = description.icon + self._last_value = None + + @property + def attribute_value(self) -> StateType: + """Value of sensor.""" + if callable(self.description.value): + self._last_value = self.description.value( + self.wrapper.device.status[self.key], self._last_value + ) + return self._last_value + + @property + def available(self) -> bool: + """Available.""" + available = super().available + + if not available or not self.description.available: + return available + + return self.description.available(self.wrapper.device.status[self.key]) + + @property + def extra_state_attributes(self) -> dict[str, Any] | None: + """Return the state attributes.""" + if self.description.extra_state_attributes is None: + return None + + return self.description.extra_state_attributes( + self.wrapper.device.status[self.key] + ) + + class ShellySleepingBlockAttributeEntity(ShellyBlockAttributeEntity, RestoreEntity): """Represent a shelly sleeping block attribute entity.""" diff --git a/homeassistant/components/shelly/light.py b/homeassistant/components/shelly/light.py index bb636013361..6a1035816a5 100644 --- a/homeassistant/components/shelly/light.py +++ b/homeassistant/components/shelly/light.py @@ -51,7 +51,7 @@ from .const import ( STANDARD_RGB_EFFECTS, ) from .entity import ShellyBlockEntity, ShellyRpcEntity -from .utils import async_remove_shelly_entity, get_device_entry_gen +from .utils import async_remove_shelly_entity, get_device_entry_gen, get_rpc_key_ids _LOGGER: Final = logging.getLogger(__name__) @@ -106,25 +106,22 @@ async def async_setup_rpc_entry( ) -> None: """Set up entities for RPC device.""" wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][RPC] + switch_key_ids = get_rpc_key_ids(wrapper.device.status, "switch") - switch_keys = [] - for i in range(4): - key = f"switch:{i}" - if not wrapper.device.status.get(key): - continue - + switch_ids = [] + for id_ in switch_key_ids: con_types = wrapper.device.config["sys"]["ui_data"].get("consumption_types") - if con_types is None or con_types[i] != "lights": + if con_types is None or con_types[id_] != "lights": continue - switch_keys.append((key, i)) - unique_id = f"{wrapper.mac}-{key}" + switch_ids.append(id_) + unique_id = f"{wrapper.mac}-switch:{id_}" await async_remove_shelly_entity(hass, "switch", unique_id) - if not switch_keys: + if not switch_ids: return - async_add_entities(RpcShellyLight(wrapper, key, id_) for key, id_ in switch_keys) + async_add_entities(RpcShellyLight(wrapper, id_) for id_ in switch_ids) class BlockShellyLight(ShellyBlockEntity, LightEntity): @@ -417,9 +414,9 @@ class BlockShellyLight(ShellyBlockEntity, LightEntity): class RpcShellyLight(ShellyRpcEntity, LightEntity): """Entity that controls a light on RPC based Shelly devices.""" - def __init__(self, wrapper: RpcDeviceWrapper, key: str, id_: int) -> None: + def __init__(self, wrapper: RpcDeviceWrapper, id_: int) -> None: """Initialize light.""" - super().__init__(wrapper, key) + super().__init__(wrapper, f"switch:{id_}") self._id = id_ @property diff --git a/homeassistant/components/shelly/sensor.py b/homeassistant/components/shelly/sensor.py index 7ffaae82daa..8a1ac340d31 100644 --- a/homeassistant/components/shelly/sensor.py +++ b/homeassistant/components/shelly/sensor.py @@ -16,6 +16,7 @@ from homeassistant.const import ( PERCENTAGE, POWER_WATT, SIGNAL_STRENGTH_DECIBELS_MILLIWATT, + TEMP_CELSIUS, ) from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback @@ -25,13 +26,16 @@ from .const import SHAIR_MAX_WORK_HOURS from .entity import ( BlockAttributeDescription, RestAttributeDescription, + RpcAttributeDescription, ShellyBlockAttributeEntity, ShellyRestAttributeEntity, + ShellyRpcAttributeEntity, ShellySleepingBlockAttributeEntity, async_setup_entry_attribute_entities, async_setup_entry_rest, + async_setup_entry_rpc, ) -from .utils import get_device_uptime, temperature_unit +from .utils import get_device_entry_gen, get_device_uptime, temperature_unit SENSORS: Final = { ("device", "battery"): BlockAttributeDescription( @@ -220,7 +224,60 @@ REST_SENSORS: Final = { ), "uptime": RestAttributeDescription( name="Uptime", - value=get_device_uptime, + value=lambda status, last: get_device_uptime(status["uptime"], last), + device_class=sensor.DEVICE_CLASS_TIMESTAMP, + default_enabled=False, + ), +} + + +RPC_SENSORS: Final = { + "power": RpcAttributeDescription( + key="switch", + name="Power", + unit=POWER_WATT, + value=lambda status, _: round(float(status["apower"]), 1), + device_class=sensor.DEVICE_CLASS_POWER, + state_class=sensor.STATE_CLASS_MEASUREMENT, + ), + "voltage": RpcAttributeDescription( + key="switch", + name="Voltage", + unit=ELECTRIC_POTENTIAL_VOLT, + value=lambda status, _: round(float(status["voltage"]), 1), + device_class=sensor.DEVICE_CLASS_VOLTAGE, + state_class=sensor.STATE_CLASS_MEASUREMENT, + ), + "energy": RpcAttributeDescription( + key="switch", + name="Energy", + unit=ENERGY_KILO_WATT_HOUR, + value=lambda status, _: round(status["aenergy"]["total"] / 1000, 2), + device_class=sensor.DEVICE_CLASS_ENERGY, + state_class=sensor.STATE_CLASS_TOTAL_INCREASING, + ), + "temperature": RpcAttributeDescription( + key="switch", + name="Temperature", + unit=TEMP_CELSIUS, + value=lambda status, _: round(status["temperature"]["tC"], 1), + device_class=sensor.DEVICE_CLASS_TEMPERATURE, + state_class=sensor.STATE_CLASS_MEASUREMENT, + default_enabled=False, + ), + "rssi": RpcAttributeDescription( + key="wifi", + name="RSSI", + unit=SIGNAL_STRENGTH_DECIBELS_MILLIWATT, + value=lambda status, _: status["rssi"], + device_class=sensor.DEVICE_CLASS_SIGNAL_STRENGTH, + state_class=sensor.STATE_CLASS_MEASUREMENT, + default_enabled=False, + ), + "uptime": RpcAttributeDescription( + key="sys", + name="Uptime", + value=lambda status, last: get_device_uptime(status["uptime"], last), device_class=sensor.DEVICE_CLASS_TIMESTAMP, default_enabled=False, ), @@ -233,21 +290,26 @@ async def async_setup_entry( async_add_entities: AddEntitiesCallback, ) -> None: """Set up sensors for device.""" + if get_device_entry_gen(config_entry) == 2: + return await async_setup_entry_rpc( + hass, config_entry, async_add_entities, RPC_SENSORS, RpcSensor + ) + if config_entry.data["sleep_period"]: await async_setup_entry_attribute_entities( - hass, config_entry, async_add_entities, SENSORS, ShellySleepingSensor + hass, config_entry, async_add_entities, SENSORS, BlockSleepingSensor ) else: await async_setup_entry_attribute_entities( - hass, config_entry, async_add_entities, SENSORS, ShellySensor + hass, config_entry, async_add_entities, SENSORS, BlockSensor ) await async_setup_entry_rest( - hass, config_entry, async_add_entities, REST_SENSORS, ShellyRestSensor + hass, config_entry, async_add_entities, REST_SENSORS, RestSensor ) -class ShellySensor(ShellyBlockAttributeEntity, SensorEntity): - """Represent a shelly sensor.""" +class BlockSensor(ShellyBlockAttributeEntity, SensorEntity): + """Represent a block sensor.""" @property def native_value(self) -> StateType: @@ -265,8 +327,8 @@ class ShellySensor(ShellyBlockAttributeEntity, SensorEntity): return cast(str, self._unit) -class ShellyRestSensor(ShellyRestAttributeEntity, SensorEntity): - """Represent a shelly REST sensor.""" +class RestSensor(ShellyRestAttributeEntity, SensorEntity): + """Represent a REST sensor.""" @property def native_value(self) -> StateType: @@ -284,8 +346,27 @@ class ShellyRestSensor(ShellyRestAttributeEntity, SensorEntity): return self.description.unit -class ShellySleepingSensor(ShellySleepingBlockAttributeEntity, SensorEntity): - """Represent a shelly sleeping sensor.""" +class RpcSensor(ShellyRpcAttributeEntity, SensorEntity): + """Represent a RPC sensor.""" + + @property + def native_value(self) -> StateType: + """Return value of sensor.""" + return self.attribute_value + + @property + def state_class(self) -> str | None: + """State class of sensor.""" + return self.description.state_class + + @property + def native_unit_of_measurement(self) -> str | None: + """Return unit of sensor.""" + return self.description.unit + + +class BlockSleepingSensor(ShellySleepingBlockAttributeEntity, SensorEntity): + """Represent a block sleeping sensor.""" @property def native_value(self) -> StateType: diff --git a/homeassistant/components/shelly/switch.py b/homeassistant/components/shelly/switch.py index 97f44d9c40e..d6e8fa11798 100644 --- a/homeassistant/components/shelly/switch.py +++ b/homeassistant/components/shelly/switch.py @@ -13,7 +13,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback from . import BlockDeviceWrapper, RpcDeviceWrapper from .const import BLOCK, DATA_CONFIG_ENTRY, DOMAIN, RPC from .entity import ShellyBlockEntity, ShellyRpcEntity -from .utils import async_remove_shelly_entity, get_device_entry_gen +from .utils import async_remove_shelly_entity, get_device_entry_gen, get_rpc_key_ids async def async_setup_entry( @@ -72,25 +72,22 @@ async def async_setup_rpc_entry( ) -> None: """Set up entities for RPC device.""" wrapper = hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry.entry_id][RPC] + switch_key_ids = get_rpc_key_ids(wrapper.device.status, "switch") - switch_keys = [] - for i in range(4): - key = f"switch:{i}" - if not wrapper.device.status.get(key): - continue - + switch_ids = [] + for id_ in switch_key_ids: con_types = wrapper.device.config["sys"]["ui_data"].get("consumption_types") - if con_types is not None and con_types[i] == "lights": + if con_types is not None and con_types[id_] == "lights": continue - switch_keys.append((key, i)) - unique_id = f"{wrapper.mac}-{key}" + switch_ids.append(id_) + unique_id = f"{wrapper.mac}-switch:{id_}" await async_remove_shelly_entity(hass, "light", unique_id) - if not switch_keys: + if not switch_ids: return - async_add_entities(RpcRelaySwitch(wrapper, key, id_) for key, id_ in switch_keys) + async_add_entities(RpcRelaySwitch(wrapper, id_) for id_ in switch_ids) class BlockRelaySwitch(ShellyBlockEntity, SwitchEntity): @@ -129,9 +126,9 @@ class BlockRelaySwitch(ShellyBlockEntity, SwitchEntity): class RpcRelaySwitch(ShellyRpcEntity, SwitchEntity): """Entity that controls a relay on RPC based Shelly devices.""" - def __init__(self, wrapper: RpcDeviceWrapper, key: str, id_: int) -> None: + def __init__(self, wrapper: RpcDeviceWrapper, id_: int) -> None: """Initialize relay switch.""" - super().__init__(wrapper, key) + super().__init__(wrapper, f"switch:{id_}") self._id = id_ @property diff --git a/homeassistant/components/shelly/utils.py b/homeassistant/components/shelly/utils.py index 10046ccd4b0..13b34ef5aea 100644 --- a/homeassistant/components/shelly/utils.py +++ b/homeassistant/components/shelly/utils.py @@ -21,6 +21,7 @@ from .const import ( CONF_COAP_PORT, DEFAULT_COAP_PORT, DOMAIN, + MAX_RPC_KEY_INSTANCES, SHBTN_INPUTS_EVENTS_TYPES, SHBTN_MODELS, SHIX3_1_INPUTS_EVENTS_TYPES, @@ -88,7 +89,7 @@ def get_block_entity_name( description: str | None = None, ) -> str: """Naming for block based switch and sensors.""" - channel_name = get_device_channel_name(device, block) + channel_name = get_block_channel_name(device, block) if description: return f"{channel_name} {description}" @@ -96,7 +97,7 @@ def get_block_entity_name( return channel_name -def get_device_channel_name(device: BlockDevice, block: Block | None) -> str: +def get_block_channel_name(device: BlockDevice, block: Block | None) -> str: """Get name based on device and channel name.""" entity_name = get_block_device_name(device) @@ -125,8 +126,8 @@ def get_device_channel_name(device: BlockDevice, block: Block | None) -> str: return f"{entity_name} channel {chr(int(block.channel)+base)}" -def is_momentary_input(settings: dict[str, Any], block: Block) -> bool: - """Return true if input button settings is set to a momentary type.""" +def is_block_momentary_input(settings: dict[str, Any], block: Block) -> bool: + """Return true if block input button settings is set to a momentary type.""" # Shelly Button type is fixed to momentary and no btn_type if settings["device"]["type"] in SHBTN_MODELS: return True @@ -147,9 +148,9 @@ def is_momentary_input(settings: dict[str, Any], block: Block) -> bool: return button_type in ["momentary", "momentary_on_release"] -def get_device_uptime(status: dict[str, Any], last_uptime: str | None) -> str: +def get_device_uptime(uptime: float, last_uptime: str | None) -> str: """Return device uptime string, tolerate up to 5 seconds deviation.""" - delta_uptime = utcnow() - timedelta(seconds=status["uptime"]) + delta_uptime = utcnow() - timedelta(seconds=uptime) if ( not last_uptime @@ -166,7 +167,7 @@ def get_input_triggers(device: BlockDevice, block: Block) -> list[tuple[str, str if "inputEvent" not in block.sensor_ids or "inputEventCnt" not in block.sensor_ids: return [] - if not is_momentary_input(device.settings, block): + if not is_block_momentary_input(device.settings, block): return [] triggers = [] @@ -240,21 +241,64 @@ def get_model_name(info: dict[str, Any]) -> str: return cast(str, MODEL_NAMES.get(info["type"], info["type"])) +def get_rpc_channel_name(device: RpcDevice, key: str) -> str: + """Get name based on device and channel name.""" + key = key.replace("input", "switch") + device_name = get_rpc_device_name(device) + entity_name: str | None = device.config[key].get("name", device_name) + + if entity_name is None: + return f"{device_name} {key.replace(':', '_')}" + + return entity_name + + def get_rpc_entity_name( device: RpcDevice, key: str, description: str | None = None ) -> str: """Naming for RPC based switch and sensors.""" - entity_name: str | None = device.config[key].get("name") - - if entity_name is None: - entity_name = f"{get_rpc_device_name(device)} {key.replace(':', '_')}" + channel_name = get_rpc_channel_name(device, key) if description: - return f"{entity_name} {description}" + return f"{channel_name} {description}" - return entity_name + return channel_name def get_device_entry_gen(entry: ConfigEntry) -> int: """Return the device generation from config entry.""" return entry.data.get("gen", 1) + + +def get_rpc_key_instances(keys_dict: dict[str, Any], key: str) -> list[str]: + """Return list of key instances for RPC device from a dict.""" + if key in keys_dict: + return [key] + + keys_list: list[str] = [] + for i in range(MAX_RPC_KEY_INSTANCES): + key_inst = f"{key}:{i}" + if key_inst not in keys_dict: + return keys_list + + keys_list.append(key_inst) + + return keys_list + + +def get_rpc_key_ids(keys_dict: dict[str, Any], key: str) -> list[int]: + """Return list of key ids for RPC device from a dict.""" + key_ids: list[int] = [] + for i in range(MAX_RPC_KEY_INSTANCES): + key_inst = f"{key}:{i}" + if key_inst not in keys_dict: + return key_ids + + key_ids.append(i) + + return key_ids + + +def is_rpc_momentary_input(config: dict[str, Any], key: str) -> bool: + """Return true if rpc input button settings is set to a momentary type.""" + return cast(bool, config[key]["type"] == "button")