From 9e0230731b39c85806af6a7d1cf71f2de4710cad Mon Sep 17 00:00:00 2001 From: Franzz Date: Mon, 2 Nov 2026 23:59:33 +0100 Subject: [PATCH] Remove legacy code for format conversion --- README.md | 33 ++++-------- config.example.toml | 2 +- solar_relay/cli.py | 8 +-- solar_relay/config.py | 8 +-- solar_relay/coverage.py | 3 -- solar_relay/history.py | 113 +++++++++------------------------------- solar_relay/storage.py | 36 +++---------- tests/test_history.py | 93 ++++++++++++--------------------- tests/test_parquet.py | 30 ++++++++++- tests/test_relay.py | 15 +++--- 10 files changed, 118 insertions(+), 223 deletions(-) diff --git a/README.md b/README.md index 00f6662..2a77547 100644 --- a/README.md +++ b/README.md @@ -100,14 +100,9 @@ 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 -``` +chart data distinguish recovered values from live snapshots. Both sources use the +same validation and completeness checks. The relay stores power directly; energy +totals are calculated by the receiving application. These common power readings match the sibling `solar` application's existing model: its `stats` table stores timestamped watts for Production and Consumption, @@ -177,7 +172,7 @@ for the local API's archive limitations. Optional recovery signs in to Solar.web with a real Fronius account and reads its private daily charts after the inverter uploads its backlog, using the same login flow and chart series as the sibling `solar` website. No cloud calls occur while -disabled. A public guest link and Query API subscription/keys are not needed. +disabled. A public guest link is not needed. Create a dedicated Solar.web / Fronius account and grant it permission to view this installation's metrics. Keep public sharing disabled. The account can have guest @@ -218,16 +213,14 @@ not prevent other dates from being recovered. Login and network failures stop th attempt and retry on the next scheduled check. Set `collector.timezone` to the installation's Solar.web timezone; local chart dates -include whole 23/25-hour DST days. The old `history.lookback_days` setting is accepted but -ignored; `collector.retention_days` controls the automatic recovery window. +include whole 23/25-hour DST days. `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. 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. +Grid import/export are their nonnegative differences. The relay keeps the original +chart watts without converting them to interval energy. 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 @@ -238,12 +231,6 @@ disk. An expired session is renewed once. Rejected credentials, MFA, CAPTCHA, consent, or a changed sign-in form stop that attempt with a redacted error. Complete any required interactive account setup in a browser. -When migrating from Query API recovery, replace `access_key_id_env` and -`access_key_value_env` with `username` and `password` in `[history]`, or use the -environment fallback above. Obsolete configuration keys are accepted and ignored -for migration. Restart the service after updating its configuration or environment. -Recovery requires the account credentials even if API keys remain set. - ```bash .venv/bin/python -m solar_relay --config config.toml recover \ --from 2026-10-29 --until 2026-10-31 @@ -254,8 +241,8 @@ 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 common power readings, retaining `history_interval` -as provenance and compatibility metadata. A future solar-side importer can put both +Recovery stores stable UUIDs and common power readings with their chart provenance. +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. diff --git a/config.example.toml b/config.example.toml index ec1f2ba..2d10912 100644 --- a/config.example.toml +++ b/config.example.toml @@ -13,7 +13,7 @@ request_timeout_seconds = 5 min_request_interval_seconds = 4 [upload] -url = "" # Set to https://your-solar-host/index.php?a=relay to enable uploads. +url = "" # Set a compatible HTTPS receiver URL to enable uploads. token = "" # Shared bearer token; enter the real value only in private config.toml. batch_size = 100 request_timeout_seconds = 10 diff --git a/solar_relay/cli.py b/solar_relay/cli.py index 0de0030..a1c8229 100644 --- a/solar_relay/cli.py +++ b/solar_relay/cli.py @@ -40,9 +40,6 @@ 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 @@ -106,11 +103,10 @@ 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"), - ("recover", "Retrieve missing energy intervals from the authenticated Solar.web API"), + ("recover", "Retrieve missing power samples from authenticated Solar.web charts"), ]: command = commands.add_parser(name, help=help_text) command.add_argument("--from", dest="from_date", help="Inclusive local date or ISO timestamp") @@ -141,8 +137,6 @@ 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/config.py b/solar_relay/config.py index ed18b4e..53568c7 100644 --- a/solar_relay/config.py +++ b/solar_relay/config.py @@ -9,8 +9,6 @@ import tomllib from urllib.parse import urlsplit from zoneinfo import ZoneInfo -LEGACY_HISTORY_SETTINGS = {"access_key_id_env", "access_key_value_env", "lookback_days"} - def utc_text(value: datetime) -> str: if value.tzinfo is None: @@ -131,8 +129,7 @@ def load_config(path: Path) -> Config: "inverter": set(InverterConfig.__dataclass_fields__), "upload": set(UploadConfig.__dataclass_fields__), "http": set(HttpConfig.__dataclass_fields__), - # Keep existing private configurations valid while using the full retention window. - "history": set(HistoryConfig.__dataclass_fields__) | LEGACY_HISTORY_SETTINGS, + "history": set(HistoryConfig.__dataclass_fields__), }.items(): if not isinstance(data.get(section, {}), dict): raise ValueError(f"{section} must be a table") @@ -193,8 +190,7 @@ def load_config(path: Path) -> Config: raise ValueError("http.base_path must be empty or a URL path without a trailing slash") if isinstance(http.history_days, bool) or not isinstance(http.history_days, int) or not 1 <= http.history_days <= 90: raise ValueError("http.history_days must be an integer between 1 and 90") - history = HistoryConfig(**{key: value for key, value in data.get("history", {}).items() - if key not in LEGACY_HISTORY_SETTINGS}) + history = HistoryConfig(**data.get("history", {})) if not isinstance(history.enabled, bool): raise ValueError("history.enabled must be true or false") if not isinstance(history.pv_system_id, str) or (history.pv_system_id and not re.fullmatch(r"[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}", history.pv_system_id)): diff --git a/solar_relay/coverage.py b/solar_relay/coverage.py index dbc2872..6ec31d1 100644 --- a/solar_relay/coverage.py +++ b/solar_relay/coverage.py @@ -24,9 +24,6 @@ def _finite(value): def sample_coverage(sample): observed = milliseconds(sample["observed_at"]) - if "history_interval" in sample: - # Cloud history has interval energy, but no cumulative meter or relay history. - return (sample["id"], sample["source_id"], observed, observed, 1, observed, 0, observed, 0) readings = sample.get("readings", {}) result = [sample["id"], sample["source_id"], observed] for source, fields in [ diff --git a/solar_relay/history.py b/solar_relay/history.py index ebfde3c..d632f91 100644 --- a/solar_relay/history.py +++ b/solar_relay/history.py @@ -1,8 +1,4 @@ -"""Optional signed-in Solar.web chart recovery of missing five-minute energy. - -GEN24's local Solar API exposes no archive. Use the same private account login, -chart series and five-minute W-to-Wh conversion as the sibling solar website. -""" +"""Recover missing five-minute power samples from signed-in Solar.web charts.""" from datetime import datetime, timedelta, timezone import logging @@ -17,91 +13,33 @@ from .http import TransportError from .solarweb import ChartDataError, SolarWebClient LOG = logging.getLogger(__name__) -CHANNELS = { - "production": "EnergyProductionTotal", "consumption": "EnergyConsumptionTotal", - "self_consumption": "EnergySelfConsumptionTotal", "bought": "EnergyPurchased", "sold": "EnergyFeedIn", -} -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: - raise ValueError("Historical record is not an aligned five-minute interval") - channels = {} - for channel in entry["channels"]: - name, value = channel.get("channelName"), channel.get("value") - if name not in CHANNELS.values(): - continue - if name in channels or channel.get("unit") != "Wh" or channel.get("channelType") != "Energy": - raise ValueError("Unexpected historical channel metadata") - if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or not 0 <= value <= 1e18: - raise ValueError("Historical energy is missing or invalid") - channels[name] = value - energy = {field: channels[name] for field, name in CHANNELS.items()} +def history_sample(stamp, generated, consumed, source_id, system_id): + """Store paired chart watts in the same readings as a live power snapshot.""" + if isinstance(stamp, bool) or not isinstance(stamp, int) or stamp % 300_000: + raise ValueError("Historical record is not an aligned five-minute sample") + if any(isinstance(value, bool) or not isinstance(value, (int, float)) + or not math.isfinite(value) or not 0 <= value <= 1e18 for value in (generated, consumed)): + raise ValueError("Historical chart power is missing or invalid") when = utc_text(datetime.fromtimestamp(stamp / 1000, timezone.utc)) - return normalize_history_sample({ + grid = consumed - generated + return { "schema_version": 1, - # Preserve IDs of intervals already recovered with the former Query API. + # This immutable UUID name is part of the record ID format, never a request URL. "id": str(uuid5(NAMESPACE_URL, f"https://swqapi.solarweb.com/{system_id}/{source_id}/{when}")), - "source_id": source_id, "observed_at": when, - "readings": dict.fromkeys([ - "production_w", "production_ac_w", "consumption_w", "grid_w", "grid_import_w", "grid_export_w", - "production_ac_total_wh", "grid_import_total_wh", "grid_export_total_wh", "boiler_enabled", "boiler_reason", - ]), - "source_timestamps": {}, - "errors": {"meter": "Cumulative meter readings are unavailable in recovered energy intervals", + "source_id": source_id, "observed_at": when, "poll_interval_seconds": 300, + "readings": { + "production_w": None, "production_ac_w": generated, "consumption_w": consumed, + "grid_w": grid, "grid_import_w": max(0, grid), "grid_export_w": max(0, -grid), + "production_ac_total_wh": None, "grid_import_total_wh": None, "grid_export_total_wh": None, + "boiler_enabled": None, "boiler_reason": None, + }, + "source_timestamps": {"powerflow": when}, + "errors": {"meter": "Cumulative meter readings are unavailable in recovered chart samples", "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): - """Pair real chart samples; assume constant W over five minutes, as solar does.""" - for stamp in sorted(production.keys() & consumption.keys()): - generated, consumed = production[stamp], consumption[stamp] - 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} - yield {"logDateTime": utc_text(datetime.fromtimestamp(stamp / 1000, timezone.utc)), "logDuration": 300, - "channels": [{"channelName": name, "channelType": "Energy", "unit": "Wh", "value": values[field]} - for field, name in CHANNELS.items()], - "chart_power_w": {"production": generated, "consumption": consumed}, - "energy_method": "constant_power_over_five_minutes"} + "raw": {"solarweb_history": {"chart_power_w": {"production": generated, "consumption": consumed}}}, + } class HistoryRecovery: @@ -158,11 +96,12 @@ class HistoryRecovery: # A day with no usable cloud data must not block recovery of newer gaps. LOG.warning("Solar.web chart data unavailable for %s; leaving its gaps for the next check", day) continue - for entry in chart_entries(production, consumption): - epoch = milliseconds(entry["logDateTime"]) // 1000 + for stamp in sorted(production.keys() & consumption.keys()): + epoch = stamp // 1000 if epoch not in missing or datetime.fromtimestamp(epoch, self.zone).date() != day: continue - sample = history_sample(entry, self.source_id, self.config.pv_system_id) + sample = history_sample(stamp, production[stamp], consumption[stamp], + self.source_id, self.config.pv_system_id) self.store.add(sample) missing.remove(epoch) recovered += 1 diff --git a/solar_relay/storage.py b/solar_relay/storage.py index b327c7d..63da9d7 100644 --- a/solar_relay/storage.py +++ b/solar_relay/storage.py @@ -20,7 +20,6 @@ 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", @@ -34,7 +33,6 @@ SCHEMA = pa.schema([ *((field, pa.float64()) for field in NUMBERS), ("boiler_enabled", pa.bool_()), ("boiler_reason", pa.string()), ("record_kind", pa.string()), - ("interval_energy_wh", pa.struct([(field, pa.float64()) for field in ("production", "consumption", "self_consumption", "bought", "sold")])), ("power_ms", pa.int64()), ("power_ok", pa.bool_()), ("meter_ms", pa.int64()), ("meter_ok", pa.bool_()), ("boiler_ms", pa.int64()), ("boiler_ok", pa.bool_()), @@ -82,9 +80,13 @@ class Store: if not path.exists(): return [] parquet = pq.ParquetFile(path, page_checksum_verification=True) - if not parquet.schema_arrow.equals(SCHEMA, check_metadata=True): + schema = parquet.schema_arrow + # Validate the fields this version uses; read only those columns. + if schema.metadata != SCHEMA.metadata or any( + schema.get_field_index(field.name) < 0 or schema.field(field.name) != field for field in SCHEMA + ): raise StoreError("Unsupported Parquet archive schema") - return parquet.read(columns=columns).to_pylist() + return parquet.read(columns=SCHEMA.names if columns is None else columns).to_pylist() @staticmethod def _sync_directory(path): @@ -125,8 +127,7 @@ class Store: "payload": json.dumps(sample, allow_nan=False, separators=(",", ":")), **{name: readings.get(name) for name in NUMBERS}, "boiler_enabled": readings.get("boiler_enabled"), "boiler_reason": readings.get("boiler_reason"), - "record_kind": "history" if "history_interval" in sample else "live", - "interval_energy_wh": sample.get("history_interval", {}).get("energy_wh"), + "record_kind": "history" if "solarweb_history" in sample.get("raw", {}) else "live", **{name: bool(value) if name.endswith("_ok") else value for name, value in zip(COVERAGE_COLUMNS[1:], coverage[3:])}, } @@ -137,32 +138,11 @@ class Store: rows = self._read(path) for existing in rows: if existing["id"] == row["id"]: - if normalize_history_sample(json.loads(existing["payload"])) != normalize_history_sample(sample): + if json.loads(existing["payload"]) != 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/tests/test_history.py b/tests/test_history.py index c3f3815..3d14db6 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -8,7 +8,7 @@ import pyarrow.parquet as pq from zoneinfo import ZoneInfo from solar_relay.config import HistoryConfig, utc_text -from solar_relay.history import CHANNELS, HistoryRecovery, chart_entries, history_sample +from solar_relay.history import HistoryRecovery, history_sample from solar_relay.http import TransportError from solar_relay.solarweb import ChartDataError from solar_relay.storage import Store @@ -17,12 +17,6 @@ from tests.test_status import observed SYSTEM = "73733052-6f10-47a5-9746-7f7e76ebcb8c" -def entry(when): - values = {"production": 100, "consumption": 120, "self_consumption": 80, "bought": 40, "sold": 20} - return {"logDateTime": utc_text(when), "logDuration": 300, - "channels": [{"channelName": channel, "channelType": "Energy", "unit": "Wh", "value": values[field]} - for field, channel in CHANNELS.items()]} - class FakeCharts: def __init__(self, when): @@ -55,9 +49,9 @@ class HistoryTests(unittest.TestCase): self.store.add(observed(self.when + timedelta(minutes=offset))) self.assertEqual(self.recover(), 1) records = list(self.store.records()) - recovered = [s for s in records if "history_interval" in s] + recovered = [s for s in records if "solarweb_history" in s["raw"]] self.assertEqual(len(recovered), 1) - self.assertEqual(recovered[0]["history_interval"]["energy_wh"]["production"], 100) + self.assertEqual(recovered[0]["readings"]["production_ac_w"], 1200) self.assertIsNone(recovered[0]["readings"]["boiler_enabled"]) self.assertNotIn("boiler", recovered[0]["source_timestamps"]) self.assertIn("boiler", recovered[0]["errors"]) @@ -78,41 +72,20 @@ class HistoryTests(unittest.TestCase): self.assertEqual(day["complete"], 0) 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_recovered_power_uses_the_same_completeness_checks_as_live_power(self): + sample = history_sample(int(self.when.timestamp() * 1000), 1200, 1440, self.source, SYSTEM) + sample["readings"]["consumption_w"] = None + self.store.add(sample) + day = self.store.coverage(self.source, "UTC", 1, self.when + timedelta(days=1))[1] + self.assertEqual(day["energy_complete"], 0) + self.assertEqual(day["boiler_complete"], 0) + self.assertEqual(day["slots"][120], 0) def test_partial_cloud_records_remain_missing_for_retry(self): stamp = int((self.when + timedelta(minutes=5)).timestamp() * 1000) @@ -123,20 +96,23 @@ class HistoryTests(unittest.TestCase): self.assertEqual(self.recover(), 1) def test_recovered_identifiers_are_stable_for_duplicate_delivery(self): - first = history_sample(entry(self.when), self.source, SYSTEM) - second = history_sample(entry(self.when), self.source, SYSTEM) + stamp = int(self.when.timestamp() * 1000) + first = history_sample(stamp, 1200, 1440, self.source, SYSTEM) + second = history_sample(stamp, 1200, 1440, self.source, SYSTEM) self.assertEqual(first, second) + self.store.add(first) + self.store.add(second) + self.assertEqual(self.store.status()["samples"], 1) - def test_non_five_minute_and_bad_units_are_rejected(self): - for change in [{"logDuration": 599}, {"logDateTime": utc_text(self.when + timedelta(seconds=1))}]: - value = entry(self.when) - value.update(change) - with self.assertRaises(ValueError): - history_sample(value, self.source, SYSTEM) - value = entry(self.when) - value["channels"][0]["unit"] = "W" - with self.assertRaises(ValueError): - history_sample(value, self.source, SYSTEM) + def test_unaligned_timestamps_and_invalid_power_are_rejected(self): + stamp = int(self.when.timestamp() * 1000) + for bad_stamp in (stamp + 1000, float(stamp), True): + with self.subTest(stamp=bad_stamp), self.assertRaises(ValueError): + history_sample(bad_stamp, 1200, 1440, self.source, SYSTEM) + for invalid in (None, True, "1200", -1, float("nan"), float("inf"), 1e19): + for generated, consumed in ((invalid, 1440), (1200, invalid)): + with self.subTest(generated=generated, consumed=consumed), self.assertRaises(ValueError): + history_sample(stamp, generated, consumed, self.source, SYSTEM) def test_missing_private_credentials_fail_before_requests(self): with self.assertRaisesRegex(ValueError, "SOLARWEB_USERNAME and SOLARWEB_PASSWORD"): @@ -190,7 +166,7 @@ class HistoryTests(unittest.TestCase): self.client.points = {"production": points, "consumption": points} self.store.add(observed(now.replace(minute=0, second=0))) self.assertEqual(self.recovery.recover(now=now), 3) - recovered = [record for record in self.store.records() if "history_interval" in record] + recovered = [record for record in self.store.records() if "solarweb_history" in record["raw"]] self.assertEqual([record["observed_at"] for record in recovered], [utc_text(when) for when in recoverable]) self.assertEqual(min(day for _, _, day in self.client.requests), cutoff.astimezone(self.recovery.zone).date()) @@ -266,22 +242,19 @@ class HistoryTests(unittest.TestCase): self.recovery.recover(now=self.when + timedelta(days=2)) self.assertEqual(client.call_count, 1) - def test_chart_conversion_matches_solar_and_retains_provenance(self): + def test_chart_watts_are_stored_directly_and_retain_provenance(self): 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})) - 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") + sample = history_sample(stamp, generated, consumed, self.source, SYSTEM) self.assertEqual(sample["readings"]["production_ac_w"], generated) self.assertEqual(sample["readings"]["consumption_w"], consumed) self.assertEqual(sample["readings"]["grid_w"], consumed - generated) + self.assertEqual(sample["readings"]["grid_import_w"], max(0, consumed - generated)) + self.assertEqual(sample["readings"]["grid_export_w"], max(0, generated - consumed)) self.assertIsNone(sample["readings"]["production_w"]) + self.assertEqual(sample["raw"]["solarweb_history"]["chart_power_w"], + {"production": generated, "consumption": consumed}) def test_dst_days_recover_every_actual_five_minute_slot(self): for day, expected in [("2026-03-29", 276), ("2026-10-25", 300)]: diff --git a/tests/test_parquet.py b/tests/test_parquet.py index c3c8612..3e67a6a 100644 --- a/tests/test_parquet.py +++ b/tests/test_parquet.py @@ -7,11 +7,12 @@ import unittest from unittest.mock import patch from zoneinfo import ZoneInfo +import pyarrow as pa import pyarrow.parquet as pq from solar_relay.cli import next_poll_at from solar_relay.config import retention_cutoff -from solar_relay.storage import Store, StoreError +from solar_relay.storage import SCHEMA, Store, StoreError from tests.test_status import observed @@ -49,6 +50,33 @@ class ParquetTests(unittest.TestCase): self.assertEqual(self.store.prune(now=self.now + timedelta(days=31)), 2) self.assertEqual(list(self.store.records()), [items[3]]) + def test_additional_archive_columns_do_not_change_records_or_acknowledgements(self): + value = observed(self.now) + self.store.add(value) + path = next(self.path.glob("date=*/samples.parquet")) + table = pq.ParquetFile(path).read() + table = table.append_column("unused_estimate", pa.array([100.0])) + pq.write_table(table, path) + self.assertEqual(self.store.pending(100), [value]) + self.assertEqual(list(self.store.records()), [value]) + self.store.acknowledge([value["id"]], now=self.now) + self.assertEqual(self.store.pending(100), []) + self.assertEqual(list(self.store.records()), [value]) + self.assertTrue(pq.ParquetFile(path).schema_arrow.equals(SCHEMA, check_metadata=True)) + + def test_missing_or_mistyped_archive_columns_are_rejected(self): + value = observed(self.now) + self.store.add(value) + path = next(self.path.glob("date=*/samples.parquet")) + table = pq.ParquetFile(path).read() + index = table.schema.get_field_index("production_ac_w") + for invalid in (table.drop_columns(["production_ac_w"]), + table.set_column(index, "production_ac_w", pa.array(["invalid"]))): + with self.subTest(schema=invalid.schema): + pq.write_table(invalid, path) + with self.assertRaises(StoreError): + self.store.pending(100) + def test_retention_boundary_is_local_midnight_regardless_of_execution_time(self): zone = ZoneInfo("Europe/Zurich") for hour in (0, 12, 23): diff --git a/tests/test_relay.py b/tests/test_relay.py index fd51129..2aa3d7b 100644 --- a/tests/test_relay.py +++ b/tests/test_relay.py @@ -402,27 +402,28 @@ class ConfigTests(unittest.TestCase): with self.assertRaises(ValueError): self.load('\n[http]\nbase_path="/../config"\n') - def test_history_account_settings_and_migration_from_query_api_keys(self): + def test_history_account_environment_settings(self): config, _ = self.load('\n[history]\nusername_env="BACKUP_USER"\npassword_env="BACKUP_PASSWORD"\n') self.assertEqual(config.history.username_env, "BACKUP_USER") self.assertEqual(config.history.password_env, "BACKUP_PASSWORD") self.assertFalse(config.history.enabled) - config, _ = self.load('\n[history]\naccess_key_id_env="OLD_ID"\naccess_key_value_env="OLD_KEY"\n') - self.assertEqual(config.history.username_env, "SOLARWEB_USERNAME") - self.assertEqual(config.history.password_env, "SOLARWEB_PASSWORD") for name in ("username_env", "password_env"): with self.subTest(name=name), self.assertRaisesRegex(ValueError, "environment variable"): self.load(f'\n[history]\n{name}="invalid name"\n') - def test_history_uses_retention_instead_of_legacy_lookback_setting(self): + def test_history_uses_collector_retention(self): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "synthetic-test.toml" path.write_text('[collector]\nretention_days=2\n[history]\nenabled=true\n' - 'pv_system_id="73733052-6f10-47a5-9746-7f7e76ebcb8c"\nlookback_days=7\n') + 'pv_system_id="73733052-6f10-47a5-9746-7f7e76ebcb8c"\n') config = load_config(path) self.assertEqual(config.collector.retention_days, 2) self.assertTrue(config.history.enabled) - self.assertFalse(hasattr(config.history, "lookback_days")) + + def test_obsolete_history_options_are_rejected(self): + for name in ("access_key_id_env", "access_key_value_env", "lookback_days"): + with self.subTest(name=name), self.assertRaisesRegex(ValueError, "Unknown history settings"): + self.load(f'\n[history]\n{name}=7\n') def test_private_config_accepts_credentials_and_redacts_them_from_repr(self): config, _ = self.load('\n[upload]\ntoken="synthetic-shared-token"\n'