Migrate records to solar web format

README.md

@@ -96,6 +96,25 @@ atomically replaces the daily file, and fsyncs the directory. File permissions a
 uploader, status request or CLI replay from overwriting another operation. HTTP
 requests happen outside the archive lock.
 
+Live and recovered records share the same Parquet table and watt columns:
+`production_ac_w`, `consumption_w`, `grid_w`, `grid_import_w`, and `grid_export_w`.
+Recovered chart values are stored directly in those fields; `production_w` (PV DC),
+lifetime counters, and boiler state remain unknown. `record_kind` and the original
+chart data distinguish recovered values from live snapshots. The older
+`history_interval.energy_wh` metadata is retained for compatibility. Existing archives
+are normalized on `run` startup, preserving IDs, recording times and upload acknowledgements.
+To normalize an archive before exporting without starting collection:
+
+```bash
+.venv/bin/python -m solar_relay --config config.toml normalize-history
+```
+
+These common power readings match the sibling `solar` application's existing
+model: its `stats` table stores timestamped watts for Production and Consumption,
+then applies `5/60` only when calculating Wh totals. `production_ac_w` maps to
+Production and `consumption_w` maps to Consumption. The solar-side relay importer
+is separate work; this relay change prepares both record sources for ingestion.
+
 The upload request is a JSON object with `schema_version: 1` and a `samples` array,
 authenticated with `upload.token` from private `config.toml`, or the environment
 variable named by `upload.token_env` when that setting is empty. The website must
@@ -108,8 +127,10 @@ commit its data before responding:
 Only explicitly acknowledged IDs receive an `uploaded_at` timestamp. A plain HTTP
 200, malformed response, timeout or unknown ID never clears the queue. Partial
 acknowledgements mark only the accepted records; remaining records retry with
-exponential backoff. Delivery is **at least once**: the website deduplicates stable
+exponential backoff. Delivery is **at least once**: a receiver must deduplicate stable
 UUIDs when a response is lost, so retries and replays do not double-count energy.
+The sibling solar app does not yet provide an HTTP receiver; leave uploads
+disabled until a compatible receiver is available.
 
 `collector.retention_days` defaults to 30. All retention cutoffs are midnight in
 `collector.timezone`. For example, on November 2 with 30 days retained, the cutoff
@@ -202,11 +223,11 @@ ignored; `collector.retention_days` controls the automatic recovery window.
 
 Recovery pairs the `FromGen` Production series from the consumption view with the
 `ToConsumer` Consumption series from the production view, regardless of the display
-language. As in `solar`, each five-minute W value is treated as constant over that
+language. Paired power values are stored directly in the common watt readings.
-interval and divided by 12 to obtain Wh. Self-consumption is the smaller of production
+Grid import/export are their nonnegative differences. As in `solar`, multiplying
-and consumption; purchase/export are their nonnegative differences. These are chart
+each five-minute power value by `5/60` yields estimated Wh. The compatibility
-estimates, rather than the former Query API's measured Wh channels. The original
+`history_interval` field retains those estimates and the conversion method;
-paired power values and conversion method are retained in each recovered record.
+the original chart watts remain available without converting Wh back to W.
 Null or unpaired readings stay missing and are tried again later.
 
 Login uses HTTPS and in-memory cookies. Credentials are submitted only to the Fronius
@@ -233,9 +254,10 @@ per invocation and can query dates older than the automatic retention window. Un
 the whole retention window, even when longer than 30 days. Only aligned five-minute
 chart samples in W are accepted;
 longer aggregates and unexpected units are rejected.
-Recovery stores stable UUIDs and `history_interval` records, which the main website
+Recovery stores stable UUIDs and common power readings, retaining `history_interval`
-uses directly without integrating them as live power. Install the corresponding
+as provenance and compatibility metadata. A future solar-side importer can put both
-`solar` receiver update before enabling cloud recovery.
+live and recovered production/consumption in the existing `stats` watt table and
+use its normal five-minute Wh conversion exactly once.
 
 **Boiler switch history and cumulative meter readings are not recovered.** Historical
 energy intervals leave those sources unknown, and the status page shows partial

solar_relay/cli.py

@@ -40,6 +40,9 @@ def next_poll_at(now, interval):
 
 
 def _run(config, store):
+    normalized = store.normalize_history()
+    if normalized:
+        LOG.info("Normalized %d recovered records to common power readings", normalized)
     stop = threading.Event()
     previous = {}
     uploader = Uploader(config.upload, store) if config.upload.url else None
@@ -103,6 +106,7 @@ def _parser():
     commands.add_parser("upload", help="Upload the pending queue, stopping on the first failure")
     commands.add_parser("status", help="Show local archive/queue status without contacting the inverter")
     commands.add_parser("prune", help="Delete expired whole local days after acknowledgement retention")
+    commands.add_parser("normalize-history", help="Backfill common power readings in previously recovered records")
     for name, help_text in [
         ("export", "Export archived samples as JSON Lines, including already uploaded samples"),
         ("replay", "Requeue archived samples for upload, preserving their IDs"),
@@ -137,6 +141,8 @@ def main(argv=None):
             print(json.dumps(store.status(), indent=2))
         elif args.command == "prune":
             print(json.dumps({"deleted": store.prune()}))
+        elif args.command == "normalize-history":
+            print(json.dumps({"normalized": store.normalize_history()}))
         elif args.command == "recover":
             if not config.history.enabled:
                 raise ValueError("Enable history and configure Solar.web account credentials before recovery")

solar_relay/history.py

@@ -23,6 +23,40 @@ CHANNELS = {
 }
 
 
+def normalize_history_sample(sample):
+    """Expose recovered power in the same fields as live samples, including old archives."""
+    if "history_interval" not in sample:
+        return sample
+    interval = sample["history_interval"]
+    if interval["duration_seconds"] != 300:
+        raise ValueError("Historical record is not a five-minute interval")
+    energy = interval["energy_wh"]
+    if any(isinstance(energy[field], bool) or not isinstance(energy[field], (int, float))
+           or not math.isfinite(energy[field]) or not 0 <= energy[field] <= 1e18 for field in CHANNELS):
+        raise ValueError("Historical energy is missing or invalid")
+    power = {field: energy[field] * 12 for field in CHANNELS}
+    chart = sample.get("raw", {}).get("solarweb_history", {}).get("chart_power_w")
+    if chart is not None:
+        # Retain the chart's original watts instead of a W -> Wh -> W round trip.
+        generated, consumed = chart["production"], chart["consumption"]
+        if any(isinstance(value, bool) or not isinstance(value, (int, float))
+               or not math.isfinite(value) or value < 0 for value in (generated, consumed)):
+            raise ValueError("Historical chart power is missing or invalid")
+        if not all(math.isclose(value, power[field]) for field, value in
+                   (("production", generated), ("consumption", consumed))):
+            raise ValueError("Historical chart power disagrees with interval energy")
+        power = {"production": generated, "consumption": consumed,
+                 "bought": max(0, consumed - generated), "sold": max(0, generated - consumed)}
+    return {
+        **sample, "poll_interval_seconds": 300,
+        "readings": {**sample["readings"],
+                     "production_ac_w": power["production"], "consumption_w": power["consumption"],
+                     "grid_w": power["bought"] - power["sold"],
+                     "grid_import_w": power["bought"], "grid_export_w": power["sold"]},
+        "source_timestamps": {**sample.get("source_timestamps", {}), "powerflow": sample["observed_at"]},
+    }
+
+
 def history_sample(entry, source_id, system_id):
     stamp = milliseconds(entry["logDateTime"])
     if stamp % 300_000 or entry.get("logDuration") != 300:
@@ -39,7 +73,7 @@ def history_sample(entry, source_id, system_id):
         channels[name] = value
     energy = {field: channels[name] for field, name in CHANNELS.items()}
     when = utc_text(datetime.fromtimestamp(stamp / 1000, timezone.utc))
-    return {
+    return normalize_history_sample({
         "schema_version": 1,
         # Preserve IDs of intervals already recovered with the former Query API.
         "id": str(uuid5(NAMESPACE_URL, f"https://swqapi.solarweb.com/{system_id}/{source_id}/{when}")),
@@ -53,7 +87,7 @@ def history_sample(entry, source_id, system_id):
                    "boiler": "Historical boiler switch state is unavailable"},
         "raw": {"solarweb_history": entry},
         "history_interval": {"provider": "fronius-solarweb", "start": when, "duration_seconds": 300, "energy_wh": energy},
-    }
+    })
 
 
 def chart_entries(production, consumption):
@@ -106,12 +140,15 @@ class HistoryRecovery:
         missing = set(range(lower, upper, 300)) - present
         LOG.info("Checking %d historical energy intervals; %d are missing", (upper - lower) // 300, len(missing))
         # Chart dates use the installation's local timezone, including 23/25-hour DST days.
-        days = sorted({datetime.fromtimestamp(epoch, self.zone).date() for epoch in missing})
+        days = {}
+        for epoch in sorted(missing):
+            local = datetime.fromtimestamp(epoch, self.zone)
+            days.setdefault(local.date(), []).append(local.strftime("%H:%M:%S"))
         recovered = 0
-        for day in days:
+        for day in sorted(days):
             if stop and stop.is_set():
                 break
-            LOG.info("Recovering Solar.web energy history for %s", day)
+            LOG.info("Recovering Solar.web energy history for %s %s", day, ", ".join(days[day]))
             try:
                 production = self.client.get_chart(self.config.pv_system_id, "production", day)
                 if stop and stop.is_set():

solar_relay/storage.py

@@ -20,6 +20,7 @@ import pyarrow.parquet as pq
 
 from .config import retention_cutoff, utc_text
 from .coverage import daily_coverage, milliseconds, sample_coverage
+from .history import normalize_history_sample
 
 NUMBERS = (
     "production_w", "production_ac_w", "consumption_w", "grid_w",
@@ -136,11 +137,32 @@ class Store:
             rows = self._read(path)
             for existing in rows:
                 if existing["id"] == row["id"]:
-                    if existing["payload"] != row["payload"]:
+                    if normalize_history_sample(json.loads(existing["payload"])) != normalize_history_sample(sample):
                         raise StoreError("Sample ID already contains different data")
                     return
             self._write(path, [*rows, row])
 
+    def normalize_history(self):
+        """Backfill common power columns and queued payloads without changing IDs or acknowledgements."""
+        changed = 0
+        with self._locked(write=True):
+            for path in self._files():
+                rows = self._read(path)
+                dirty = False
+                for index, row in enumerate(rows):
+                    if row["record_kind"] != "history":
+                        continue
+                    sample = json.loads(row["payload"])
+                    normalized = normalize_history_sample(sample)
+                    if normalized == sample:
+                        continue
+                    rows[index] = {**self._row(normalized, row["uploaded_at"]), "recorded_ns": row["recorded_ns"]}
+                    changed += 1
+                    dirty = True
+                if dirty:
+                    self._write(path, rows)
+        return changed
+
     def pending(self, limit):
         with self._locked():
             rows = []

solar_relay/web/index.html

@@ -4,15 +4,15 @@
   <meta charset="utf-8">
   <meta name="viewport" content="width=device-width, initial-scale=1">
   <meta name="color-scheme" content="dark">
-  <title>Solar relay · Collection status</title>
+  <title>Solar relay</title>
-  <link rel="icon" href="data:image/svg+xml,%3Csvg xmlns='http://www.w3.org/2000/svg' viewBox='0 0 64 64'%3E%3Crect width='64' height='64' rx='16' fill='%23171d20'/%3E%3Ccircle cx='32' cy='32' r='15' fill='%23e8ba70'/%3E%3C/svg%3E">
+  <link rel="icon" href="data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHZpZXdCb3g9IjAgMCA2NCA2NCI+CiAgPHRpdGxlPlNvbGFyIFJlbGF5PC90aXRsZT4KICA8cmVjdCB3aWR0aD0iNjQiIGhlaWdodD0iNjQiIHJ4PSIxNSIgZmlsbD0iIzBBMTMzMCIvPgogIDxjaXJjbGUgY3g9IjI4IiBjeT0iMjkiIHI9IjEzIiBmaWxsPSIjRkZDQTQ1Ii8+CiAgPHBhdGggZD0iTTExIDQzQzIwIDM0IDM4IDMwIDUzIDI5IiBmaWxsPSJub25lIiBzdHJva2U9IiNGRkY3RTgiIHN0cm9rZS13aWR0aD0iNSIgc3Ryb2tlLWxpbmVjYXA9InJvdW5kIi8+CiAgPHBhdGggZD0iTTQ2IDIzTDU0IDI5TDQ2IDM1IiBmaWxsPSJub25lIiBzdHJva2U9IiNGRkY3RTgiIHN0cm9rZS13aWR0aD0iNCIgc3Ryb2tlLWxpbmVjYXA9InJvdW5kIiBzdHJva2UtbGluZWpvaW49InJvdW5kIi8+Cjwvc3ZnPg==">
   <link rel="stylesheet" href="assets/status.css">
   <script src="assets/status.js" defer></script>
 </head>
 <body>
   <main>
     <header class="masthead">
-      <div class="brand"><span class="sun" aria-hidden="true">☀</span><span>SOLAR RELAY<small>RPI OFFICE</small></span></div>
+      <div class="brand"><span class="sun" aria-hidden="true">☀️</span><span>SOLAR RELAY<small>Raspberry Pi · Office</small></span></div>
       <div class="header-actions"><span id="overall" class="badge neutral" role="status">Connecting</span><button id="refresh" class="refresh">↻ <span>Refresh</span></button></div>
     </header>
     <section class="intro">

tests/test_history.py

@@ -79,6 +79,40 @@ class HistoryTests(unittest.TestCase):
         row = pq.ParquetFile(next(self.store.path.glob("date=*/samples.parquet"))).read().to_pylist()[0]
         self.assertEqual(row["record_kind"], "history")
         self.assertEqual(row["interval_energy_wh"]["production"], 100)
+        self.assertEqual(row["production_ac_w"], 1200)
+        self.assertEqual(row["consumption_w"], 1440)
+        self.assertEqual(row["grid_import_w"], 240)
+        self.assertEqual(row["grid_export_w"], 0)
+        self.assertEqual(day["slots"][120:123], [1, 1, 1])
+
+    def test_old_archive_normalization_preserves_delivery_state_and_live_records(self):
+        old = history_sample(entry(self.when), self.source, SYSTEM)
+        old["readings"] = dict.fromkeys(old["readings"])
+        old["source_timestamps"] = {}
+        old.pop("poll_interval_seconds")
+        live = observed(self.when + timedelta(minutes=5))
+        self.store.add(old)
+        self.store.add(live)
+        self.store.acknowledge([old["id"]], now=self.when + timedelta(hours=1))
+        path = next(self.store.path.glob("date=*/samples.parquet"))
+        before = {row["id"]: row for row in pq.ParquetFile(path).read().to_pylist()}
+
+        self.assertEqual(self.store.normalize_history(), 1)
+        self.assertEqual(self.store.normalize_history(), 0)
+        after = {row["id"]: row for row in pq.ParquetFile(path).read().to_pylist()}
+        self.assertEqual(after[live["id"]], before[live["id"]])
+        row = after[old["id"]]
+        for field in ("id", "observed_at", "recorded_ns", "uploaded_at"):
+            self.assertEqual(row[field], before[old["id"]][field])
+        self.assertEqual(row["production_ac_w"], 1200)
+        self.assertEqual(row["consumption_w"], 1440)
+        self.assertIsNone(row["production_w"])
+        self.assertIsNone(row["boiler_enabled"])
+        self.assertEqual(self.store.pending(100), [live])
+        self.assertEqual(list(self.store.records())[0]["source_timestamps"]["powerflow"], utc_text(self.when))
+        # Replaying the original legacy payload still identifies the same record.
+        self.store.add(old)
+        self.assertEqual(self.store.status()["samples"], 2)
 
     def test_partial_cloud_records_remain_missing_for_retry(self):
         stamp = int((self.when + timedelta(minutes=5)).timestamp() * 1000)
@@ -233,16 +267,21 @@ class HistoryTests(unittest.TestCase):
         self.assertEqual(client.call_count, 1)
 
     def test_chart_conversion_matches_solar_and_retains_provenance(self):
-        for generated, consumed in [(1200, 1440), (2400, 1200), (0, 0)]:
+        for generated, consumed in [(1200, 1440), (2400, 1200), (0, 0), (1234.56789, 3456.78901)]:
             with self.subTest(generated=generated, consumed=consumed):
                 stamp = int(self.when.timestamp() * 1000)
                 record = next(chart_entries({stamp: generated}, {stamp: consumed}))
-                values = history_sample(record, self.source, SYSTEM)["history_interval"]["energy_wh"]
+                sample = history_sample(record, self.source, SYSTEM)
+                values = sample["history_interval"]["energy_wh"]
                 self.assertEqual(values, {"production": generated / 12, "consumption": consumed / 12,
                                           "self_consumption": min(generated, consumed) / 12,
                                           "bought": max(0, consumed - generated) / 12,
                                           "sold": max(0, generated - consumed) / 12})
                 self.assertEqual(record["energy_method"], "constant_power_over_five_minutes")
+                self.assertEqual(sample["readings"]["production_ac_w"], generated)
+                self.assertEqual(sample["readings"]["consumption_w"], consumed)
+                self.assertEqual(sample["readings"]["grid_w"], consumed - generated)
+                self.assertIsNone(sample["readings"]["production_w"])
 
     def test_dst_days_recover_every_actual_five_minute_slot(self):
         for day, expected in [("2026-03-29", 276), ("2026-10-25", 300)]: