Add configurable source heartbeat verification and bounded Modbus recovery

KONFIGURACJA.txt

@@ -83,3 +83,20 @@ Po edycji: .venv/bin/python settings.py. Nie wyświetla haseł.
 Zmiany wczytywane są podczas kolejnego uruchomienia; usługa stale działająca
 nie jest potrzebna, a serwera nie trzeba restartować.
 W razie niepewności sprawdź jeden odczyt z lemmy.enabled=false.
+
+
+WERSJA 1.1.0 — ODZYSKIWANIE ODCZYTU I HEARTBEAT
+Po szybkich próbach aplikacja ponawia odczyt do 5 razy co 60 sekund.
+Sekcja recovery: attempts (0..10, 0 wyłącza), delay_seconds (1..600).
+Sekcja heartbeat: enabled=true włącza kontrolę; address to adres rejestru
+licznika stacji, liczony od zera. timeout_seconds=70, poll_interval_seconds=5.
+Licznik musi pochodzić ze stacji pomiarowej, a nie z cyklu ekranu HMI.
+checks to opcjonalna lista {"address": 101, "allowed_values": [1]} —
+adres jest przykładowy; dopasuj do dokumentacji własnego urządzenia.
+Zmiana licznika musi nastąpić podczas bieżącego odczytu. Zawinięcie licznika
+jest poprawne. Brak zmiany lub błędna diagnostyka wyklucza zielony status.
+Ostatnie poprawne wartości zostają zachowane z ich rzeczywistym czasem.
+Bez konfiguracji heartbeat kontrola jest wyłączona (zgodność ze starszą wersją).
+Odczyty odbywają się wg harmonogramu, a nie stale. Heartbeat potwierdza pracę
+źródła, nie sprawność każdego czujnika. Do tego służą dodatkowe checks.
+Szczegóły, przyczyny zmian i aktualizacja: RELEASE_v1.1.0.md.

README.md

@@ -221,3 +221,7 @@ The screenshot below was captured from the actual public post on 13 October 2026
 
 This project is licensed under the GNU General Public License v3.0 (GPL-3.0-only).
 See [LICENSE](LICENSE) for the full license text.
+
+## v1.1.0: recovery and heartbeat verification
+
+See [release notes and upgrade instructions](RELEASE_v1.1.0.md) for the reasons for these changes, configuration examples and monitoring limitations. Heartbeat checking is optional and disabled by default; enable it to detect frozen source data behind a responding HMI.

RELEASE_v1.1.0.md

@@ -0,0 +1,24 @@
+## v1.1.0 — Recovery and source freshness (2026-10-19)
+
+### What changed and why
+- Added bounded recovery after the initial fast Modbus retries (default: five extra attempts, 60 seconds apart). Temporary HMI/network outages no longer force waiting until the next hourly cycle when the station recovers within this window.
+- Added optional, configurable source heartbeat verification. A responding HMI may still contain frozen sensor values; a successful Modbus reply alone is not evidence of fresh measurements.
+- Added optional diagnostic register guards, checked against configured allowed values before accepting a heartbeat snapshot.
+- Successful collection with heartbeat enabled requires observing a counter change during that collection. Counter wrap is supported. Unchanged temperature/humidity is valid when the source counter advances.
+- Failed collection preserves the last good readings and their timestamp. The public post explains that they are historical, and cannot show green when heartbeat verification is required but missing.
+- Status JSON now includes `error`, `heartbeat_enabled` and `freshness_verified`. Existing SQLite history and private configuration are retained.
+
+### Upgrade and configuration
+Back up the installation and database, update the program files, then run `python settings.py` and the tests. Do not replace an existing private config with the example. Existing configurations remain valid: heartbeat defaults to disabled and recovery defaults to five attempts with a 60-second delay.
+
+Add `recovery: {"attempts": 5, "delay_seconds": 60}` to configure recovery; use zero attempts to disable it. Each recovery attempt performs one collection, including heartbeat verification if enabled.
+
+To enable source freshness, add a `heartbeat` object like the one in config.example.json, set `enabled` to true and supply a zero-based 16-bit counter register `address`. The counter must originate from the measuring station and be forwarded by the HMI, rather than count HMI screen refreshes. `timeout_seconds` must exceed the normal source/HMI update interval; `poll_interval_seconds` controls checks. Optional `checks` contains objects such as `{"address": 101, "allowed_values": [1]}` (illustrative address only). All reads use the configured Modbus host, unit and FC03/FC04; no writes are performed.
+
+The 70-second default heartbeat window is suitable for a counter forwarded every 30 seconds. Adapt it to your device. Diagnostic checks and the counter are separate Modbus reads, not an atomic snapshot. A heartbeat proves source activity, not the validity of every individual sensor; configure diagnostic guards for that purpose.
+
+### Limits
+This remains a scheduled publisher, not a continuous alarm monitor. A post changes only when a cycle publishes it. Recovery extends cycle duration; overlapping runs are prevented by existing locks. With heartbeat disabled, the post explicitly says source heartbeat is not checked. Enabling it rejects older status files without positive verification until the next successful collection.
+
+### Validation
+All 27 automated tests passed on Linux with Python 3.11. Tests cover signed measurements, fragmented Modbus replies, failed reads, recovery limits, counter changes/wrap/timeouts, diagnostic guards, compatibility and publication status. Tests use fixtures and mocked services, not live public posts.

config.example.json

@@ -46,6 +46,17 @@
     "title": "Stacja meteo",
     "heading": "Stacja meteo",
     "pin_in_community": false,
-    "update_note": "Aktualizacja co godzinę. Wykresy obejmują bieżący miesiąc."
+    "update_note": "Aktualizacja co godzinÄ™. Wykresy obejmujÄ… bieĹĽÄ…cy miesiÄ…c."
+  },
+  "recovery": {
+    "attempts": 5,
+    "delay_seconds": 60
+  },
+  "heartbeat": {
+    "enabled": false,
+    "address": null,
+    "timeout_seconds": 70,
+    "poll_interval_seconds": 5,
+    "checks": []
   }
 }

meteo.py

@@ -76,13 +76,64 @@ def decode(registers, start, sensors):
     return values, raw
 
 
+class FreshnessError(ValueError):
+    """The station is reachable but its source data is not confirmed fresh."""
+
+
+def verify_freshness(config):
+    h = config.get('heartbeat', {})
+    if not h.get('enabled', False):
+        return False
+    deadline = time.monotonic() + h['timeout_seconds']
+    settings = dict(config['modbus'])
+
+    def read(address):
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise FreshnessError('Heartbeat: brak zmiany licznika w wymaganym czasie')
+        settings['timeout_seconds'] = min(config['modbus']['timeout_seconds'], remaining)
+        return read_registers(settings, address, 1)[0]
+
+    def snapshot():
+        for check in h.get('checks', []):
+            value = read(check['address'])
+            if value not in check['allowed_values']:
+                raise FreshnessError('Heartbeat: nieprawidłowy stan diagnostyczny źródła')
+        return read(h['address'])
+
+    previous = snapshot()
+    while True:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise FreshnessError('Heartbeat: brak zmiany licznika w wymaganym czasie')
+        time.sleep(min(h['poll_interval_seconds'], remaining))
+        current = snapshot()
+        if time.monotonic() <= deadline and current != previous:
+            return True
+
+
 def collect(config):
+    result = collect_fast(config)
+    total = result[3]
+    recovery = config.get('recovery', {'attempts': 5, 'delay_seconds': 60})
+    for _ in range(recovery['attempts']):
+        if result[1] is not None:
+            break
+        time.sleep(recovery['delay_seconds'])
+        retry_config = dict(config, modbus=dict(config['modbus'], attempts=1))
+        result = collect_fast(retry_config)
+        total += result[3]
+    return result[:3] + (total, result[4])
+
+
+def collect_fast(config):
     settings = config["modbus"]
     attempts = settings["attempts"]
     if not 1 <= attempts <= 10:
         raise ValueError("Liczba prób musi wynosić od 1 do 10")
     for attempt in range(1, attempts + 1):
         try:
+            verify_freshness(config)
             # Read only configured registers, including widely separated addresses.
             registers = {}
             for name in FIELDS:
@@ -91,6 +142,8 @@ def collect(config):
                     registers[address] = read_registers(settings, address, 1)[0]
             values, raw = decode(registers, 0, config['sensors'])
             return utc_now(), values, raw, attempt, None
+        except FreshnessError as exc:
+            return utc_now(), None, None, attempt, str(exc)
         except (OSError, ValueError) as exc:
             logging.warning("Próba %s/%s: %s", attempt, attempts, exc)
             if attempt < attempts:
@@ -122,8 +175,11 @@ def save_result(db_path, checked_at, values, raw, attempts, error):
     return last_good
 
 
-def write_status(path, checked_at, values, last_good, local_timezone):
+def write_status(path, checked_at, values, last_good, local_timezone, error=None, heartbeat_enabled=False):
     status = {
+        "error": error,
+        "heartbeat_enabled": heartbeat_enabled,
+        "freshness_verified": heartbeat_enabled and values is not None,
         "status": "ok" if values is not None else "error",
         "label": "Działa poprawnie" if values is not None else "Błąd stacji",
         "checked_at": checked_at,
@@ -163,7 +219,8 @@ def main():
             last_good = save_result(BASE / "data/meteo.sqlite3", checked_at,
                                     values, raw, attempts, error)
             status = write_status(BASE / "data/status.json", checked_at,
-                                  values, last_good, config["timezone"])
+                                  values, last_good, config["timezone"], error,
+                                  config.get("heartbeat", {}).get("enabled", False))
             if values is not None:
                 logging.info("Pomiar: temperatura=%s °C; wilgotność=%s %%; ciśnienie=%s hPa",
                              values["temperature"], values["humidity"], values["pressure"])

publish_lemmy.py

@@ -120,6 +120,8 @@ def render_section(status, images, now=None, config=None):
     checked = datetime.fromisoformat(status['checked_at'])
     age = (now - checked).total_seconds()
     healthy = status['status'] == 'ok' and 0 <= age <= config['stale_after_minutes'] * 60
+    if config.get('heartbeat', {}).get('enabled', False):
+        healthy = healthy and status.get('freshness_verified') is True
     values = status.get('values') if healthy else status.get('last_success_values')
     def value(name, unit):
         return ('—' if not values else f"{values[name]:.1f}".replace('.', ',') + ' ' + unit)
@@ -134,6 +136,10 @@ def render_section(status, images, now=None, config=None):
     lines.append('**Ostatni poprawny pomiar:** ' + shown)
     if not healthy and values:
         lines.append('\nPodane wartości pochodzą z ostatniego poprawnego pomiaru.')
+    if not healthy:
+        lines.append('\nBrak aktualnego, poprawnego odczytu lub potwierdzenia świeżości danych stacji.')
+    elif not config.get('heartbeat', {}).get('enabled', False):
+        lines.append('\nKontrola heartbeatu źródła jest wyłączona; status dotyczy odczytu Modbus.')
     lines.append('\n' + config['lemmy']['update_note'])
     for slug, title in CHARTS:
         lines.extend(['', f'### {title}', '', f"![{title} — bieżący miesiąc]({images[slug]['url']})"])

settings.py

@@ -32,6 +32,31 @@ def validate(config):
                 raise ValueError(name + '.' + key + ': expected finite number')
         if s['multiplier'] == 0 or s['min'] > s['max']:
             raise ValueError(name + ': invalid multiplier or range')
+    recovery = config.get('recovery', {'attempts': 5, 'delay_seconds': 60})
+    if type(recovery['attempts']) is not int or not 0 <= recovery['attempts'] <= 10:
+        raise ValueError('recovery.attempts: expected 0..10')
+    if type(recovery['delay_seconds']) not in (int, float) or not 1 <= recovery['delay_seconds'] <= 600:
+        raise ValueError('recovery.delay_seconds: expected 1..600')
+    h = config.get('heartbeat', {'enabled': False})
+    if type(h['enabled']) is not bool:
+        raise ValueError('heartbeat.enabled: expected boolean')
+    if h['enabled']:
+        if type(h['address']) is not int or not 0 <= h['address'] <= 65535:
+            raise ValueError('heartbeat.address: expected zero-based register address')
+        for key, lo, hi in [('timeout_seconds', 1, 600), ('poll_interval_seconds', .1, 60)]:
+            if type(h[key]) not in (int, float) or not lo <= h[key] <= hi:
+                raise ValueError('heartbeat.' + key + ': invalid value')
+        if h['poll_interval_seconds'] >= h['timeout_seconds']:
+            raise ValueError('heartbeat: polling interval must be shorter than timeout')
+        checks = h.get('checks', [])
+        if not isinstance(checks, list) or len(checks) > 16:
+            raise ValueError('heartbeat.checks: expected up to 16 checks')
+        for check in checks:
+            if type(check['address']) is not int or not 0 <= check['address'] <= 65535:
+                raise ValueError('heartbeat.checks.address: invalid address')
+            allowed = check['allowed_values']
+            if not isinstance(allowed, list) or not allowed or any(type(v) is not int or not 0 <= v <= 65535 for v in allowed):
+                raise ValueError('heartbeat.checks.allowed_values: expected unsigned register values')
     l = config['lemmy']
     if type(l['enabled']) is not bool:
         raise ValueError('lemmy.enabled: expected boolean')

test_freshness.py

@@ -0,0 +1,70 @@
+import unittest
+import itertools
+from unittest.mock import patch
+from datetime import datetime, timezone
+import meteo
+import settings
+from test_support import config
+from publish_lemmy import render_section
+
+class FreshnessTests(unittest.TestCase):
+    def setUp(self):
+        self.c = config()
+        self.c['heartbeat'].update(enabled=True, address=100, timeout_seconds=10, poll_interval_seconds=1)
+
+    def test_changed_and_wrapped_counter(self):
+        for values in [(5, 6), (65535, 0)]:
+            with patch('meteo.read_registers', side_effect=[(v,) for v in values]), patch('meteo.time.sleep'):
+                self.assertTrue(meteo.verify_freshness(self.c))
+
+    def test_stuck_counter(self):
+        with patch('meteo.time.monotonic', side_effect=itertools.count(0, 2)), patch('meteo.time.sleep'), patch('meteo.read_registers', return_value=(7,)):
+            with self.assertRaises(meteo.FreshnessError):
+                meteo.verify_freshness(self.c)
+
+    def test_diagnostic_guard(self):
+        self.c['heartbeat']['checks'] = [{'address': 101, 'allowed_values': [1]}]
+        with patch('meteo.read_registers', return_value=(0,)):
+            with self.assertRaises(meteo.FreshnessError):
+                meteo.verify_freshness(self.c)
+
+    def test_disabled_needs_no_reads(self):
+        self.c.pop('heartbeat')
+        with patch('meteo.read_registers') as reader:
+            self.assertFalse(meteo.verify_freshness(self.c))
+            reader.assert_not_called()
+
+    def test_failure_does_not_read_measurements(self):
+        with patch('meteo.verify_freshness', side_effect=meteo.FreshnessError('stale')), patch('meteo.read_registers') as reader:
+            self.assertIsNone(meteo.collect_fast(self.c)[1])
+            reader.assert_not_called()
+
+    def test_recovery_stops_after_success(self):
+        bad = ('time', None, None, 3, 'timeout')
+        good = ('time', {'temperature': 1}, {}, 1, None)
+        with patch('meteo.collect_fast', side_effect=[bad, good]) as collect, patch('meteo.time.sleep'):
+            result = meteo.collect(self.c)
+            self.assertEqual(result[3], 4)
+            self.assertEqual(collect.call_count, 2)
+
+    def test_recovery_is_bounded(self):
+        self.c['recovery']['attempts'] = 2
+        with patch('meteo.collect_fast', return_value=('time', None, None, 1, 'timeout')) as collect, patch('meteo.time.sleep'):
+            self.assertEqual(meteo.collect(self.c)[3], 3)
+            self.assertEqual(collect.call_count, 3)
+
+    def test_configuration_validation(self):
+        settings.validate(self.c)
+        for address in [None, -1, 65536, True]:
+            self.c['heartbeat']['address'] = address
+            with self.assertRaises(ValueError):
+                settings.validate(self.c)
+
+    def test_no_green_without_verified_heartbeat(self):
+        status = {'status': 'ok', 'checked_at': '2026-10-19T12:00:00+00:00',
+                  'values': {'temperature': 8, 'humidity': 72, 'pressure': 1032}}
+        images = {name: {'url': '/fixture.png'} for name in ('temperatura', 'wilgotnosc', 'cisnienie')}
+        now = datetime(2026, 10, 19, 12, 1, tzinfo=timezone.utc)
+        self.assertNotIn('🟢', render_section(status, images, now, self.c))
+        status['freshness_verified'] = True
+        self.assertIn('🟢', render_section(status, images, now, self.c))

test_meteo.py

@@ -65,7 +65,7 @@ class CollectorTests(unittest.TestCase):
     def test_failure_retries_and_preserves_last_good(self):
         with patch("meteo.read_registers", side_effect=TimeoutError("test timeout")) as reader:
             with patch("meteo.time.sleep"):
-                checked_at, values, raw, attempts, error = meteo.collect(CONFIG)
+                checked_at, values, raw, attempts, error = meteo.collect_fast(CONFIG)
         self.assertEqual(reader.call_count, 3)
         self.assertIsNone(values)
         with tempfile.TemporaryDirectory(dir=meteo.BASE) as temporary: