localhorst/vw-eu-data-act-prometheus-exporter
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

inital impl

Browse Source
This commit is contained in:
localhorst
2026-08-13 17:41:04 +02:00
parent 1c54a6b7b2
commit 05b2ad9f8d
6 changed files with 901 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2
View File
@@ -203,3 +203,5 @@ cython_debug/
# Built Visual Studio Code Extensions
*.vsix
notes.md
*.json
README.md
+132 -1
View File
@@ -1,3 +1,134 @@
# vw-eu-data-act-prometheus-exporter
VW EU Data Act Prometheus Exporter
A Prometheus exporter for vehicle data from the
[VW Group EU Data Act Portal](https://eu-data-act.drivesomethinggreater.com/)
(Volkswagen; not supported for Audi/Skoda/SEAT/CUPRA without adjustments).
For each configured VIN, a dedicated background thread logs into the portal,
downloads the latest available dataset, and derives Prometheus metrics from
it - following the classic multi-target-exporter pattern via a `vin` label on
a shared `/metrics` endpoint (no Docker, plain systemd deployment).
## Features
- Multiple vehicles (VINs) under one VW account, each with its own
scrape-interval timing in the background
- **No persistent state by default**: no ZIPs, no JSON, no values are
written to disk - everything lives in process memory and is empty again
after a restart. Optionally, raw JSON datasets can be persisted to disk via
`config.persist_raw_json` (see [Configuration](#configuration-configpy))
- Fields are only updated when they are present in the current dataset - if
a field is missing, the last known value is kept
- The `health` metric cleanly distinguishes between "portal ok, but no new
data (yet)" (not a failure) and actual failures (login, HTTP, timeout, or
parse errors)
## Metrics
All metrics carry the `vin` label; all names carry the configurable prefix
(default: `vw_eu_data_act_exporter_`).
| Metric | Type | Description |
|---|---|---|
| `mileage_km` | Gauge | Odometer reading |
| `hvsoc_percent` | Gauge | HV battery state of charge in % |
| `driver_present` | Gauge (bool) | Driver detected in vehicle |
| `cruising_range_km` | Gauge | Remaining range in km |
| `hvbattery_temperature_max_celsius` | Gauge | Max. HV battery temperature |
| `hvbattery_temperature_min_celsius` | Gauge | Min. HV battery temperature |
| `charging_state` | Enum | `chargingStatus.currentChargeState` |
| `charge_power_kw` | Gauge | Current charging power in kW |
| `plug_connection_state` | Enum | Plug connection status |
| `target_soc_percent` | Gauge | Charge target in % |
| `position_longitude` / `position_latitude` | Gauge | Last known position (only set when **both** coordinates were present in the same dataset) |
| `position_created_timestamp_seconds` | Gauge | Unix timestamp of the last known position |
| `locked` | Gauge (bool) | Combined lock status of all doors, trunk and hood (only `unlocked` if at least one component actively reports `UNLOCKED`) |
| `is_parked` | Gauge (bool) | Vehicle parked |
| `parking_brake_engaged` | Gauge (bool) | Parking brake engaged |
| `driving_mode` | Enum | Active driving mode |
| `next_service_type` | Enum | Next due service type |
| `service_due_in_days` | Gauge | Remaining days until the next service |
| `last_vehicle_signal_timestamp_seconds` | Gauge | Timestamp of the last vehicle signal (`carCapturedUTCTimestamp`) |
| `uptime_seconds` | Gauge | Uptime of the exporter process |
| `last_successful_scrape_timestamp_seconds` | Gauge | Last scrape with **new** data |
| `health` | Gauge (bool) | `0` initially and after a failure, `1` once new data has been received at least once |
**Note on enum metrics:** The state space (e.g. possible values of
`drivingMode` or `chargingStatus.currentChargeState`) is compiled
best-effort from the sample dataset and comparable projects, and each one
includes an `UNKNOWN` fallback. If an unknown raw value shows up, it is
exported as `UNKNOWN` and additionally logged as `WARNING` - the state list
in `vw-eu-data-act-exporter.py` (the `*_STATES` constants) can then be
extended.
## Installation (systemd)
```bash
# 1. Create directory + user
sudo mkdir -p /opt/vw-eu-data-act-exporter
sudo useradd --system --no-create-home --shell /usr/sbin/nologin vw-exporter
# 2. Copy files
sudo cp vw-eu-data-act-exporter.py config.py requirements.txt /opt/vw-eu-data-act-exporter/
# 3. Virtualenv + dependencies
sudo python3 -m venv /opt/vw-eu-data-act-exporter/venv
sudo /opt/vw-eu-data-act-exporter/venv/bin/pip install -r /opt/vw-eu-data-act-exporter/requirements.txt
# 4. Adjust config.py (VW account, VINs, interval)
sudo nano /opt/vw-eu-data-act-exporter/config.py
# 5. Permissions (config.py contains the VW account password in plain text)
sudo chown -R vw-exporter:vw-exporter /opt/vw-eu-data-act-exporter
sudo chmod 600 /opt/vw-eu-data-act-exporter/config.py
# 6. Install systemd unit
sudo cp vw-eu-data-act-exporter.service /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl enable --now vw-eu-data-act-exporter
sudo systemctl status vw-eu-data-act-exporter
```
## Configuration (`config.py`)
| Field | Meaning |
|---|---|
| `hostName` / `serverPort` | Bind address of the `/metrics` endpoint |
| `exporter_prefix` | Prefix for all metric names (only `[a-zA-Z0-9_:]`, e.g. `vw_eu_data_act_exporter_`) |
| `vw_account` | E-mail + password of the VW account under which **all** configured VINs are registered |
| `vins` | List of vehicle VINs to scrape |
| `scrape_interval_minutes` | Poll interval per vehicle. The portal only delivers new datasets for "continuous" data requests roughly every ~15 min anyway - a shorter interval will not yield fresher data, but will increase the login frequency against the account |
| `persist_raw_json` | `False` by default. When set to `True`, the raw JSON of every downloaded dataset is written to disk (one file per VIN per scrape) - useful while tuning the field mapping in `vw-eu-data-act-exporter.py`. Not needed for normal operation |
| `persist_raw_json_dir` | Directory the raw JSON files are written to when `persist_raw_json` is enabled (default: `./raw_json`, relative to the working directory) |
Prerequisite in the portal (one-time, in a browser): connect vehicle ->
"Get customised data" -> enable continuous, 15-minute frequency (see the
docstring in `vw-eu-data-act-exporter.py`).
## Prometheus integration
```yaml
scrape_configs:
- job_name: vw-eu-data-act-exporter
static_configs:
- targets: ["127.0.0.1:9109"]
```
Since the exporter itself polls in the background (it does not hit the
portal live on every Prometheus scrape), the Prometheus scrape interval can
be independent of, and significantly shorter than, `scrape_interval_minutes`
- it will simply repeatedly return the same cached value.
## Troubleshooting
- `health` stays `0` permanently: check
`journalctl -u vw-eu-data-act-exporter -f` - usually a login failure
(wrong password) or no datasets available in the portal yet (see
prerequisite above).
- A single metric is missing entirely: the corresponding field has never
been present in any dataset fetched so far (e.g.
`position_longitude`/`_latitude`, if both coordinates were never
delivered at the same time).
- `WARNING ... Unknown enum value`: a new, not-yet-listed value for an enum
metric - the state is exported as `UNKNOWN`, extend the list in the
source code if needed.
config.py
+52
View File
@@ -0,0 +1,52 @@
"""Configuration for the vw-eu-data-act-exporter."""
# --- HTTP Server -------------------------------------------------------
hostName = "127.0.0.1"
serverPort = 9109
# Prefix for all exported Prometheus metrics.
# Must be a valid Prometheus metric name prefix: [a-zA-Z_:][a-zA-Z0-9_:]*
exporter_prefix = "vw_eu_data_act_exporter_"
# --- VW Group EU Data Act Portal Account --------------------------------
# https://eu-data-act.drivesomethinggreater.com/
# All VINs configured below must belong to this account.
vw_account = {
"email": "you@example.com",
"password": "changeme",
}
# --- Vehicles --------------------------------------------------------------
# List of vehicle VINs to scrape (17 characters).
vins = [
"AAABBCCEEF75854df",
]
# --- HTTP Client -------------------------------------------------------------
# Pool of User-Agent strings. A random entry is picked for each login
# session against the portal.
user_agents = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.2 Safari/605.1.15",
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
]
# --- Scrape behavior ---------------------------------------------------------
# Interval in minutes at which each VIN is polled in the background.
# The portal only produces new "continuous" data records roughly every
# ~15 minutes - a shorter interval will not yield fresher data, but will
# increase the login frequency against the VW account.
scrape_interval_minutes = 15
# --- Raw JSON persistence ---------------------------------------------------
# Optional: write every downloaded raw JSON dataset to disk (one file per VIN
# per scrape), e.g. to inspect the portal's payload while improving the field
# mapping in vw-eu-data-act-exporter.py. Disabled by default, since the
# exporter is designed to be stateless (see README.md).
persist_raw_json = False
persist_raw_json_dir = "./raw_json"
# --- Logging ---------------------------------------------------------------
log_level = "INFO"
requirements.txt
+3
View File
@@ -0,0 +1,3 @@
requests>=2.31
beautifulsoup4>=4.12
prometheus_client>=0.20
vw-eu-data-act-exporter.py
+684
View File
@@ -0,0 +1,684 @@
"""
vw-eu-data-act-exporter
========================
Prometheus exporter for vehicle data from the VW Group EU Data Act Portal
(https://eu-data-act.drivesomethinggreater.com/).
For each configured VIN, a dedicated background thread logs into the portal,
downloads the latest available dataset, and exposes the fields it contains as
Prometheus metrics (label ``vin``) on a shared ``/metrics`` endpoint (the
classic Prometheus multi-target pattern via labels instead of separate
processes).
No raw data or ZIP files are written to disk - all state lives purely in
process memory and is lost on restart (health/last-scrape metrics then start
at 0 again). Optionally, the raw JSON of every downloaded dataset can be
persisted to disk via ``config.persist_raw_json`` to help refine the field
mapping below.
"""
from __future__ import annotations
import io
import json
import logging
import random
import re
import sys
import threading
import time
import zipfile
from datetime import datetime
from pathlib import Path
from urllib.parse import urlencode, urljoin, urlparse
import requests
from bs4 import BeautifulSoup
from prometheus_client import Enum, Gauge, start_http_server
from prometheus_client.core import GaugeMetricFamily
from prometheus_client.registry import REGISTRY
import config
# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------
logging.basicConfig(
level=getattr(logging, config.log_level.upper(), logging.INFO),
format="%(asctime)s %(levelname)-8s %(message)s",
datefmt="%H:%M:%S",
)
log = logging.getLogger("vw_eu_data_act_exporter")
# ---------------------------------------------------------------------------
# Portal & OIDC constants (brand "volkswagen" only)
# ---------------------------------------------------------------------------
BASE_URL = "https://eu-data-act.drivesomethinggreater.com"
IDENTITY_BASE = "https://identity.vwgroup.io"
OIDC_AUTHORIZE = IDENTITY_BASE + "/oidc/v1/authorize"
OIDC_SCOPE = "openid cars profile"
OIDC_REDIRECT = BASE_URL + "/login"
OIDC_CLIENT_ID = "9b58543e-1c15-4193-91d5-8a14145bebb0@apps_vw-dilab_com"
OIDC_STATE_KEY = "VOLKSWAGEN_PASSENGER_CARS"
DEFAULT_COUNTRY = "de"
DEFAULT_LANGUAGE = "en"
METADATA_PATH = "/proxy_api/euda-apim/datarequest/vehicles/{vin}/metadata/partial"
LIST_PATH = "/proxy_api/euda-apim/datadelivery/vehicles/{vin}/{identifier}/list"
DOWNLOAD_PATH = "/proxy_api/euda-apim/datadelivery/vehicles/{vin}/{identifier}/download"
NO_CONTENT_SUFFIX = "_no_content_found.zip"
HTTP_TIMEOUT = 30
IDENTIFIER_KEYS = {"identifier", "datarequestid", "id"}
# ---------------------------------------------------------------------------
# HTML / JS login-parsing helpers (adopted from vw_eu_data_act_downloader.py)
# ---------------------------------------------------------------------------
def _extract_template_model(html: str) -> dict:
idx = html.find("templateModel")
if idx == -1:
return {}
brace = html.find("{", idx)
if brace == -1:
return {}
depth = 0
for i in range(brace, len(html)):
c = html[i]
if c == "{":
depth += 1
elif c == "}":
depth -= 1
if depth == 0:
try:
return json.loads(html[brace : i + 1])
except ValueError:
return {}
return {}
def _extract_csrf(html: str) -> str | None:
m = re.search(r"csrf_token\s*[:=]\s*['\"]([^'\"]+)['\"]", html)
return m.group(1) if m else None
def _parse_form_action(html: str, base_url: str) -> str | None:
soup = BeautifulSoup(html, "html.parser")
form = soup.find("form")
if not form:
return None
action = form.get("action")
if not action:
return None
return urljoin(base_url, action)
def _collect_login_fields(html: str) -> dict[str, str]:
soup = BeautifulSoup(html, "html.parser")
fields: dict[str, str] = {}
form = soup.find("form")
if form:
for inp in form.find_all("input"):
name = inp.get("name")
if name:
fields[name] = inp.get("value") or ""
model = _extract_template_model(html)
if model:
for key in ("hmac", "relayState"):
if model.get(key):
fields[key] = model[key]
email_val = (model.get("emailPasswordForm") or {}).get("email")
if email_val:
fields.setdefault("email", email_val)
csrf = _extract_csrf(html)
if csrf:
fields.setdefault("_csrf", csrf)
return fields
def _extract_login_error(html: str) -> str | None:
model = _extract_template_model(html)
err = model.get("error") or model.get("errorCode")
if isinstance(err, dict):
return err.get("text") or err.get("errorCode") or str(err)
return str(err) if err else None
# ---------------------------------------------------------------------------
# Portal client
# ---------------------------------------------------------------------------
def _build_authorize_url() -> str:
state = f"{DEFAULT_COUNTRY}__{DEFAULT_LANGUAGE}__{OIDC_STATE_KEY}"
params = {
"client_id": OIDC_CLIENT_ID,
"response_type": "code",
"scope": OIDC_SCOPE,
"state": state,
"redirect_uri": OIDC_REDIRECT,
"prompt": "login",
}
return f"{OIDC_AUTHORIZE}?{urlencode(params)}"
def login(session: requests.Session, email: str, password: str) -> None:
"""Runs the full OIDC login flow. Raises RuntimeError on failure."""
try:
session.get(BASE_URL + "/", timeout=15)
except requests.RequestException as exc:
log.debug("Priming request failed (ignored): %s", exc)
authorize_url = _build_authorize_url()
resp = session.get(authorize_url, timeout=20)
resp.raise_for_status()
signin_url = resp.url
signin_html = resp.text
fields = _collect_login_fields(signin_html)
if "hmac" not in fields or "_csrf" not in fields:
raise RuntimeError(
f"Could not parse the sign-in form (fields found: {sorted(fields)})."
)
fields["email"] = email
action = _parse_form_action(signin_html, signin_url)
identifier_action = action or signin_url
resp = session.post(
identifier_action, data=fields, headers={"Referer": signin_url}, timeout=20
)
resp.raise_for_status()
authenticate_url = resp.url
authenticate_html = resp.text
fields2 = _collect_login_fields(authenticate_html)
if "hmac" not in fields2 or "_csrf" not in fields2:
err = _extract_login_error(authenticate_html)
raise RuntimeError(err or "IDP did not return a password form.")
fields2["email"] = email
fields2["password"] = password
action2 = _parse_form_action(authenticate_html, authenticate_url)
authenticate_action = action2 or authenticate_url.split("?", 1)[0]
resp = session.post(
authenticate_action,
data=fields2,
headers={"Referer": authenticate_url},
timeout=20,
)
if resp.status_code >= 400:
err = _extract_login_error(resp.text)
raise RuntimeError(err or f"Login rejected with HTTP {resp.status_code}.")
landing = resp.url
portal_host = urlparse(BASE_URL).netloc
if "signin-service" in landing or "/error" in landing:
err = _extract_login_error(resp.text)
raise RuntimeError(f"Login failed - {err or 'check email/password'}")
if urlparse(landing).netloc != portal_host:
raise RuntimeError(f"Login did not complete - landed at {landing!r}.")
def _api_get(session: requests.Session, url: str, *, headers: dict | None = None):
resp = session.get(url, headers=headers or {}, timeout=HTTP_TIMEOUT)
if resp.status_code >= 400:
raise RuntimeError(
f"API GET {url} -> HTTP {resp.status_code}: {resp.text[:300]}"
)
return resp.json()
def get_metadata(session: requests.Session, vin: str) -> dict:
url = BASE_URL + METADATA_PATH.format(vin=vin)
return _api_get(session, url)
def find_identifier(node) -> str | None:
if isinstance(node, dict):
for k, v in node.items():
if k.lower() in IDENTIFIER_KEYS and isinstance(v, str) and len(v) > 8:
return v
for v in node.values():
r = find_identifier(v)
if r:
return r
elif isinstance(node, list):
for v in node:
r = find_identifier(v)
if r:
return r
return None
def list_datasets(session: requests.Session, vin: str, identifier: str) -> list[dict]:
url = BASE_URL + LIST_PATH.format(vin=vin, identifier=identifier)
data = _api_get(session, url, headers={"type": "partial"})
return data if isinstance(data, list) else data.get("files", [])
def download_zip(
session: requests.Session, vin: str, identifier: str, name: str
) -> bytes:
url = BASE_URL + DOWNLOAD_PATH.format(vin=vin, identifier=identifier)
resp = session.get(url, headers={"filename": name, "type": "partial"}, timeout=60)
if resp.status_code >= 400:
raise RuntimeError(
f"Download {name} -> HTTP {resp.status_code}: {resp.text[:200]}"
)
return resp.content
def unzip_json(raw: bytes, name: str) -> dict:
with zipfile.ZipFile(io.BytesIO(raw)) as zf:
members = [m for m in zf.namelist() if m.lower().endswith(".json")]
if not members:
raise ValueError(f"No JSON file in {name}")
with zf.open(members[0]) as fh:
return json.loads(fh.read().decode("utf-8"))
def pick_latest_dataset(datasets: list[dict]) -> dict:
def sort_key(d: dict) -> str:
return d.get("createdOn") or d.get("created") or d.get("name") or ""
return sorted(datasets, key=sort_key)[-1]
def flatten_dataset(record: dict) -> dict[str, str]:
"""Builds a dataFieldName -> value mapping. On duplicates, the last
entry in the Data array wins."""
fields: dict[str, str] = {}
for entry in record.get("Data", []):
name = entry.get("dataFieldName")
if name is None:
continue
fields[name] = entry.get("value")
return fields
def persist_raw_dataset(vin: str, record: dict) -> None:
"""Writes the raw dataset JSON to disk, so it can be inspected later to
refine the field mapping below. Only called when
``config.persist_raw_json`` is enabled."""
out_dir = Path(config.persist_raw_json_dir)
out_dir.mkdir(parents=True, exist_ok=True)
timestamp = datetime.utcnow().strftime("%Y%m%d%H%M%S")
dest = out_dir / f"{vin}_{timestamp}.json"
dest.write_text(json.dumps(record, indent=2, ensure_ascii=False), encoding="utf-8")
log.debug("Persisted raw dataset for VIN %s to %s", vin, dest)
def fetch_latest_dataset_fields(vin: str) -> dict[str, str] | None:
"""Logs in, downloads the latest dataset for *vin* and returns the
flattened fields. Returns ``None`` if no datasets are (yet) available -
this is explicitly NOT a failure. Raises an exception on login/HTTP/parse
errors (= failure)."""
session = requests.Session()
session.headers.update({"User-Agent": random.choice(config.user_agents)})
login(session, config.vw_account["email"], config.vw_account["password"])
meta = get_metadata(session, vin)
identifier = find_identifier(meta)
if not identifier:
raise RuntimeError(f"No data-request identifier in metadata for VIN {vin}.")
datasets = list_datasets(session, vin, identifier)
datasets = [
d
for d in datasets
if not (d.get("name") or d.get("fileName") or "").endswith(NO_CONTENT_SUFFIX)
]
if not datasets:
return None
latest = pick_latest_dataset(datasets)
name = latest.get("name") or latest.get("fileName") or str(latest)
raw = download_zip(session, vin, identifier, name)
record = unzip_json(raw, name)
if config.persist_raw_json:
try:
persist_raw_dataset(vin, record)
except OSError as exc:
log.warning("Could not persist raw dataset for VIN %s: %s", vin, exc)
return flatten_dataset(record)
# ---------------------------------------------------------------------------
# Value conversion (VW API strings -> Prometheus types)
# ---------------------------------------------------------------------------
def parse_bool(raw) -> bool:
return str(raw).strip().lower() == "true"
def parse_float(raw) -> float:
return float(raw)
def parse_iso8601_to_unix(raw) -> float:
text = str(raw).strip()
if text.endswith("Z"):
text = text[:-1] + "+00:00"
return datetime.fromisoformat(text).timestamp()
def parse_bool_as_float(raw) -> float:
return 1.0 if parse_bool(raw) else 0.0
# ---------------------------------------------------------------------------
# Prometheus metrics
# ---------------------------------------------------------------------------
P = config.exporter_prefix
g_mileage_km = Gauge(f"{P}mileage_km", "Odometer reading", ["vin"])
g_hvsoc_percent = Gauge(
f"{P}hvsoc_percent", "HV battery state of charge in percent", ["vin"]
)
g_driver_present = Gauge(
f"{P}driver_present", "Driver detected in vehicle (1=yes)", ["vin"]
)
g_cruising_range_km = Gauge(f"{P}cruising_range_km", "Remaining range in km", ["vin"])
g_hvbattery_temp_max_celsius = Gauge(
f"{P}hvbattery_temperature_max_celsius",
"Max. HV battery temperature in Celsius",
["vin"],
)
g_hvbattery_temp_min_celsius = Gauge(
f"{P}hvbattery_temperature_min_celsius",
"Min. HV battery temperature in Celsius",
["vin"],
)
g_charge_power_kw = Gauge(
f"{P}charge_power_kw", "Current charging power in kW", ["vin"]
)
g_target_soc_percent = Gauge(
f"{P}target_soc_percent", "Charge target in percent", ["vin"]
)
g_position_longitude = Gauge(
f"{P}position_longitude", "Longitude of the last known position", ["vin"]
)
g_position_latitude = Gauge(
f"{P}position_latitude", "Latitude of the last known position", ["vin"]
)
g_position_created_timestamp_seconds = Gauge(
f"{P}position_created_timestamp_seconds",
"Unix timestamp of when the last position was recorded",
["vin"],
)
g_locked = Gauge(
f"{P}locked",
"Combined lock status of all doors, trunk and hood (1=locked)",
["vin"],
)
g_is_parked = Gauge(f"{P}is_parked", "Vehicle parked (1=yes)", ["vin"])
g_parking_brake_engaged = Gauge(
f"{P}parking_brake_engaged", "Parking brake engaged (1=yes)", ["vin"]
)
g_service_due_in_days = Gauge(
f"{P}service_due_in_days", "Remaining days until the next service", ["vin"]
)
g_last_vehicle_signal_timestamp_seconds = Gauge(
f"{P}last_vehicle_signal_timestamp_seconds",
"Unix timestamp of the last vehicle signal (carCapturedUTCTimestamp)",
["vin"],
)
g_last_successful_scrape_timestamp_seconds = Gauge(
f"{P}last_successful_scrape_timestamp_seconds",
"Unix timestamp of the last successful scrape with new data",
["vin"],
)
g_health = Gauge(
f"{P}health",
"1 once new data has been received at least once via the EU Data Act API, "
"0 initially and after every failure (not a failure: portal ok but no new data)",
["vin"],
)
# Enum metrics. The state space is best-effort (compiled from the sample
# dataset and comparable projects) plus an "UNKNOWN" fallback for unknown raw
# values - these are additionally logged as WARNING so the list can be
# extended if needed.
DRIVING_MODE_STATES = ["standard", "eco", "comfort", "sport", "individual", "UNKNOWN"]
CHARGING_STATE_STATES = [
"OFF",
"READY_FOR_CHARGING",
"NOT_READY_FOR_CHARGING",
"CHARGING",
"CONSERVING",
"ERROR",
"UNKNOWN",
]
PLUG_CONNECTION_STATES = ["CONNECTED", "DISCONNECTED", "UNKNOWN"]
SERVICE_TYPE_STATES = [
"SERVICE_TYPE_INSPECTION",
"SERVICE_TYPE_OIL_CHANGE",
"SERVICE_TYPE_BRAKE_FLUID",
"SERVICE_TYPE_TIMING_BELT",
"UNKNOWN",
]
e_driving_mode = Enum(
f"{P}driving_mode", "Active driving mode", ["vin"], states=DRIVING_MODE_STATES
)
e_charging_state = Enum(
f"{P}charging_state",
"Current charging state",
["vin"],
states=CHARGING_STATE_STATES,
)
e_plug_connection_state = Enum(
f"{P}plug_connection_state",
"Plug connection status",
["vin"],
states=PLUG_CONNECTION_STATES,
)
e_service_type = Enum(
f"{P}next_service_type",
"Next due service type",
["vin"],
states=SERVICE_TYPE_STATES,
)
# Field name (dataFieldName in the dataset) -> (Gauge, conversion function)
GAUGE_FIELD_MAP: list[tuple[str, Gauge, callable]] = [
("mileage_info.value", g_mileage_km, parse_float),
("hvsoc_info.value", g_hvsoc_percent, parse_float),
("Driver Presence", g_driver_present, parse_bool_as_float),
("batteryStatus.cruisingRange.range", g_cruising_range_km, parse_float),
(
"hvbatterytemperature_info.max_temperature.value",
g_hvbattery_temp_max_celsius,
parse_float,
),
(
"hvbatterytemperature_info.min_temperature.value",
g_hvbattery_temp_min_celsius,
parse_float,
),
("chargingStatus.chargePower_kW", g_charge_power_kw, parse_float),
("targetSoc_pct", g_target_soc_percent, parse_float),
("positionCreated", g_position_created_timestamp_seconds, parse_iso8601_to_unix),
("isParked", g_is_parked, parse_bool_as_float),
("parking_brake_info.value", g_parking_brake_engaged, parse_bool_as_float),
("service_maintenance_info.due_in_time.value", g_service_due_in_days, parse_float),
(
"carCapturedUTCTimestamp",
g_last_vehicle_signal_timestamp_seconds,
parse_iso8601_to_unix,
),
]
ENUM_FIELD_MAP: list[tuple[str, Enum, list[str]]] = [
("drivingMode", e_driving_mode, DRIVING_MODE_STATES),
("chargingStatus.currentChargeState", e_charging_state, CHARGING_STATE_STATES),
(
"plugStatusItem.plugConnectionState",
e_plug_connection_state,
PLUG_CONNECTION_STATES,
),
("service_maintenance_info.service_type", e_service_type, SERVICE_TYPE_STATES),
]
# Lock components -> internal name. If a component is missing from a dataset
# entirely (even across all scrapes so far), it is treated as "LOCKED" (safe
# default), so a single door actively reporting locked does not on its own
# report "vehicle locked" if another component is still unknown.
LOCK_FIELDS = {
"door_info.front_left.door_lock_status.value": "front_left",
"door_info.front_right.door_lock_status.value": "front_right",
"door_info.rear_left.door_lock_status.value": "rear_left",
"door_info.rear_right.door_lock_status.value": "rear_right",
"trunk_lid_info.trunk_lid_lock_status.value": "trunk",
"hood_info.hood_lock_status.value": "hood",
}
class UptimeCollector:
"""Computes uptime freshly on every scrape instead of maintaining a
ticker thread."""
def __init__(self, start_time: float) -> None:
self._start_time = start_time
def collect(self):
g = GaugeMetricFamily(
f"{P}uptime_seconds", "Uptime of the exporter process in seconds"
)
g.add_metric([], time.time() - self._start_time)
yield g
# ---------------------------------------------------------------------------
# Per-VIN background poller
# ---------------------------------------------------------------------------
class VehiclePoller:
def __init__(self, vin: str) -> None:
self.vin = vin
self.lock_components: dict[str, str] = {
name: "LOCKED" for name in LOCK_FIELDS.values()
}
self.last_signal_ts_raw: str | None = None
def apply_fields(self, fields: dict[str, str]) -> None:
vin = self.vin
for key, gauge, transform in GAUGE_FIELD_MAP:
if key not in fields:
continue
try:
gauge.labels(vin=vin).set(transform(fields[key]))
except (TypeError, ValueError) as exc:
log.warning("Could not parse field %s (VIN %s): %s", key, vin, exc)
for key, enum_metric, states in ENUM_FIELD_MAP:
if key not in fields:
continue
raw = str(fields[key])
state = raw if raw in states else "UNKNOWN"
if state == "UNKNOWN":
log.warning("Unknown enum value for %s (VIN %s): %r", key, vin, raw)
enum_metric.labels(vin=vin).state(state)
for key, comp_name in LOCK_FIELDS.items():
if key in fields:
self.lock_components[comp_name] = str(fields[key])
locked = (
0.0 if any(v == "UNLOCKED" for v in self.lock_components.values()) else 1.0
)
g_locked.labels(vin=vin).set(locked)
if "longitude" in fields and "latitude" in fields:
try:
g_position_longitude.labels(vin=vin).set(
parse_float(fields["longitude"])
)
g_position_latitude.labels(vin=vin).set(parse_float(fields["latitude"]))
except (TypeError, ValueError) as exc:
log.warning("Could not parse position (VIN %s): %s", vin, exc)
new_ts = fields.get("carCapturedUTCTimestamp")
if new_ts is not None and new_ts != self.last_signal_ts_raw:
self.last_signal_ts_raw = new_ts
g_last_successful_scrape_timestamp_seconds.labels(vin=vin).set(time.time())
g_health.labels(vin=vin).set(1.0)
def run_once(self) -> None:
fields = fetch_latest_dataset_fields(self.vin)
if fields is None:
log.info("No new datasets available for VIN %s.", self.vin)
return
self.apply_fields(fields)
def loop(self, stop_event: threading.Event) -> None:
g_health.labels(vin=self.vin).set(0.0)
while not stop_event.is_set():
try:
self.run_once()
except Exception as exc: # login/HTTP/parse errors = failure
log.error("Scrape for VIN %s failed: %s", self.vin, exc)
g_health.labels(vin=self.vin).set(0.0)
stop_event.wait(config.scrape_interval_minutes * 60)
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main() -> int:
if not config.vins:
log.error("config.vins is empty - no vehicles configured to scrape.")
return 1
REGISTRY.register(UptimeCollector(time.time()))
stop_event = threading.Event()
pollers = [VehiclePoller(vin) for vin in config.vins]
threads = []
for poller in pollers:
t = threading.Thread(
target=poller.loop,
args=(stop_event,),
daemon=True,
name=f"poll-{poller.vin}",
)
t.start()
threads.append(t)
start_http_server(config.serverPort, addr=config.hostName)
log.info(
"vw-eu-data-act-exporter running on http://%s:%s/metrics (%d VIN(s), interval %d min)",
config.hostName,
config.serverPort,
len(config.vins),
config.scrape_interval_minutes,
)
try:
while True:
time.sleep(3600)
except KeyboardInterrupt:
log.info("Shutting down...")
stop_event.set()
return 0
if __name__ == "__main__":
sys.exit(main())
vw-eu-data-act-exporter.service
+28
View File
@@ -0,0 +1,28 @@
[Unit]
Description=VW EU Data Act Prometheus Exporter
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
User=vw-exporter
Group=vw-exporter
WorkingDirectory=/opt/vw-eu-data-act-exporter
ExecStart=/opt/vw-eu-data-act-exporter/venv/bin/python3 /opt/vw-eu-data-act-exporter/vw-eu-data-act-exporter.py
Restart=on-failure
RestartSec=30
# Hardening
ReadWritePaths=/opt/vw-eu-data-act-exporter/raw_json
NoNewPrivileges=true
ProtectSystem=strict
ProtectHome=true
PrivateTmp=true
PrivateDevices=true
ProtectKernelTunables=true
ProtectKernelModules=true
ProtectControlGroups=true
RestrictSUIDSGID=true
[Install]
WantedBy=multi-user.target