diff --git a/README.md b/README.md index eb34061..00f6662 100644 --- a/README.md +++ b/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 -interval and divided by 12 to obtain Wh. Self-consumption is the smaller of production -and consumption; purchase/export are their nonnegative differences. These are chart -estimates, rather than the former Query API's measured Wh channels. The original -paired power values and conversion method are retained in each recovered record. +language. Paired power values are stored directly in the common watt readings. +Grid import/export are their nonnegative differences. As in `solar`, multiplying +each five-minute power value by `5/60` yields estimated Wh. The compatibility +`history_interval` field retains those estimates and the conversion method; +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 -uses directly without integrating them as live power. Install the corresponding -`solar` receiver update before enabling cloud recovery. +Recovery stores stable UUIDs and common power readings, retaining `history_interval` +as provenance and compatibility metadata. A future solar-side importer can put both +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 diff --git a/solar_relay/cli.py b/solar_relay/cli.py index b419358..0de0030 100644 --- a/solar_relay/cli.py +++ b/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") diff --git a/solar_relay/history.py b/solar_relay/history.py index 26d7fe4..ebfde3c 100644 --- a/solar_relay/history.py +++ b/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(): diff --git a/solar_relay/storage.py b/solar_relay/storage.py index 3810015..b327c7d 100644 --- a/solar_relay/storage.py +++ b/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 = [] diff --git a/solar_relay/web/index.html b/solar_relay/web/index.html index 3b1a033..3235cc3 100644 --- a/solar_relay/web/index.html +++ b/solar_relay/web/index.html @@ -4,15 +4,15 @@ - Solar relay · Collection status - + Solar relay +
-
SOLAR RELAYRPI OFFICE
+
SOLAR RELAYRaspberry Pi · Office
Connecting
diff --git a/tests/test_history.py b/tests/test_history.py index 22add9e..c3f3815 100644 --- a/tests/test_history.py +++ b/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)]: