Remove legacy code for format conversion

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")

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)):

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 [

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

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 = []