Update victron_mqtt to 2026.7.3

custom_components/victron_mqtt/_vendor/__init__.py

@@ -1,3 +1,3 @@
 """Vendored third-party packages."""
 
-VICTRON_MQTT_VERSION = "2026.7.2"
+VICTRON_MQTT_VERSION = "2026.7.3"

custom_components/victron_mqtt/_vendor/victron_mqtt/_unwrappers.py

@@ -1,8 +1,9 @@
 """Functions to unwrap the data from the JSON string."""
 
 import json
-from collections.abc import Iterable
+from collections.abc import Callable, Iterable
 from datetime import UTC, datetime
+from typing import Any
 
 from .constants import BITMASK_SEPARATOR, ValueType, VictronEnum
 
@@ -199,7 +200,7 @@ def wrap_epoch(value: datetime | None) -> str:
     return json.dumps({"value": datetime.timestamp(value)})
 
 
-VALUE_TYPE_UNWRAPPER = {
+VALUE_TYPE_UNWRAPPER: dict[ValueType, Callable[..., Any]] = {
     ValueType.INT: unwrap_int,
     ValueType.INT_DEFAULT_0: unwrap_int_default_0,
     ValueType.FLOAT: unwrap_float,
@@ -213,7 +214,7 @@ def wrap_epoch(value: datetime | None) -> str:
     ValueType.FLOAT_M3_TO_LITERS: unwrap_float_m3_to_liters,
 }
 
-VALUE_TYPE_WRAPPER = {
+VALUE_TYPE_WRAPPER: dict[ValueType, Callable[..., str]] = {
     ValueType.INT: wrap_int,
     ValueType.INT_DEFAULT_0: wrap_int_default_0,
     ValueType.FLOAT: wrap_float,

custom_components/victron_mqtt/_vendor/victron_mqtt/_victron_formulas.py

@@ -73,7 +73,7 @@ def schedule_charge_enabled(
 
 
 def schedule_charge_enabled_set(
-    value: str, depends_on: dict[str, Metric], _transient_state: FormulaTransientState | None
+    value: str | GenericOnOff, depends_on: dict[str, Metric], _transient_state: FormulaTransientState | None
 ) -> tuple[GenericOnOff, None]:
     """Set schedule charge enabled state."""
 
@@ -108,7 +108,7 @@ def dvcc_enabled(
     metric = next(iter(depends_on.values()))
     if metric.value is None:
         return None, None
-    code = metric.value.code if isinstance(metric.value, DVCCMode) else int(metric.value)
+    code = int(metric.value.code) if isinstance(metric.value, DVCCMode) else int(metric.value)
     return (GenericOnOff.ON if code & 1 else GenericOnOff.OFF), None
 
 
@@ -125,7 +125,7 @@ def dvcc_enabled_set(
     metric = next(iter(depends_on.values()))
     assert isinstance(metric, WritableMetric), "Expected WritableMetric for dvcc_enabled_set"
     # Do not override BMS/system-forced state (FORCED_OFF=2, FORCED_ON=3)
-    current_code = metric.value.code if isinstance(metric.value, DVCCMode) else int(metric.value or 0)
+    current_code = int(metric.value.code) if isinstance(metric.value, DVCCMode) else int(metric.value or 0)
     if current_code & 2:
         # Forced by system — return current state without writing
         return (GenericOnOff.ON if current_code & 1 else GenericOnOff.OFF), None
@@ -232,7 +232,7 @@ def ess_user_mode(
         return ESSUserMode.EXTERNAL_CONTROL, None
 
     state_value = state_metric.value
-    code = state_value.code if isinstance(state_value, ESSState) else int(state_value)
+    code = int(state_value.code) if isinstance(state_value, ESSState) else int(state_value)
 
     if code == 9:
         return ESSUserMode.KEEP_BATTERIES_CHARGED, None

custom_components/victron_mqtt/_vendor/victron_mqtt/_victron_topics.py

@@ -3175,6 +3175,28 @@
         name="Grid power {phase}",
         metric_type=MetricType.POWER,
     ),
+    TopicDescriptor(
+        topic="N/{installation_id}/system/{device_id}/Ac/PvOnGrid/NumberOfPhases",
+        message_type=MetricKind.SENSOR,
+        short_id="system_pv_on_grid_phases",
+        name="PV on grid phases",
+        value_type=ValueType.INT,
+        unit_of_measurement="phases",
+    ),
+    TopicDescriptor(
+        topic="N/{installation_id}/system/{device_id}/Ac/PvOnGrid/{phase}/Current",
+        message_type=MetricKind.SENSOR,
+        short_id="system_pv_on_grid_current_{phase}",
+        name="PV on grid current {phase}",
+        metric_type=MetricType.CURRENT,
+    ),
+    TopicDescriptor(
+        topic="N/{installation_id}/system/{device_id}/Ac/PvOnGrid/{phase}/Power",
+        message_type=MetricKind.SENSOR,
+        short_id="system_pv_on_grid_power_{phase}",
+        name="PV on grid power {phase}",
+        metric_type=MetricType.POWER,
+    ),
     TopicDescriptor(
         topic="N/{installation_id}/system/{device_id}/Ac/PvOnOutput/NumberOfPhases",
         message_type=MetricKind.SENSOR,

custom_components/victron_mqtt/_vendor/victron_mqtt/data_classes.py

@@ -406,7 +406,7 @@ def finalize_topic_fields(self, topic_desc: TopicDescriptor, device_unique_id: s
         self._key_values = self.get_key_values(topic_desc)
         self._key_values.update(topic_desc.key_values)
         self._short_id = self._replace_ids(topic_desc.short_id).lower()
-        effective_device_id = device_unique_id if device_unique_id else self.get_device_unique_id()
+        effective_device_id = device_unique_id or self.get_device_unique_id()
         self._unique_id = ParsedTopic.make_unique_id(effective_device_id, self._short_id)
         assert topic_desc.name is not None, f"TopicDescriptor name is None for topic: {topic_desc.topic}"
         self._name = self._replace_ids(topic_desc.name)

custom_components/victron_mqtt/_vendor/victron_mqtt/device.py

@@ -52,11 +52,11 @@ def __init__(
         self._device_type = parsed_topic.device_type
         self._device_id = parsed_topic.device_id
         self._installation_id = parsed_topic.installation_id
-        self._model = None
-        self._manufacturer = None
-        self._serial_number = None
-        self._firmware_version = None
-        self._custom_name = None
+        self._model: str | None = None
+        self._manufacturer: str | None = None
+        self._serial_number: str | None = None
+        self._firmware_version: str | None = None
+        self._custom_name: str | None = None
         self._parent_device: Device | None = parent_device
 
         _LOGGER.debug(
@@ -128,15 +128,17 @@ def handle_message(
 
         parsed_topic.finalize_topic_fields(topic_desc, device_unique_id=self._unique_id)
         if fallback_to_metric_topic:
-            value = unwrap_bool(payload)
-            if value is None:
+            fallback_value = unwrap_bool(payload)
+            if fallback_value is None:
                 log_debug(
                     "Ignoring null fallback_to_metric_topic value for device %s metric %s",
                     self.unique_id,
                     topic_desc.short_id,
                 )
                 return None
-            return FallbackPlaceholder(device=self, parsed_topic=parsed_topic, topic_descriptor=topic_desc, value=value)
+            return FallbackPlaceholder(
+                device=self, parsed_topic=parsed_topic, topic_descriptor=topic_desc, value=fallback_value
+            )
         value = Device._unwrap_payload(topic_desc, payload)
         if value is None:
             log_debug("Ignoring null topic value for device %s metric %s", self.unique_id, topic_desc.short_id)
@@ -154,6 +156,7 @@ def _unwrap_payload(topic_desc: TopicDescriptor, payload: str) -> str | float |
         assert topic_desc.value_type is not None
         unwrapper = VALUE_TYPE_UNWRAPPER[topic_desc.value_type]
         if unwrapper in [unwrap_enum, unwrap_bitmask]:
+            assert topic_desc.enum is not None, f"enum must be set for topic: {topic_desc.topic}"
             return unwrapper(payload, topic_desc.enum)
         if unwrapper in [
             unwrap_float,
@@ -245,6 +248,7 @@ def _create_metric_from_placeholder(
         )
         assert metric_placeholder.parsed_topic.device_type is not None, "device_type must be set for metric"
 
+        metric: Metric
         if new_topic_desc.message_type in [
             MetricKind.SWITCH,
             MetricKind.NUMBER,
@@ -282,6 +286,7 @@ def _add_formula_metric(self, topic_desc: TopicDescriptor, hub: Hub, key_values:
         name = ParsedTopic.replace_ids(topic_desc.name, key_values)
         short_id = ParsedTopic.replace_ids(topic_desc.short_id, key_values)
         unique_id = ParsedTopic.make_unique_id(self.unique_id, short_id)
+        metric: FormulaMetric
         if topic_desc.message_type in [
             MetricKind.SWITCH,
             MetricKind.NUMBER,

custom_components/victron_mqtt/_vendor/victron_mqtt/hub.py

@@ -35,6 +35,14 @@
 FULL_PUBLISH_MIN_INTERVAL_SECONDS = 180
 FIRST_FULL_PUBLISH_MIN_INTERVAL_SECONDS = 30
 MINIMUM_FULLY_SUPPORTED_VERSION = 3.5
+# The keepalive loop ticks every 30s. Every Nth tick we send a forced full republish so the
+# broker re-sends every current value, refreshing last_seen for otherwise-constant metrics.
+FULL_REPUBLISH_STALENESS_INTERVAL_CYCLES = 6
+# A metric not seen for this long is considered stale (its source stopped publishing) and is
+# marked unavailable. Must be larger than the forced full republish interval
+# (FULL_REPUBLISH_STALENESS_INTERVAL_CYCLES * 30s) so healthy but constant metrics are not
+# wrongly invalidated between republishes.
+STALE_METRIC_TIMEOUT_SECONDS = 360
 
 # Modify the logger to include instance_id without changing the tracing level
 # class InstanceIDFilter(logging.Filter):
@@ -187,7 +195,7 @@ def __init__(
         self._first_refresh_event: asyncio.Event = asyncio.Event()
         self._installation_id_event: asyncio.Event = asyncio.Event()
         self._snapshot: dict[str, Any] = {}
-        self._keepalive_task = None
+        self._keepalive_task: asyncio.Task[None] | None = None
         self._connected_event = asyncio.Event()
         self._on_new_metric: CallbackOnNewMetric | None = None
         self._on_new_device: CallbackOnNewDevice | None = None
@@ -786,13 +794,10 @@ def _handle_normal_message(self, topic: str, payload: str, log_debug: Callable[.
         if desc_list is None:
             log_debug("Ignoring message - no descriptor found for topic: %s", topic)
             return
-        if len(desc_list) == 1:
-            desc = desc_list[0]
-        else:
-            desc = parsed_topic.match_from_list(desc_list)
-            if desc is None:
-                log_debug("Ignoring message - no matching descriptor found for list of topic: %s", topic)
-                return
+        desc = desc_list[0] if len(desc_list) == 1 else parsed_topic.match_from_list(desc_list)
+        if desc is None:
+            log_debug("Ignoring message - no matching descriptor found for list of topic: %s", topic)
+            return
 
         device = self._get_or_create_device(parsed_topic, desc)
         placeholder = device.handle_message(fallback_to_metric_topic, topic, parsed_topic, desc, payload, log_debug)
@@ -802,9 +807,9 @@ def _handle_normal_message(self, topic: str, payload: str, log_debug: Callable[.
                 log_debug("Replacing existing metric placeholder: %s", existing_placeholder)
             self._metrics_placeholders[placeholder.parsed_topic.unique_id] = placeholder
         elif isinstance(placeholder, FallbackPlaceholder):
-            existing_placeholder = self._fallback_placeholders.get(placeholder.parsed_topic.unique_id)
-            if existing_placeholder:
-                log_debug("Replacing existing fallback placeholder: %s", existing_placeholder)
+            existing_fallback = self._fallback_placeholders.get(placeholder.parsed_topic.unique_id)
+            if existing_fallback:
+                log_debug("Replacing existing fallback placeholder: %s", existing_fallback)
             self._fallback_placeholders[placeholder.parsed_topic.unique_id] = placeholder
 
     async def disconnect(self) -> None:
@@ -852,24 +857,30 @@ async def _keepalive_loop(self) -> None:
                 # We should keep alive all metrics every 60 seconds
                 count += 1
                 # Old firmwars dont resend values after the keepalive message so we cant use this logic of invalidation if there is no new value
-                if self._firmware_version >= MINIMUM_FULLY_SUPPORTED_VERSION and count % 2 == 0:
-                    self._keepalive_metrics()
+                if self._firmware_version >= MINIMUM_FULLY_SUPPORTED_VERSION:
+                    if count % 2 == 0:
+                        # Republish throttled values and mark long-silent sources unavailable (issue #454).
+                        self._keepalive_metrics(stale_timeout=STALE_METRIC_TIMEOUT_SECONDS)
+                    # Periodically force a full republish so the broker re-sends every current
+                    # value, refreshing last_seen for constant metrics that would otherwise look stale.
+                    if count % FULL_REPUBLISH_STALENESS_INTERVAL_CYCLES == 0:
+                        self._keepalive(force=True)
             except asyncio.CancelledError:
                 _LOGGER.info("Keepalive loop canceled")
                 raise
             except Exception as exc:
                 _LOGGER.exception("Error in keepalive loop: %s", exc)
                 await asyncio.sleep(5)  # Short delay before retrying
 
-    def _keepalive_metrics(self, force_invalidate: bool = False) -> None:
+    def _keepalive_metrics(self, force_invalidate: bool = False, stale_timeout: float | None = None) -> None:
         """Keep alive all metrics."""
         _LOGGER.debug("Keeping alive all metrics")
         for metric in self._all_metrics.values():
             # Determine log level based on the substring
             is_info_level = self._topic_log_info and self._topic_log_info in metric._descriptor.topic
             log_debug = _LOGGER.info if is_info_level else _LOGGER.debug
 
-            metric._keepalive(force_invalidate, log_debug)
+            metric._keepalive(force_invalidate, log_debug, stale_timeout=stale_timeout)
 
     def _start_keep_alive_loop(self) -> None:
         """Start the keep_alive loop."""

custom_components/victron_mqtt/_vendor/victron_mqtt/metric.py

@@ -217,12 +217,34 @@ def on_update(self, value: CallbackOnUpdate | None) -> None:
         """Sets the on_update callback."""
         self._on_update = value
 
-    def _keepalive(self, force_invalidate: bool, log_debug: Callable[..., None]):
-        """Reset metrics value if no updates or send last values if they got skipped"""
+    def _keepalive(
+        self,
+        force_invalidate: bool,
+        log_debug: Callable[..., None],
+        stale_timeout: float | None = None,
+    ):
+        """Reset metrics value if no updates or send last values if they got skipped.
+
+        stale_timeout, when provided, is a number of seconds: if the metric has not been seen
+        for longer than that, its source is considered to have stopped publishing and the
+        metric is reset to None (unavailable). This relies on the hub periodically forcing a
+        full republish, so the timeout must be larger than that republish interval.
+        """
         if force_invalidate and self._value is not None:
             log_debug("Metric %s is being forced reset", self.unique_id)
             self._handle_message(None, log_debug, update_last_seen=False)  # Dont update the last_seen as it wasnt seen
             return
+        if stale_timeout is not None and self._value is not None:
+            elapsed = time.monotonic() - self._last_seen
+            if elapsed > stale_timeout:
+                log_debug(
+                    "Metric %s has been silent for %.2fs (> %.2fs), resetting to unavailable",
+                    self.unique_id,
+                    elapsed,
+                    stale_timeout,
+                )
+                self._handle_message(None, log_debug, update_last_seen=False)  # Dont update last_seen as it wasnt seen
+                return
         if self._last_seen > self._last_notified:
             log_debug(
                 "Metric %s has been updated at %.2f but not published since %.2fs, re-publishing",

custom_components/victron_mqtt/_vendor/victron_mqtt/testing/hub_helpers.py

@@ -95,10 +95,6 @@ async def _async_noop(_self: Any) -> None:
             # Set the mocked client explicitly to prevent overwriting
             hub._client = mocked_client
 
-            # Dynamically mock undefined attributes
-            hub._process_device = MagicMock(name="_process_device")
-            hub._process_metric = MagicMock(name="_process_metric")
-
             # Mock connect_async to trigger the _on_connect callback
             def mock_connect_async(*_args: Any, **_kwargs: Any) -> None:
                 hub._on_connect(

custom_components/victron_mqtt/_vendor/victron_mqtt/writable_formula_metric.py

@@ -22,24 +22,30 @@ def __init__(self, *, descriptor: TopicDescriptor, **kwargs: Any) -> None:
         """Initialize the FormulaMetric."""
         _LOGGER.debug(
             "Creating new FormulaMetric: unique_id=%s, type=%s, nature=%s",
-            descriptor.short_id, descriptor.metric_type, descriptor.metric_nature
+            descriptor.short_id,
+            descriptor.metric_type,
+            descriptor.metric_nature,
         )
         assert descriptor.topic.startswith("$$func")
-        func_name = descriptor.topic.split('/')[-1]
+        func_name = descriptor.topic.split("/")[-1]
         assert ":" in func_name
         write_func_name = func_name.split(":", 1)[1]
         self._write_func = getattr(formulas, write_func_name)
 
-        super().__init__(descriptor = descriptor, **kwargs)
-
+        super().__init__(descriptor=descriptor, **kwargs)
 
     def __str__(self) -> str:
         return f"WritableFormulaMetric({super().__str__()}, transient_state={self.transient_state})"
 
     def __repr__(self) -> str:
         return self.__str__()
 
-    def _keepalive(self, force_invalidate: bool, log_debug: Callable[..., None]):
+    def _keepalive(
+        self,
+        force_invalidate: bool,
+        log_debug: Callable[..., None],
+        stale_timeout: float | None = None,
+    ):
         log_debug("Metric is WritableFormulaMetric so no keepalive for now: %s", self.unique_id)
 
     def set(self, value: str | float | int | bool | VictronEnum) -> None:

custom_components/victron_mqtt/_vendor/victron_mqtt/writable_metric.py

@@ -4,9 +4,9 @@
 
 import json
 import logging
-from collections.abc import Callable, Iterable
+from collections.abc import Iterable
 from enum import Enum
-from typing import Any, cast
+from typing import Any
 
 from ._unwrappers import VALUE_TYPE_WRAPPER, wrap_bitmask, wrap_enum
 from ._victron_enums import SwitchableOutputType
@@ -187,7 +187,7 @@ def set(self, value: str | float | int | bool | VictronEnum) -> None:
             payload = json.dumps({"value": self._labels.index(value)})  # type: ignore[union-attr]
         else:
             payload = WritableMetric._wrap_payload(self._descriptor, value)
-        self._hub._publish(self._write_topic, payload)
+        self._hub._publish(self._write_topic, payload)  # pylint: disable=protected-access
 
     @staticmethod
     def _wrap_payload(topic_desc: TopicDescriptor, value: str | float | int | bool | Enum) -> str:
@@ -206,7 +206,7 @@ def _wrap_payload(topic_desc: TopicDescriptor, value: str | float | int | bool |
             )
             return wrap_bitmask(value, topic_desc.enum)
 
-        wrapper = cast("Callable[[Any], str]", VALUE_TYPE_WRAPPER[value_type])
+        wrapper = VALUE_TYPE_WRAPPER[value_type]
         return wrapper(value)
 
     @property

custom_components/victron_mqtt/translations/en.json

@@ -2266,6 +2266,16 @@
       "system_heartbeat": {
         "name": "GX system heartbeat"
       },
+      "system_pv_on_grid_current_phase": {
+        "name": "PV on grid current {phase}"
+      },
+      "system_pv_on_grid_phases": {
+        "name": "PV on grid phases",
+        "unit_of_measurement": "phases"
+      },
+      "system_pv_on_grid_power_phase": {
+        "name": "PV on grid power {phase}"
+      },
       "system_pv_on_output_current_phase": {
         "name": "PV on output current {phase}"
       },