diff --git a/netatmo_service/collector.py b/netatmo_service/collector.py index 778fd3a..07b2534 100644 --- a/netatmo_service/collector.py +++ b/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) - diff --git a/netatmo_service/rrd.py b/netatmo_service/rrd.py index 11c27ef..6328528 100644 --- a/netatmo_service/rrd.py +++ b/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]: diff --git a/tests/test_rrd.py b/tests/test_rrd.py index bb8858e..10889f7 100644 --- a/tests/test_rrd.py +++ b/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