Add count_covered for coverage between valid samples

This returns the count of samples that are covered between the oldest and newest
valid samples.

Signed-off-by: cwasicki <[email protected]>

Author: cwasicki

src/frequenz/sdk/timeseries/_ringbuffer/buffer.py

@@ -566,6 +566,33 @@ def __getitem__(self, index_or_slice: SupportsIndex | slice) -> float | FloatArr
         """
         return self._buffer.__getitem__(index_or_slice)
 
+    def _covered_time_range(self) -> timedelta:
+        """Return the time range that is covered by the oldest and newest valid samples.
+
+        Returns:
+            The time range between the oldest and newest valid samples or 0 if
+                there are is no time range covered.
+        """
+        if not self.oldest_timestamp:
+            return timedelta(0)
+
+        assert (
+            self.newest_timestamp is not None
+        ), "Newest timestamp cannot be None here."
+        return self.newest_timestamp - self.oldest_timestamp + self._sampling_period
+
+    def count_covered(self) -> int:
+        """Count the number of samples that are covered by the oldest and newest valid samples.
+
+        Returns:
+            The count of samples between the oldest and newest (inclusive) valid samples
+                or 0 if there are is no time range covered.
+        """
+        return int(
+            self._covered_time_range().total_seconds()
+            // self._sampling_period.total_seconds()
+        )
+
     def count_valid(self) -> int:
         """Count the number of valid items that this buffer currently holds.
 

tests/timeseries/test_ringbuffer.py

@@ -210,18 +210,21 @@ def test_gaps() -> None:  # pylint: disable=too-many-statements
     assert buffer.oldest_timestamp is None
     assert buffer.newest_timestamp is None
     assert buffer.count_valid() == 0
+    assert buffer.count_covered() == 0
     assert len(buffer.gaps) == 0
 
     buffer.update(Sample(dt(0), Quantity(0)))
     assert buffer.oldest_timestamp == dt(0)
     assert buffer.newest_timestamp == dt(0)
     assert buffer.count_valid() == 1
+    assert buffer.count_covered() == 1
     assert len(buffer.gaps) == 1
 
     buffer.update(Sample(dt(6), Quantity(0)))
     assert buffer.oldest_timestamp == dt(6)
     assert buffer.newest_timestamp == dt(6)
     assert buffer.count_valid() == 1
+    assert buffer.count_covered() == 1
     assert len(buffer.gaps) == 1
 
     buffer.update(Sample(dt(2), Quantity(2)))
@@ -230,48 +233,57 @@ def test_gaps() -> None:  # pylint: disable=too-many-statements
     assert buffer.oldest_timestamp == dt(2)
     assert buffer.newest_timestamp == dt(6)
     assert buffer.count_valid() == 4
+    assert buffer.count_covered() == 5
     assert len(buffer.gaps) == 1
 
     buffer.update(Sample(dt(3), None))
     assert buffer.oldest_timestamp == dt(2)
     assert buffer.newest_timestamp == dt(6)
     assert buffer.count_valid() == 3
+    assert buffer.count_covered() == 5
     assert len(buffer.gaps) == 2
 
     buffer.update(Sample(dt(3), Quantity(np.nan)))
     assert buffer.oldest_timestamp == dt(2)
     assert buffer.newest_timestamp == dt(6)
     assert buffer.count_valid() == 3
+    assert buffer.count_covered() == 5
     assert len(buffer.gaps) == 2
 
     buffer.update(Sample(dt(2), Quantity(np.nan)))
     assert buffer.oldest_timestamp == dt(4)
     assert buffer.newest_timestamp == dt(6)
     assert buffer.count_valid() == 2
+    assert buffer.count_covered() == 3
     assert len(buffer.gaps) == 2
 
     buffer.update(Sample(dt(3), Quantity(3)))
     assert buffer.oldest_timestamp == dt(3)
     assert buffer.newest_timestamp == dt(6)
     assert buffer.count_valid() == 3
+    assert buffer.count_covered() == 4
     assert len(buffer.gaps) == 2
 
     buffer.update(Sample(dt(2), Quantity(2)))
     assert buffer.oldest_timestamp == dt(2)
     assert buffer.newest_timestamp == dt(6)
     assert buffer.count_valid() == 4
+    assert buffer.count_covered() == 5
     assert len(buffer.gaps) == 1
 
     buffer.update(Sample(dt(5), Quantity(5)))
     assert buffer.oldest_timestamp == dt(2)
     assert buffer.newest_timestamp == dt(6)
     assert buffer.count_valid() == 5
+    assert buffer.count_covered() == 5
     assert len(buffer.gaps) == 0
 
+    # whole range gap suffers from sdk#646
     buffer.update(Sample(dt(99), None))
     assert buffer.oldest_timestamp == dt(95)  # bug: should be None
     assert buffer.newest_timestamp == dt(99)  # bug: should be None
     assert buffer.count_valid() == 4  # bug: should be 0 (whole range gap)
+    assert buffer.count_covered() == 5  # bug: should be 0
     assert len(buffer.gaps) == 1