Fix OCPP subprotocol negotiation and version handling across reconnects

custom_components/ocpp/api.py

@@ -25,7 +25,9 @@
 from .const import (
     CentralSystemSettings,
     DOMAIN,
+    OCPP_1_6,
     OCPP_2_0,
+    OCPP_VERSION_AUTO,
     ChargerSystemSettings,
 )
 from .enums import (
@@ -102,7 +104,7 @@ def __init__(self, hass: HomeAssistant, entry: ConfigEntry):
         self.hass = hass
         self.entry = entry
         self.settings = CentralSystemSettings(**entry.data)
-        self.subprotocols = self.settings.subprotocols
+        self.subprotocols = self._resolve_subprotocols(self.settings)
         self._server = None
         self.id = self.settings.csid
         self.charge_points = {}  # uses cp_id as reference to charger instance
@@ -203,27 +205,86 @@ def _norm_conn(connector_id: int | None) -> int:
         except Exception:
             return 0
 
+    @staticmethod
+    def _resolve_subprotocols(settings: CentralSystemSettings) -> list:
+        """Return the websocket subprotocols the server should advertise.
+
+        When the user pins an OCPP version (config-flow "OCPP version" field is
+        not ``auto``), advertise only that version's subprotocol so a charger
+        that offers several versions cannot negotiate the wrong one. ``auto``
+        advertises everything the integration supports, preserving the previous
+        behaviour.
+        """
+        ocpp_version = settings.ocpp_version
+        if ocpp_version and ocpp_version != OCPP_VERSION_AUTO:
+            return [f"ocpp{ocpp_version}"]
+        return list(settings.subprotocols)
+
     def select_subprotocol(
         self, connection: ServerConnection, subprotocols
     ) -> Subprotocol | None:
         """Override default subprotocol selection."""
 
-        # Server offers at least one subprotocol but client doesn't offer any.
-        # Default to None
+        _LOGGER.debug("Charger offered subprotocols: %s", list(subprotocols or []))
+
+        # Returning None here (rather than raising, as the websockets default
+        # does) is a deliberate deviation that lets a charger offering no
+        # subprotocol default to OCPP 1.6, and it is kept. It is only valid
+        # while the server is actually willing to talk 1.6: when the version is
+        # pinned to 2.x, accepting such a connection would create a v1.6
+        # ChargePoint that then poisons the charge point cache for the
+        # follow-up connection, so reject instead.
         if not subprotocols:
-            return None
+            if OCPP_1_6 in self.subprotocols:
+                return None
+            raise NegotiationError(
+                "no subprotocol offered; expected one of "
+                + ", ".join(self.subprotocols)
+            )
 
-        # Server and client both offer subprotocols. Look for a shared one.
+        # Server and client both offer subprotocols. Look for a shared one,
+        # iterating the server's ordered list - the documented websockets
+        # behaviour this function overrides ("pick the first one in the list
+        # declared the server"). This loop previously iterated the client's
+        # offer as a set instead, so for a charger offering several
+        # subprotocols the negotiated version depended on set-iteration order
+        # and varied between handshakes. Pinning an OCPP version leaves a
+        # single entry in self.subprotocols, which is how a charger is held to
+        # a version other than the default preference.
         proposed_subprotocols = set(subprotocols)
-        for subprotocol in proposed_subprotocols:
-            if subprotocol in self.subprotocols:
+        for subprotocol in self.subprotocols:
+            if subprotocol in proposed_subprotocols:
                 return subprotocol
 
         # No common subprotocol was found.
         raise NegotiationError(
             "invalid subprotocol; expected one of " + ", ".join(self.subprotocols)
         )
 
+    @staticmethod
+    def _negotiated_ocpp_version(websocket: ServerConnection) -> str:
+        """Return the ocpp version string implied by the negotiated subprotocol.
+
+        Mirrors how ``ChargePoint._ocpp_version`` is derived in the v16/v201
+        subclasses: ``ocpp1.6`` -> ``1.6``, ``ocpp2.0.1`` -> ``2.0.1``,
+        ``ocpp2.1`` -> ``2.1``. A charger that offers no subprotocol defaults
+        to 1.6.
+        """
+        subprotocol = websocket.subprotocol
+        if not subprotocol:
+            return "1.6"
+        return subprotocol.replace("ocpp", "")
+
+    def _build_charge_point(self, cp_id: str, websocket: ServerConnection, cp_settings):
+        """Construct the ChargePoint matching this connection's subprotocol."""
+        if websocket.subprotocol and websocket.subprotocol.startswith(OCPP_2_0):
+            return ChargePointv201(
+                cp_id, websocket, self.hass, self.entry, self.settings, cp_settings
+            )
+        return ChargePointv16(
+            cp_id, websocket, self.hass, self.entry, self.settings, cp_settings
+        )
+
     async def on_connect(self, websocket: ServerConnection):
         """Request handler executed for every new OCPP connection."""
         if websocket.subprotocol is not None:
@@ -264,14 +325,7 @@ async def on_connect(self, websocket: ServerConnection):
                 _LOGGER.error(f"Failed to setup charger {cp_id}: {str(e)}")
                 return
 
-            if websocket.subprotocol and websocket.subprotocol.startswith(OCPP_2_0):
-                charge_point = ChargePointv201(
-                    cp_id, websocket, self.hass, self.entry, self.settings, cp_settings
-                )
-            else:
-                charge_point = ChargePointv16(
-                    cp_id, websocket, self.hass, self.entry, self.settings, cp_settings
-                )
+            charge_point = self._build_charge_point(cp_id, websocket, cp_settings)
             self.charge_points[cp_id] = charge_point
             self.connections += 1
             _LOGGER.info(
@@ -282,11 +336,52 @@ async def on_connect(self, websocket: ServerConnection):
             )
             await charge_point.start()
         else:
-            _LOGGER.info(
-                f"Charger {cp_id} reconnected to {self.settings.host}:{self.settings.port}."
-            )
             charge_point = self.charge_points[cp_id]
-            await charge_point.reconnect(websocket)
+            negotiated_version = self._negotiated_ocpp_version(websocket)
+            if negotiated_version != charge_point._ocpp_version:
+                # The charger reconnected negotiating a different OCPP version
+                # than the cached ChargePoint was built with. The cached object
+                # carries a fixed validator / message set / entity model, so
+                # reusing it would validate the new version's payloads against
+                # the old version's schema (e.g. a 2.0.1 BootNotification
+                # against the 1.6 schema), raising FormatViolationError on
+                # every reconnect until the config entry is reloaded. Some
+                # chargers (e.g. FoxESS A-series) make a short-lived 1.6 probe
+                # connection right after a version switch, which is exactly
+                # what plants the mismatched object. Rebuild the ChargePoint
+                # from this handshake's negotiated subprotocol instead. See
+                # issue #2008.
+                _LOGGER.info(
+                    f"Charger {cp_id} reconnected with a different OCPP version "
+                    f"({charge_point._ocpp_version} -> {negotiated_version}); "
+                    f"rebuilding charge point."
+                )
+                # stop() closes the websocket before cancelling its tasks, so a
+                # failure to close would otherwise leave the stale instance's
+                # monitor_connection running against a charger that has already
+                # been replaced here - duplicate pings and metric writes.
+                # Cancel them explicitly if stop() does not get that far.
+                try:
+                    await charge_point.stop()
+                except Exception:
+                    _LOGGER.debug(
+                        "Error stopping stale charge point %s during rebuild; "
+                        "cancelling its tasks directly",
+                        cp_id,
+                        exc_info=True,
+                    )
+                    for task in getattr(charge_point, "tasks", None) or []:
+                        task.cancel()
+                charge_point = self._build_charge_point(
+                    cp_id, websocket, charge_point.settings
+                )
+                self.charge_points[cp_id] = charge_point
+                await charge_point.start()
+            else:
+                _LOGGER.info(
+                    f"Charger {cp_id} reconnected to {self.settings.host}:{self.settings.port}."
+                )
+                await charge_point.reconnect(websocket)
 
     def _get_metrics(self, id: str):
         """Return (cp_id, metrics mapping, cp instance, safe int num_connectors)."""

custom_components/ocpp/config_flow.py

@@ -22,6 +22,7 @@
     CONF_MONITORED_VARIABLES,
     CONF_MONITORED_VARIABLES_AUTOCONFIG,
     CONF_NUM_CONNECTORS,
+    CONF_OCPP_VERSION,
     CONF_PORT,
     CONF_SKIP_SCHEMA_VALIDATION,
     CONF_SSL,
@@ -42,6 +43,7 @@
     DEFAULT_MONITORED_VARIABLES,
     DEFAULT_MONITORED_VARIABLES_AUTOCONFIG,
     DEFAULT_NUM_CONNECTORS,
+    DEFAULT_OCPP_VERSION,
     DEFAULT_PORT,
     DEFAULT_SKIP_SCHEMA_VALIDATION,
     DEFAULT_SSL,
@@ -53,6 +55,7 @@
     DEFAULT_WEBSOCKET_PING_TRIES,
     DOMAIN,
     MEASURANDS,
+    OCPP_VERSIONS,
 )
 
 STEP_USER_CS_DATA_SCHEMA = vol.Schema(
@@ -63,6 +66,9 @@
         vol.Required(CONF_SSL_CERTFILE_PATH, default=DEFAULT_SSL_CERTFILE_PATH): str,
         vol.Required(CONF_SSL_KEYFILE_PATH, default=DEFAULT_SSL_KEYFILE_PATH): str,
         vol.Required(CONF_CSID, default=DEFAULT_CSID): vol.All(str, vol.Length(max=20)),
+        vol.Required(CONF_OCPP_VERSION, default=DEFAULT_OCPP_VERSION): vol.In(
+            OCPP_VERSIONS
+        ),
         vol.Required(
             CONF_WEBSOCKET_CLOSE_TIMEOUT, default=DEFAULT_WEBSOCKET_CLOSE_TIMEOUT
         ): int,
@@ -139,6 +145,39 @@ async def async_step_user(self, user_input=None) -> ConfigFlowResult:
             description_placeholders={"docs_url": "https://github.com/lbbrhzn/ocpp"},
         )
 
+    async def async_step_reconfigure(self, user_input=None) -> ConfigFlowResult:
+        """Allow reconfiguring the central system settings of an existing entry.
+
+        Without this, settings added after an entry was created (such as the
+        OCPP version pin) could only be changed by deleting and re-adding the
+        integration.
+        """
+        entry = self._get_reconfigure_entry()
+
+        if user_input is not None:
+            # Don't allow servers to use same websocket port (the entry being
+            # reconfigured is excluded from the match).
+            self._async_abort_entries_match({CONF_PORT: user_input[CONF_PORT]})
+            # Updating the entry already triggers a reload, via the
+            # add_update_listener(async_reload_entry) registered in
+            # async_setup_entry. async_update_reload_and_abort() would schedule
+            # a second one on top of it, and the two overlap: the websocket
+            # server is rebound while the first setup is still in flight and
+            # the platform forwards then fail with "config entry ... has
+            # already been setup". Update only, and let the listener reload.
+            self.hass.config_entries.async_update_entry(
+                entry, data={**entry.data, **user_input}
+            )
+            return self.async_abort(reason="reconfigure_successful")
+
+        return self.async_show_form(
+            step_id="reconfigure",
+            data_schema=self.add_suggested_values_to_schema(
+                STEP_USER_CS_DATA_SCHEMA, entry.data
+            ),
+            description_placeholders={"docs_url": "https://github.com/lbbrhzn/ocpp"},
+        )
+
     async def async_step_integration_discovery(
         self, discovery_info=None
     ) -> ConfigFlowResult:

custom_components/ocpp/const.py

@@ -26,6 +26,7 @@
 CONF_MONITORED_VARIABLES_AUTOCONFIG = "monitored_variables_autoconfig"
 CONF_NAME = ha.CONF_NAME
 CONF_NUM_CONNECTORS = "num_connectors"
+CONF_OCPP_VERSION = "ocpp_version"
 CONF_PASSWORD = ha.CONF_PASSWORD
 CONF_PORT = ha.CONF_PORT
 CONF_SKIP_SCHEMA_VALIDATION = "skip_schema_validation"
@@ -54,7 +55,16 @@
 DEFAULT_SSL_CERTFILE_PATH = pathlib.Path.cwd().joinpath("fullchain.pem")
 DEFAULT_SSL_KEYFILE_PATH = pathlib.Path.cwd().joinpath("privkey.pem")
 DEFAULT_SUBPROTOCOLS = ["ocpp1.6", "ocpp2.0.1", "ocpp2.1"]
+OCPP_1_6 = "ocpp1.6"
 OCPP_2_0 = "ocpp2"
+OCPP_VERSION_AUTO = "auto"
+DEFAULT_OCPP_VERSION = OCPP_VERSION_AUTO
+# Selectable values for the config-flow "OCPP version" field. "auto" advertises
+# every supported subprotocol (DEFAULT_SUBPROTOCOLS) and negotiates in that
+# order, so a charger offering several versions gets the first entry; any other
+# value restricts negotiation to that single OCPP version, so such a charger is
+# held to it and cannot fall back to (and then crash on) the wrong one.
+OCPP_VERSIONS = [OCPP_VERSION_AUTO, "1.6", "2.0.1", "2.1"]
 DEFAULT_METER_INTERVAL = 60
 DEFAULT_IDLE_INTERVAL = 900
 DEFAULT_WEBSOCKET_CLOSE_TIMEOUT = 10
@@ -172,6 +182,8 @@ class CentralSystemSettings:
     websocket_ping_tries: int
     cpids: list = field(default_factory=list)  # holds cpid config flow settings
     subprotocols: list = field(default_factory=lambda: DEFAULT_SUBPROTOCOLS)
+    # "auto" (advertise all) or a specific OCPP version to pin negotiation to.
+    ocpp_version: str = DEFAULT_OCPP_VERSION
 
     # def __post_init__(self):
     #     i = 0

custom_components/ocpp/translations/en.json

@@ -11,10 +11,34 @@
                     "ssl_certfile_path": "Path to SSL certificate or (None)",
                     "ssl_keyfile_path": "Path to SSL key or (None)",
                     "csid": "Central system identity",
+                    "ocpp_version": "OCPP version",
                     "websocket_close_timeout": "Websocket close timeout (seconds)",
                     "websocket_ping_tries": "Websocket successive times to try connection before closing",
                     "websocket_ping_interval": "Websocket ping interval (seconds)",
                     "websocket_ping_timeout": "Websocket ping timeout (seconds)"
+                },
+                "data_description": {
+                    "ocpp_version": "Auto negotiates with the charger using the integration's default order (OCPP 1.6 first). Choose a specific version to restrict negotiation to it, so a charger that advertises several versions is held to the one you pick."
+                }
+            },
+            "reconfigure": {
+                "title": "OCPP Central System Configuration",
+                "description": "If you need help with the configuration have a look [here]({docs_url})",
+                "data": {
+                    "host": "Central system host address",
+                    "port": "Central system port number",
+                    "ssl": "Secure connection",
+                    "ssl_certfile_path": "Path to SSL certificate or (None)",
+                    "ssl_keyfile_path": "Path to SSL key or (None)",
+                    "csid": "Central system identity",
+                    "ocpp_version": "OCPP version",
+                    "websocket_close_timeout": "Websocket close timeout (seconds)",
+                    "websocket_ping_tries": "Websocket successive times to try connection before closing",
+                    "websocket_ping_interval": "Websocket ping interval (seconds)",
+                    "websocket_ping_timeout": "Websocket ping timeout (seconds)"
+                },
+                "data_description": {
+                    "ocpp_version": "Auto negotiates with the charger using the integration's default order (OCPP 1.6 first). Choose a specific version to restrict negotiation to it, so a charger that advertises several versions is held to the one you pick."
                 }
             },
             "cp_user": {
@@ -66,7 +90,8 @@
         },
         "abort": {
             "single_instance_allowed": "Only a single instance is allowed",
-            "reauth_successful": "New charger configured"
+            "reauth_successful": "New charger configured",
+            "reconfigure_successful": "Configuration updated"
         }
     },
     "exceptions": {

custom_components/ocpp/translations/i-default.json

@@ -11,10 +11,14 @@
                     "ssl_certfile_path": "Path to SSL certificate or (None)",
                     "ssl_keyfile_path": "Path to SSL key or (None)",
                     "csid": "Central system identity",
+                    "ocpp_version": "OCPP version",
                     "websocket_close_timeout": "Websocket close timeout (seconds)",
                     "websocket_ping_tries": "Websocket successive times to try connection before closing",
                     "websocket_ping_interval": "Websocket ping interval (seconds)",
                     "websocket_ping_timeout": "Websocket ping timeout (seconds)"
+                },
+                "data_description": {
+                    "ocpp_version": "Auto negotiates with the charger using the integration's default order (OCPP 1.6 first). Choose a specific version to restrict negotiation to it, so a charger that advertises several versions is held to the one you pick."
                 }
             },
             "cp_user": {
@@ -57,6 +61,26 @@
                     "Temperature": "Temperature: Temperature reading inside Charge Point",
                     "Voltage": "Voltage: Instantaneous AC RMS supply voltage"
                 }
+            },
+            "reconfigure": {
+                "title": "OCPP Central System Configuration",
+                "description": "If you need help with the configuration have a look [here]({docs_url})",
+                "data": {
+                    "host": "Central system host address",
+                    "port": "Central system port number",
+                    "ssl": "Secure connection",
+                    "ssl_certfile_path": "Path to SSL certificate or (None)",
+                    "ssl_keyfile_path": "Path to SSL key or (None)",
+                    "csid": "Central system identity",
+                    "ocpp_version": "OCPP version",
+                    "websocket_close_timeout": "Websocket close timeout (seconds)",
+                    "websocket_ping_tries": "Websocket successive times to try connection before closing",
+                    "websocket_ping_interval": "Websocket ping interval (seconds)",
+                    "websocket_ping_timeout": "Websocket ping timeout (seconds)"
+                },
+                "data_description": {
+                    "ocpp_version": "Auto negotiates with the charger using the integration's default order (OCPP 1.6 first). Choose a specific version to restrict negotiation to it, so a charger that advertises several versions is held to the one you pick."
+                }
             }
         },
         "error": {
@@ -66,7 +90,8 @@
         },
         "abort": {
             "single_instance_allowed": "Only a single instance is allowed",
-            "reauth_successful": "New charger configured"
+            "reauth_successful": "New charger configured",
+            "reconfigure_successful": "Configuration updated"
         }
     },
     "exceptions": {
@@ -86,4 +111,4 @@
             "message": "Charger is unavailable: {message}"
         }
     }
-}
+}