pluginbattery/scripts/export_from_influx.py

#!/usr/bin/env python3
"""Export hourly prices, P1 power, and solar irradiance from data-vm InfluxDB.

SSHs into data-vm and queries localhost:8086 there. The InfluxDB v2 token is
read from sync/json2influx.py on data-vm — no secrets stored locally or in git.

Outputs three CSVs in data/raw/ with columns:
    timestamp (UTC ISO-8601), <value>

Usage:
    python scripts/export_from_influx.py
    python scripts/export_from_influx.py --start 2024-09-01T00:00:00Z --end 2025-09-01T00:00:00Z
"""
from __future__ import annotations

import argparse
import csv
import io
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path

REMOTE = "data-vm"
INFLUX_URL = "http://localhost:8086"
INFLUX_DB = "default"

# Default window: cleanest 365-day overlap of prices + P1 + solar.
# 2024-09→2025-09 has a 19-day P1 outage in March 2025; the year before is clean.
DEFAULT_START = "2023-09-01T00:00:00Z"
DEFAULT_END = "2024-09-01T00:00:00Z"

QUERIES = {
    # Raw day-ahead EPEX price, mean per hour, in EUR/kWh.
    # Taxes and supplier markup are layered on downstream — not here.
    "prices_hourly.csv": (
        "eur_per_kwh",
        "SELECT mean(\"epex-price\")/1000 AS eur_per_kwh "
        "FROM \"measurements\" WHERE \"market\" = 'nl' "
        "AND time >= '{start}' AND time < '{end}' "
        "GROUP BY time(1h) fill(none)",
    ),
    # Net power draw at the meter, mean over the hour, in watts.
    # Sign convention to be confirmed (likely positive = import).
    "p1_hourly.csv": (
        "power_w",
        "SELECT mean(\"power-current\") AS power_w "
        "FROM \"measurements\" "
        "WHERE time >= '{start}' AND time < '{end}' "
        "GROUP BY time(1h) fill(none)",
    ),
    # Horizontal solar irradiance, mean over the hour, W/m^2.
    "solar_hourly.csv": (
        "irradiance_w_m2",
        "SELECT mean(\"ws1-solarradiation\") AS irradiance_w_m2 "
        "FROM \"measurements\" "
        "WHERE time >= '{start}' AND time < '{end}' "
        "GROUP BY time(1h) fill(none)",
    ),
}


def fetch_csv(query: str) -> str:
    """Run InfluxQL on data-vm and return CSV body."""
    # Escape internal double quotes so they survive the bash double-quoted arg.
    q_escaped = query.replace("\\", "\\\\").replace('"', r"\"")
    remote_script = f"""set -euo pipefail
TOKEN=$(grep -oP "token='[^']+'" /home/michielb/sync/json2influx.py | head -1 | cut -d"'" -f2)
HTTP_STATUS=$(curl -sS -o /tmp/influx_resp.txt -w "%{{http_code}}" -G "{INFLUX_URL}/query" \\
  --data-urlencode "db={INFLUX_DB}" \\
  --data-urlencode "q={q_escaped}" \\
  -H "Authorization: Token $TOKEN" \\
  -H "Accept: application/csv")
if [ "$HTTP_STATUS" != "200" ]; then
  echo "HTTP $HTTP_STATUS:" >&2
  cat /tmp/influx_resp.txt >&2
  exit 1
fi
cat /tmp/influx_resp.txt
"""
    result = subprocess.run(
        ["ssh", REMOTE, "bash -s"],
        input=remote_script,
        capture_output=True,
        text=True,
    )
    if result.returncode != 0:
        sys.stderr.write(f"--- remote stderr ---\n{result.stderr}\n")
        sys.stderr.write(f"--- remote stdout ---\n{result.stdout}\n")
        raise SystemExit(f"remote command failed (exit {result.returncode})")
    return result.stdout


def write_clean(raw_csv: str, value_col: str, out_path: Path) -> int:
    """Strip InfluxDB's name/tags columns, convert ns epoch → ISO-8601 UTC."""
    reader = csv.DictReader(io.StringIO(raw_csv))
    rows_out = 0
    with out_path.open("w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["timestamp", value_col])
        for row in reader:
            ts_ns = int(row["time"])
            iso = datetime.fromtimestamp(ts_ns / 1e9, tz=timezone.utc).strftime(
                "%Y-%m-%dT%H:%M:%SZ"
            )
            writer.writerow([iso, row[value_col]])
            rows_out += 1
    return rows_out


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--start", default=DEFAULT_START)
    parser.add_argument("--end", default=DEFAULT_END)
    parser.add_argument("--out", default="data/raw")
    args = parser.parse_args()

    out_dir = Path(args.out)
    out_dir.mkdir(parents=True, exist_ok=True)

    expected_hours = int(
        (
            datetime.strptime(args.end, "%Y-%m-%dT%H:%M:%SZ")
            - datetime.strptime(args.start, "%Y-%m-%dT%H:%M:%SZ")
        ).total_seconds()
        // 3600
    )
    print(f"Window: {args.start} → {args.end}  ({expected_hours} hours)")

    for filename, (value_col, query_template) in QUERIES.items():
        query = query_template.format(start=args.start, end=args.end)
        print(f"Fetching {filename} ...", flush=True)
        raw = fetch_csv(query)
        rows = write_clean(raw, value_col, out_dir / filename)
        coverage = rows / expected_hours * 100 if expected_hours else 0
        print(f"  {rows} rows ({coverage:.1f}% coverage) -> {out_dir / filename}")
    return 0


if __name__ == "__main__":
    sys.exit(main())