Add test for v2x multi connector. Handle meter values properly. max_current working per connector.

Author: Jan Thunqvist

custom_components/ocpp/api.py

@@ -5,6 +5,7 @@
 import contextlib
 import json
 import logging
+import re
 import ssl
 
 from functools import partial
@@ -90,6 +91,10 @@
 )
 
 
+def _norm(s: str) -> str:
+    return re.sub(r"[^a-z0-9]", "", str(s).lower())
+
+
 class CentralSystem:
     """Server for handling OCPP connections."""
 
@@ -192,7 +197,7 @@ async def create(hass: HomeAssistant, entry: ConfigEntry):
 
     @staticmethod
     def _norm_conn(connector_id: int | None) -> int:
-        if connector_id in (None, 0):
+        if connector_id is None:
             return 0
         try:
             return int(connector_id)
@@ -293,20 +298,50 @@ def _get_metrics(self, id: str):
     def get_metric(self, id: str, measurand: str, connector_id: int | None = None):
         """Return last known value for given measurand."""
         # allow id to be either cpid or cp_id
-        cp_id, m = self._get_metrics(id)
-
-        if m is None:
+        cp_id = self.cpids.get(id, id)
+        if cp_id not in self.charge_points:
             return None
 
-        conn = self._norm_conn(connector_id)
-        try:
-            return m[(conn, measurand)].value
-        except Exception:
-            if conn == 0:
-                with contextlib.suppress(Exception):
-                    return m[measurand].value
+        cp = self.charge_points[cp_id]
+        m = cp._metrics
+        n_connectors = getattr(cp, "num_connectors", 1) or 1
+
+        def _try_val(key):
+            with contextlib.suppress(Exception):
+                val = m[key].value
+                return val
             return None
 
+        # 1) Explicit connector_id (including 0): just get it
+        if connector_id is not None:
+            conn = 0 if connector_id == 0 else connector_id
+            return _try_val((conn, measurand))
+
+        # 2) No connector_id: try CHARGER level (conn=0)
+        val = _try_val((0, measurand))
+        if val is not None:
+            return val
+
+        # 3) Legacy "flat" key (before the connector support)
+        with contextlib.suppress(Exception):
+            val = m[measurand].value
+            if val is not None:
+                return val
+
+        # 4) Fallback to connector 1 (old tests often expect this)
+        if n_connectors >= 1:
+            val = _try_val((1, measurand))
+            if val is not None:
+                return val
+
+        # 5) Last resort: find the first connector 2..N with value
+        for c in range(2, int(n_connectors) + 1):
+            val = _try_val((c, measurand))
+            if val is not None:
+                return val
+
+        return None
+
     def del_metric(self, id: str, measurand: str, connector_id: int | None = None):
         """Set given measurand to None."""
         # allow id to be either cpid or cp_id
@@ -326,47 +361,125 @@ def del_metric(self, id: str, measurand: str, connector_id: int | None = None):
     def get_unit(self, id: str, measurand: str, connector_id: int | None = None):
         """Return unit of given measurand."""
         # allow id to be either cpid or cp_id
-        cp_id, m = self._get_metrics(id)
-        if m is None:
+        cp_id = self.cpids.get(id, id)
+        if cp_id not in self.charge_points:
             return None
-        conn = self._norm_conn(connector_id)
-        try:
-            return m[(conn, measurand)].unit
-        except Exception:
-            if conn == 0:
-                with contextlib.suppress(Exception):
-                    return m[measurand].unit
+
+        cp = self.charge_points[cp_id]
+        m = cp._metrics
+        n_connectors = getattr(cp, "num_connectors", 1) or 1
+
+        def _try_unit(key):
+            with contextlib.suppress(Exception):
+                return m[key].unit
             return None
 
+        if connector_id is not None:
+            conn = 0 if connector_id == 0 else connector_id
+            return _try_unit((conn, measurand))
+
+        val = _try_unit((0, measurand))
+        if val is not None:
+            return val
+
+        with contextlib.suppress(Exception):
+            val = m[measurand].unit
+            if val is not None:
+                return val
+
+        if n_connectors >= 1:
+            val = _try_unit((1, measurand))
+            if val is not None:
+                return val
+
+        for c in range(2, int(n_connectors) + 1):
+            val = _try_unit((c, measurand))
+            if val is not None:
+                return val
+
+        return None
+
     def get_ha_unit(self, id: str, measurand: str, connector_id: int | None = None):
         """Return home assistant unit of given measurand."""
-        cp_id, m = self._get_metrics(id)
-        if m is None:
+        cp_id = self.cpids.get(id, id)
+        if cp_id not in self.charge_points:
             return None
-        conn = self._norm_conn(connector_id)
-        try:
-            return m[(conn, measurand)].ha_unit
-        except Exception:
-            if conn == 0:
-                with contextlib.suppress(Exception):
-                    return m[measurand].ha_unit
+
+        cp = self.charge_points[cp_id]
+        m = cp._metrics
+        n_connectors = getattr(cp, "num_connectors", 1) or 1
+
+        def _try_ha_unit(key):
+            with contextlib.suppress(Exception):
+                return m[key].ha_unit
             return None
 
+        if connector_id is not None:
+            conn = 0 if connector_id == 0 else connector_id
+            return _try_ha_unit((conn, measurand))
+
+        val = _try_ha_unit((0, measurand))
+        if val is not None:
+            return val
+
+        with contextlib.suppress(Exception):
+            val = m[measurand].ha_unit
+            if val is not None:
+                return val
+
+        if n_connectors >= 1:
+            val = _try_ha_unit((1, measurand))
+            if val is not None:
+                return val
+
+        for c in range(2, int(n_connectors) + 1):
+            val = _try_ha_unit((c, measurand))
+            if val is not None:
+                return val
+
+        return None
+
     def get_extra_attr(self, id: str, measurand: str, connector_id: int | None = None):
-        """Return last known extra attributes for given measurand."""
+        """Return extra attributes for given measurand."""
         # allow id to be either cpid or cp_id
-        cp_id, m = self._get_metrics(id)
-        if m is None:
+        cp_id = self.cpids.get(id, id)
+        if cp_id not in self.charge_points:
             return None
-        conn = self._norm_conn(connector_id)
-        try:
-            return m[(conn, measurand)].extra_attr
-        except Exception:
-            if conn == 0:
-                with contextlib.suppress(Exception):
-                    return m[measurand].extra_attr
+
+        cp = self.charge_points[cp_id]
+        m = cp._metrics
+        n_connectors = getattr(cp, "num_connectors", 1) or 1
+
+        def _try_extra(key):
+            with contextlib.suppress(Exception):
+                return m[key].extra_attr
             return None
 
+        if connector_id is not None:
+            conn = 0 if connector_id == 0 else connector_id
+            return _try_extra((conn, measurand))
+
+        val = _try_extra((0, measurand))
+        if val is not None:
+            return val
+
+        with contextlib.suppress(Exception):
+            val = m[measurand].extra_attr
+            if val is not None:
+                return val
+
+        if n_connectors >= 1:
+            val = _try_extra((1, measurand))
+            if val is not None:
+                return val
+
+        for c in range(2, int(n_connectors) + 1):
+            val = _try_extra((c, measurand))
+            if val is not None:
+                return val
+
+        return None
+
     def get_available(self, id: str, connector_id: int | None = None):
         """Return whether the charger (or a specific connector) is available."""
         # allow id to be either cpid or cp_id
@@ -397,7 +510,7 @@ def get_available(self, id: str, connector_id: int | None = None):
         if not status_val:
             return cp.status == STATE_OK
 
-        ok_statuses = {
+        ok_statuses_norm = {
             "available",
             "preparing",
             "charging",
@@ -408,7 +521,7 @@ def get_available(self, id: str, connector_id: int | None = None):
             "reserved",
         }
 
-        ret = str(status_val).lower() in ok_statuses
+        ret = _norm(status_val) in ok_statuses_norm
         return ret
 
     def get_supported_features(self, id: str):
@@ -420,32 +533,46 @@ def get_supported_features(self, id: str):
             return self.charge_points[cp_id].supported_features
         return 0
 
-    async def set_max_charge_rate_amps(self, id: str, value: float):
+    async def set_max_charge_rate_amps(
+        self, id: str, value: float, connector_id: int = 0
+    ):
         """Set the maximum charge rate in amps."""
         # allow id to be either cpid or cp_id
         cp_id = self.cpids.get(id, id)
 
         if cp_id in self.charge_points:
-            return await self.charge_points[cp_id].set_charge_rate(limit_amps=value)
+            return await self.charge_points[cp_id].set_charge_rate(
+                limit_amps=value, conn_id=connector_id
+            )
         return False
 
-    async def set_charger_state(self, id: str, service_name: str, state: bool = True):
+    async def set_charger_state(
+        self,
+        id: str,
+        service_name: str,
+        state: bool = True,
+        connector_id: int | None = 1,
+    ):
         """Carry out requested service/state change on connected charger."""
         # allow id to be either cpid or cp_id
         cp_id = self.cpids.get(id, id)
 
         resp = False
         if cp_id in self.charge_points:
             if service_name == csvcs.service_availability.name:
-                resp = await self.charge_points[cp_id].set_availability(state)
+                resp = await self.charge_points[cp_id].set_availability(
+                    state, connector_id=connector_id
+                )
             if service_name == csvcs.service_charge_start.name:
-                resp = await self.charge_points[cp_id].start_transaction()
+                resp = await self.charge_points[cp_id].start_transaction(
+                    connector_id=connector_id
+                )
             if service_name == csvcs.service_charge_stop.name:
                 resp = await self.charge_points[cp_id].stop_transaction()
             if service_name == csvcs.service_reset.name:
                 resp = await self.charge_points[cp_id].reset()
             if service_name == csvcs.service_unlock.name:
-                resp = await self.charge_points[cp_id].unlock()
+                resp = await self.charge_points[cp_id].unlock(connector_id=connector_id)
         return resp
 
     def device_info(self):

custom_components/ocpp/button.py

@@ -58,7 +58,7 @@ async def async_setup_entry(hass, entry, async_add_devices):
         num_connectors = int(cp_id_settings.get(CONF_NUM_CONNECTORS, 1))
 
         for desc in BUTTONS:
-            if desc.per_connector and num_connectors > 1:
+            if desc.per_connector:
                 for connector_id in range(1, num_connectors + 1):
                     entities.append(
                         ChargePointButton(
@@ -124,5 +124,7 @@ def available(self) -> bool:
     async def async_press(self) -> None:
         """Triggers the charger press action service."""
         await self.central_system.set_charger_state(
-            self.cpid, self.entity_description.press_action
+            self.cpid,
+            self.entity_description.press_action,
+            connector_id=self.connector_id,
         )