Add the Sofar active power limit binary sensor (#181198)

homeassistant/components/sofar/binary_sensor.py

@@ -1,8 +1,10 @@
 """Support for Sofar binary sensors."""
 
 from dataclasses import dataclass
+from enum import IntFlag
 from typing import override
 
+from sofar_modbus.modern.enums import PowerControlFlags
 from sofar_modbus.modern.faults import FaultCategory
 
 from homeassistant.components.binary_sensor import (
@@ -52,6 +54,29 @@ FAULT_SENSOR_DESCRIPTIONS: tuple[SofarFaultBinarySensorDescription, ...] = tuple
 )
 
 
+@dataclass(frozen=True, kw_only=True)
+class SofarFlagBinarySensorDescription(
+    SofarEntityDescription, BinarySensorEntityDescription
+):
+    """Describe a Sofar binary sensor backed by one flags-register bit."""
+
+    attribute: str
+    flag: IntFlag
+
+
+FLAG_SENSOR_DESCRIPTIONS: tuple[SofarFlagBinarySensorDescription, ...] = (
+    SofarFlagBinarySensorDescription(
+        key="active_power_limit_enabled",
+        component="active_power_control",
+        translation_key="active_power_limit_enabled",
+        entity_category=EntityCategory.DIAGNOSTIC,
+        entity_registry_enabled_default=False,
+        attribute="power_control",
+        flag=PowerControlFlags.ACTIVE_POWER,
+    ),
+)
+
+
 async def async_setup_entry(
     hass: HomeAssistant,
     entry: SofarConfigEntry,
@@ -65,6 +90,11 @@ async def async_setup_entry(
         for description in FAULT_SENSOR_DESCRIPTIONS
         if description.component in served
     )
+    async_add_entities(
+        SofarFlagBinarySensor(runtime_data, description)
+        for description in FLAG_SENSOR_DESCRIPTIONS
+        if description.component in served
+    )
 
 
 class SofarFaultBinarySensor(SofarEntity, BinarySensorEntity):
@@ -80,3 +110,16 @@ class SofarFaultBinarySensor(SofarEntity, BinarySensorEntity):
             fault.category is self.entity_description.category
             for fault in component.active_faults
         )
+
+
+class SofarFlagBinarySensor(SofarEntity, BinarySensorEntity):
+    """Reports whether one bit of a flags register is set."""
+
+    entity_description: SofarFlagBinarySensorDescription
+
+    @property
+    @override
+    def is_on(self) -> bool:
+        component = getattr(self.coordinator.device, self.entity_description.component)
+        flags = getattr(component, self.entity_description.attribute)
+        return self.entity_description.flag in flags

homeassistant/components/sofar/strings.json

@@ -47,6 +47,9 @@
   },
   "entity": {
     "binary_sensor": {
+      "active_power_limit_enabled": {
+        "name": "Active power limit enabled"
+      },
       "fault_ac_output": {
         "name": "AC output fault"
       },

tests/components/sofar/snapshots/test_binary_sensor.ambr

@@ -50,6 +50,56 @@
     'state': 'off',
   })
 # ---
+# name: test_all_entities[binary_sensor.4_4_ktlx_g3_active_power_limit_enabled-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': list([
+      None,
+    ]),
+    'area_id': None,
+    'capabilities': None,
+    'config_entry_id': <ANY>,
+    'config_subentry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'binary_sensor',
+    'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
+    'entity_id': 'binary_sensor.4_4_ktlx_g3_active_power_limit_enabled',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'object_id_base': 'Active power limit enabled',
+    'options': dict({
+    }),
+    'original_device_class': None,
+    'original_icon': None,
+    'original_name': 'Active power limit enabled',
+    'platform': 'sofar',
+    'previous_unique_id': None,
+    'suggested_object_id': None,
+    'supported_features': 0,
+    'translation_key': 'active_power_limit_enabled',
+    'unique_id': 'SS2ES104N5S445_active_power_limit_enabled',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_all_entities[binary_sensor.4_4_ktlx_g3_active_power_limit_enabled-state]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      <EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: '4.4 KTLX-G3 Active power limit enabled',
+    }),
+    'context': <ANY>,
+    'entity_id': 'binary_sensor.4_4_ktlx_g3_active_power_limit_enabled',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'off',
+  })
+# ---
 # name: test_all_entities[binary_sensor.4_4_ktlx_g3_arc_fault_afci-entry]
   EntityRegistryEntrySnapshot({
     'aliases': list([

tests/components/sofar/test_binary_sensor.py

@@ -108,8 +108,9 @@ async def test_enabled_by_default_excludes_commercial_hardware(
             entity_registry, init_integration.entry_id
         )
         if e.domain == BINARY_SENSOR_DOMAIN
+        and e.unique_id.startswith(f"{MOCK_SERIAL}_fault_")
     ]
-    assert len(entries) == len(FAULT_SENSOR_DESCRIPTIONS)
+    assert len(entries) == 17
 
     disabled_categories = {
         FaultCategory(e.unique_id.removeprefix(f"{MOCK_SERIAL}_fault_"))
@@ -122,3 +123,43 @@ async def test_enabled_by_default_excludes_commercial_hardware(
         FaultCategory.INPUT_FUSE,
         FaultCategory.STRING_FUSE,
     }
+
+
+@pytest.mark.parametrize(
+    ("armed", "expected"),
+    [
+        pytest.param(0b0, STATE_OFF, id="not_armed"),
+        pytest.param(0b1, STATE_ON, id="armed"),
+        pytest.param(0b10, STATE_OFF, id="unrelated_flag_only"),
+    ],
+)
+@pytest.mark.usefixtures("entity_registry_enabled_by_default")
+async def test_active_power_limit_enabled_follows_its_bit(
+    hass: HomeAssistant,
+    entity_registry: er.EntityRegistry,
+    mock_connection: MockModbusConnection,
+    mock_config_entry: MockConfigEntry,
+    armed: int,
+    expected: str,
+) -> None:
+    """Test the arm bit of 1105 drives the state, not the limit itself."""
+    unit = mock_connection.for_unit(1)
+    unit.holding[0x1105] = armed
+    unit.holding[0x1106] = 800  # 80% limit, ignored while the bit is unset
+
+    with patch(
+        "homeassistant.components.sofar.async_get_unit",
+        side_effect=lambda hass, entry, params, unit_id: mock_connection.for_unit(
+            unit_id
+        ),
+    ):
+        mock_config_entry.add_to_hass(hass)
+        await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done(wait_background_tasks=True)
+
+    entity_id = entity_registry.async_get_entity_id(
+        BINARY_SENSOR_DOMAIN, DOMAIN, f"{MOCK_SERIAL}_active_power_limit_enabled"
+    )
+    assert entity_id is not None
+    assert (state := hass.states.get(entity_id)) is not None
+    assert state.state == expected