First iteration on improved battery support

custom_components/enphase_envoy/binary_sensor.py

@@ -95,6 +95,24 @@ async def async_setup_entry(
                         )
                     )
 
+        elif sensor_description.key.startswith("batteries_"):
+            if coordinator.data.get("batteries") is not None:
+                for battery in coordinator.data["batteries"].keys():
+                    device_name = f"Battery {battery}"
+                    entity_name = f"{device_name} {sensor_description.name}"
+                    serial_number = battery
+                    entities.append(
+                        EnvoyBatteryEntity(
+                            sensor_description,
+                            entity_name,
+                            device_name,
+                            serial_number,
+                            None,
+                            coordinator,
+                            config_entry.unique_id,
+                        )
+                    )
+
         elif sensor_description.key == "firmware":
             if coordinator.data.get("envoy_info", {}).get("update_status") is not None:
                 entity_name = f"{name} {sensor_description.name}"
@@ -420,3 +438,91 @@ def is_on(self) -> bool | None:
     @property
     def extra_state_attributes(self) -> dict | None:
         return None
+
+
+class EnvoyBatteryEntity(CoordinatorEntity, BinarySensorEntity):
+    """Envoy battery entity."""
+
+    def __init__(
+        self,
+        description,
+        name,
+        device_name,
+        device_serial_number,
+        serial_number,
+        coordinator,
+        parent_device,
+    ):
+        self.entity_description = description
+        self._name = name
+        self._serial_number = serial_number
+        self._device_name = device_name
+        self._device_serial_number = device_serial_number
+        self._parent_device = parent_device
+        CoordinatorEntity.__init__(self, coordinator)
+
+    @property
+    def name(self):
+        """Return the name of the sensor."""
+        return self._name
+
+    @property
+    def unique_id(self):
+        """Return the unique id of the sensor."""
+        if self._serial_number:
+            return self._serial_number
+        if self._device_serial_number:
+            return f"{self._device_serial_number}_{self.entity_description.key}"
+
+    @property
+    def is_on(self) -> bool:
+        """Return the status of the requested attribute."""
+        if self.coordinator.data.get("batteries") is not None:
+            return (
+                self.coordinator.data.get("batteries")
+                .get(self._device_serial_number)
+                .get(self.entity_description.key[10:])
+            )
+
+    @property
+    def extra_state_attributes(self):
+        """Return the state attributes."""
+        if self.coordinator.data.get("batteries") is not None:
+            battery = self.coordinator.data.get("batteries").get(
+                self._device_serial_number
+            )
+            return {"last_reported": battery.get("report_date")}
+
+        return None
+
+    @property
+    def device_info(self) -> DeviceInfo | None:
+        """Return the device_info of the device."""
+        if not self._device_serial_number:
+            return None
+
+        sw_version = None
+        hw_version = None
+        if self.coordinator.data.get("batteries") and self.coordinator.data.get(
+            "batteries"
+        ).get(self._device_serial_number):
+            sw_version = (
+                self.coordinator.data.get("batteries")
+                .get(self._device_serial_number)
+                .get("img_pnum_running")
+            )
+            hw_version = (
+                self.coordinator.data.get("batteries")
+                .get(self._device_serial_number)
+                .get("part_num")
+            )
+
+        return DeviceInfo(
+            identifiers={(DOMAIN, str(self._device_serial_number))},
+            manufacturer="Enphase",
+            model=get_model_name("Battery", hw_version),
+            name=self._device_name,
+            via_device=(DOMAIN, self._parent_device),
+            sw_version=sw_version,
+            hw_version=resolve_hardware_id(hw_version),
+        )

custom_components/enphase_envoy/const.py

@@ -50,7 +50,7 @@
     "800-00656-r06": {"name": "Envoy-S-Standard-EU", "sku": "ENV-S-WB-230"},
     "800-00656-r07": {"name": "Envoy-S-Standard-EU", "sku": "ENV-S-WB-230"},
     "800-01359-r02": {"name": "IQ8+ Microinverter", "sku": "IQ8PLUS-72-M-INT"},
-    "800-01391-r03 ": {"name": "IQ8HC Microinverter", "sku": "IQ8HC-72-M-INT"},
+    "800-01391-r03": {"name": "IQ8HC Microinverter", "sku": "IQ8HC-72-M-INT"},
     "800-01736-r02": {"name": "IQ7+ Microinverter", "sku": "IQ7PLUS-72-M-INT"},
     "800-00631-r02": {"name": "IQ7+ Microinverter", "sku": "IQ7PLUS-72-2-INT"},
     "800-01127-r02": {"name": "IQ7A Microinverter", "sku": "IQ7A-72-M-INT"},
@@ -174,24 +174,52 @@ def get_model_name(model, hardware_id):
         device_class=SensorDeviceClass.TEMPERATURE,
     ),
     SensorEntityDescription(
-        key="batteries",
-        name="Battery",
+        key="batteries_percentFull",
+        name="Percentage Full",
         native_unit_of_measurement=PERCENTAGE,
         state_class=SensorStateClass.MEASUREMENT,
         device_class=SensorDeviceClass.BATTERY,
     ),
     SensorEntityDescription(
-        key="total_battery_percentage",
-        name="Total Battery Percentage",
+        key="batteries_temperature",
+        name="Temperature",
+        native_unit_of_measurement=UnitOfTemperature.CELSIUS,
+        state_class=SensorStateClass.MEASUREMENT,
+        device_class=SensorDeviceClass.TEMPERATURE,
+    ),
+    SensorEntityDescription(
+        key="batteries_encharge_capacity",
+        name="Capacity",
+        native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
+        state_class=SensorStateClass.TOTAL,
+        device_class=SensorDeviceClass.ENERGY,
+    ),
+    SensorEntityDescription(
+        key="batteries_encharge_capacity_current",
+        name="Current Capacity",
+        native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
+        state_class=SensorStateClass.TOTAL,
+        device_class=SensorDeviceClass.ENERGY,
+    ),
+    SensorEntityDescription(
+        key="avg_batteries_percentFull",
+        name="Batteries Percentage Full",
         native_unit_of_measurement=PERCENTAGE,
         state_class=SensorStateClass.MEASUREMENT,
         device_class=SensorDeviceClass.BATTERY,
     ),
     SensorEntityDescription(
-        key="current_battery_capacity",
-        name="Current Battery Capacity",
+        key="total_batteries_encharge_capacity",
+        name="Batteries Capacity",
         native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
-        state_class=SensorStateClass.MEASUREMENT,
+        state_class=SensorStateClass.TOTAL,
+        device_class=SensorDeviceClass.ENERGY,
+    ),
+    SensorEntityDescription(
+        key="total_batteries_encharge_capacity_current",
+        name="Batteries Current Capacity",
+        native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
+        state_class=SensorStateClass.TOTAL,
         device_class=SensorDeviceClass.ENERGY,
     ),
     SensorEntityDescription(
@@ -380,22 +408,18 @@ def get_model_name(model, hardware_id):
         name="Firmware",
         device_class=BinarySensorDeviceClass.UPDATE,
     ),
-)
-
-BATTERY_ENERGY_DISCHARGED_SENSOR = SensorEntityDescription(
-    key="battery_energy_discharged",
-    name="Battery Energy Discharged",
-    native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
-    state_class=SensorStateClass.TOTAL,
-    device_class=SensorDeviceClass.ENERGY,
-)
-
-BATTERY_ENERGY_CHARGED_SENSOR = SensorEntityDescription(
-    key="battery_energy_charged",
-    name="Battery Energy Charged",
-    native_unit_of_measurement=UnitOfEnergy.WATT_HOUR,
-    state_class=SensorStateClass.TOTAL,
-    device_class=SensorDeviceClass.ENERGY,
+    BinarySensorEntityDescription(
+        key="batteries_operating",
+        name="Operating",
+        device_class=BinarySensorDeviceClass.POWER,
+        entity_category=EntityCategory.DIAGNOSTIC,
+    ),
+    BinarySensorEntityDescription(
+        key="batteries_communicating",
+        name="Communicating",
+        device_class=BinarySensorDeviceClass.CONNECTIVITY,
+        entity_category=EntityCategory.DIAGNOSTIC,
+    ),
 )
 
 PRODUCION_POWER_SWITCH = SwitchEntityDescription(

custom_components/enphase_envoy/envoy_reader.py

@@ -149,6 +149,23 @@ class EnvoyError(EnvoyReaderError):
     pass
 
 
+class TestData:
+    def __init__(self, file):
+        with open(url) as json_file:
+            self.json_data = json.load(json_file)
+
+    @property
+    def status_code(self):
+        return 200
+
+    @property
+    def headers(self):
+        return {"content-type": "application/json"}
+
+    def json(self):
+        return self.json_data
+
+
 class StreamData:
     class PhaseData:
         def __init__(self, phase_data):
@@ -530,6 +547,15 @@ def batteries(self):
         if isinstance(battery_data, list) and len(battery_data) > 0:
             battery_dict = {}
             for item in battery_data:
+                if "last_rpt_date" in item:
+                    item["report_date"] = time.strftime(
+                        "%Y-%m-%d %H:%M:%S", time.localtime(item["last_rpt_date"])
+                    )
+                if "encharge_capacity" in item and "percentFull" in item:
+                    item["encharge_capacity_current"] = item["encharge_capacity"] * (
+                        item["percentFull"] / 100
+                    )
+
                 battery_dict[item["serial_num"]] = item
 
             return battery_dict
@@ -750,12 +776,16 @@ def async_client(self):
 
     async def _update_endpoint(self, attr, url, only_on_success=False):
         """Update a property from an endpoint."""
-        formatted_url = url.format(self.host)
-        response = await self._async_fetch_with_retry(
-            formatted_url, follow_redirects=False
-        )
-        if not only_on_success or response.status_code == 200:
-            setattr(self, attr, response)
+        if url.startswith("https://"):
+            formatted_url = url.format(self.host)
+            response = await self._async_fetch_with_retry(
+                formatted_url, follow_redirects=False
+            )
+            if not only_on_success or response.status_code == 200:
+                setattr(self, attr, response)
+        else:
+            data = TestData(url)
+            setattr(self, attr, data)
 
     async def _async_fetch_with_retry(self, url, **kwargs):
         """Retry 3 times to fetch the url if there is a transport error."""