Add current limit fallback for OCPP 1.6 chargers

custom_components/ocpp/ocppv16.py

@@ -62,6 +62,8 @@
 
 _LOGGER: logging.Logger = logging.getLogger(__package__)
 
+CURRENT_LIMIT_CONFIGURATION_KEY = "ChargeRate"
+
 
 def _to_message_trigger(name: str) -> MessageTrigger | None:
     if isinstance(name, MessageTrigger):
@@ -432,7 +434,7 @@ async def set_charge_rate(
 
         if not (int(self.supported_features or 0) & prof.SMART):
             _LOGGER.info("Smart charging is not supported by this charger")
-            return False
+            return await self._set_charge_rate_via_configuration(limit_amps)
 
         # Determine allowed unit (default to Amps if not reported)
         units_resp = await self.get_configuration(
@@ -577,7 +579,53 @@ def _profile_id(purpose: str, cid: int) -> int:
                     f"Note: Active TxProfile applied, but TxDefaultProfile failed: {ex}"
                 )
 
-        return bool(txp_ok or txd_ok)
+        if txp_ok or txd_ok:
+            return True
+
+        return await self._set_charge_rate_via_configuration(limit_amps)
+
+    async def _set_charge_rate_via_configuration(self, limit_amps: int) -> bool:
+        """Set charge current using an optional configuration key if available."""
+        try:
+            amps = int(float(limit_amps))
+        except (TypeError, ValueError):
+            return False
+
+        if amps <= 0:
+            return False
+
+        try:
+            current = await self.get_configuration(CURRENT_LIMIT_CONFIGURATION_KEY)
+        except Exception as ex:
+            _LOGGER.debug(
+                "Could not read %s configuration key: %s",
+                CURRENT_LIMIT_CONFIGURATION_KEY,
+                ex,
+            )
+            return False
+
+        if not current or str(current).strip().lower() == "unknown":
+            return False
+
+        target = str(amps)
+        if str(current).strip() == target:
+            return True
+
+        try:
+            result = await self.configure(CURRENT_LIMIT_CONFIGURATION_KEY, target)
+        except Exception as ex:
+            _LOGGER.debug(
+                "Could not set %s configuration key: %s",
+                CURRENT_LIMIT_CONFIGURATION_KEY,
+                ex,
+            )
+            return False
+
+        return result in (
+            SetVariableResult.accepted,
+            SetVariableResult.reboot_required,
+            None,
+        )
 
     async def set_availability(self, state: bool = True, connector_id: int | None = 0):
         """Change availability."""

docs/supported-devices.md

@@ -50,6 +50,15 @@ This list is based on the overview of OCPP 1.6 implementation for ABB Terra AC (
 
 ## [Alfen - Eve Single S-line](https://alfen.com/en/ev-charge-points/alfen-product-range)
 
+## BRC Opticharge 11
+Configure OCPP 1.6-J with a WebSocket URL that includes the charge point
+identity, for example `ws://homeassistant.local:9000/<charge-point-id>`.
+
+If needed, disable automatic measurand detection and manually select current,
+energy, power, temperature, and voltage import/export measurands. Dynamic current
+control uses standard OCPP smart charging first, with a configuration fallback
+when supported by the charger.
+
 ## [CTEK Chargestorm Connected 1, dual connectors](https://www.ctek.com/uk/ev-charging/chargestorm-connected-1)
 See CTEK Chargestorm Connected 2 below for getting started instructions.
 

tests/test_set_charge_rate_v16.py

@@ -14,6 +14,7 @@
 import pytest
 
 from custom_components.ocpp.ocppv16 import ChargePoint as ChargePointv16
+from custom_components.ocpp.chargepoint import SetVariableResult
 from custom_components.ocpp.enums import (
     Profiles as prof,
     ConfigurationKey as ckey,
@@ -89,7 +90,7 @@ async def fake_get_conf(_key):
 async def test_smart_charging_not_supported_returns_false_no_notify(
     cp_v16, monkeypatch
 ):
-    """2) If the charger doesn't advertise SMART profile, return False without notifications."""
+    """2) If SMART and ChargeRate are unavailable, return False without notifications."""
     cp_v16._attr_supported_features = prof.NONE
 
     notices = []
@@ -98,22 +99,81 @@ async def fake_notify(msg, title="Ocpp integration"):
         notices.append(msg)
         return True
 
-    # get_configuration and call should not be called
-    async def fake_get_conf(_key):
-        pytest.fail("get_configuration should not be called when SMART not supported")
+    async def fake_get_conf(key):
+        assert key == "ChargeRate"
+        return "Unknown"
 
     async def fake_call(_req):
         pytest.fail("call should not be called when SMART not supported")
 
+    async def fake_configure(_key, _value):
+        pytest.fail("configure should not be called for an unknown key")
+
     monkeypatch.setattr(cp_v16, "notify_ha", fake_notify)
     monkeypatch.setattr(cp_v16, "get_configuration", fake_get_conf)
     monkeypatch.setattr(cp_v16, "call", fake_call)
+    monkeypatch.setattr(cp_v16, "configure", fake_configure)
 
     ok = await cp_v16.set_charge_rate(limit_amps=16, conn_id=2)
     assert ok is False
     assert notices == []
 
 
+@pytest.mark.asyncio
+async def test_smart_charging_not_supported_uses_charge_rate_key(
+    cp_v16, monkeypatch
+):
+    """If SMART is missing, fall back to the optional current configuration key."""
+    cp_v16._attr_supported_features = prof.NONE
+    configured = []
+
+    async def fake_get_conf(key):
+        assert key == "ChargeRate"
+        return "32"
+
+    async def fake_configure(key, value):
+        configured.append((key, value))
+        return SetVariableResult.accepted
+
+    async def fake_call(_req):
+        pytest.fail("SetChargingProfile should not be called when SMART is missing")
+
+    monkeypatch.setattr(cp_v16, "get_configuration", fake_get_conf)
+    monkeypatch.setattr(cp_v16, "configure", fake_configure)
+    monkeypatch.setattr(cp_v16, "call", fake_call)
+
+    ok = await cp_v16.set_charge_rate(limit_amps=16, conn_id=2)
+
+    assert ok is True
+    assert configured == [("ChargeRate", "16")]
+
+
+@pytest.mark.asyncio
+async def test_smart_charging_not_supported_charge_rate_unknown_returns_false(
+    cp_v16, monkeypatch
+):
+    """Unknown ChargeRate support leaves the old failure behavior intact."""
+    cp_v16._attr_supported_features = prof.NONE
+
+    async def fake_get_conf(key):
+        assert key == "ChargeRate"
+        return "Unknown"
+
+    async def fake_configure(_key, _value):
+        pytest.fail("configure should not be called for an unknown key")
+
+    async def fake_call(_req):
+        pytest.fail("SetChargingProfile should not be called when SMART is missing")
+
+    monkeypatch.setattr(cp_v16, "get_configuration", fake_get_conf)
+    monkeypatch.setattr(cp_v16, "configure", fake_configure)
+    monkeypatch.setattr(cp_v16, "call", fake_call)
+
+    ok = await cp_v16.set_charge_rate(limit_amps=16, conn_id=2)
+
+    assert ok is False
+
+
 @pytest.mark.asyncio
 async def test_cpmax_exception_falls_back_to_txdefault_accepted_returns_true(
     cp_v16, monkeypatch
@@ -185,3 +245,40 @@ async def fake_notify(msg, title="Ocpp integration"):
     ok = await cp_v16.set_charge_rate(limit_amps=10, conn_id=2)
     assert ok is True
     assert notices == []
+
+
+@pytest.mark.asyncio
+async def test_rejected_smart_profiles_fall_back_to_charge_rate_key(
+    cp_v16, monkeypatch
+):
+    """If standard smart profiles are rejected, use the optional current key."""
+    configured = []
+
+    async def fake_get_conf(key: str):
+        if key == ckey.charging_schedule_allowed_charging_rate_unit.value:
+            return "Current"
+        if key == ckey.charge_profile_max_stack_level.value:
+            return "3"
+        if key == "ChargeRate":
+            return "32"
+        pytest.fail(f"Unexpected get_configuration key: {key}")
+
+    async def fake_call(req):
+        assert req.cs_charging_profiles["chargingProfilePurpose"] in (
+            ChargingProfilePurposeType.charge_point_max_profile.value,
+            ChargingProfilePurposeType.tx_default_profile.value,
+        )
+        return SimpleNamespace(status=ChargingProfileStatus.rejected)
+
+    async def fake_configure(key, value):
+        configured.append((key, value))
+        return SetVariableResult.accepted
+
+    monkeypatch.setattr(cp_v16, "get_configuration", fake_get_conf)
+    monkeypatch.setattr(cp_v16, "call", fake_call)
+    monkeypatch.setattr(cp_v16, "configure", fake_configure)
+
+    ok = await cp_v16.set_charge_rate(limit_amps=12, conn_id=2)
+
+    assert ok is True
+    assert configured == [("ChargeRate", "12")]