Identify EVSE devices by model

custom_components/solaredge_modbus_multi/const.py

@@ -35,6 +35,11 @@
     re.IGNORECASE,
 )
 
+DETECT_EVSE_REGEX = re.compile(
+    r"^(?:SE-EV-SA)",  # Add additional prefixes with |OTHER-PREFIX
+    re.IGNORECASE,
+)
+
 STATUS_VENDOR4_VERSION = "3.20.0"  # solaredge firmware version
 INVERTED_POWER_VERSION = "2026.2.0"  # home assistant core version
 

custom_components/solaredge_modbus_multi/diagnostics.py

@@ -15,6 +15,7 @@
 REDACT_INVERTER = {"identifiers", "C_SerialNumber", "serial_number"}
 REDACT_METER = {"identifiers", "C_SerialNumber", "serial_number", "via_device"}
 REDACT_BATTERY = {"identifiers", "B_SerialNumber", "serial_number", "via_device"}
+REDACT_EVSE = {"identifiers", "C_SerialNumber", "serial_number"}
 
 
 def format_values(format_input) -> Any:
@@ -88,4 +89,14 @@ async def async_get_config_entry_diagnostics(
         }
         data.update(async_redact_data(battery, REDACT_BATTERY))
 
+    for evse in hub.evses:
+        evse: dict[str, Any] = {
+            f"evse_unit_id_{evse.evse_unit_id}": {
+                "device_info": evse.device_info,
+                "common": evse.decoded_common,
+                "model": format_values(evse.decoded_model),
+            }
+        }
+        data.update(async_redact_data(evse, REDACT_EVSE))
+
     return data

custom_components/solaredge_modbus_multi/hub.py

@@ -30,6 +30,7 @@
 
 from .const import (
     BATTERY_REG_BASE,
+    DETECT_EVSE_REGEX,
     DOMAIN,
     METER_REG_BASE,
     PYMODBUS_REQUIRED_VERSION,
@@ -68,6 +69,12 @@ class DeviceInitFailed(SolarEdgeException):
     pass
 
 
+class DeviceIsEVSE(SolarEdgeException):
+    """Raised when an inverter device matches a EVSE model"""
+
+    pass
+
+
 class ModbusReadError(SolarEdgeException):
     """Raised when a modbus read fails (generic)"""
 
@@ -186,6 +193,7 @@ def __init__(
         self.inverters = []
         self.meters = []
         self.batteries = []
+        self.evses = []
         self.inverter_common = {}
         self.mmppt_common = {}
         self.has_write = None
@@ -283,6 +291,17 @@ async def _async_init_solaredge(self) -> None:
                 _LOGGER.error(f"Inverter at {self.hub_host} ID {inverter_unit_id}: {e}")
                 raise HubInitFailed(f"{e}")
 
+            except DeviceIsEVSE as e:
+                _LOGGER.debug(
+                    f"Device model matches EVSE at {self.hub_host} ID {inverter_unit_id}: {e}"
+                )
+                new_evse = SolarEdgeEVSE(inverter_unit_id, self)
+                await new_evse.init_device()
+                self.evses.append(new_evse)
+
+                # Skip meter and battery detection if DeviceIsEVSE
+                continue
+
             if self._detect_meters:
                 for meter_id in METER_REG_BASE:
                     try:
@@ -357,6 +376,8 @@ async def _async_init_solaredge(self) -> None:
                 await meter.read_modbus_data()
             for battery in self.batteries:
                 await battery.read_modbus_data()
+            for evse in self.evses:
+                await evse.read_modbus_data()
 
             timestamp = dt.now()
             for inverter in self.inverters:
@@ -447,6 +468,8 @@ async def async_refresh_modbus_data(self) -> bool:
                     await meter.read_modbus_data()
                 for battery in self.batteries:
                     await battery.read_modbus_data()
+                for evse in self.evses:
+                    await evse.read_modbus_data()
 
         except ModbusReadError as e:
             self.disconnect()
@@ -551,7 +574,7 @@ async def modbus_read_holding_registers(self, unit, address, rcount):
         sig = inspect.signature(self._client.read_holding_registers)
 
         _LOGGER.debug(
-            f"I{self._rr_unit}: modbus_read_holding_registers "
+            f"unit={self._rr_unit}: modbus_read_holding_registers "
             f"address={self._rr_address} count={self._rr_count}"
         )
 
@@ -564,38 +587,40 @@ async def modbus_read_holding_registers(self, unit, address, rcount):
                 address=self._rr_address, count=self._rr_count, slave=self._rr_unit
             )
 
-        _LOGGER.debug(f"I{self._rr_unit}: result is error: {result.isError()} ")
+        _LOGGER.debug(f"unit={self._rr_unit}: result is error: {result.isError()} ")
 
         if result.isError():
-            _LOGGER.debug(f"I{self._rr_unit}: error result: {type(result)} ")
+            _LOGGER.debug(f"unit={self._rr_unit}: error result: {type(result)} ")
 
             if type(result) is ModbusIOException:
                 raise ModbusIOError(result)
 
             if type(result) is ExceptionResponse:
                 if result.exception_code == ModbusExceptions.IllegalAddress:
-                    _LOGGER.debug(f"I{unit} Read IllegalAddress: {result}")
+                    _LOGGER.debug(f"unit={self._rr_unit} Read IllegalAddress: {result}")
                     raise ModbusIllegalAddress(result)
 
                 if result.exception_code == ModbusExceptions.IllegalFunction:
-                    _LOGGER.debug(f"I{unit} Read IllegalFunction: {result}")
+                    _LOGGER.debug(
+                        f"unit={self._rr_unit} Read IllegalFunction: {result}"
+                    )
                     raise ModbusIllegalFunction(result)
 
                 if result.exception_code == ModbusExceptions.IllegalValue:
-                    _LOGGER.debug(f"I{unit} Read IllegalValue: {result}")
+                    _LOGGER.debug(f"unit={self._rr_unit} Read IllegalValue: {result}")
                     raise ModbusIllegalValue(result)
 
             raise ModbusReadError(result)
 
         _LOGGER.debug(
-            f"I{self._rr_unit}: Registers received={len(result.registers)} "
+            f"unit={self._rr_unit}: Registers received={len(result.registers)} "
             f"requested={self._rr_count} address={self._rr_address} "
             f"result={result}"
         )
 
         if len(result.registers) != rcount:
             raise ModbusReadError(
-                f"I{self._rr_unit}: Registers received != requested : "
+                f"unit={self._rr_unit}: Registers received != requested : "
                 f"{len(result.registers)} != {self._rr_count} at {self._rr_address}"
             )
 
@@ -961,6 +986,9 @@ async def init_device(self) -> None:
                 f"ID {self.inverter_unit_id} is not a SunSpec inverter."
             )
 
+        if DETECT_EVSE_REGEX.match(self.decoded_common["C_Model"]):
+            raise DeviceIsEVSE(f"Model {self.decoded_common['C_Model']}")
+
         if (
             self.decoded_common["C_SunSpec_ID"] == SunSpecNotImpl.UINT32
             or self.decoded_common["C_SunSpec_DID"] == SunSpecNotImpl.UINT16
@@ -2653,3 +2681,191 @@ def battery_energy_reset_cycles(self) -> int:
     @property
     def last_update(self) -> datetime.datetime | None:
         return self._last_update_timestamp
+
+
+class SolarEdgeEVSE:
+    """Class that defines a SolarEdge EVSE."""
+
+    def __init__(self, device_id: int, hub: SolarEdgeModbusMultiHub) -> None:
+        self.evse_unit_id = device_id
+        self.hub = hub
+        self.decoded_common = {}
+        self.decoded_model = {}
+        self.has_parent = False
+
+    async def init_device(self) -> None:
+        """Set up data about the device from modbus."""
+
+        try:
+            evse_data = await self.hub.modbus_read_holding_registers(
+                unit=self.evse_unit_id, address=40000, rcount=69
+            )
+
+            self.decoded_common = dict(
+                [
+                    (
+                        "C_SunSpec_ID",
+                        ModbusClientMixin.convert_from_registers(
+                            evse_data.registers[0:2],
+                            data_type=ModbusClientMixin.DATATYPE.UINT32,
+                        ),
+                    )
+                ]
+            )
+
+            uint16_fields = [
+                "C_SunSpec_DID",
+                "C_SunSpec_Length",
+                "C_Device_address",
+            ]
+            uint16_data = evse_data.registers[2:4] + [evse_data.registers[68]]
+            self.decoded_common.update(
+                dict(
+                    zip(
+                        uint16_fields,
+                        ModbusClientMixin.convert_from_registers(
+                            uint16_data,
+                            data_type=ModbusClientMixin.DATATYPE.UINT16,
+                        ),
+                    )
+                )
+            )
+
+            self.decoded_common.update(
+                dict(
+                    [
+                        (
+                            "C_Manufacturer",  # string(32)
+                            int_list_to_string(
+                                ModbusClientMixin.convert_from_registers(
+                                    evse_data.registers[4:20],
+                                    data_type=ModbusClientMixin.DATATYPE.UINT16,
+                                )
+                            ),
+                        ),
+                        (
+                            "C_Model",  # string(32)
+                            int_list_to_string(
+                                ModbusClientMixin.convert_from_registers(
+                                    evse_data.registers[20:36],
+                                    data_type=ModbusClientMixin.DATATYPE.UINT16,
+                                )
+                            ),
+                        ),
+                        (
+                            "C_Option",  # string(16)
+                            int_list_to_string(
+                                ModbusClientMixin.convert_from_registers(
+                                    evse_data.registers[36:44],
+                                    data_type=ModbusClientMixin.DATATYPE.UINT16,
+                                )
+                            ),
+                        ),
+                        (
+                            "C_Version",  # string(16)
+                            int_list_to_string(
+                                ModbusClientMixin.convert_from_registers(
+                                    evse_data.registers[44:52],
+                                    data_type=ModbusClientMixin.DATATYPE.UINT16,
+                                )
+                            ),
+                        ),
+                        (
+                            "C_SerialNumber",  # string(32)
+                            int_list_to_string(
+                                ModbusClientMixin.convert_from_registers(
+                                    evse_data.registers[52:68],
+                                    data_type=ModbusClientMixin.DATATYPE.UINT16,
+                                )
+                            ),
+                        ),
+                    ]
+                )
+            )
+
+            for name, value in iter(self.decoded_common.items()):
+                _LOGGER.debug(
+                    (
+                        f"E{self.evse_unit_id}: "
+                        f"{name} {hex(value) if isinstance(value, int) else value}"
+                        f"{type(value)}"
+                    ),
+                )
+
+        except ModbusIOError:
+            raise DeviceInvalid(f"No response from evse ID {self.evse_unit_id}")
+
+        except ModbusIllegalAddress:
+            raise DeviceInvalid(f"ID {self.evse_unit_id} is not SunSpec.")
+
+        if (
+            self.decoded_common["C_SunSpec_ID"] == SunSpecNotImpl.UINT32
+            or self.decoded_common["C_SunSpec_DID"] == SunSpecNotImpl.UINT16
+            or self.decoded_common["C_SunSpec_ID"] != 0x53756E53
+            or self.decoded_common["C_SunSpec_DID"] != 0x0001
+            or self.decoded_common["C_SunSpec_Length"] != 65
+        ):
+            raise DeviceInvalid(f"ID {self.evse_unit_id} is not SunSpec.")
+
+        self.manufacturer = self.decoded_common["C_Manufacturer"]
+        self.model = self.decoded_common["C_Model"]
+        self.option = self.decoded_common["C_Option"]
+        self.serial = self.decoded_common["C_SerialNumber"]
+        self.device_address = self.decoded_common["C_Device_address"]
+        self.name = f"{self.hub.hub_id.capitalize()} E{self.evse_unit_id}"
+        self.uid_base = f"{self.model}_{self.serial}"
+
+    async def read_modbus_data(self) -> None:
+        """Read and update dynamic modbus registers."""
+
+        try:
+            evse_data = await self.hub.modbus_read_holding_registers(
+                unit=self.evse_unit_id, address=40044, rcount=16
+            )
+
+            self.decoded_common["C_Version"] = int_list_to_string(
+                ModbusClientMixin.convert_from_registers(
+                    evse_data.registers[0:8],
+                    data_type=ModbusClientMixin.DATATYPE.UINT16,
+                )
+            )
+
+            for name, value in iter(self.decoded_model.items()):
+                if isinstance(value, float):
+                    display_value = float_to_hex(value)
+                else:
+                    display_value = hex(value) if isinstance(value, int) else value
+                _LOGGER.debug(
+                    f"E{self.evse_unit_id}: {name} {display_value} {type(value)}"
+                )
+
+        except ModbusIllegalAddress:
+            _LOGGER.error(f"E{self.evse_unit_id}: EVSE register(s) NOT available")
+
+        except ModbusIOError:
+            raise ModbusReadError(f"No response from EVSE ID {self.evse_unit_id}")
+
+    @property
+    def online(self) -> bool:
+        """Device is online."""
+        return self.hub.online
+
+    @property
+    def fw_version(self) -> str | None:
+        if "C_Version" in self.decoded_common:
+            return self.decoded_common["C_Version"]
+
+        return None
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        """Return the device info."""
+        return DeviceInfo(
+            identifiers={(DOMAIN, self.uid_base)},
+            name=self.name,
+            manufacturer=self.manufacturer,
+            model=self.model,
+            serial_number=self.serial,
+            sw_version=self.fw_version,
+            hw_version=self.option,
+        )

custom_components/solaredge_modbus_multi/manifest.json

@@ -10,5 +10,5 @@
   "issue_tracker": "https://github.com/WillCodeForCats/solaredge-modbus-multi/issues",
   "loggers": ["custom_components.solaredge_modbus_multi"],
   "requirements": ["pymodbus>=3.8.3", "awesomeversion>=25.5.0"],
-  "version": "3.3.0-pre.2"
+  "version": "3.3.0-evse.6"
 }

custom_components/solaredge_modbus_multi/sensor.py

@@ -242,6 +242,9 @@ async def async_setup_entry(
         entities.append(SolarEdgeBatterySOE(battery, config_entry, coordinator))
         entities.append(SolarEdgeBatteryStatus(battery, config_entry, coordinator))
 
+    for evse in hub.evses:
+        entities.append(Version(evse, config_entry, coordinator))
+
     if entities:
         async_add_entities(entities)