diff --git a/.gitea/workflows/deploy.yml b/.gitea/workflows/deploy.yml
new file mode 100644
index 0000000..483c057
--- /dev/null
+++ b/.gitea/workflows/deploy.yml
@@ -0,0 +1,100 @@
+# Register the Pi runner with the label solar-relay:host.
+# Install the systemd unit and the restricted sudoers rule described in README.md once.
+name: Deploy Solar Relay
+
+on:
+ push:
+ branches:
+ - main
+ workflow_dispatch:
+
+jobs:
+ deploy:
+ runs-on: solar-relay
+
+ env:
+ DEPLOY_PATH: /var/www/html/solar-relay
+ SERVICE: solar-relay.service
+ HEALTH_URL: https://rpioffice.nest.lutran.ch/solar-relay/healthz
+
+ steps:
+ - name: Checkout
+ uses: actions/checkout@v4
+ with:
+ persist-credentials: false
+
+ - name: Check runner and deployment setup
+ shell: bash
+ run: |
+ set -euo pipefail
+ command -v python3
+ command -v rsync
+ command -v sudo
+ command -v systemctl
+ command -v flock
+ command -v curl
+ python3 -c 'import sys, tomllib; assert sys.version_info >= (3, 11)'
+ test -d "$DEPLOY_PATH"
+ test -w "$DEPLOY_PATH"
+ python3 - <<'PY'
+ import os
+ from pathlib import Path
+ checkout = Path.cwd().resolve()
+ deployment = Path(os.environ['DEPLOY_PATH']).resolve()
+ if checkout == deployment or checkout in deployment.parents or deployment in checkout.parents:
+ raise SystemExit('The runner checkout and deployment directory must not overlap')
+ PY
+ test -r /etc/solar-relay/config.toml
+ cmp deploy/solar-relay.service /etc/systemd/system/solar-relay.service || {
+ echo 'Install the updated deploy/solar-relay.service and run systemctl daemon-reload first.'
+ exit 1
+ }
+ sudo -n -l "$(command -v systemctl)" restart "$SERVICE" >/dev/null
+
+ - name: Create isolated test environment
+ run: |
+ python3 -m venv .venv-ci
+ .venv-ci/bin/python -m pip install --only-binary=pyarrow -r requirements.txt
+
+ - name: Run tests
+ run: .venv-ci/bin/python -m unittest discover -s tests -v
+
+ - name: Deploy and restart
+ shell: bash
+ run: |
+ set -euo pipefail
+ # Also serializes deployments on Gitea versions that lack workflow concurrency.
+ exec 9>"$DEPLOY_PATH/.deploy.lock"
+ flock -w 120 9
+ rsync -azc --no-times --delete \
+ --exclude '/.git/' \
+ --exclude '/.gitea/' \
+ --exclude '/.venv/' \
+ --exclude '/.venv-ci/' \
+ --exclude '/.deploy.lock' \
+ --exclude '/config.toml' \
+ --exclude '/.env' \
+ --exclude '/data/' \
+ --exclude '__pycache__/' \
+ --exclude '*.parquet' \
+ ./ "$DEPLOY_PATH/"
+ if [ ! -x "$DEPLOY_PATH/.venv/bin/python" ]; then
+ python3 -m venv "$DEPLOY_PATH/.venv"
+ fi
+ "$DEPLOY_PATH/.venv/bin/python" -c 'import sys; assert sys.version_info >= (3, 11)'
+ "$DEPLOY_PATH/.venv/bin/python" -m pip install --only-binary=pyarrow -r "$DEPLOY_PATH/requirements.txt"
+ sudo -n "$(command -v systemctl)" restart "$SERVICE"
+ # Catch an invalid config or missing token that triggers an immediate restart loop.
+ sleep 10
+ if ! systemctl is-active --quiet "$SERVICE"; then
+ systemctl status "$SERVICE" --no-pager --lines 30 || true
+ exit 1
+ fi
+ if [ "$(systemctl show --property=NRestarts --value "$SERVICE")" != "0" ]; then
+ echo 'The collector restarted unexpectedly after deployment.'
+ systemctl status "$SERVICE" --no-pager --lines 30 || true
+ exit 1
+ fi
+ systemctl status "$SERVICE" --no-pager --lines 10
+ curl --fail --silent --show-error --connect-timeout 3 --max-time 5 \
+ --retry 4 --retry-delay 2 "$HEALTH_URL"
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..fb808a2
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,10 @@
+.venv/
+.venv-ci/
+.deploy.lock
+__pycache__/
+*.py[cod]
+config.toml
+.env
+data/
+*.parquet
+
diff --git a/README.md b/README.md
index 5296364..716eb61 100644
--- a/README.md
+++ b/README.md
@@ -1,3 +1,289 @@
# solar-relay
-Fronius data extractor and relay for the solar app
\ No newline at end of file
+Local Fronius GEN24 recorder and HTTPS relay for the sibling `solar` website.
+Python 3.11+ and a project-local virtual environment are required. License:
+AGPL-3.0-or-later, see [LICENSE](LICENSE).
+
+The collector takes one observation every **five minutes**, aligned to the clock
+(`:00`, `:05`, …). Startup and restarts wait for the next five-minute boundary.
+`poll_interval_seconds` must be 300. Each observation
+contains production, consumption, grid power, cumulative energy counters, boiler
+switch state, individual source timestamps, errors, and the original JSON responses.
+Requests to the inverter remain at least four seconds apart, so these readings are
+not simultaneous. The default inverter is `http://fronius.local`, meter `0`, boiler `pin1`.
+
+An HTTP/network request failure retries the affected endpoint up to three times
+(four attempts total), keeping the same request spacing. Successful endpoints are
+not repeated. If retries are exhausted, that source is recorded as unavailable.
+Retries belong to the same observation and do not reset the five-minute clock.
+Scheduled collection remains anchored to `:00`, `:05`, `:10`, … boundaries.
+
+Observations are saved durably to **Parquet** before upload. Separate workers handle
+delivery and optional history recovery; an internet outage does not block recording.
+Acknowledged observations are deleted **30 days after acknowledgement**. Pending
+observations never expire, even if their measurement dates are much older.
+
+## Installation
+
+```bash
+python3 -m venv .venv
+source .venv/bin/activate
+python -m pip install --only-binary=pyarrow -r requirements.txt
+cp config.example.toml config.toml
+python -m solar_relay --config config.toml status
+python -m solar_relay --config config.toml run
+```
+
+Use `.venv/bin/python` directly if you prefer not to activate the environment.
+All packages stay inside the venv. Parquet storage uses the pinned **PyArrow**
+dependency in `requirements.txt`. Use a 64-bit Raspberry Pi OS
+with an available PyArrow wheel; `--only-binary=pyarrow` avoids a large source build.
+The deployment workflow installs dependencies in both its test and runtime venvs.
+
+Configuration, credentials, `.venv` and Parquet data are ignored by Git.
+Relative paths resolve against the configuration file, rather than the working directory.
+Uploads default to disabled until `[upload].url` is configured.
+
+## Sources and measurement limits
+
+| Source | Endpoint |
+| --- | --- |
+| Solar API: PV, AC output, household load, grid power | `/solar_api/v1/GetPowerFlowRealtimeData.fcgi` |
+| Solar API: grid meter counters | `/solar_api/v1/GetMeterRealtimeData.cgi?Scope=System` |
+| Internal GEN24 API: boiler switch | `/api/status/emrs` |
+
+Solar API must be enabled in the inverter's Communication settings. The collector
+only reads these endpoints. The boiler endpoint is an internal firmware interface;
+if it stops working, boiler state is recorded as unknown rather than off.
+
+| Reading | Meaning |
+| --- | --- |
+| `production_w` | `Site.P_PV`, PV generator DC power |
+| `production_ac_w` | Sum of inverter `P`, AC output |
+| `consumption_w` | `-Site.P_Load`, household load on this GEN24 |
+| `grid_w` | Positive import, negative export at the feed-in meter |
+| `grid_import_w`, `grid_export_w` | Nonnegative directional grid power |
+| `production_ac_total_wh` | Lifetime AC production counter |
+| `grid_import_total_wh`, `grid_export_total_wh` | Grid meter lifetime import/export counters |
+| `boiler_enabled`, `boiler_reason` | Output state (`true`, `false`, or `null`) and reason code |
+
+The meter must be installed at the grid feed-in point. Missing/stale readings and
+source errors never become zero energy or an off boiler. DC panel power differs
+from AC output by conversion losses. The website uses AC output for its energy balance.
+
+The boiler switch enables heating but does not measure electricity use: the thermostat
+can stop heating while the output remains enabled. Five-minute polling can miss short
+transitions; enabled durations are estimates. A branch meter is needed for boiler kWh.
+The 2000 W threshold is power (2 kW), not energy (kWh).
+
+## Parquet archive, delivery and retention
+
+`collector.archive` is a directory, containing one partition per **UTC date**:
+
+```text
+archive/
+ .lock
+ date=2026-11-01/samples.parquet
+ date=2026-11-02/samples.parquet
+```
+
+The files contain typed numeric/boolean readings, IDs, UTC observation timestamps,
+coverage flags, `uploaded_at`, and the complete original sample as JSON. They use
+Zstandard compression. Each update writes a temporary file, flushes/fsyncs it,
+atomically replaces the daily file, and fsyncs the directory. File permissions are
+0600 and new archive directories are 0700. Thread/process locks prevent a collector,
+uploader, status request or CLI replay from overwriting another operation. HTTP
+requests happen outside the archive lock.
+
+The upload request is a JSON object with `schema_version: 1` and a `samples` array,
+authenticated with the bearer token named by `upload.token_env`. The website must
+commit its data before responding:
+
+```json
+{"accepted_ids": ["1599c148-612c-46ab-aa9a-cf9a13271a72"]}
+```
+
+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
+UUIDs when a response is lost, so retries and replays do not double-count energy.
+
+`collector.retention_days` defaults to 30. Automatic cleanup runs at service startup
+and hourly, deleting records acknowledged more than that many days ago. Old pending
+records remain. Replay clears acknowledgement and therefore prevents expiry until
+another successful transfer. The main website retains its own imported history.
+
+```bash
+.venv/bin/python -m solar_relay --config config.toml prune
+.venv/bin/python -m solar_relay --config config.toml export \
+ --from 2026-10-29 --until 2026-10-31 > readings.jsonl
+.venv/bin/python -m solar_relay --config config.toml replay \
+ --from 2026-10-29 --until 2026-10-31
+.venv/bin/python -m solar_relay --config config.toml upload
+```
+
+Date bounds use `collector.timezone`, including DST transitions. `--from` is inclusive;
+`--until` is exclusive. ISO timestamps may include an explicit offset. Unbounded
+export/replay covers all retained observations. Export provides a consistent snapshot;
+stop the service and other CLI writers before copying the archive directory as a backup.
+
+## Recovering history after Wi-Fi loss
+
+GEN24 records historical data for transfer to Solar.web after reconnecting. This is
+separate from its local Solar API: Fronius explicitly lists archive access for GEN24
+as unavailable, and this inverter's `GetArchiveData.cgi` returns 404. Inspecting its
+current web interface also found no energy-history retrieval endpoint. Its SDP support
+download is an encrypted `.cry` diagnostic file, not a documented energy-history API.
+
+See the official [Solar API manual, section 5.1.1](https://www.fronius.com/~/downloads/Solar%20Energy/Operating%20Instructions/42%2C0410%2C2012.pdf)
+and [Solar.web Query API manual, section 6.5](https://www.fronius.com/~/downloads/Solar%20Energy/User%20Information/SE_UI_API_InterfaceDocumentation_EN.pdf).
+
+Optional recovery reads missing five-minute **measured energy** through the separate,
+authenticated Solar.web Query API, after the inverter uploads its backlog. This uses
+no public guest page. Obtain Query API access and keys from Fronius; its commercial
+terms are separate from normal Solar.web access. No cloud calls occur while disabled.
+
+Set `[history].enabled = true` and `pv_system_id` in the configuration, and put these
+in the service's root-readable `/etc/solar-relay/solar-relay.env`:
+
+```text
+SOLARWEB_ACCESS_KEY_ID=your-access-key-id
+SOLARWEB_ACCESS_KEY_VALUE=your-access-key-value
+```
+
+The recovery worker checks hourly, looking back seven days by default and allowing
+a two-hour delay for the inverter's hourly uploads. It fetches only UTC days with
+missing energy extractions, requests five energy channels, limits each request to
+24 hours, and skips intervals already present. Requests use HTTPS, headers for
+credentials, timeouts and no redirects. Missing/partial cloud data stays missing and
+is tried again later. The configured lookback cannot exceed the retention period.
+
+```bash
+.venv/bin/python -m solar_relay --config config.toml recover \
+ --from 2026-10-29 --until 2026-10-31
+```
+
+Manual recovery accepts at most 30 days per invocation and can query dates older than
+the automatic lookback. Only aligned 300-second energy records with all required Wh
+channels are accepted; longer aggregates are not split into invented five-minute values.
+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.
+
+**Boiler switch history and cumulative meter readings are not recovered.** Historical
+energy intervals leave those sources unknown, and the status page shows partial
+coverage. Counter differences can describe a total across an outage, but cannot
+recover individual five-minute values. Fully local operation leaves Wi-Fi gaps visible.
+
+## Raspberry Pi service
+
+Deploy the checkout to `/var/www/html/solar-relay` and create/install its venv as above.
+Place configuration outside the web directory:
+
+```bash
+sudo install -d -m 0755 /etc/solar-relay
+sudo install -m 0644 config.example.toml /etc/solar-relay/config.toml
+sudoedit /etc/solar-relay/config.toml
+```
+
+Use these collector settings, retaining the example's other sections:
+
+```toml
+[collector]
+source_id = "home-gen24"
+archive = "/var/lib/solar-relay/archive"
+timezone = "Europe/Zurich"
+poll_interval_seconds = 300
+retention_days = 30
+```
+
+Set `[upload].url` to the main website's `https://your-solar-host/index.php?a=relay`.
+Put `SOLAR_RELAY_TOKEN=your-shared-token` in `/etc/solar-relay/solar-relay.env` with
+permissions 0600, readable by root. Configure the same token and source ID on the
+main website as described in its README. Tokens and API keys stay out of TOML/Git.
+
+```bash
+sudo cp deploy/solar-relay.service /etc/systemd/system/
+sudo systemctl daemon-reload
+sudo systemctl enable --now solar-relay
+sudo journalctl -u solar-relay -f
+```
+
+systemd provides `/var/lib/solar-relay` to a dynamic, unprivileged service user.
+Recording/uploading remains separate from serving status. Running tests or `sample`
+does not install/start the service. Initial systemd/configuration installation is manual.
+
+## HTTPS reverse proxy and status page
+
+Python listens at **`http://127.0.0.1:8765`**. Choose one proxy configuration and include
+it inside the existing HTTPS virtual host for `rpioffice.nest.lutran.ch`:
+
+- Nginx: [`deploy/nginx-solar-relay.conf`](deploy/nginx-solar-relay.conf).
+- Apache: [`deploy/apache-solar-relay.conf`](deploy/apache-solar-relay.conf).
+ Enable `proxy`, `proxy_http` and `headers` (`sudo a2enmod proxy proxy_http headers`).
+
+Keep your existing TLS certificates on the proxy. Proxy the entire `/solar-relay/`
+prefix; do not serve the repository/archive as static files. The collector accepts
+both a stripped proxy path and a preserved `/solar-relay/` prefix. Validate the web
+server configuration before reloading it (`nginx -t` or `apachectl configtest`).
+
+The status page is **https://rpioffice.nest.lutran.ch/solar-relay/**. It shows last
+extraction/last success, upload backlog, retained observations, and daily five-minute
+completeness with a selectable interval grid and gap ranges. Dates use the configured
+timezone. DST days have 276 or 300 intervals; ordinary days have 288.
+
+Complete means at least one fresh, successful extraction for energy, meter and boiler
+in that five-minute slot. It does not mean continuously observed power or switch state.
+The unfinished current slot is excluded from missing counts. Historical recovery
+contributes energy only. After retention cleanup, pruned records no longer contribute
+to this local page; the website keeps the transferred history. Refresh is every 15 seconds.
+
+Endpoints under `/solar-relay/`:
+
+- `/`: status page.
+- `/api/status?days=14`: status and extraction completeness JSON.
+- `/healthz`: process liveness, also used by deployment checks.
+
+All routes are read-only and expose no raw readings, credentials or filesystem paths.
+Disable `[http].enabled` and remove the workflow's health probe if HTTP is not needed.
+
+## Gitea deployment
+
+`.gitea/workflows/deploy.yml` deploys pushes to `main` or manual workflow runs to
+`/var/www/html/solar-relay` and restarts `solar-relay.service`. Register the Pi runner
+with **`solar-relay:host`**; its checkout must be separate from production.
+
+The runner needs Python 3.11+, `python3-venv`/pip, a supported 64-bit PyArrow platform,
+`rsync`, `flock`, `curl`, `sudo`, systemd and Node.js 20+ for `actions/checkout@v4`.
+Give its account write access to the deployment directory. Install the config and unit
+before the first deployment, then add a restricted sudoers rule using the actual runner
+account and the path from `command -v systemctl`:
+
+```sudoers
+gitea-runner ALL=(root) NOPASSWD: /usr/bin/systemctl restart solar-relay.service
+```
+
+The workflow installs dependencies and runs tests in `.venv-ci`, checks the installed
+unit, serializes deployment with `flock`, syncs source, installs runtime dependencies
+inside `.venv`, restarts and checks the service, then probes the public HTTPS health
+endpoint with certificate verification. Runtime venv/configuration, `data/` and
+Parquet files are preserved. Recommended state under `/var/lib` and
+configuration under `/etc` are entirely outside the sync and web directory.
+
+If the deployment directory is initially empty, enable the unit without `--now`; the
+first deployment creates the venv and starts it. Set up the proxy before deployment.
+For unit changes, install the updated file and run `systemctl daemon-reload`; the
+workflow detects a mismatch and does not grant itself broader sudo access.
+
+## Checks
+
+```bash
+.venv/bin/python -m unittest discover -s tests -v
+node --check solar_relay/web/status.js
+```
+
+Tests cover normalization, real local HTTP delivery/retries, Parquet round-trips,
+concurrent writers, interrupted writes, retention, DST completeness,
+mocked authenticated history recovery, status routes and the deployment script.
+Live Solar.web recovery requires real Query API credentials and is not exercised by tests.
diff --git a/config.example.toml b/config.example.toml
new file mode 100644
index 0000000..08cb51e
--- /dev/null
+++ b/config.example.toml
@@ -0,0 +1,41 @@
+[collector]
+source_id = "home-gen24"
+archive = "data/archive" # Parquet directory, relative to this configuration file.
+timezone = "Europe/Zurich" # Used for date-only export/replay arguments.
+poll_interval_seconds = 300
+retention_days = 30 # Days AFTER acknowledgement; pending records never expire.
+
+[inverter]
+base_url = "http://fronius.local"
+meter_id = "0" # Primary meter, as reported by Solar API.
+boiler_pin = "pin1"
+request_timeout_seconds = 5
+min_request_interval_seconds = 4
+
+[upload]
+url = "" # Set to https://your-solar-host/index.php?a=relay to enable uploads.
+token_env = "SOLAR_RELAY_TOKEN"
+batch_size = 100
+request_timeout_seconds = 10
+retry_initial_seconds = 5
+retry_max_seconds = 300
+
+[http]
+enabled = true
+bind = "127.0.0.1"
+port = 8765
+base_path = "/solar-relay"
+history_days = 14
+
+[history]
+# Optional, authenticated Solar.web Query API. No public guest link is used.
+# GEN24's local Solar API has no historical-data endpoint.
+enabled = false
+pv_system_id = "" # Your Solar.web PV system UUID; required when enabled.
+access_key_id_env = "SOLARWEB_ACCESS_KEY_ID"
+access_key_value_env = "SOLARWEB_ACCESS_KEY_VALUE"
+lookback_days = 7
+lag_seconds = 7200 # Allow the inverter to upload its hourly historical data.
+check_interval_seconds = 3600
+request_timeout_seconds = 10
+
diff --git a/deploy/apache-solar-relay.conf b/deploy/apache-solar-relay.conf
new file mode 100644
index 0000000..f437090
--- /dev/null
+++ b/deploy/apache-solar-relay.conf
@@ -0,0 +1,11 @@
+# Include inside the existing HTTPS VirtualHost for rpioffice.nest.lutran.ch.
+# Enable modules: sudo a2enmod proxy proxy_http headers
+# Keep the host's existing SSLEngine and certificate configuration.
+ProxyRequests Off
+ProxyPreserveHost On
+RedirectMatch 308 ^/solar-relay$ /solar-relay/
+ProxyPass /solar-relay/ http://127.0.0.1:8765/ connectiontimeout=3 timeout=15
+ProxyPassReverse /solar-relay/ http://127.0.0.1:8765/
+
+ RequestHeader set X-Forwarded-Proto "https"
+
diff --git a/deploy/nginx-solar-relay.conf b/deploy/nginx-solar-relay.conf
new file mode 100644
index 0000000..01d5a6d
--- /dev/null
+++ b/deploy/nginx-solar-relay.conf
@@ -0,0 +1,16 @@
+# Include these locations in the existing HTTPS server block for rpioffice.nest.lutran.ch.
+# The trailing slash on proxy_pass removes /solar-relay/ before forwarding.
+location = /solar-relay {
+ return 308 /solar-relay/;
+}
+
+location ^~ /solar-relay/ {
+ proxy_pass http://127.0.0.1:8765/;
+ proxy_http_version 1.1;
+ proxy_set_header Host $host;
+ proxy_set_header X-Real-IP $remote_addr;
+ proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+ proxy_set_header X-Forwarded-Proto $scheme;
+ proxy_connect_timeout 3s;
+ proxy_read_timeout 15s;
+}
diff --git a/deploy/solar-relay.service b/deploy/solar-relay.service
new file mode 100644
index 0000000..e17729d
--- /dev/null
+++ b/deploy/solar-relay.service
@@ -0,0 +1,25 @@
+[Unit]
+Description=Fronius local recorder and solar website relay
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+DynamicUser=yes
+WorkingDirectory=/var/www/html/solar-relay
+ExecStart=/var/www/html/solar-relay/.venv/bin/python -m solar_relay --config /etc/solar-relay/config.toml run
+EnvironmentFile=-/etc/solar-relay/solar-relay.env
+StateDirectory=solar-relay
+StateDirectoryMode=0700
+UMask=0077
+Restart=on-failure
+RestartSec=10
+TimeoutStopSec=45
+NoNewPrivileges=yes
+PrivateTmp=yes
+ProtectSystem=strict
+ProtectHome=yes
+ReadWritePaths=/var/lib/solar-relay
+
+[Install]
+WantedBy=multi-user.target
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..a314af5
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,2 @@
+# Runtime dependency; install inside .venv, never into the system Python.
+pyarrow==26.0.0
diff --git a/solar_relay/__init__.py b/solar_relay/__init__.py
new file mode 100644
index 0000000..8c2010c
--- /dev/null
+++ b/solar_relay/__init__.py
@@ -0,0 +1,4 @@
+"""Local Fronius recorder and durable HTTP relay (AGPL-3.0-or-later)."""
+
+__version__ = "0.2.0"
+
diff --git a/solar_relay/__main__.py b/solar_relay/__main__.py
new file mode 100644
index 0000000..8f0c1d5
--- /dev/null
+++ b/solar_relay/__main__.py
@@ -0,0 +1,4 @@
+from .cli import main
+
+raise SystemExit(main())
+
diff --git a/solar_relay/cli.py b/solar_relay/cli.py
new file mode 100644
index 0000000..42ba05c
--- /dev/null
+++ b/solar_relay/cli.py
@@ -0,0 +1,166 @@
+"""Commands for recording, uploading, inspecting and replaying the local archive."""
+
+import argparse
+import json
+import logging
+import math
+from pathlib import Path
+import signal
+import sys
+import threading
+import time
+from zoneinfo import ZoneInfoNotFoundError
+
+from . import __version__
+from .config import date_bound, load_config
+from .fronius import FroniusClient
+from .history import HistoryRecovery
+from .http import TransportError
+from .storage import Store
+from .status import StatusServer
+from .upload import Uploader, UploadError
+
+LOG = logging.getLogger(__name__)
+
+
+def _record(client, store, source_id, interval=300):
+ sample = client.collect(source_id, interval)
+ # Commit before attempting any upload. Even a completely failed poll is a gap marker.
+ store.add(sample)
+ if sample["errors"]:
+ LOG.warning("Recorded observation %s with errors: %s", sample["id"], json.dumps(sample["errors"]))
+ else:
+ LOG.info("Recorded observation %s", sample["id"])
+ return sample
+
+
+def next_poll_at(now, interval):
+ """UTC wall-clock boundaries: xx:00, xx:05, ... for the default interval."""
+ return (math.floor(now / interval) + 1) * interval
+
+
+def _run(config, store):
+ stop = threading.Event()
+ previous = {}
+ uploader = Uploader(config.upload, store) if config.upload.url else None
+ recovery = HistoryRecovery(config.history, config.collector.source_id, store) if config.history.enabled else None
+ worker = None
+ history_worker = None
+ status_server = None
+ try:
+ for name in (signal.SIGINT, signal.SIGTERM):
+ previous[name] = signal.signal(name, lambda *_: stop.set())
+ if config.http.enabled:
+ status_server = StatusServer(config, store)
+ status_server.start()
+ LOG.info("Status page listening on http://%s:%s", *status_server.address)
+ if uploader:
+ # Remote timeouts and retry backoff must not interrupt local recording.
+ worker = threading.Thread(target=uploader.run, args=(stop,), name="solar-upload", daemon=True)
+ worker.start()
+ else:
+ LOG.info("Upload URL is empty; recording locally only")
+ if recovery:
+ history_worker = threading.Thread(target=recovery.run, args=(stop,), name="solar-history", daemon=True)
+ history_worker.start()
+ client = FroniusClient(config.inverter)
+ next_prune = 0
+ while not stop.is_set():
+ if time.monotonic() >= next_prune:
+ deleted = store.prune()
+ if deleted:
+ LOG.info("Deleted %d acknowledged observations older than the retention period", deleted)
+ next_prune = time.monotonic() + 3600
+ interval = config.collector.poll_interval_seconds
+ now = time.time()
+ # Wait before every observation, including the first after a restart.
+ # Recompute after each poll so a slow poll never triggers a catch-up burst.
+ if stop.wait(min(interval, max(0, next_poll_at(now, interval) - now))):
+ break
+ _record(client, store, config.collector.source_id, interval)
+ finally:
+ stop.set()
+ if status_server:
+ status_server.close()
+ if worker:
+ worker.join(timeout=config.upload.request_timeout_seconds + 1)
+ if history_worker:
+ history_worker.join(timeout=config.history.request_timeout_seconds + 1)
+ for name, handler in previous.items():
+ signal.signal(name, handler)
+ return 0
+
+
+def _parser():
+ parser = argparse.ArgumentParser(description="Record local Fronius readings and relay them to a website")
+ parser.add_argument("--version", action="version", version=__version__)
+ parser.add_argument("--config", type=Path, default=Path("config.toml"))
+ parser.add_argument("--log-level", choices=("DEBUG", "INFO", "WARNING", "ERROR"), default="INFO")
+ commands = parser.add_subparsers(dest="command", required=True)
+ commands.add_parser("run", help="Record continuously, uploading in a separate worker when configured")
+ commands.add_parser("sample", help="Record one observation and print its normalized readings")
+ 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 records acknowledged more than retention_days ago")
+ 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"),
+ ]:
+ command = commands.add_parser(name, help=help_text)
+ command.add_argument("--from", dest="from_date", help="Inclusive local date or ISO timestamp")
+ command.add_argument("--until", help="Exclusive local date or ISO timestamp")
+ return parser
+
+
+def main(argv=None):
+ args = _parser().parse_args(argv)
+ logging.basicConfig(level=args.log_level, format="%(asctime)s %(levelname)s %(message)s")
+ try:
+ config = load_config(args.config)
+ start = until = None
+ if args.command in ("export", "replay", "recover"):
+ start = date_bound(args.from_date, config.collector.timezone)
+ until = date_bound(args.until, config.collector.timezone)
+ if start is not None and until is not None and start >= until:
+ raise ValueError("--from must be earlier than --until")
+ store = Store(config.collector.archive, config.collector.retention_days)
+ if args.command == "run":
+ return _run(config, store)
+ if args.command == "sample":
+ sample = _record(FroniusClient(config.inverter), store, config.collector.source_id, config.collector.poll_interval_seconds)
+ print(json.dumps({key: value for key, value in sample.items() if key != "raw"}, indent=2))
+ return 1 if sample["errors"] else 0
+ if args.command == "status":
+ print(json.dumps(store.status(), indent=2))
+ elif args.command == "prune":
+ print(json.dumps({"deleted": store.prune()}))
+ elif args.command == "recover":
+ if not config.history.enabled:
+ raise ValueError("Enable history and configure Solar.web API credentials before recovery")
+ recovery = HistoryRecovery(config.history, config.collector.source_id, store)
+ print(json.dumps({"recovered": recovery.recover(start, until)}))
+ elif args.command == "export":
+ for sample in store.records(start, until):
+ print(json.dumps(sample, allow_nan=False, separators=(",", ":")))
+ elif args.command == "replay":
+ print(json.dumps({"requeued": store.replay(start, until)}))
+ elif args.command == "upload":
+ store.prune()
+ uploader = Uploader(config.upload, store)
+ count = 0
+ while uploaded := uploader.upload_once():
+ count += uploaded
+ print(json.dumps({"uploaded": count, "pending": store.status()["pending"]}))
+ return 0
+ except BrokenPipeError:
+ return 0
+ except (ValueError, TypeError, OSError, UploadError, TransportError, ZoneInfoNotFoundError) as error:
+ LOG.error("%s", error)
+ return 1
+ except KeyboardInterrupt:
+ return 130
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/solar_relay/config.py b/solar_relay/config.py
new file mode 100644
index 0000000..7fa2228
--- /dev/null
+++ b/solar_relay/config.py
@@ -0,0 +1,195 @@
+"""Configuration contains environment variable names, never upload credentials."""
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import math
+from pathlib import Path
+import re
+import tomllib
+from urllib.parse import urlsplit
+from zoneinfo import ZoneInfo
+
+
+def utc_text(value: datetime) -> str:
+ if value.tzinfo is None:
+ raise ValueError("A timestamp must include its timezone")
+ return value.astimezone(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
+
+
+def date_bound(value: str | None, timezone_name: str) -> str | None:
+ """Date bounds are local midnights; timestamp bounds may specify an offset."""
+ if value is None:
+ return None
+ parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+ if parsed.tzinfo is None:
+ parsed = parsed.replace(tzinfo=ZoneInfo(timezone_name))
+ return utc_text(parsed)
+
+
+@dataclass(frozen=True)
+class CollectorConfig:
+ source_id: str
+ archive: Path
+ timezone: str = "Europe/Zurich"
+ poll_interval_seconds: float = 300
+ retention_days: int = 30
+
+
+@dataclass(frozen=True)
+class InverterConfig:
+ base_url: str = "http://fronius.local"
+ meter_id: str = "0"
+ boiler_pin: str = "pin1"
+ request_timeout_seconds: float = 5
+ min_request_interval_seconds: float = 4
+
+
+@dataclass(frozen=True)
+class UploadConfig:
+ url: str = ""
+ token_env: str = "SOLAR_RELAY_TOKEN"
+ batch_size: int = 100
+ request_timeout_seconds: float = 10
+ retry_initial_seconds: float = 5
+ retry_max_seconds: float = 300
+
+
+@dataclass(frozen=True)
+class HttpConfig:
+ enabled: bool = True
+ bind: str = "127.0.0.1"
+ port: int = 8765
+ base_path: str = "/solar-relay"
+ history_days: int = 14
+
+
+@dataclass(frozen=True)
+class HistoryConfig:
+ enabled: bool = False
+ pv_system_id: str = ""
+ access_key_id_env: str = "SOLARWEB_ACCESS_KEY_ID"
+ access_key_value_env: str = "SOLARWEB_ACCESS_KEY_VALUE"
+ lookback_days: int = 7
+ lag_seconds: int = 7200
+ check_interval_seconds: int = 3600
+ request_timeout_seconds: float = 10
+
+
+@dataclass(frozen=True)
+class Config:
+ collector: CollectorConfig
+ inverter: InverterConfig
+ upload: UploadConfig
+ http: HttpConfig = HttpConfig()
+ history: HistoryConfig = HistoryConfig()
+
+
+def _number(value: object, name: str, minimum: float) -> float:
+ if isinstance(value, bool) or not isinstance(value, (int, float)):
+ raise ValueError(f"{name} must be a number")
+ if not math.isfinite(value) or value < minimum:
+ raise ValueError(f"{name} must be at least {minimum}")
+ return float(value)
+
+
+def _url(value: str, name: str, *, upload: bool = False) -> None:
+ if not isinstance(value, str):
+ raise ValueError(f"{name} must be a URL string")
+ parsed = urlsplit(value)
+ if parsed.scheme not in ("http", "https") or not parsed.hostname:
+ raise ValueError(f"{name} must be an HTTP(S) URL")
+ if parsed.username or parsed.password or parsed.fragment:
+ raise ValueError(f"{name} cannot contain credentials or a fragment")
+ if upload and parsed.scheme != "https" and parsed.hostname not in ("localhost", "127.0.0.1", "::1"):
+ raise ValueError("The upload URL must use HTTPS (HTTP is allowed only on loopback for testing)")
+ # Validate the port now, rather than failing inside the recorder loop.
+ parsed.port
+
+
+def load_config(path: Path) -> Config:
+ path = path.resolve()
+ with path.open("rb") as handle:
+ data = tomllib.load(handle)
+ unknown = set(data) - {"collector", "inverter", "upload", "http", "history"}
+ if unknown:
+ raise ValueError(f"Unknown configuration sections: {', '.join(sorted(unknown))}")
+ for section, keys in {
+ "collector": set(CollectorConfig.__dataclass_fields__),
+ "inverter": set(InverterConfig.__dataclass_fields__),
+ "upload": set(UploadConfig.__dataclass_fields__),
+ "http": set(HttpConfig.__dataclass_fields__),
+ "history": set(HistoryConfig.__dataclass_fields__),
+ }.items():
+ if not isinstance(data.get(section, {}), dict):
+ raise ValueError(f"{section} must be a table")
+ extra = set(data.get(section, {})) - keys
+ if extra:
+ raise ValueError(f"Unknown {section} settings: {', '.join(sorted(extra))}")
+ values = data.get("collector", {})
+ source_id = values.get("source_id", "home-gen24")
+ if not isinstance(source_id, str) or not re.fullmatch(r"[A-Za-z0-9_.-]{1,80}", source_id):
+ raise ValueError("collector.source_id must contain 1–80 letters, digits, dots, underscores or hyphens")
+ archive = Path(values.get("archive", "data/archive")).expanduser()
+ if not archive.is_absolute():
+ archive = path.parent / archive
+ timezone_name = values.get("timezone", "Europe/Zurich")
+ ZoneInfo(timezone_name)
+ interval = _number(values.get("poll_interval_seconds", 300), "collector.poll_interval_seconds", 10)
+ if interval != 300:
+ raise ValueError("collector.poll_interval_seconds must be 300 (one observation every five minutes)")
+ retention = values.get("retention_days", 30)
+ if isinstance(retention, bool) or not isinstance(retention, int) or retention < 1:
+ raise ValueError("collector.retention_days must be a positive integer")
+ collector = CollectorConfig(source_id, archive.resolve(), timezone_name, interval, retention)
+ inverter = InverterConfig(**data.get("inverter", {}))
+ _url(inverter.base_url, "inverter.base_url")
+ parsed = urlsplit(inverter.base_url)
+ if parsed.path not in ("", "/") or parsed.query:
+ raise ValueError("inverter.base_url must be the inverter origin, without a path or query")
+ if not isinstance(inverter.meter_id, str) or not inverter.meter_id.isdecimal():
+ raise ValueError('inverter.meter_id must be a string such as "0"')
+ if not isinstance(inverter.boiler_pin, str) or not re.fullmatch(r"pin[1-4]", inverter.boiler_pin):
+ raise ValueError("inverter.boiler_pin must be pin1, pin2, pin3 or pin4")
+ _number(inverter.request_timeout_seconds, "inverter.request_timeout_seconds", 0.1)
+ _number(inverter.min_request_interval_seconds, "inverter.min_request_interval_seconds", 4)
+ upload = UploadConfig(**data.get("upload", {}))
+ if not isinstance(upload.url, str):
+ raise ValueError("upload.url must be a string")
+ if upload.url:
+ _url(upload.url, "upload.url", upload=True)
+ if not isinstance(upload.token_env, str) or not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", upload.token_env):
+ raise ValueError("upload.token_env must be an environment variable name")
+ if isinstance(upload.batch_size, bool) or not isinstance(upload.batch_size, int) or not 1 <= upload.batch_size <= 500:
+ raise ValueError("upload.batch_size must be an integer between 1 and 500")
+ _number(upload.request_timeout_seconds, "upload.request_timeout_seconds", 0.1)
+ _number(upload.retry_initial_seconds, "upload.retry_initial_seconds", 1)
+ _number(upload.retry_max_seconds, "upload.retry_max_seconds", upload.retry_initial_seconds)
+ http = HttpConfig(**data.get("http", {}))
+ if not isinstance(http.enabled, bool):
+ raise ValueError("http.enabled must be true or false")
+ if http.bind != "127.0.0.1":
+ raise ValueError("http.bind must be 127.0.0.1; expose the page through a reverse proxy")
+ if isinstance(http.port, bool) or not isinstance(http.port, int) or not 1024 <= http.port <= 65535:
+ raise ValueError("http.port must be an integer between 1024 and 65535")
+ if not isinstance(http.base_path, str) or (http.base_path and not re.fullmatch(r"/[A-Za-z0-9_-]+(?:/[A-Za-z0-9_-]+)*", http.base_path)):
+ 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(**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)):
+ raise ValueError("history.pv_system_id must be a UUID")
+ if history.enabled and not history.pv_system_id:
+ raise ValueError("Set history.pv_system_id before enabling Solar.web recovery")
+ for name in ("access_key_id_env", "access_key_value_env"):
+ if not isinstance(getattr(history, name), str) or not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", getattr(history, name)):
+ raise ValueError(f"history.{name} must name an environment variable")
+ for name, minimum in [("lookback_days", 1), ("lag_seconds", 3600), ("check_interval_seconds", 3600)]:
+ value = getattr(history, name)
+ if isinstance(value, bool) or not isinstance(value, int) or value < minimum:
+ raise ValueError(f"history.{name} must be an integer of at least {minimum}")
+ if history.lookback_days > 30 or (history.enabled and history.lookback_days > retention):
+ raise ValueError("history.lookback_days must be at most 30 and no longer than retention_days when enabled")
+ _number(history.request_timeout_seconds, "history.request_timeout_seconds", 0.1)
+ return Config(collector, inverter, upload, http, history)
diff --git a/solar_relay/coverage.py b/solar_relay/coverage.py
new file mode 100644
index 0000000..abdef41
--- /dev/null
+++ b/solar_relay/coverage.py
@@ -0,0 +1,88 @@
+"""Five-minute extraction completeness, distinct from continuous observed coverage."""
+
+from collections import Counter
+from datetime import datetime, timedelta, timezone
+import math
+from zoneinfo import ZoneInfo
+
+from .config import utc_text
+
+BUCKET_MS = 300_000
+MAX_SOURCE_AGE_MS = 90_000
+
+
+def milliseconds(value):
+ parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+ if parsed.tzinfo is None:
+ raise ValueError("Observation timestamp has no timezone")
+ return round(parsed.timestamp() * 1000)
+
+
+def _finite(value):
+ return not isinstance(value, bool) and isinstance(value, (int, float)) and math.isfinite(value)
+
+
+def sample_coverage(sample):
+ observed = milliseconds(sample["observed_at"])
+ if "history_interval" in sample:
+ # Cloud history has measured 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 [
+ ("powerflow", ["production_ac_w", "consumption_w", "grid_import_w", "grid_export_w"]),
+ ("meter", ["grid_import_total_wh", "grid_export_total_wh"]),
+ ("boiler", []),
+ ]:
+ try:
+ stamp = milliseconds(sample["source_timestamps"][source])
+ valid = source not in sample.get("errors", {}) and abs(stamp - observed) <= MAX_SOURCE_AGE_MS
+ except (KeyError, ValueError, TypeError, AttributeError):
+ stamp, valid = observed, False
+ valid = valid and all(_finite(readings.get(field)) for field in fields)
+ if source == "boiler":
+ valid = valid and isinstance(readings.get("boiler_enabled"), bool)
+ result.extend([stamp if valid else observed, int(valid)])
+ return tuple(result)
+
+
+def daily_coverage(rows, timezone_name, days, now=None):
+ now = now or datetime.now(timezone.utc)
+ zone = ZoneInfo(timezone_name)
+ today = now.astimezone(zone).date()
+ first = today - timedelta(days=days)
+ upper = round(now.timestamp() * 1000)
+ slots_by_time = {}
+ observations = Counter()
+ for row in rows:
+ observed = row[0]
+ if observed > upper + 1000:
+ continue
+ date = datetime.fromtimestamp(observed / 1000, zone).date().isoformat()
+ observations[date] += 1
+ bucket = observed // BUCKET_MS * BUCKET_MS
+ mask = sum(1 << index for index in range(3) if row[2 + index * 2])
+ slots_by_time[bucket] = slots_by_time.get(bucket, 0) | mask
+ result = []
+ for offset in range(days + 1):
+ date = today - timedelta(days=offset)
+ start = datetime.combine(date, datetime.min.time(), zone)
+ end = datetime.combine(date + timedelta(days=1), datetime.min.time(), zone)
+ start_ms = round(start.timestamp() * 1000)
+ expected = round((end.timestamp() - start.timestamp()) * 1000) // BUCKET_MS
+ elapsed = expected if date < today else max(0, min(expected, (upper - start_ms) // BUCKET_MS))
+ slots = []
+ for index in range(expected):
+ slots.append(slots_by_time.get(start_ms + index * BUCKET_MS, 0) if index < elapsed else -1)
+ complete = sum(mask == 7 for mask in slots)
+ count = observations[date.isoformat()]
+ status = "in_progress" if date == today else "complete" if complete == expected else "partial" if count else "missing"
+ result.append({
+ "date": date.isoformat(), "day_start": utc_text(start), "expected": expected, "elapsed": elapsed,
+ "complete": complete, "missing": elapsed - complete, "samples": count, "status": status,
+ "energy_complete": sum(mask >= 0 and bool(mask & 1) for mask in slots),
+ "meter_complete": sum(mask >= 0 and bool(mask & 2) for mask in slots),
+ "boiler_complete": sum(mask >= 0 and bool(mask & 4) for mask in slots),
+ "slots": slots,
+ })
+ return result
diff --git a/solar_relay/fronius.py b/solar_relay/fronius.py
new file mode 100644
index 0000000..a1e70e9
--- /dev/null
+++ b/solar_relay/fronius.py
@@ -0,0 +1,154 @@
+"""Read Solar API and GEN24 load-management state without changing the inverter."""
+
+from datetime import datetime, timezone
+import logging
+import math
+import time
+from uuid import uuid4
+
+from .config import InverterConfig, utc_text
+from .http import JsonHttp, TransportError
+
+LOG = logging.getLogger(__name__)
+REQUEST_RETRIES = 3
+
+
+def _number(value):
+ if value is None:
+ return None
+ if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
+ raise ValueError("Invalid numeric reading")
+ return value
+
+
+def _sum_known(values):
+ values = list(values)
+ return sum(values) if values and all(v is not None for v in values) else None
+
+
+def _data(response):
+ head = response.get("Head", {})
+ if head.get("Status", {}).get("Code") != 0:
+ raise ValueError("Solar API reported an unsuccessful status")
+ data = response.get("Body", {}).get("Data")
+ if not isinstance(data, dict):
+ raise ValueError("Solar API data is missing")
+ timestamp = datetime.fromisoformat(head["Timestamp"].replace("Z", "+00:00"))
+ return data, utc_text(timestamp)
+
+
+def powerflow_readings(response):
+ data, timestamp = _data(response)
+ site = data.get("Site")
+ inverters = data.get("Inverters", {})
+ if not isinstance(site, dict) or not isinstance(inverters, dict):
+ raise ValueError("Solar API site/inverter data is missing")
+ if not {"P_PV", "P_Load", "P_Grid"}.issubset(site):
+ raise ValueError("Required Solar API power fields are missing")
+ load = _number(site.get("P_Load"))
+ grid = _number(site.get("P_Grid"))
+ total = _number(site.get("E_Total"))
+ if total is None:
+ total = _sum_known(_number(device.get("E_Total")) for device in inverters.values())
+ return {
+ # GEN24 P_PV is DC production; keep AC output separately instead of mixing them.
+ "production_w": _number(site.get("P_PV")),
+ "production_ac_w": _sum_known(_number(device.get("P")) for device in inverters.values()),
+ "consumption_w": -load if load is not None else None,
+ "grid_w": grid, # Positive: import. Negative: export (feed-in point meter).
+ "grid_import_w": max(0, grid) if grid is not None else None,
+ "grid_export_w": max(0, -grid) if grid is not None else None,
+ "production_ac_total_wh": total,
+ }, timestamp
+
+
+def meter_readings(response, meter_id):
+ data, timestamp = _data(response)
+ meter = data.get(meter_id)
+ if not isinstance(meter, dict):
+ raise ValueError(f"Configured meter {meter_id} is missing")
+ if meter.get("Enable") not in (1, 1.0):
+ raise ValueError("Configured meter is disabled")
+ if meter.get("Meter_Location_Current") != 0:
+ raise ValueError("Configured meter must be at the grid feed-in point")
+ if not {"EnergyReal_WAC_Sum_Consumed", "EnergyReal_WAC_Sum_Produced"}.issubset(meter):
+ raise ValueError("Required meter energy counters are missing")
+ if meter.get("TimeStamp") is not None:
+ try:
+ timestamp = utc_text(datetime.fromtimestamp(_number(meter["TimeStamp"]), timezone.utc))
+ except (OverflowError, OSError):
+ raise ValueError("Invalid meter measurement timestamp") from None
+ return {
+ "grid_import_total_wh": _number(meter.get("EnergyReal_WAC_Sum_Consumed")),
+ "grid_export_total_wh": _number(meter.get("EnergyReal_WAC_Sum_Produced")),
+ }, timestamp
+
+
+def boiler_readings(response, pin):
+ state = response.get(pin)
+ if not isinstance(state, dict) or not isinstance(state.get("state"), bool):
+ raise ValueError(f"Boiler output {pin} has no boolean state")
+ reason = state.get("reason")
+ if reason is not None and not isinstance(reason, str):
+ raise ValueError("Boiler reason is not a string")
+ return {"boiler_enabled": state["state"], "boiler_reason": reason}
+
+
+class FroniusClient:
+ def __init__(self, config: InverterConfig, *, http=None):
+ self.config = config
+ self.http = http or JsonHttp(config.request_timeout_seconds, local=True)
+ self.last_request_started = None
+
+ def _read(self, path):
+ for attempt in range(REQUEST_RETRIES + 1):
+ if self.last_request_started is not None:
+ wait = self.config.min_request_interval_seconds - (time.monotonic() - self.last_request_started)
+ if wait > 0:
+ time.sleep(wait)
+ self.last_request_started = time.monotonic()
+ try:
+ return self.http.request(self.config.base_url.rstrip("/") + path)
+ except TransportError as error:
+ if attempt == REQUEST_RETRIES:
+ raise
+ LOG.warning("Fronius GET %s failed (%s); retry %d/%d", path, error, attempt + 1, REQUEST_RETRIES)
+
+ def collect(self, source_id, poll_interval_seconds=300):
+ sample = {
+ "schema_version": 1,
+ "id": str(uuid4()),
+ "source_id": source_id,
+ "poll_interval_seconds": poll_interval_seconds,
+ "observed_at": utc_text(datetime.now(timezone.utc)),
+ "inverter": {"meter_id": self.config.meter_id, "boiler_pin": self.config.boiler_pin},
+ "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": {},
+ "raw": {},
+ }
+ for name, path in [
+ ("powerflow", "/solar_api/v1/GetPowerFlowRealtimeData.fcgi"),
+ ("meter", "/solar_api/v1/GetMeterRealtimeData.cgi?Scope=System"),
+ ("boiler", "/api/status/emrs"),
+ ]:
+ try:
+ response = self._read(path)
+ sample["raw"][name] = response
+ if name == "powerflow":
+ readings, timestamp = powerflow_readings(response)
+ elif name == "meter":
+ readings, timestamp = meter_readings(response, self.config.meter_id)
+ else:
+ readings = boiler_readings(response, self.config.boiler_pin)
+ timestamp = utc_text(datetime.now(timezone.utc))
+ sample["readings"].update(readings)
+ sample["source_timestamps"][name] = timestamp
+ except (TransportError, ValueError, KeyError, TypeError, AttributeError) as error:
+ sample["errors"][name] = str(error)
+ return sample
diff --git a/solar_relay/history.py b/solar_relay/history.py
new file mode 100644
index 0000000..fc34918
--- /dev/null
+++ b/solar_relay/history.py
@@ -0,0 +1,129 @@
+"""Optional private Solar.web Query API recovery of missing five-minute energy.
+
+GEN24's local Solar API exposes no archive. This reads the independent cloud API
+after the inverter has uploaded its internal backlog; it never requests a guest page.
+"""
+
+from datetime import datetime, timedelta, timezone
+import logging
+import math
+import os
+from urllib.parse import urlencode
+from uuid import NAMESPACE_URL, uuid5
+
+from .config import utc_text
+from .coverage import milliseconds
+from .http import JsonHttp, TransportError
+
+LOG = logging.getLogger(__name__)
+CHANNELS = {
+ "production": "EnergyProductionTotal", "consumption": "EnergyConsumptionTotal",
+ "self_consumption": "EnergySelfConsumptionTotal", "bought": "EnergyPurchased", "sold": "EnergyFeedIn",
+}
+
+
+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()}
+ when = utc_text(datetime.fromtimestamp(stamp / 1000, timezone.utc))
+ return {
+ "schema_version": 1,
+ "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",
+ "boiler": "Historical boiler switch state is unavailable"},
+ "raw": {"solarweb_history": entry},
+ "history_interval": {"provider": "fronius-solarweb", "start": when, "duration_seconds": 300, "energy_wh": energy},
+ }
+
+
+class HistoryRecovery:
+ def __init__(self, config, source_id, store, *, http=None, credentials=None):
+ self.config, self.source_id, self.store = config, source_id, store
+ names = (config.access_key_id_env, config.access_key_value_env)
+ values = credentials if credentials is not None else [os.environ.get(name, "") for name in names]
+ if len(values) != 2 or any(not isinstance(value, str) or not value or any(ord(c) < 33 or ord(c) > 126 for c in value) for value in values):
+ raise ValueError(f"Set {names[0]} and {names[1]} before enabling Solar.web recovery")
+ self.headers = dict(zip(("AccessKeyId", "AccessKeyValue"), values))
+ self.http = http or JsonHttp(config.request_timeout_seconds)
+
+ def recover(self, start=None, until=None, *, now=None, stop=None):
+ now = now or datetime.now(timezone.utc)
+ latest = int((now - timedelta(seconds=self.config.lag_seconds)).timestamp()) // 300 * 300
+ first = self.store.status()["first_observation"]
+ earliest = max(int((now - timedelta(days=self.config.lookback_days)).timestamp()) // 300 * 300,
+ milliseconds(first) // 300_000 * 300 if first else latest)
+ lower = milliseconds(start) // 300_000 * 300 if start else earliest
+ upper = min(milliseconds(until) // 300_000 * 300, latest) if until else latest
+ if lower >= upper:
+ return 0
+ if upper - lower > 30 * 86400:
+ raise ValueError("Recover at most 30 days at once")
+ stamp = lambda epoch: utc_text(datetime.fromtimestamp(epoch, timezone.utc))
+ present = self.store.energy_slots(self.source_id, stamp(lower), stamp(upper))
+ missing = set(range(lower, upper, 300)) - present
+ # Query only UTC days containing gaps. No request spans the API's 24-hour limit.
+ days = sorted({epoch // 86400 * 86400 for epoch in missing})
+ recovered = 0
+ for day in days:
+ if stop and stop.is_set():
+ break
+ begin, end = max(day, lower), min(day + 86400, upper)
+ page = 0
+ while True:
+ if stop and stop.is_set():
+ break
+ query = urlencode({"from": stamp(begin), "to": stamp(end), "timezone": "zulu",
+ "channel": ",".join(CHANNELS.values()), "offset": page * 300, "limit": 300})
+ url = f"https://swqapi.solarweb.com/pvsystems/{self.config.pv_system_id}/histdata?{query}"
+ response = self.http.request(url, extra_headers=self.headers)
+ system_id = response.get("pvSystemId")
+ if not isinstance(system_id, str) or system_id.lower() != self.config.pv_system_id.lower() or not isinstance(response.get("data"), list):
+ raise ValueError("Unexpected Solar.web history response")
+ entries = response["data"]
+ if len(entries) > 300:
+ raise ValueError("Solar.web history response exceeds the requested page size")
+ for entry in entries:
+ try:
+ sample = history_sample(entry, self.source_id, self.config.pv_system_id)
+ epoch = milliseconds(sample["observed_at"]) // 1000
+ except (KeyError, TypeError, ValueError, AttributeError):
+ # Partial/unavailable cloud records remain missing and are tried again later.
+ continue
+ if begin <= epoch < end and epoch in missing:
+ self.store.add(sample)
+ missing.remove(epoch)
+ recovered += 1
+ if len(entries) < 300:
+ break
+ page += 1
+ if page > 2:
+ raise ValueError("Solar.web history pagination did not finish")
+ return recovered
+
+ def run(self, stop):
+ while not stop.is_set():
+ try:
+ recovered = self.recover(stop=stop)
+ if recovered:
+ LOG.info("Recovered %d historical energy intervals from Solar.web", recovered)
+ except (TransportError, OSError, ValueError) as error:
+ LOG.warning("Historical recovery failed (%s); retrying on the next scheduled check", type(error).__name__)
+ stop.wait(self.config.check_interval_seconds)
diff --git a/solar_relay/http.py b/solar_relay/http.py
new file mode 100644
index 0000000..89e8017
--- /dev/null
+++ b/solar_relay/http.py
@@ -0,0 +1,63 @@
+"""Bounded JSON requests; redirects cannot forward upload credentials."""
+
+import json
+from http.client import HTTPException
+from urllib.error import HTTPError, URLError
+from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener
+
+from . import __version__
+
+
+class TransportError(Exception):
+ pass
+
+
+class NoRedirects(HTTPRedirectHandler):
+ def redirect_request(self, request, response, code, message, headers, new_url):
+ return None
+
+
+def _invalid_constant(value):
+ raise ValueError("Non-finite JSON number")
+
+
+class JsonHttp:
+ MAX_RESPONSE_BYTES = 2 * 1024 * 1024
+
+ def __init__(self, timeout: float, *, local: bool = False):
+ handlers = [NoRedirects()]
+ if local:
+ # LAN traffic must not go through a workstation's HTTP(S) proxy.
+ handlers.append(ProxyHandler({}))
+ self.opener = build_opener(*handlers)
+ self.timeout = timeout
+
+ def request(self, url: str, *, payload: dict | None = None, token: str | None = None, extra_headers=None) -> dict:
+ headers = {"Accept": "application/json", "User-Agent": f"solar-relay/{__version__}"}
+ headers.update(extra_headers or {})
+ body = None
+ if payload is not None:
+ body = json.dumps(payload, allow_nan=False, separators=(",", ":")).encode("utf-8")
+ headers["Content-Type"] = "application/json"
+ if token:
+ headers["Authorization"] = f"Bearer {token}"
+ request = Request(url, data=body, headers=headers, method="POST" if body is not None else "GET")
+ try:
+ with self.opener.open(request, timeout=self.timeout) as response:
+ raw = response.read(self.MAX_RESPONSE_BYTES + 1)
+ except HTTPError as error:
+ # Do not log server bodies: they can echo request headers or credentials.
+ error.close()
+ raise TransportError(f"HTTP {error.code}") from None
+ except (URLError, OSError, TimeoutError, HTTPException) as error:
+ raise TransportError(f"Network request failed ({type(error).__name__})") from None
+ if len(raw) > self.MAX_RESPONSE_BYTES:
+ raise TransportError("JSON response exceeds 2 MiB")
+ try:
+ value = json.loads(raw, parse_constant=_invalid_constant)
+ except (ValueError, UnicodeError, RecursionError):
+ raise TransportError("Response is not valid JSON") from None
+ if not isinstance(value, dict):
+ raise TransportError("Expected a JSON object")
+ return value
+
diff --git a/solar_relay/status.py b/solar_relay/status.py
new file mode 100644
index 0000000..7f9a667
--- /dev/null
+++ b/solar_relay/status.py
@@ -0,0 +1,129 @@
+"""Loopback-only status page; Nginx provides HTTPS and the public path."""
+
+from datetime import datetime, timezone
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+from pathlib import Path
+import threading
+import time
+from urllib.parse import parse_qs, urlsplit
+
+from . import __version__
+from .config import utc_text
+from .coverage import milliseconds
+
+
+class StatusServer:
+ def __init__(self, config, store):
+ self.config, self.store = config, store
+ self.started = time.monotonic()
+ self.cache = {}
+ self.cache_lock = threading.Lock()
+ folder = Path(__file__).parent / "web"
+ self.assets = {
+ "/": ("text/html; charset=utf-8", (folder / "index.html").read_bytes()),
+ "/assets/status.css": ("text/css; charset=utf-8", (folder / "status.css").read_bytes()),
+ "/assets/status.js": ("text/javascript; charset=utf-8", (folder / "status.js").read_bytes()),
+ }
+ owner = self
+
+ class Handler(BaseHTTPRequestHandler):
+ server_version = "solar-relay"
+ sys_version = ""
+
+ def log_message(self, *_):
+ pass
+
+ def send(self, status, body, kind="application/json; charset=utf-8", *, head=False, headers=None):
+ if not isinstance(body, bytes):
+ body = json.dumps(body, allow_nan=False, separators=(",", ":")).encode()
+ self.send_response(status)
+ self.send_header("Content-Type", kind)
+ self.send_header("Content-Length", str(len(body)))
+ self.send_header("Cache-Control", "no-store")
+ self.send_header("X-Content-Type-Options", "nosniff")
+ self.send_header("Content-Security-Policy", "default-src 'self'; img-src 'self' data:; script-src 'self'; style-src 'self' 'unsafe-inline'; object-src 'none'; frame-ancestors 'none'; base-uri 'self'")
+ for key, value in (headers or {}).items():
+ self.send_header(key, value)
+ self.end_headers()
+ if not head:
+ self.wfile.write(body)
+
+ def page(self, head=False):
+ url = urlsplit(self.path)
+ path = url.path
+ prefix = owner.config.http.base_path
+ if prefix and path == prefix:
+ self.send(308, {}, head=head, headers={"Location": prefix + "/"})
+ return
+ if prefix and path.startswith(prefix + "/"):
+ path = path[len(prefix):]
+ if path in owner.assets:
+ kind, body = owner.assets[path]
+ self.send(200, body, kind, head=head)
+ elif path == "/healthz":
+ self.send(200, {"status": "ok", "service": "solar-relay", "version": __version__}, head=head)
+ elif path == "/api/status":
+ try:
+ values = parse_qs(url.query).get("days", [str(owner.config.http.history_days)])
+ if len(values) != 1 or not values[0].isdecimal() or not 1 <= int(values[0]) <= 90:
+ self.send(400, {"error": "days must be between 1 and 90"}, head=head)
+ return
+ self.send(200, owner.snapshot(int(values[0])), head=head)
+ except (OSError, ValueError):
+ self.send(503, {"error": "Archive status is temporarily unavailable"}, head=head)
+ else:
+ self.send(404, {"error": "Not found"}, head=head)
+
+ def do_GET(self):
+ self.page()
+
+ def do_HEAD(self):
+ self.page(head=True)
+
+ def do_POST(self):
+ self.send(405, {"error": "Read-only status endpoint"}, headers={"Allow": "GET, HEAD"})
+
+ self.server = ThreadingHTTPServer((config.http.bind, config.http.port), Handler)
+ self.thread = threading.Thread(target=self.server.serve_forever, kwargs={"poll_interval": 0.2}, name="solar-status", daemon=True)
+
+ @property
+ def address(self):
+ return self.server.server_address
+
+ def start(self):
+ self.thread.start()
+
+ def close(self):
+ self.server.shutdown()
+ self.server.server_close()
+ self.thread.join(timeout=2)
+
+ def snapshot(self, days):
+ # Repeated page refreshes don't repeatedly scan the historical coverage index.
+ with self.cache_lock:
+ cached = self.cache.get(days)
+ if cached and time.monotonic() - cached[0] < 10:
+ return cached[1]
+ archive = self.store.status()
+ now = datetime.now(timezone.utc)
+ attempt = archive["last_observation"]
+ age = (now.timestamp() * 1000 - milliseconds(attempt)) / 1000 if attempt else None
+ errors = list(archive["last_errors"])
+ success = self.store.last_success(self.config.collector.source_id)
+ status = "waiting" if age is None else "stale" if age > max(90, self.config.collector.poll_interval_seconds * 3) or age < -90 else "partial" if errors or success != attempt else "ok"
+ body = {
+ "service": "solar-relay", "version": __version__, "generated_at": utc_text(now),
+ "timezone": self.config.collector.timezone, "source_id": self.config.collector.source_id,
+ "status": status, "uptime_seconds": int(time.monotonic() - self.started),
+ "collector": {"last_attempt": attempt, "last_success": success,
+ "age_seconds": round(max(0, age), 1) if age is not None else None, "source_errors": errors,
+ "poll_interval_seconds": self.config.collector.poll_interval_seconds},
+ "archive": {"samples": archive["samples"], "pending": archive["pending"], "last_upload": archive["last_upload"],
+ "upload_enabled": bool(self.config.upload.url), "format": "parquet", "retention_days": self.config.collector.retention_days},
+ "history": {"enabled": self.config.history.enabled, "provider": "solarweb_query_api",
+ "boiler_history_available": False},
+ "days": self.store.coverage(self.config.collector.source_id, self.config.collector.timezone, days, now),
+ }
+ self.cache[days] = time.monotonic(), body
+ return body
diff --git a/solar_relay/storage.py b/solar_relay/storage.py
new file mode 100644
index 0000000..45495b9
--- /dev/null
+++ b/solar_relay/storage.py
@@ -0,0 +1,263 @@
+"""Durable, date-partitioned Parquet archive and acknowledged upload queue.
+
+Each mutation replaces a daily partition atomically. A process/thread lock serializes
+readers and writers; it is never held during an HTTP request.
+"""
+
+from contextlib import contextmanager
+from datetime import datetime, timedelta, timezone
+import fcntl
+import json
+import os
+from pathlib import Path
+import tempfile
+import threading
+import time
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+
+from .config import utc_text
+from .coverage import daily_coverage, milliseconds, sample_coverage
+
+NUMBERS = (
+ "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",
+)
+SCHEMA = pa.schema([
+ ("id", pa.string()), ("source_id", pa.string()), ("observed_at", pa.string()),
+ ("observed_ms", pa.int64()), ("recorded_ns", pa.int64()),
+ ("uploaded_at", pa.string()), ("payload", pa.string()),
+ *((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_()),
+], metadata={b"solar_relay_schema": b"1"})
+COVERAGE_COLUMNS = ["observed_ms", "power_ms", "power_ok", "meter_ms", "meter_ok", "boiler_ms", "boiler_ok"]
+
+
+class StoreError(OSError):
+ """An archive could not be read or updated safely."""
+
+
+class Store:
+ def __init__(self, path: Path, retention_days=30):
+ self.path = Path(path)
+ if self.path.is_file():
+ raise StoreError("collector.archive must be a directory")
+ self.path.mkdir(parents=True, exist_ok=True, mode=0o700)
+ self.retention_days = retention_days
+ self._thread_lock = threading.RLock()
+ descriptor = os.open(self.path / ".lock", os.O_CREAT | os.O_WRONLY, 0o600)
+ os.close(descriptor)
+
+ @contextmanager
+ def _locked(self, *, write=False):
+ with self._thread_lock, (self.path / ".lock").open("rb") as handle:
+ fcntl.flock(handle, fcntl.LOCK_EX if write else fcntl.LOCK_SH)
+ try:
+ yield
+ except pa.ArrowException as error:
+ raise StoreError(f"Parquet archive error ({type(error).__name__})") from error
+ finally:
+ fcntl.flock(handle, fcntl.LOCK_UN)
+
+ def _files(self):
+ return sorted(self.path.glob("date=????-??-??/samples.parquet"))
+
+ def _partition(self, stamp):
+ date = datetime.fromtimestamp(milliseconds(stamp) / 1000, timezone.utc).date()
+ return self.path / f"date={date.isoformat()}" / "samples.parquet"
+
+ @staticmethod
+ def _read(path, columns=None):
+ # Avoid implicit Hive partition/schema inference from the directory name.
+ if not path.exists():
+ return []
+ parquet = pq.ParquetFile(path, page_checksum_verification=True)
+ if not parquet.schema_arrow.equals(SCHEMA, check_metadata=True):
+ raise StoreError("Unsupported Parquet archive schema")
+ return parquet.read(columns=columns).to_pylist()
+
+ @staticmethod
+ def _sync_directory(path):
+ descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY)
+ try:
+ os.fsync(descriptor)
+ finally:
+ os.close(descriptor)
+
+ def _write(self, path, rows):
+ new_directory = not path.parent.exists()
+ path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+ if new_directory:
+ self._sync_directory(self.path)
+ if not rows:
+ path.unlink(missing_ok=True)
+ self._sync_directory(path.parent)
+ return
+ descriptor, temporary = tempfile.mkstemp(prefix=".samples-", suffix=".tmp", dir=path.parent)
+ try:
+ with os.fdopen(descriptor, "wb") as handle:
+ table = pa.Table.from_pylist(sorted(rows, key=lambda row: (row["observed_ms"], row["recorded_ns"], row["id"])), schema=SCHEMA)
+ pq.write_table(table, handle, compression="zstd", write_page_checksum=True)
+ handle.flush()
+ os.fsync(handle.fileno())
+ os.replace(temporary, path)
+ self._sync_directory(path.parent)
+ finally:
+ Path(temporary).unlink(missing_ok=True)
+
+ @staticmethod
+ def _row(sample, uploaded_at=None):
+ coverage = sample_coverage(sample)
+ readings = sample.get("readings", {})
+ return {
+ "id": sample["id"], "source_id": sample["source_id"], "observed_at": sample["observed_at"],
+ "observed_ms": coverage[2], "recorded_ns": time.time_ns(), "uploaded_at": uploaded_at,
+ "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"),
+ **{name: bool(value) if name.endswith("_ok") else value
+ for name, value in zip(COVERAGE_COLUMNS[1:], coverage[3:])},
+ }
+
+ def add(self, sample):
+ path, row = self._partition(sample["observed_at"]), self._row(sample)
+ with self._locked(write=True):
+ rows = self._read(path)
+ for existing in rows:
+ if existing["id"] == row["id"]:
+ if existing["payload"] != row["payload"]:
+ raise StoreError("Sample ID already contains different data")
+ return
+ self._write(path, [*rows, row])
+
+ def pending(self, limit):
+ with self._locked():
+ rows = []
+ for path in self._files():
+ index = self._read(path, ["uploaded_at"])
+ if any(row["uploaded_at"] is None for row in index):
+ rows.extend(row for row in self._read(path, ["recorded_ns", "id", "payload", "uploaded_at"]) if row["uploaded_at"] is None)
+ rows.sort(key=lambda row: (row["recorded_ns"], row["id"]))
+ return [json.loads(row["payload"]) for row in rows[:limit]]
+
+ def acknowledge(self, ids, *, now=None):
+ wanted, stamp = set(ids), utc_text(now or datetime.now(timezone.utc))
+ if not wanted:
+ return
+ with self._locked(write=True):
+ for path in self._files():
+ index = self._read(path, ["id", "uploaded_at"])
+ if not any(row["id"] in wanted and row["uploaded_at"] is None for row in index):
+ continue
+ rows = self._read(path)
+ for row in rows:
+ if row["id"] in wanted and row["uploaded_at"] is None:
+ row["uploaded_at"] = stamp
+ self._write(path, rows)
+
+ def records(self, start=None, until=None):
+ lower, upper = milliseconds(start) if start else None, milliseconds(until) if until else None
+ with self._locked():
+ rows = []
+ for path in self._files():
+ rows.extend(row for row in self._read(path, ["observed_ms", "recorded_ns", "id", "payload"])
+ if (lower is None or row["observed_ms"] >= lower) and (upper is None or row["observed_ms"] < upper))
+ rows.sort(key=lambda row: (row["observed_ms"], row["recorded_ns"], row["id"]))
+ return (json.loads(row["payload"]) for row in rows)
+
+ def replay(self, start=None, until=None):
+ lower, upper = milliseconds(start) if start else None, milliseconds(until) if until else None
+ count = 0
+ with self._locked(write=True):
+ for path in self._files():
+ rows, changed = self._read(path), False
+ for row in rows:
+ if (lower is None or row["observed_ms"] >= lower) and (upper is None or row["observed_ms"] < upper):
+ row["uploaded_at"] = None
+ count += 1
+ changed = True
+ if changed:
+ self._write(path, rows)
+ return count
+
+ def prune(self, *, now=None):
+ """Retain pending records; age acknowledged records from their receipt time."""
+ cutoff = utc_text((now or datetime.now(timezone.utc)) - timedelta(days=self.retention_days))
+ deleted = 0
+ with self._locked(write=True):
+ for path in self._files():
+ index = self._read(path, ["uploaded_at"])
+ if not any(row["uploaded_at"] is not None and row["uploaded_at"] < cutoff for row in index):
+ continue
+ rows = self._read(path)
+ retained = [row for row in rows if row["uploaded_at"] is None or row["uploaded_at"] >= cutoff]
+ self._write(path, retained)
+ deleted += len(rows) - len(retained)
+ return deleted
+
+ def status(self):
+ with self._locked():
+ rows = []
+ latest = None
+ for path in self._files():
+ index = self._read(path, ["observed_at", "observed_ms", "recorded_ns", "uploaded_at"])
+ rows.extend(index)
+ if index:
+ candidate = max(index, key=lambda row: (row["observed_ms"], row["recorded_ns"]))
+ if latest is None or (candidate["observed_ms"], candidate["recorded_ns"]) > latest[0]:
+ latest = ((candidate["observed_ms"], candidate["recorded_ns"]), path)
+ errors = {}
+ if latest:
+ newest = max(self._read(latest[1], ["observed_ms", "recorded_ns", "payload"]), key=lambda row: (row["observed_ms"], row["recorded_ns"]))
+ errors = json.loads(newest["payload"]).get("errors", {})
+ return {
+ "archive": str(self.path), "format": "parquet", "retention_days": self.retention_days,
+ "samples": len(rows), "pending": sum(row["uploaded_at"] is None for row in rows),
+ "first_observation": min((row["observed_at"] for row in rows), default=None),
+ "last_observation": max((row["observed_at"] for row in rows), default=None),
+ "last_upload": max((row["uploaded_at"] for row in rows if row["uploaded_at"]), default=None),
+ "last_errors": errors,
+ }
+
+ def coverage(self, source_id, timezone_name, days, now=None):
+ from zoneinfo import ZoneInfo
+ now = now or datetime.now(timezone.utc)
+ first = now.astimezone(ZoneInfo(timezone_name)).date() - timedelta(days=days)
+ lower = round(datetime.combine(first, datetime.min.time(), ZoneInfo(timezone_name)).timestamp() * 1000)
+ upper = round(now.timestamp() * 1000)
+ with self._locked():
+ rows = []
+ for path in self._files():
+ if path.parent.name[5:] < (now - timedelta(days=days + 2)).date().isoformat():
+ continue
+ rows.extend(tuple(row[name] for name in COVERAGE_COLUMNS)
+ for row in self._read(path, ["source_id", *COVERAGE_COLUMNS])
+ if row["source_id"] == source_id and lower <= row["observed_ms"] <= upper)
+ return daily_coverage(sorted(rows), timezone_name, days, now)
+
+ def last_success(self, source_id):
+ with self._locked():
+ for path in reversed(self._files()):
+ rows = self._read(path, ["source_id", "observed_ms", "power_ok", "meter_ok", "boiler_ok"])
+ stamps = [row["observed_ms"] for row in rows if row["source_id"] == source_id and row["power_ok"] and row["meter_ok"] and row["boiler_ok"]]
+ if stamps:
+ return utc_text(datetime.fromtimestamp(max(stamps) / 1000, timezone.utc))
+ return None
+
+ def energy_slots(self, source_id, start, until):
+ """Return UTC bucket starts already having valid local or recovered energy."""
+ lower, upper = milliseconds(start), milliseconds(until)
+ with self._locked():
+ return {row["observed_ms"] // 300_000 * 300
+ for path in self._files()
+ for row in self._read(path, ["source_id", "observed_ms", "power_ok"])
+ if row["source_id"] == source_id and row["power_ok"] and lower <= row["observed_ms"] < upper}
diff --git a/solar_relay/upload.py b/solar_relay/upload.py
new file mode 100644
index 0000000..4d6e26d
--- /dev/null
+++ b/solar_relay/upload.py
@@ -0,0 +1,64 @@
+"""Retryable delivery with explicit per-sample acknowledgement."""
+
+import logging
+import os
+
+from .config import UploadConfig
+from .http import JsonHttp, TransportError
+from .storage import Store
+
+LOG = logging.getLogger(__name__)
+
+
+class UploadError(Exception):
+ pass
+
+
+class Uploader:
+ def __init__(self, config: UploadConfig, store: Store, *, token=None, http=None):
+ if not config.url:
+ raise ValueError("Set upload.url to the website's ingestion endpoint first")
+ self.config = config
+ self.store = store
+ self.token = token if token is not None else os.environ.get(config.token_env, "")
+ if not self.token or any(ord(c) < 33 or ord(c) > 126 for c in self.token):
+ raise ValueError(f"Set {config.token_env} to a nonempty bearer token without whitespace")
+ self.http = http or JsonHttp(config.request_timeout_seconds)
+
+ def upload_once(self) -> int:
+ batch = self.store.pending(self.config.batch_size)
+ if not batch:
+ return 0
+ try:
+ response = self.http.request(
+ self.config.url, payload={"schema_version": 1, "samples": batch}, token=self.token,
+ )
+ except TransportError as error:
+ raise UploadError(str(error)) from None
+ accepted = response.get("accepted_ids")
+ if not isinstance(accepted, list) or not all(isinstance(value, str) for value in accepted):
+ raise UploadError("Receiver must return an accepted_ids array")
+ requested = {sample["id"] for sample in batch}
+ if len(set(accepted)) != len(accepted) or not set(accepted).issubset(requested):
+ raise UploadError("Receiver acknowledged duplicate or unknown sample IDs")
+ self.store.acknowledge(accepted)
+ if set(accepted) != requested:
+ raise UploadError(f"Receiver accepted {len(accepted)}/{len(batch)} samples; others remain queued")
+ return len(accepted)
+
+ def run(self, stop):
+ delay = self.config.retry_initial_seconds
+ while not stop.is_set():
+ try:
+ count = self.upload_once()
+ except (UploadError, OSError) as error:
+ LOG.warning("Upload failed: %s; retrying in %.0fs", error, delay)
+ stop.wait(delay)
+ delay = min(delay * 2, self.config.retry_max_seconds)
+ else:
+ delay = self.config.retry_initial_seconds
+ if count:
+ LOG.info("Uploaded and acknowledged %d samples", count)
+ else:
+ stop.wait(5)
+
diff --git a/solar_relay/web/index.html b/solar_relay/web/index.html
new file mode 100644
index 0000000..096f210
--- /dev/null
+++ b/solar_relay/web/index.html
@@ -0,0 +1,46 @@
+
+
+
+
+
+
+ Solar relay · Collection status
+
+
+
+
+
+
+
+
☀SOLAR RELAYRPI OFFICE
+
Connecting
+
+
+
COLLECTION MONITOR
+
Solar relay status
+
Extraction, delivery and five-minute coverage of your local Fronius data.
+
—/Europe/Zurich/Waiting for the relay
+
+
+
+
Last extraction
—
No observation received yet
Last successful extraction: —
+
Yesterday’s coverage
— / —
Five-minute intervals
Energy, meter and boiler status
+
Waiting to upload
—
Checking delivery
Last acknowledgement: —
+
Local archive
—
Observations retained in Parquet
Acknowledged readings kept for 30 days
+
+
+
DAILY COVERAGE
Five-minute completeness
+
Complete means a valid energy, meter and boiler extraction in that five-minute interval. It does not mean continuous measurement. Recovered cloud history supplies energy only.
+
Day
Complete intervals
Energy
Meter
Boiler
Status
Waiting for the local archive…
+
+
+
FIVE-MINUTE INTERVALS
Select a day
CompletePartialMissingUpcoming
+
Each column is one hour. Each cell is a five-minute interval.