Allow ComponentDataSamples carrying only states

If a `ComponentDataSample` doesn't have any `metric_samples` we raise
an exception, but that's not correct, as it is perfectly valid for them
to carry only states.

This commits allows this case, using the timestamp of the last state (as
we do with metrics). The only caveat is the data itself will be created
in `*Data` object using defaults (zeros).

We also add some regression tests, although the affected code should be
gone soon.

Signed-off-by: Leandro Lucarella <luca-frequenz@llucax.com>

Author: llucax

RELEASE_NOTES.md

@@ -14,4 +14,4 @@
 
 ## Bug Fixes
 
-<!-- Here goes notable bug fixes that are worth a special mention or explanation -->
+* Accept `ComponentDataSamples` that only carry state changes (the state changes were dropped before).

src/frequenz/sdk/microgrid/_old_component_data.py

@@ -104,19 +104,26 @@ def to_samples(self: Self) -> ComponentDataSamples:
     @staticmethod
     def _from_samples(class_: type[T], /, samples: ComponentDataSamples) -> T:
         """Create a new instance from a component data object."""
-        if not samples.metric_samples:
-            raise ValueError("No metrics in the samples.")
+        if not samples.metric_samples and not samples.states:
+            raise ValueError("No metrics and no states in the samples.")
 
         # FIXME: This might not be true forever, but the service sends all metrics with
         # the same timestamp for now, and it is very convenient to map the received data
         # to the old component data metrics packets, which had only one timestamp.
         # When we move away frome these old_component_data wrappers, we should not
         # assume only one metric sample can come per telemetry message anymore.
-        timestamp = samples.metric_samples[-1].sampled_at
-        for sample in samples.metric_samples[:-1]:
-            if sample.sampled_at != timestamp:
+        # When there are no metric samples, fall back to the last state sample's
+        # timestamp so that state-only messages (for example, a component error
+        # state without any metric values) are still propagated downstream.
+        source = samples.metric_samples or samples.states
+        kind = "metric" if samples.metric_samples else "state"
+        timestamp = source[-1].sampled_at
+        for item in source[:-1]:
+            if item.sampled_at != timestamp:
                 _logger.warning(
-                    "ComponentData has multiple timestamps. Using the last one. Samples: %r",
+                    "ComponentData has multiple %s sample timestamps. Using the last"
+                    " one. Samples: %r",
+                    kind,
                     samples,
                 )
                 break
@@ -298,9 +305,6 @@ class MeterData(ComponentData):
     # pylint: disable-next=too-many-branches
     def from_samples(cls, samples: ComponentDataSamples) -> Self:
         """Create a new instance from a component data object."""
-        if not samples.metric_samples:
-            raise ValueError("No metrics in the samples.")
-
         self = cls._from_samples(cls, samples)
 
         active_power_per_phase: list[float] = [0.0, 0.0, 0.0]
@@ -486,9 +490,6 @@ class BatteryData(ComponentData):  # pylint: disable=too-many-instance-attribute
     @classmethod
     def from_samples(cls, samples: ComponentDataSamples) -> Self:
         """Create a new instance from a component data object."""
-        if not samples.metric_samples:
-            raise ValueError("No metrics in the samples.")
-
         self = cls._from_samples(cls, samples)
 
         for sample in samples.metric_samples:
@@ -704,9 +705,6 @@ class InverterData(ComponentData):  # pylint: disable=too-many-instance-attribut
     # pylint: disable-next=too-many-branches
     def from_samples(cls, samples: ComponentDataSamples) -> Self:
         """Create a new instance from a component data object."""
-        if not samples.metric_samples:
-            raise ValueError("No metrics in the samples.")
-
         self = cls._from_samples(cls, samples)
 
         active_power_per_phase: list[float] = [0.0, 0.0, 0.0]
@@ -966,9 +964,6 @@ class ChpData(ComponentData):  # pylint: disable=too-many-instance-attributes
     # pylint: disable-next=too-many-branches
     def from_samples(cls, samples: ComponentDataSamples) -> Self:
         """Create a new instance from a component data object."""
-        if not samples.metric_samples:
-            raise ValueError("No metrics in the samples.")
-
         self = cls._from_samples(cls, samples)
 
         active_power_per_phase: list[float] = [0.0, 0.0, 0.0]
@@ -1229,9 +1224,6 @@ class EVChargerData(ComponentData):  # pylint: disable=too-many-instance-attribu
     # pylint: disable-next=too-many-branches
     def from_samples(cls, samples: ComponentDataSamples) -> Self:
         """Create a new instance from a component data object."""
-        if not samples.metric_samples:
-            raise ValueError("No metrics in the samples.")
-
         self = cls._from_samples(cls, samples)
 
         active_power_per_phase: list[float] = [0.0, 0.0, 0.0]

tests/microgrid/test_old_component_data.py

@@ -0,0 +1,113 @@
+# License: MIT
+# Copyright © 2026 Frequenz Energy-as-a-Service GmbH
+
+"""Tests for `frequenz.sdk.microgrid._old_component_data`."""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from frequenz.client.common.microgrid.components import ComponentId
+from frequenz.client.microgrid.component import (
+    ComponentDataSamples,
+    ComponentStateCode,
+    ComponentStateSample,
+)
+
+from frequenz.sdk.microgrid._old_component_data import (
+    BatteryData,
+    ChpData,
+    ComponentData,
+    EVChargerData,
+    InverterData,
+    MeterData,
+)
+
+
+@pytest.mark.parametrize(
+    "cls",
+    [BatteryData, ChpData, EVChargerData, InverterData, MeterData],
+)
+def test_from_samples_preserves_state_without_metrics(
+    cls: type[ComponentData],
+) -> None:
+    """`from_samples` must not drop a component state when no metrics are present.
+
+    Regression test for
+    https://github.com/frequenz-floss/frequenz-sdk-python/issues/1406:
+    a `ComponentDataSamples` message that carries only a state sample (for example
+    `ComponentStateCode.ERROR`) and no metric samples must still be converted into
+    a concrete `ComponentData` instance carrying that state, so downstream code
+    learns about the component's error status as soon as it happens.
+
+    Previously, `from_samples` raised `ValueError("No metrics in the samples.")`
+    and the state was silently swallowed by `_receive_logging_errors`.
+    """
+    component_id = ComponentId(1503)
+    sampled_at = datetime(2026, 5, 21, 12, 37, 40, 107899, tzinfo=timezone.utc)
+    samples = ComponentDataSamples(
+        component_id=component_id,
+        metric_samples=[],
+        states=[
+            ComponentStateSample(
+                sampled_at=sampled_at,
+                states=frozenset({ComponentStateCode.ERROR}),
+                warnings=frozenset(),
+                errors=frozenset(),
+            ),
+        ],
+    )
+
+    data = cls.from_samples(samples)
+
+    assert isinstance(data, cls)
+    assert data.component_id == component_id
+    assert data.timestamp == sampled_at
+    assert ComponentStateCode.ERROR in data.states
+    assert not data.warnings
+    assert not data.errors
+
+
+def test_from_samples_warns_on_multiple_state_timestamps(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Multiple state samples with differing timestamps trigger a warning.
+
+    When falling back to a state sample's timestamp (because there are no metric
+    samples), `_from_samples` should warn if the state samples disagree on
+    timestamps, mirroring the existing behaviour for metric samples.
+    """
+    component_id = ComponentId(1503)
+    first = datetime(2026, 5, 21, 12, 37, 40, tzinfo=timezone.utc)
+    second = first + timedelta(seconds=1)
+    samples = ComponentDataSamples(
+        component_id=component_id,
+        metric_samples=[],
+        states=[
+            ComponentStateSample(
+                sampled_at=first,
+                states=frozenset({ComponentStateCode.ERROR}),
+                warnings=frozenset(),
+                errors=frozenset(),
+            ),
+            ComponentStateSample(
+                sampled_at=second,
+                states=frozenset({ComponentStateCode.ERROR}),
+                warnings=frozenset(),
+                errors=frozenset(),
+            ),
+        ],
+    )
+
+    with caplog.at_level(
+        logging.WARNING, logger="frequenz.sdk.microgrid._old_component_data"
+    ):
+        data = InverterData.from_samples(samples)
+
+    assert data.timestamp == second
+    assert any(
+        "multiple state sample timestamps" in record.message
+        for record in caplog.records
+    ), caplog.records