From db0abd97ce92889a06a08a8cd477d2c497dec9fc Mon Sep 17 00:00:00 2001 From: pailozka Date: Tue, 13 Oct 2026 15:03:33 +0200 Subject: [PATCH] feat: CO2-only regulation via Wattnet by default, with login-based tokens - Ship a CO2-only [PIPELINE] (read power,cpufreq; score co2; act cpu_freq) as the default; no price fetch. Default CO2 backend is now Wattnet. - Wattnet auth: WATTNET_EMAIL/WATTNET_PASSWORD replace WATTNET_API_KEY (Wattnet has no API keys). The 24 h token is fetched from /token-request/get_token, cached in /var/lib/hpc_eff/wattnet_token.json (mode 600), and renewed 10 min before expiry. - A rejected token is answered with a 302 to the Wattnet login page, not a 401. Redirects are no longer followed: 302/401 triggers one re-login, then a clear error. - Wattnet parsing: merge all response series (final, pending, preview) so the current value is ~now instead of ~4 h old, and average per clock hour. Previously only the first series was read, so live calls always failed with "only 21/24 hourly values". - CO2 fetch failure: reuse the last real grade if at most MAX_AGE_MIN (120) minutes old, else rating 1 (no slowdown without evidence). Fallback grades are never chained. The reason is logged as a WARNING and written to the state.json summary. - deb postinst no longer enables cron on install (matches the rpm); it prints the setup steps instead. Upgrades keep an existing cron job. - Docs: Wattnet login requirement and registration (https://api.wattnet.eu/token-request/), token handling, CO2 fallback, troubleshooting. --- .gitignore | 3 +- README.md | 4 +- debian/postinst | 11 +- docs/configuration.md | 75 ++++++++++-- docs/deployment.md | 14 +-- docs/install.md | 19 +-- docs/troubleshooting.md | 6 + src/hpc_eff/config.ini.example | 32 +++-- src/hpc_eff/pipeline/builtins.py | 49 ++++++++ src/hpc_eff/utils/co2_value.py | 167 +++++++++++++++++++------ tests/test_pipeline.py | 203 +++++++++++++++++++++++++++++-- 11 files changed, 501 insertions(+), 82 deletions(-) diff --git a/.gitignore b/.gitignore index 44107a9..995a8d0 100644 --- a/.gitignore +++ b/.gitignore @@ -1,7 +1,7 @@ __pycache__ config.ini config/.env - +dev/ # Python build / install artifacts (pip install ., setup.py build) build/ dist/ @@ -10,6 +10,7 @@ src/*.egg-info/ *.egg *.whl + # Debian packaging build artifacts (dpkg-buildpackage / debhelper / pybuild) .pybuild/ debian/.debhelper/ diff --git a/README.md b/README.md index c77cd74..d3a5c8f 100644 --- a/README.md +++ b/README.md @@ -91,7 +91,7 @@ sudo apt install -y build-essential devscripts debhelper dh-python \ python3-all python3-setuptools fakeroot sudo apt install -y ipmitool cpufrequtils python3-numpy python3-requests make deb -sudo dpkg -i ../hpc-eff_*.deb # this also enables the cron job +sudo dpkg -i ../hpc-eff_*.deb # does not enable cron; see below ``` Then configure and start: @@ -149,7 +149,7 @@ JSON on stdout. Contract, built-in plugins, security rules and starter files: | Signal | Source | Notes | |---|---|---| | Electricity price | [spotovaelektrina.cz](https://spotovaelektrina.cz) (Czech OTE spot market) | Current price via API; the 12-month baseline is scraped from HTML | -| Carbon intensity | [Nowtricity](https://www.nowtricity.com/) (default) or [Wattnet](https://api.wattnet.eu) | Selectable via `[CO2_API] TYPE`; Nowtricity needs an API key | +| Carbon intensity | [Wattnet](https://api.wattnet.eu) (default) or [Nowtricity](https://www.nowtricity.com/) | Selectable via `[CO2_API] TYPE`; Nowtricity needs an API key, Wattnet a login (email + password, register at [api.wattnet.eu/token-request](https://api.wattnet.eu/token-request/)) | | Temperature | IPMI sensor, HTTP API, or your own Python module | "Bring your own reader": see [docs/configuration.md](docs/configuration.md#temperature_source-pluggable-temperature-reading) | --- diff --git a/debian/postinst b/debian/postinst index 5fdd143..7bd55a2 100644 --- a/debian/postinst +++ b/debian/postinst @@ -6,11 +6,12 @@ set -e case "$1" in configure) - # Enable the cron job when package is installed - if [ -x /usr/bin/hpc-eff ]; then - echo "Setting up cron job for hpc-eff..." - /usr/bin/hpc-eff --enable 2>/dev/null || true - fi + # Cron is NOT enabled here: the shipped config has placeholder API keys, + # so regulating before the admin configures it would cap the CPU on no + # data. Upgrades keep an existing /etc/cron.d/hpc-eff untouched. + echo "hpc-eff installed. Set WATTNET_EMAIL and WATTNET_PASSWORD in" + echo "/etc/hpc_eff/config.ini (chmod 600), test with 'sudo hpc-eff', then" + echo "enable cron with 'sudo hpc-eff --enable'." ;; abort-upgrade|abort-remove|abort-deconfigure) ;; diff --git a/docs/configuration.md b/docs/configuration.md index d21869a..f8be489 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -80,12 +80,12 @@ only, never acted on. ```ini [CO2_API] -TYPE=nowtricity +TYPE=wattnet ``` `TYPE` selects the backend; the other backend's keys are ignored. -### `TYPE=nowtricity` (default) +### `TYPE=nowtricity` | Key | Default | |---|---| @@ -98,22 +98,81 @@ Two endpoints are called per run: `emissions-previous-24h//` and node per day: check that against your plan's quota before a cluster-wide rollout. -An API key for the default `nowtricity` backend is sent as `X-Api-Key`; get one +An API key for the `nowtricity` backend is sent as `X-Api-Key`; get one at nowtricity.com. -### `TYPE=wattnet` +### `TYPE=wattnet` (default) + +**Wattnet requires a login (email and password). There are no API keys.** | Key | Default | |---|---| +| `WATTNET_EMAIL` | none, **required** | +| `WATTNET_PASSWORD` | none, **required** | +| `WATTNET_TOKEN_CACHE` | `/var/lib/hpc_eff/wattnet_token.json` | +| `WATTNET_TOKEN_URL` | `https://api.wattnet.eu/token-request/get_token` | | `WATTNET_URL` | `https://api.wattnet.eu/v1/footprints` | | `WATTNET_ZONE` | `CZ` | | `WATTNET_FOOTPRINT_TYPE` | `carbon` | | `WATTNET_SCOPE` | `operational` | -The 24-hour window is computed automatically. Wattnet returns 15-minute -samples, which are averaged into 24 hourly values; the newest hourly average -doubles as the "current" value. **If fewer than 24 hourly values come back the -call fails** and the run falls back to the price rating. +```ini +[CO2_API] +TYPE=wattnet +WATTNET_EMAIL=throttlepoint@your-site.example +WATTNET_PASSWORD= +``` + +#### Getting a login + +1. Request API access at [wattnet.eu/contact](https://wattnet.eu/contact). + Registration only works once your email is approved. +2. Register once with the token service at + [https://api.wattnet.eu/token-request/](https://api.wattnet.eu/token-request/): + ```bash + curl -X POST https://api.wattnet.eu/token-request/register \ + -H "Content-Type: application/json" \ + -d '{"email": "", "password": ""}' + ``` +3. Put the same email and password in `[CO2_API]`. + +Wattnet suggests a dedicated account for unattended services, not a personal +one. The email does not need to be a real mailbox. + +#### How the token is handled + +Wattnet tokens are valid for 24 hours and there are no refresh tokens. On the +first run ThrottlePoint logs in at `/token-request/get_token` and caches the +token in `WATTNET_TOKEN_CACHE` (mode 600). Later runs reuse it and log in again +about 10 minutes before it expires: one login per node per day, as Wattnet +asks. A wrong email or password fails the CO₂ fetch with +`Wattnet login rejected`. + +The config file now holds a password: make it readable by root only +(`chmod 600 /etc/hpc_eff/config.ini`). + +Full API documentation: [docs.wattnet.eu](https://docs.wattnet.eu/user-guide/api/api-access/). + +#### Data + +The 24-hour window is computed automatically. Wattnet splits the response into +several series by data state: final data ends about 4 hours ago, newer data is +pending or a preview and may still be revised slightly. All series are merged +(final data wins on overlap) so the newest point is about now. The 15-minute +samples are averaged per clock hour; the newest, possibly partial, hour is the +"current" value. **If fewer than 24 hourly values come back the call fails**, +and the CO₂ fallback below applies. + +### When the CO₂ fetch fails + +The last real CO₂ grade is reused if it is at most `MAX_AGE_MIN` minutes old +(default 120, i.e. one missed hourly value plus slack). Older or none: rating 1, +no slowdown without evidence. Each failure logs a WARNING and a +`CO2 unavailable (...)` summary in `state.json`. + +| Key | Default | +|---|---| +| `MAX_AGE_MIN` | `120` | --- diff --git a/docs/deployment.md b/docs/deployment.md index 75d7a7e..5126d08 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -43,8 +43,9 @@ The four things to set beyond the mode itself: | Section / key | Set it to | Notes | |---|---|---| -| `[CO2_API] TYPE` | `nowtricity` or `wattnet` | Picks the backend. | -| `[CO2_API] NOWTRICITY_API_KEY` | your key | Also set `NOWTRICITY_USER_AGENT` to something identifying your site. | +| `[CO2_API] TYPE` | `wattnet` (default) or `nowtricity` | Picks the backend. | +| `[CO2_API] NOWTRICITY_API_KEY` | your key | Only for `TYPE=nowtricity`. Also set `NOWTRICITY_USER_AGENT` to something identifying your site. | +| `[CO2_API] WATTNET_EMAIL`, `WATTNET_PASSWORD` | your Wattnet login | Only for `TYPE=wattnet`. Wattnet has no API keys; register at [api.wattnet.eu/token-request](https://api.wattnet.eu/token-request/), see [configuration.md](configuration.md#typewattnet-default). | | `[frequency_tables]` | frequencies for your CPU, MHz, highest first | See [the frequency table](#co2-mode-the-frequency-table). | | `[aggregation]` | `rating_type` and weights | See below. | @@ -61,11 +62,10 @@ DEBUG=yes control_mode=co2 [CO2_API] -TYPE=nowtricity -NOWTRICITY_USER_AGENT=fzu-throttlepoint -NOWTRICITY_API_KEY= -NOWTRICITY_BASE_URL=https://www.nowtricity.com/api -NOWTRICITY_ZONE=czech-republic +TYPE=wattnet +WATTNET_EMAIL= +WATTNET_PASSWORD= +WATTNET_ZONE=CZ [aggregation] rating_type=average diff --git a/docs/install.md b/docs/install.md index 67ef82f..6fc13f3 100644 --- a/docs/install.md +++ b/docs/install.md @@ -6,7 +6,8 @@ configure and start it. ThrottlePoint is not a daemon. It is a short-lived program that root runs periodically from `/etc/cron.d/hpc-eff`, every 10 minutes by default. What it -does on each run depends on `[MODE] control_mode`; see +does on each run depends on the shipped `[PIPELINE]` (CO₂ only by default) or, +if you remove it, on `[MODE] control_mode`; see [regulation-modes.md](regulation-modes.md). --- @@ -23,12 +24,12 @@ Required in both modes: | A writable cpufreq interface | Otherwise the cap silently does nothing | `cat /sys/devices/system/cpu/cpu0/cpufreq/scaling_max_freq` | | `ipmitool` + DCMI-capable BMC | Power reading for the log (optional) | `ipmitool dcmi power reading` | -Additionally, for `control_mode = co2`: +Additionally, for CO₂ regulation (the default): | Requirement | Why | Check | |---|---|---| | Outbound HTTPS | Price and CO₂ data | see below | -| A Nowtricity API key | CO₂ rating | https://www.nowtricity.com/ | +| A Wattnet login (default) **or** a Nowtricity API key | CO₂ rating | Wattnet: email + password, registered at https://api.wattnet.eu/token-request/ (see [configuration.md](configuration.md#typewattnet-default)) · Nowtricity: https://www.nowtricity.com/ | Additionally, for `control_mode = temperature`: @@ -46,8 +47,8 @@ config; set `https_proxy` in the cron environment if you need one. | Host | Purpose | Required? | |---|---|---| | `spotovaelektrina.cz` | Current price and historical monthly averages | Yes: price rating | -| `www.nowtricity.com` | CO₂ intensity (default backend) | Only if `TYPE=nowtricity` | -| `api.wattnet.eu` | CO₂ intensity (alternative backend) | Only if `TYPE=wattnet` | +| `www.nowtricity.com` | CO₂ intensity (alternative backend) | Only if `TYPE=nowtricity` | +| `api.wattnet.eu` | CO₂ intensity and the daily token login (default backend) | Only if `TYPE=wattnet` | Pre-flight check: @@ -138,10 +139,10 @@ gh run download -n hpc-eff-deb # or hpc-eff-rpm | `/var/lib/hpc_eff/state.json` | Current state and rolling history, first run | | `/etc/cron.d/hpc-eff` | Created by `hpc-eff --enable` | -> **Caution:** the DEB `postinst` runs `hpc-eff --enable` automatically, so on -> Debian and Ubuntu the cron job is live the moment you install, before you -> have configured anything. The RPM does not do this. On Debian, either -> configure immediately or run `sudo hpc-eff --disable` first. +> **Note:** neither the DEB nor the RPM enables cron on install. Nothing +> regulates until you run `sudo hpc-eff --enable`, after setting the API key +> and checking one manual `sudo hpc-eff` run. Upgrading keeps an existing +> cron job. --- diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index e0b4b32..d8fb530 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -99,6 +99,9 @@ Test each source directly: curl -v https://spotovaelektrina.cz/api/v1/price/get-actual-price-czk curl -v -H "X-Api-Key: $KEY" -H "User-Agent: test" \ https://www.nowtricity.com/api/current-emissions/czech-republic/ +# Wattnet: log in, then query with the returned token +curl -X POST https://api.wattnet.eu/token-request/get_token \ + -H "Content-Type: application/json" -d '{"email": "", "password": ""}' ``` Usual causes: no outbound HTTPS from compute nodes (very common on HPC @@ -123,6 +126,9 @@ is why the node keeps working and nothing looks broken. |---|---| | `nowtricity` | Missing/expired `NOWTRICITY_API_KEY`, or the quota is exhausted. Each run makes **two** calls: 288/day/node at a 10-minute interval. | | `nowtricity` | Wrong `NOWTRICITY_ZONE` slug. | +| `wattnet` | `Wattnet needs WATTNET_EMAIL and WATTNET_PASSWORD`: the login is not set. | +| `wattnet` | `Wattnet login rejected`: wrong email/password, or the account is not registered at [api.wattnet.eu/token-request](https://api.wattnet.eu/token-request/). | +| `wattnet` | `Wattnet rejected a freshly issued token`: login works but the account has no API access. Contact Wattnet. (A rejected *cached* token is handled automatically: Wattnet answers it with a redirect to its login page, and ThrottlePoint logs in again once.) | | `wattnet` | Fewer than 24 hourly values returned; the call raises rather than working with a partial window. | | both | Egress blocked to that host specifically. | diff --git a/src/hpc_eff/config.ini.example b/src/hpc_eff/config.ini.example index d569762..64750df 100644 --- a/src/hpc_eff/config.ini.example +++ b/src/hpc_eff/config.ini.example @@ -5,8 +5,8 @@ POWERREADINGCMD=ipmitool dcmi power reading DEBUG=yes [CO2_API] -# Which CO2 intensity API to use: nowtricity | wattnet -TYPE=nowtricity +# Which CO2 intensity API to use: wattnet (default) | nowtricity +TYPE=wattnet # Nowtricity API credentials (https://www.nowtricity.com/) NOWTRICITY_USER_AGENT=UserAgentString @@ -17,11 +17,22 @@ NOWTRICITY_ZONE=czech-republic # Wattnet API settings (used only when TYPE=wattnet) # Start/end timestamps are calculated automatically for the trailing 24h WATTNET_URL=https://api.wattnet.eu/v1/footprints -WATTNET_API_KEY=*** +# Login. Wattnet has no API keys: tokens last 24 h and cannot be refreshed, +# so ThrottlePoint requests one with these credentials and caches it in +# WATTNET_TOKEN_CACHE until shortly before it expires. Wattnet suggests a +# dedicated account (the email need not be a real mailbox), not a personal one. +WATTNET_EMAIL= +WATTNET_PASSWORD= +#WATTNET_TOKEN_CACHE=/var/lib/hpc_eff/wattnet_token.json WATTNET_ZONE=CZ WATTNET_FOOTPRINT_TYPE=carbon WATTNET_SCOPE=operational +# When the CO2 API fails, reuse the last real grade if it is at most this many +# minutes old (data is hourly, so 120 = one missed hour plus slack). Older or +# none: rating 1, i.e. no slowdown without evidence. +MAX_AGE_MIN=120 + [logging] db_path=/var/lib/hpc_eff/history.db log_level=INFO @@ -154,12 +165,15 @@ weight_co2=0.4 # Built-ins: read: power, cpufreq | score: price, co2, combine | # act: cpu_freq, cpu_thermo, gpu_power | record: db_log, state_json # -# Example: regulate by CO2 alone (Nowtricity or Wattnet per [CO2_API]), no price: -# -#[PIPELINE] -#score = co2 -#act = cpu_freq -# +# DEFAULT: regulate by CO2 alone (Nowtricity or Wattnet per [CO2_API]), no price. +# While this section is present, [MODE] control_mode is ignored. To use the +# price+CO2 blend or temperature mode instead, comment this section out. + +[PIPELINE] +read = power, cpufreq +score = co2 +act = cpu_freq + # Example: swap in your own executable. A [plugin:NAME] section with `exec` # (absolute path, root-owned, not group/world-writable) overrides a built-in # of the same name or defines a new plugin. Other keys are passed to the plugin. diff --git a/src/hpc_eff/pipeline/builtins.py b/src/hpc_eff/pipeline/builtins.py index 2ee5f44..bba0756 100644 --- a/src/hpc_eff/pipeline/builtins.py +++ b/src/hpc_eff/pipeline/builtins.py @@ -8,7 +8,9 @@ """ import json +import logging import traceback +from datetime import datetime, timezone from pathlib import Path from ..utils.energy_price import get_current_energy_price, get_averages_year, classify_price_by_median @@ -146,8 +148,37 @@ def price(req, config, rt): return {"set": values, "state": state} +def _last_fresh_co2_grade(config): + """(grade, age in minutes) of the newest real CO2 grade in state.json, or None. + + Grid CO2 is published hourly, so a grade up to [CO2_API] MAX_AGE_MIN old + (default 120) still describes the grid; older ones are a guess. + """ + max_age = config.getint("CO2_API", "MAX_AGE_MIN", fallback=120) + path = config.get("logging", "state_json_path", fallback=DEFAULT_STATE_PATH) + try: + with open(path, "r") as f: + history = json.load(f).get("history", []) + except Exception: + return None + now = datetime.now(timezone.utc) + for entry in history: + if entry.get("co2_fallback") or not isinstance(entry.get("co2_grade"), int): + continue + try: + ts = datetime.strptime(entry["timestamp"], "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) + except (KeyError, ValueError): + continue + age_min = int((now - ts).total_seconds() // 60) + if age_min <= max_age: + return entry["co2_grade"], age_min + return None + return None + + def co2(req, config, rt): """CO2 intensity (Nowtricity or Wattnet per [CO2_API]), graded 1-10 as a 24 h percentile.""" + co2_error = None try: history, current, median, grade = co2_value(config) rt.debug_log(f"Last 24 hours CO2 values (g CO2eq/kWh): {history}") @@ -157,6 +188,16 @@ def co2(req, config, rt): except Exception as e: current, median, grade = None, None, "unknown" rt.debug_log(f"Error fetching co2 values and rating: {e}") + logging.getLogger("hpc_eff").warning("CO2 data unavailable: %s", e) + # The reason goes into state.json: cron discards the log output + reason = " ".join(str(e).split())[:160] + co2_error = f"CO2 unavailable ({type(e).__name__}: {reason})" + last = _last_fresh_co2_grade(config) + if last is not None: + grade, age_min = last + co2_error += f"; reusing grade {grade} from {age_min} min ago" + else: + co2_error += "; no CO2 data within max age, rating 1 (no slowdown)" values = { "co2_current": current, @@ -166,9 +207,17 @@ def co2(req, config, rt): "rating_co2": grade if isinstance(grade, int) else None, } state = dict(values) + # A fallback grade is never reused itself, so a long outage cannot chain it forward + state["co2_fallback"] = co2_error is not None + if co2_error: + state["summary"] = co2_error if values["rating_co2"] is not None: # On its own, the CO2 grade is the rating. `combine`, when listed, overrides it. values["rating"] = state["rating"] = values["rating_co2"] + elif co2_error: + # No recent data: do not slow the node down without evidence. rating_co2 + # stays None, so `combine` (when listed) falls back to price as before. + values["rating"] = state["rating"] = 1 return {"set": values, "state": state} diff --git a/src/hpc_eff/utils/co2_value.py b/src/hpc_eff/utils/co2_value.py index 8582d5f..efdfb74 100644 --- a/src/hpc_eff/utils/co2_value.py +++ b/src/hpc_eff/utils/co2_value.py @@ -1,6 +1,15 @@ +import base64 +import json +import os import requests import statistics -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone +from pathlib import Path + +WATTNET_TOKEN_URL = 'https://api.wattnet.eu/token-request/get_token' +WATTNET_TOKEN_CACHE = '/var/lib/hpc_eff/wattnet_token.json' +# Refresh this long before expiry so a token never dies mid-request +TOKEN_REFRESH_MARGIN = timedelta(minutes=10) def _nowtricity_url(config, endpoint): @@ -44,6 +53,80 @@ def fetch_current_emission(api_headers, config): return data['emissions']['value'] +def _jwt_expiry(token): + """The `exp` claim of a JWT as an aware datetime, or None. Not verified: only used for caching.""" + try: + payload = token.split('.')[1] + payload += '=' * (-len(payload) % 4) + exp = json.loads(base64.urlsafe_b64decode(payload))['exp'] + return datetime.fromtimestamp(exp, timezone.utc) + except Exception: + return None + + +def _parse_expiry(value): + if not value: + return None + try: + dt = datetime.fromisoformat(str(value).replace('Z', '+00:00')) + return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) + except ValueError: + return None + + +def _request_wattnet_token(config, email, password): + """Exchange email/password for a 24 h token at the Wattnet token-request service.""" + url = config.get('CO2_API', 'WATTNET_TOKEN_URL', fallback=WATTNET_TOKEN_URL) + response = requests.post(url, json={'email': email, 'password': password}, timeout=15) + if response.status_code == 401: + raise ValueError("Wattnet login rejected: check WATTNET_EMAIL / WATTNET_PASSWORD") + response.raise_for_status() + data = response.json() + token = (data.get('access_token') or data.get('token')) if isinstance(data, dict) else None + if not token: + raise ValueError(f"Wattnet token response has no token field (keys: {list(data) if isinstance(data, dict) else type(data).__name__})") + expires_at = _parse_expiry(data.get('expires_at')) or _jwt_expiry(token) \ + or datetime.now(timezone.utc) + timedelta(hours=23) + return token, expires_at + + +def _wattnet_token(config, force_refresh=False): + """Bearer token for Wattnet. + + Wattnet has no long-lived API keys: tokens last 24 h and cannot be + refreshed. The token is requested with WATTNET_EMAIL + WATTNET_PASSWORD and + cached in WATTNET_TOKEN_CACHE (mode 600) until shortly before it expires, + as Wattnet asks. + """ + email = config.get('CO2_API', 'WATTNET_EMAIL', fallback='').strip() + password = config.get('CO2_API', 'WATTNET_PASSWORD', fallback='').strip() + if not (email and password): + raise ValueError("Wattnet needs WATTNET_EMAIL and WATTNET_PASSWORD in [CO2_API]") + + cache = Path(config.get('CO2_API', 'WATTNET_TOKEN_CACHE', fallback=WATTNET_TOKEN_CACHE)) + if not force_refresh: + try: + cached = json.loads(cache.read_text()) + expires_at = _parse_expiry(cached.get('expires_at')) + if cached.get('email') == email and expires_at \ + and expires_at - TOKEN_REFRESH_MARGIN > datetime.now(timezone.utc): + return cached['token'] + except Exception: + pass # missing or unreadable cache: request a new token + + token, expires_at = _request_wattnet_token(config, email, password) + try: + cache.parent.mkdir(parents=True, exist_ok=True) + tmp = cache.with_suffix('.tmp') + fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, 'w') as f: + json.dump({'email': email, 'token': token, 'expires_at': expires_at.isoformat()}, f) + os.replace(tmp, cache) + except OSError: + pass # not cacheable (e.g. not root): still usable for this run + return token + + def fetch_wattnet_24h(config): """ Fetch emissions data for the past 24 hours from Wattnet API. @@ -53,7 +136,7 @@ def fetch_wattnet_24h(config): Config keys used from [CO2_API]: - WATTNET_URL: base URL (e.g. https://api.wattnet.eu/v1/footprints) - - WATTNET_API_KEY: Bearer token + - WATTNET_EMAIL / WATTNET_PASSWORD: login, token fetched and cached - WATTNET_ZONE: default CZ - WATTNET_FOOTPRINT_TYPE: default carbon - WATTNET_SCOPE: default operational @@ -64,7 +147,6 @@ def fetch_wattnet_24h(config): - current_hourly: int, most recent hourly average (or None if no data) """ base_url = config.get('CO2_API', 'WATTNET_URL', fallback='https://api.wattnet.eu/v1/footprints') - api_key = config.get('CO2_API', 'WATTNET_API_KEY', fallback=None) zone = config.get('CO2_API', 'WATTNET_ZONE', fallback='CZ') footprint_type = config.get('CO2_API', 'WATTNET_FOOTPRINT_TYPE', fallback='carbon') scope = config.get('CO2_API', 'WATTNET_SCOPE', fallback='operational') @@ -85,45 +167,62 @@ def fetch_wattnet_24h(config): 'use_global': 'false' } - headers = { - 'Accept': 'application/json' - } - if api_key and api_key != 'YOUR_API_KEY_HERE': - headers['Authorization'] = f'Bearer {api_key}' - - response = requests.get(base_url, params=params, headers=headers, timeout=15) + def get(token): + headers = {'Accept': 'application/json', 'Authorization': f'Bearer {token}'} + # Redirects are not followed: a rejected token is answered with a 302 to + # the Wattnet login page (not the 401 its docs describe), and following + # it would return that HTML page with status 200. + return requests.get(base_url, params=params, headers=headers, timeout=15, + allow_redirects=False) + + def rejected(response): + return response.status_code == 401 or response.is_redirect + + response = get(_wattnet_token(config)) + if rejected(response): + # Cached token revoked, or expired before its stated expiry: log in again once + response = get(_wattnet_token(config, force_refresh=True)) + if rejected(response): + raise ValueError("Wattnet rejected a freshly issued token " + f"(HTTP {response.status_code}); check the account's API access") response.raise_for_status() - data = response.json() - - # Parse wattnet structure: [{"series": [{"values": [[timestamp, value], ...]}]}] try: - values_15min = data[0]['series'][0]['values'] - except (KeyError, IndexError, TypeError) as e: + data = response.json() + except ValueError: + raise ValueError(f"Wattnet returned non-JSON data ({response.headers.get('content-type')})") + + # Parse wattnet structure: [{"series": [{"valid": bool, "zone_status": str, + # "values": [[timestamp, value], ...]}, ...]}]. Series split one time range by + # data state: final (valid) data ends ~4 h ago, newer data comes in pending + # and preview series. All are merged so the newest point is "now"; pending + # values may still be revised slightly by Wattnet. + try: + series = data[0]['series'] + merged = {} + for entry in series: + for ts, value in entry['values']: + if value is None: + continue + # On overlap, final data wins over pending/preview + if ts not in merged or entry.get('valid', True): + merged[ts] = float(value) + except (KeyError, IndexError, TypeError, ValueError) as e: raise ValueError( f"Unexpected wattnet API structure: {e}. " - f"Expected: data[0]['series'][0]['values'] = [[ts, value], ...]" + f"Expected: data[0]['series'][i]['values'] = [[ts, value], ...]" ) - if not values_15min: + if not merged: raise ValueError("Wattnet API returned empty values array") - # Sort by timestamp ascending (oldest first) for proper chunking - values_15min.sort(key=lambda x: x[0]) - - # Convert 15-minute intervals to hourly averages (4 values per hour) - hourly_values = [] - for i in range(0, len(values_15min), 4): - chunk = values_15min[i:i + 4] - if chunk: - avg = sum(float(v[1]) for v in chunk) / len(chunk) - hourly_values.append(round(avg)) - - # Reverse to newest first (to match Nowtricity order) and take last 24 - result = hourly_values[::-1][:24] + # Average the 15-minute points per clock hour ("2026-10-13T12" -> mean) + by_hour = {} + for ts, value in merged.items(): + by_hour.setdefault(ts[:13], []).append(value) - # Most recent hourly average is the "current" value for wattnet - # (consistent with nowtricity where current ≈ history[0]). Read it after - # the reverse: before it, index 0 is the OLDEST hour. + # Newest hour first (to match Nowtricity order), last 24 hours. The newest + # hour may be partial (e.g. 12:00-12:30); its average is the "current" value. + result = [round(sum(v) / len(v)) for _, v in sorted(by_hour.items(), reverse=True)][:24] current_hourly = result[0] if result else None if len(result) < 24: @@ -169,7 +268,7 @@ def co2_value(config): - median_value: median of historical_values - grade: 1-10 rating based on percentile """ - api_type = config.get('CO2_API', 'TYPE', fallback='nowtricity').strip().lower() + api_type = config.get('CO2_API', 'TYPE', fallback='wattnet').strip().lower() if api_type == "wattnet": historical_values, current_hourly = fetch_wattnet_24h(config) diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index a754d96..b8902c4 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -5,6 +5,7 @@ Nothing here touches the real system: cpupower/nvidia-smi calls, the network and the config under /etc are all mocked or redirected into temp directories. """ +import base64 import configparser import json import os @@ -65,12 +66,13 @@ def test_current_is_newest_hour(self): for i in range(96): # 96 x 15 min = 24 h, rising 100 -> 195 ts = (end - timedelta(minutes=15 * (95 - i))).strftime("%Y-%m-%dT%H:%M:%SZ") samples.append([ts, 100 + i]) - response = mock.Mock() + response = mock.Mock(status_code=200, is_redirect=False) response.json.return_value = [{"series": [{"values": samples}]}] response.raise_for_status.return_value = None - cfg = make_config("[CO2_API]\nTYPE=wattnet\nWATTNET_API_KEY=x\n") + cfg = make_config("[CO2_API]\nTYPE=wattnet\n") - with mock.patch.object(co2_mod.requests, "get", return_value=response): + with mock.patch.object(co2_mod, "_wattnet_token", return_value="x"), \ + mock.patch.object(co2_mod.requests, "get", return_value=response): history, current, _median, grade = co2_mod.co2_value(cfg) self.assertEqual(len(history), 24) @@ -79,6 +81,151 @@ def test_current_is_newest_hour(self): self.assertEqual(grade, 10) +class WattnetSeriesTest(unittest.TestCase): + """Live responses split one range into final, pending and preview series.""" + + def test_merges_all_series_and_groups_by_clock_hour(self): + end = datetime(2026, 10, 13, 12, 30, tzinfo=timezone.utc) + points = [(end - timedelta(minutes=15 * i)).strftime("%Y-%m-%dT%H:%M:%SZ") for i in range(97)][::-1] + final = [[ts, 400] for ts in points[:81]] # .. 08:30 + pending = [[ts, 320] for ts in points[81:94]] # 08:45 .. 11:45 + preview = [[ts, 300] for ts in points[94:]] # 12:00 .. 12:30 + response = mock.Mock(status_code=200, is_redirect=False) + response.json.return_value = [{"series": [ + {"valid": True, "zone_status": "complete", "values": final}, + {"valid": False, "zone_status": "complete", "values": pending}, + {"valid": False, "zone_status": "preview", "values": preview}]}] + cfg = make_config("[CO2_API]\n") + with mock.patch.object(co2_mod, "_wattnet_token", return_value="x"), \ + mock.patch.object(co2_mod.requests, "get", return_value=response): + history, current = co2_mod.fetch_wattnet_24h(cfg) + self.assertEqual(len(history), 24) + self.assertEqual(current, 300) # 12:00-12:30 preview hour, not 08:30 + self.assertEqual(history[1], 320) # 11:00 hour, pending + self.assertEqual(history[4], round((400 * 3 + 320) / 4)) # 08:00 hour: final 08:00-08:30 + pending 08:45 + + def test_final_value_wins_on_overlap(self): + ts = [(datetime(2026, 10, 13, tzinfo=timezone.utc) - timedelta(hours=h)).strftime("%Y-%m-%dT%H:%M:%SZ") + for h in range(24)] + response = mock.Mock(status_code=200, is_redirect=False) + response.json.return_value = [{"series": [ + {"valid": False, "values": [[ts[0], 999]]}, + {"valid": True, "values": [[t, 100] for t in ts]}]}] + with mock.patch.object(co2_mod, "_wattnet_token", return_value="x"), \ + mock.patch.object(co2_mod.requests, "get", return_value=response): + _, current = co2_mod.fetch_wattnet_24h(make_config("[CO2_API]\n")) + self.assertEqual(current, 100) + + +class WattnetTokenTest(TempDirCase): + """Login, cache and refresh of the 24 h Wattnet token.""" + + def cfg(self, extra=""): + return make_config(f"[CO2_API]\nTYPE=wattnet\nWATTNET_EMAIL=bot@x\nWATTNET_PASSWORD=pw\n" + f"WATTNET_TOKEN_CACHE={self.tmp / 'tok.json'}\n" + extra) + + def login_response(self, token="T1", hours=24): + r = mock.Mock(status_code=200) + r.json.return_value = {"access_token": token, + "expires_at": (datetime.now(timezone.utc) + timedelta(hours=hours)).isoformat()} + return r + + def test_login_then_cache_hit(self): + with mock.patch.object(co2_mod.requests, "post", return_value=self.login_response()) as post: + self.assertEqual(co2_mod._wattnet_token(self.cfg()), "T1") + self.assertEqual(co2_mod._wattnet_token(self.cfg()), "T1") + post.assert_called_once() # second call served from cache + self.assertEqual(post.call_args.kwargs["json"], {"email": "bot@x", "password": "pw"}) + self.assertEqual(os.stat(self.tmp / "tok.json").st_mode & 0o777, 0o600) + + def test_near_expiry_logs_in_again(self): + with mock.patch.object(co2_mod.requests, "post", + side_effect=[self.login_response("OLD", hours=0.1), self.login_response("NEW")]): + co2_mod._wattnet_token(self.cfg()) + self.assertEqual(co2_mod._wattnet_token(self.cfg()), "NEW") # 6 min left < 10 min margin + + def test_expiry_from_jwt_when_response_has_none(self): + exp = int((datetime.now(timezone.utc) + timedelta(hours=24)).timestamp()) + body = base64.urlsafe_b64encode(json.dumps({"exp": exp}).encode()).decode().rstrip("=") + r = mock.Mock(status_code=200) + r.json.return_value = {"token": f"h.{body}.s"} + with mock.patch.object(co2_mod.requests, "post", return_value=r): + co2_mod._wattnet_token(self.cfg()) + cached = json.loads((self.tmp / "tok.json").read_text()) + self.assertEqual(int(datetime.fromisoformat(cached["expires_at"]).timestamp()), exp) + + def test_bad_login_is_a_clear_error(self): + with mock.patch.object(co2_mod.requests, "post", return_value=mock.Mock(status_code=401)): + with self.assertRaisesRegex(ValueError, "login rejected"): + co2_mod._wattnet_token(self.cfg()) + + def test_missing_login_is_a_clear_error(self): + with mock.patch.object(co2_mod.requests, "post", side_effect=AssertionError("login attempted")): + with self.assertRaisesRegex(ValueError, "WATTNET_EMAIL and WATTNET_PASSWORD"): + co2_mod._wattnet_token(make_config("[CO2_API]\nWATTNET_EMAIL=bot@x\n")) + + def test_401_on_data_call_relogs_once(self): + denied = mock.Mock(status_code=401, is_redirect=False) + ok = mock.Mock(status_code=200, is_redirect=False) + end = datetime.now(timezone.utc) + ok.json.return_value = [{"series": [{"values": [ + [(end - timedelta(minutes=15 * i)).strftime("%Y-%m-%dT%H:%M:%SZ"), 300] for i in range(96)]}]}] + with mock.patch.object(co2_mod.requests, "post", + side_effect=[self.login_response("A"), self.login_response("B")]), \ + mock.patch.object(co2_mod.requests, "get", side_effect=[denied, ok]) as get: + co2_mod.fetch_wattnet_24h(self.cfg()) + self.assertEqual(get.call_args.kwargs["headers"]["Authorization"], "Bearer B") + + +class WattnetRejectedTokenTest(TempDirCase): + """Wattnet answers a rejected token with a 302 to its login page, not 401.""" + + def cfg(self): + return make_config("[CO2_API]\nWATTNET_EMAIL=bot@x\nWATTNET_PASSWORD=pw\n" + f"WATTNET_TOKEN_CACHE={self.tmp / 'tok.json'}\n") + + def login(self, token): + r = mock.Mock(status_code=200) + r.json.return_value = {"access_token": token, + "expires_at": (datetime.now(timezone.utc) + timedelta(hours=24)).isoformat()} + return r + + def data_ok(self): + r = mock.Mock(status_code=200, is_redirect=False) + end = datetime.now(timezone.utc) + r.json.return_value = [{"series": [{"values": [ + [(end - timedelta(minutes=15 * i)).strftime("%Y-%m-%dT%H:%M:%SZ"), 300] for i in range(96)]}]}] + return r + + def test_redirect_to_login_triggers_relogin(self): + redirect = mock.Mock(status_code=302, is_redirect=True) + with mock.patch.object(co2_mod.requests, "post", side_effect=[self.login("OLD"), self.login("NEW")]) as post, \ + mock.patch.object(co2_mod.requests, "get", side_effect=[redirect, self.data_ok()]) as get: + _, current = co2_mod.fetch_wattnet_24h(self.cfg()) + self.assertEqual(current, 300) + self.assertEqual(post.call_count, 2) + self.assertEqual(get.call_args.kwargs["headers"]["Authorization"], "Bearer NEW") + self.assertFalse(get.call_args.kwargs["allow_redirects"]) + # the new token replaced the dead one in the cache + self.assertEqual(json.loads((self.tmp / "tok.json").read_text())["token"], "NEW") + + def test_fresh_token_still_rejected_is_a_clear_error(self): + redirect = mock.Mock(status_code=302, is_redirect=True) + with mock.patch.object(co2_mod.requests, "post", side_effect=[self.login("A"), self.login("B")]), \ + mock.patch.object(co2_mod.requests, "get", side_effect=[redirect, redirect]): + with self.assertRaisesRegex(ValueError, "rejected a freshly issued token"): + co2_mod.fetch_wattnet_24h(self.cfg()) + + def test_html_body_is_a_clear_error(self): + html = mock.Mock(status_code=200, is_redirect=False) + html.headers = {"content-type": "text/html"} + html.json.side_effect = ValueError("Expecting value") + with mock.patch.object(co2_mod.requests, "post", return_value=self.login("A")), \ + mock.patch.object(co2_mod.requests, "get", return_value=html): + with self.assertRaisesRegex(ValueError, "non-JSON data \\(text/html\\)"): + co2_mod.fetch_wattnet_24h(self.cfg()) + + class CombineRatingsTest(unittest.TestCase): def cfg(self, rating_type): return make_config(f"[aggregation]\nrating_type={rating_type}\n") @@ -201,10 +348,52 @@ def test_combine_after_co2_still_applies_aggregation(self): self.run_cfg("[aggregation]\nrating_type=max\n[PIPELINE]\nscore = co2, combine\nact = cpu_freq\n") self.assertEqual(self.set_freq.call_args[0][0], 7) # no price plugin: CO2 passes through - def test_co2_failure_falls_back_to_neutral_rating(self): - self.run_cfg("[PIPELINE]\nscore = co2\nact = cpu_freq\n", co2=RuntimeError("401")) - self.assertEqual(self.set_freq.call_args[0][0], 5) - self.assertEqual(self.entry()["co2_grade"], "unknown") + def seed_history(self, *entries): + """entries: (minutes_ago, co2_grade, co2_fallback), newest first.""" + now = datetime.now(timezone.utc) + history = [{"timestamp": (now - timedelta(minutes=m)).strftime("%Y-%m-%dT%H:%M:%SZ"), + "co2_grade": g, "co2_fallback": fb} for m, g, fb in entries] + self.state.write_text(json.dumps({"static": {}, "history": history, "current": history[0]})) + + CO2_ONLY = "[PIPELINE]\nscore = co2\nact = cpu_freq\n" + + def test_co2_failure_without_history_does_not_slow_down(self): + self.run_cfg(self.CO2_ONLY, co2=RuntimeError("401")) + self.assertEqual(self.set_freq.call_args[0][0], 1) + entry = self.entry() + self.assertEqual(entry["co2_grade"], "unknown") + self.assertTrue(entry["co2_fallback"]) + self.assertIn("no CO2 data within max age", entry["summary"]) + + def test_co2_failure_reuses_recent_grade(self): + self.seed_history((30, 9, False)) + self.run_cfg(self.CO2_ONLY, co2=RuntimeError("401")) + self.assertEqual(self.set_freq.call_args[0][0], 9) + self.assertIn("reusing grade 9 from 30 min ago", self.entry()["summary"]) + + def test_co2_failure_ignores_grade_older_than_max_age(self): + self.seed_history((121, 9, False)) + self.run_cfg(self.CO2_ONLY, co2=RuntimeError("401")) + self.assertEqual(self.set_freq.call_args[0][0], 1) + + def test_co2_max_age_is_configurable(self): + self.seed_history((170, 9, False)) + self.run_cfg("[CO2_API]\nMAX_AGE_MIN=180\n" + self.CO2_ONLY, co2=RuntimeError("401")) + self.assertEqual(self.set_freq.call_args[0][0], 9) + + def test_co2_fallback_grade_is_not_chained(self): + # 10 min ago was itself a reuse of a grade now 130 min old: must not extend it + self.seed_history((10, 9, True), (130, 9, False)) + self.run_cfg(self.CO2_ONLY, co2=RuntimeError("401")) + self.assertEqual(self.set_freq.call_args[0][0], 1) + + def test_co2_failure_with_price_still_falls_back_to_price(self): + with mock.patch.object(pb, "get_current_energy_price", return_value=3000), \ + mock.patch.object(pb, "get_averages_year", return_value=[1000] * 12), \ + mock.patch.object(pb, "classify_price_by_median", return_value=("expensive", 8)): + self.run_cfg("[aggregation]\nrating_type=average\n[PIPELINE]\n" + "score = price, co2, combine\nact = cpu_freq\n", co2=RuntimeError("401")) + self.assertEqual(self.set_freq.call_args[0][0], 8) # price only, no fake CO2 = 1 def test_executable_score_plugin_sets_rating(self): path = self.example("threshold_score.py")