feat: correlate DTEK planned windows with Shelly outage timing safely

docs/DTEK_CORRELATION.md

@@ -0,0 +1,30 @@
+# DTEK vs Shelly timing correlation (offline-only draft)
+
+This module compares **when** the meter became unavailable (or a
+separately confirmed electrical event happened) with the published
+DTEK planned schedule. It never concludes **why** power was lost.
+
+Outputs: start_within_window, partial_overlap,
+outside_published_windows, and unknown.
+The boolean cause_verified is always False.
+
+Inputs require an explicit Europe/Kyiv schedule date and validated
+half-hour outage windows. Missing provider data means unknown,
+whereas an explicitly published complete day without outage
+windows may legitimately yield outside_published_windows.
+
+An ongoing event needs an explicit observation end timestamp to avoid
+unbounded intervals or quietly assuming the current time. Historical
+events cannot be matched against a graph from a different day.
+Malformed or overlapping provider periods, invalid timestamps and
+ambiguous fall-back DST half-hours fail closed to unknown.
+
+**Important:** Shelly unreachable does not prove mains power is off.
+Router, LAN, Wi-Fi and hardware faults can be responsible. Presentation
+must describe such events as communication loss unless additional
+electrical evidence is available.
+
+This module has no IO, network, Telegram, SQLite or device operations.
+It is a library candidate only. Future acceptance requires dated
+schedule snapshots, source provenance, real device event checks and
+operator-approved Telegram copy.

dtek_correlation.py

@@ -0,0 +1,110 @@
+"""Compare Shelly outage timing with published DTEK windows, not causes."""
+
+from __future__ import annotations
+
+import math
+import re
+from dataclasses import dataclass
+from datetime import date, datetime, time, timedelta
+from zoneinfo import ZoneInfo
+
+KYIV = ZoneInfo("Europe/Kyiv")
+HALF_HOUR = re.compile(r"^(?:[01][0-9]|2[0-3]):(?:00|30)$")
+
+
+@dataclass(frozen=True)
+class TimeCorrelation:
+    """A timeline comparison with no assertion of outage cause."""
+
+    status: str
+    matching_window: tuple[str, str] | None = None
+    overlap_seconds: int = 0
+    cause_verified: bool = False
+
+
+def clock_stamp(day: date, clock: str) -> float:
+    """Convert DTEK local half-hours, rejecting DST-ambiguous times."""
+
+    if clock == "24:00":
+        day += timedelta(days=1)
+        clock = "00:00"
+    elif not HALF_HOUR.fullmatch(clock):
+        raise ValueError("invalid DTEK clock value")
+    hour, minute = map(int, clock.split(":"))
+    local = datetime.combine(day, time(hour, minute), tzinfo=KYIV)
+    if local.utcoffset() != local.replace(fold=1).utcoffset():
+        raise ValueError("DTEK time is ambiguous on DST transition")
+    return local.timestamp()
+
+
+def calendar_windows(
+    day: date,
+    periods: list[tuple[str, str]],
+) -> list[tuple[float, float, tuple[str, str]]]:
+    """Check interval order, overlap and valid day bounds."""
+
+    if len(periods) > 48:
+        raise ValueError("too many DTEK intervals")
+    found = []
+    previous_end = None
+    for start, end in periods:
+        begin_stamp = clock_stamp(day, start)
+        end_stamp = clock_stamp(day, end)
+        if end_stamp <= begin_stamp:
+            raise ValueError("invalid DTEK interval order")
+        if previous_end is not None and begin_stamp < previous_end:
+            raise ValueError("overlapping DTEK intervals")
+        previous_end = end_stamp
+        found.append((begin_stamp, end_stamp, (start, end)))
+    return found
+
+
+def valid_event(
+    started: float,
+    ended: float | None,
+    observed_until: float | None,
+) -> tuple[float, float]:
+    """Validate real event times and bound an open observation."""
+
+    begin = float(started)
+    stop = observed_until if ended is None else ended
+    if stop is None:
+        raise ValueError("open event needs an observation boundary")
+    finish = float(stop)
+    if not all(math.isfinite(value) for value in (begin, finish)):
+        raise ValueError("nonfinite event timestamp")
+    if finish < begin:
+        raise ValueError("event ends before it starts")
+    return begin, finish
+
+
+def correlate_outage(
+    started: float,
+    ended: float | None,
+    schedule_day: str,
+    periods: list[tuple[str, str]] | None,
+    observed_until: float | None = None,
+) -> TimeCorrelation:
+    """Match timing only, with unknown for missing or invalid evidence."""
+
+    unknown = TimeCorrelation("unknown")
+    if periods is None:
+        return unknown
+    try:
+        day = date.fromisoformat(schedule_day)
+        begin, finish = valid_event(started, ended, observed_until)
+        if datetime.fromtimestamp(begin, KYIV).date() != day:
+            return unknown
+        windows = calendar_windows(day, periods)
+    except (ValueError, TypeError, OverflowError, OSError):
+        return unknown
+    best = (0, None)
+    for low, high, window in windows:
+        overlap = max(0, min(finish, high) - max(begin, low))
+        if low <= begin < high:
+            return TimeCorrelation("start_within_window", window, int(overlap))
+        if overlap > best[0]:
+            best = (overlap, window)
+    if best[1] is not None:
+        return TimeCorrelation("partial_overlap", best[1], int(best[0]))
+    return TimeCorrelation("outside_published_windows")

scripts/check-quality.py

@@ -34,6 +34,7 @@ def python_files(root: Path) -> list[Path]:
         root / "app.py",
         root / "advanced_features.py",
         root / "apartment_phases.py",
+        root / "dtek_correlation.py",
         root / "update_info.py",
         root / "telegram_i18n.py",
         root / "notification_features.py",

scripts/quality.sh

@@ -42,8 +42,8 @@ if [[ "$actual_version" != "$REQUIRED_RUFF_VERSION" ]]; then
 fi
 
 cd "$ROOT"
-ruff format --line-length 79 --check app.py advanced_features.py apartment_phases.py update_info.py telegram_i18n.py notification_features.py scripts tests
-ruff check --line-length 79 --select E,F,I,UP,SIM app.py advanced_features.py apartment_phases.py update_info.py telegram_i18n.py notification_features.py scripts tests
+ruff format --line-length 79 --check app.py advanced_features.py apartment_phases.py update_info.py telegram_i18n.py notification_features.py dtek_correlation.py scripts tests
+ruff check --line-length 79 --select E,F,I,UP,SIM app.py advanced_features.py apartment_phases.py update_info.py telegram_i18n.py notification_features.py dtek_correlation.py scripts tests
 "$PYTHON_BIN" scripts/check-quality.py
 
 echo "QUALITY GATE: PASS"

tests/test_dtek_correlation.py

@@ -0,0 +1,220 @@
+"""Deterministic timing checks for Shelly and DTEK schedule comparison."""
+
+from __future__ import annotations
+
+import math
+import unittest
+from datetime import datetime
+from zoneinfo import ZoneInfo
+
+import dtek_correlation as correlation
+
+KYIV = ZoneInfo("Europe/Kyiv")
+DAY = "2026-11-06"
+WINDOWS = [("14:00", "18:00")]
+
+
+def epoch(clock: str, day: str = DAY) -> float:
+    """Construct a real Europe/Kyiv timestamp without host TZ ambiguity."""
+
+    return (
+        datetime.fromisoformat(f"{day}T{clock}")
+        .replace(tzinfo=KYIV)
+        .timestamp()
+    )
+
+
+class CorrelationTests(unittest.TestCase):
+    """Validate fail-closed semantics without devices or HTTP requests."""
+
+    def result(self, start: str, stop: str, windows=None):
+        """Run the standard one-day comparison with fixed clocks."""
+
+        return correlation.correlate_outage(
+            epoch(start),
+            epoch(stop),
+            DAY,
+            WINDOWS if windows is None else windows,
+        )
+
+    def test_outage_starts_inside_window(self):
+        match = self.result("14:30", "15:00")
+        self.assertEqual(match.status, "start_within_window")
+        self.assertEqual(match.matching_window, WINDOWS[0])
+        self.assertEqual(match.overlap_seconds, 1800)
+        self.assertFalse(match.cause_verified)
+
+    def test_outage_starts_at_window_open(self):
+        self.assertEqual(
+            self.result("14:00", "14:30").status,
+            "start_within_window",
+        )
+
+    def test_outage_starts_at_window_close(self):
+        self.assertEqual(
+            self.result("18:00", "18:30").status,
+            "outside_published_windows",
+        )
+
+    def test_early_start_partially_overlaps(self):
+        match = self.result("13:50", "14:10")
+        self.assertEqual(match.status, "partial_overlap")
+        self.assertEqual(match.overlap_seconds, 600)
+        self.assertFalse(match.cause_verified)
+
+    def test_early_start_ends_at_boundary(self):
+        self.assertEqual(
+            self.result("13:45", "14:00").status,
+            "outside_published_windows",
+        )
+
+    def test_no_overlap_after_schedule(self):
+        self.assertEqual(
+            self.result("19:00", "20:00").status,
+            "outside_published_windows",
+        )
+
+    def test_explicit_empty_plan_means_outside(self):
+        self.assertEqual(
+            self.result("15:00", "16:00", []).status,
+            "outside_published_windows",
+        )
+
+    def test_missing_schedule_is_unknown(self):
+        self.assertEqual(
+            correlation.correlate_outage(
+                epoch("15:00"), epoch("16:00"), DAY, None
+            ).status,
+            "unknown",
+        )
+
+    def test_wrong_calendar_day_is_unknown(self):
+        self.assertEqual(
+            correlation.correlate_outage(
+                epoch("15:00"), epoch("16:00"), "2026-11-05", WINDOWS
+            ).status,
+            "unknown",
+        )
+
+    def test_bad_date_is_unknown(self):
+        self.assertEqual(
+            correlation.correlate_outage(
+                epoch("15:00"), epoch("16:00"), "bad", WINDOWS
+            ).status,
+            "unknown",
+        )
+
+    def test_open_event_without_boundary_unknown(self):
+        self.assertEqual(
+            correlation.correlate_outage(
+                epoch("15:00"), None, DAY, WINDOWS
+            ).status,
+            "unknown",
+        )
+
+    def test_open_event_with_observation_boundary(self):
+        match = correlation.correlate_outage(
+            epoch("15:00"),
+            None,
+            DAY,
+            WINDOWS,
+            observed_until=epoch("16:00"),
+        )
+        self.assertEqual(match.status, "start_within_window")
+        self.assertEqual(match.overlap_seconds, 3600)
+
+    def test_open_event_invalid_observation_end(self):
+        match = correlation.correlate_outage(
+            epoch("15:00"),
+            None,
+            DAY,
+            WINDOWS,
+            observed_until=epoch("14:00"),
+        )
+        self.assertEqual(match.status, "unknown")
+
+    def test_event_end_precedes_start(self):
+        self.assertEqual(self.result("15:00", "14:00").status, "unknown")
+
+    def test_nan_start_is_unknown(self):
+        self.assertEqual(
+            correlation.correlate_outage(
+                math.nan, epoch("16:00"), DAY, WINDOWS
+            ).status,
+            "unknown",
+        )
+
+    def test_infinite_end_is_unknown(self):
+        self.assertEqual(
+            correlation.correlate_outage(
+                epoch("15:00"), math.inf, DAY, WINDOWS
+            ).status,
+            "unknown",
+        )
+
+    def test_bad_hours_unknown(self):
+        self.assertEqual(
+            self.result("15:00", "16:00", [("25:00", "26:00")]).status,
+            "unknown",
+        )
+
+    def test_reversed_window_unknown(self):
+        self.assertEqual(
+            self.result("15:00", "16:00", [("18:00", "14:00")]).status,
+            "unknown",
+        )
+
+    def test_overlapping_windows_unknown(self):
+        invalid = [("14:00", "16:00"), ("15:30", "18:00")]
+        self.assertEqual(
+            self.result("15:00", "16:00", invalid).status,
+            "unknown",
+        )
+
+    def test_unsorted_windows_unknown(self):
+        invalid = [("19:00", "20:00"), ("14:00", "16:00")]
+        self.assertEqual(
+            self.result("15:00", "16:00", invalid).status,
+            "unknown",
+        )
+
+    def test_midnight_end_is_next_day(self):
+        match = self.result("23:45", "23:59", [("23:30", "24:00")])
+        self.assertEqual(match.status, "start_within_window")
+        self.assertEqual(match.overlap_seconds, 840)
+
+    def test_midnight_crossing_observation(self):
+        match = correlation.correlate_outage(
+            epoch("23:45"),
+            epoch("00:10", "2026-11-07"),
+            DAY,
+            [("23:30", "24:00")],
+        )
+        self.assertEqual(match.status, "start_within_window")
+        self.assertEqual(match.overlap_seconds, 900)
+
+    def test_fall_dst_ambiguous_hour_is_unknown(self):
+        start = epoch("03:15", "2026-10-25")
+        end = start + 1200
+        match = correlation.correlate_outage(
+            start, end, "2026-10-25", [("03:00", "04:00")]
+        )
+        self.assertEqual(match.status, "unknown")
+
+    def test_spring_dst_missing_hour_is_unknown(self):
+        start = epoch("04:15", "2026-03-29")
+        match = correlation.correlate_outage(
+            start, start + 60, "2026-03-29", [("03:00", "05:00")]
+        )
+        self.assertEqual(match.status, "unknown")
+
+    def test_more_than_48_windows_is_unknown(self):
+        invalid = [("00:00", "00:30")] * 49
+        self.assertEqual(
+            self.result("15:00", "16:00", invalid).status,
+            "unknown",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()