solved duplicate points error

netatmo_service/collector.py

@@ -20,14 +20,15 @@ class Collector:
 
     def collect_once(self) -> int:
         samples = extract_samples(self.client.stations_data(), self.module_names)
+        updated = 0
         for name, sample in samples.items():
             try:
-                self.store.update(name, sample)
+                updated += bool(self.store.update(name, sample))
             except Exception:
                 # One stale/duplicate module must not discard the other modules.
                 LOG.exception("Could not update %s RRD", name)
-        LOG.info("Collected %d Netatmo modules", len(samples))
-        return len(samples)
+        LOG.info("Collected %d Netatmo modules; updated %d RRDs", len(samples), updated)
+        return updated
 
     def start(self) -> None:
         if self._thread and self._thread.is_alive():
@@ -48,4 +49,3 @@ class Collector:
                 LOG.exception("Netatmo collection failed")
             remaining = max(1, self.interval - (time.monotonic() - started))
             self._stop.wait(remaining)
-

netatmo_service/rrd.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import colorsys
+import logging
 import re
 import tempfile
 from pathlib import Path
@@ -10,6 +11,7 @@ import rrdtool
 from .data import Sample
 
 RRD_ERROR = rrdtool.OperationalError
+LOG = logging.getLogger(__name__)
 
 STEP = 600
 HEARTBEAT = 1500
@@ -53,15 +55,26 @@ class RRDStore:
                      "RRA:AVERAGE:0.5:36:1464", "RRA:MIN:0.5:36:1464", "RRA:MAX:0.5:36:1464"]
             rrdtool.create(str(path), *args)
 
-    def update(self, name: str, sample: Sample) -> None:
+    def update(self, name: str, sample: Sample) -> bool:
+        """Store at most one sample per RRD step; return whether it was written."""
+        path = self.path(name)
+        last_timestamp = int(rrdtool.last(str(path)))
+        if sample.timestamp // STEP <= last_timestamp // STEP:
+            LOG.info(
+                "Skipping %s RRD update: sample interval already stored "
+                "(sample_timestamp=%d last_timestamp=%d)",
+                name, sample.timestamp, last_timestamp,
+            )
+            return False
         schema = SCHEMAS[name]
         fields = list(schema)
         values = ["U" if sample.values.get(field) is None else str(sample.values[field]) for field in fields]
         rrdtool.update(
-            str(self.path(name)),
+            str(path),
             "--template", ":".join(fields),
             f"{sample.timestamp}:{':'.join(values)}",
         )
+        return True
 
     def last(self, name: str, field: str) -> dict:
         if name not in SCHEMAS or field not in SCHEMAS[name]:

tests/test_rrd.py

@@ -1,3 +1,4 @@
+from netatmo_service.data import Sample
 from netatmo_service.rrd import RRDStore
 
 
@@ -5,3 +6,16 @@ def test_rrd_files_have_netatmo_prefix(tmp_path):
     store = RRDStore(tmp_path)
     assert store.path("outdoor") == tmp_path / "netatmo_outdoor.rrd"
     assert store.path("wind") == tmp_path / "netatmo_wind.rrd"
+
+
+def test_rrd_is_updated_only_once_per_collection_interval(tmp_path):
+    store = RRDStore(tmp_path)
+    store.ensure_all()
+    now = int(__import__("time").time())
+    timestamp = ((now // 600) - 1) * 600 + 100
+    first = Sample(timestamp, {"temperature": 20, "co2": 500, "humidity": 50})
+    same_interval = Sample(timestamp + 1, {"temperature": 21, "co2": 510, "humidity": 51})
+
+    assert store.update("study", first) is True
+    assert store.update("study", first) is False
+    assert store.update("study", same_interval) is False