Files
we_monitor/collect.py
T
jensandClaude Sonnet 4.6 bb1e087867 collect: add --dry flag to run push server without WeConnect
When --dry is set, WeConnect login and data collection are skipped.
The push server still starts and serves preloaded 24 h history to
connecting clients — useful for GUI testing without a live vehicle.
Credentials are not required in dry-run mode.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-05-26 13:59:58 +02:00

245 lines
8.2 KiB
Python

#!/usr/bin/env python3
"""
WeConnect vehicle data collector.
Periodically fetches selected domains from a VW WeConnect vehicle and
appends timestamped records to a JSON file. The JSON structure mirrors
the WeConnect domain hierarchy: domain → status-object → field.
Usage examples
--------------
# Credentials file with all settings:
python collect.py -c credentials/alex.json
# Override username/password via env vars:
export WC_USER=me@example.com WC_PASS=secret
python collect.py -c credentials/alex.json
# Override domains on the CLI:
python collect.py -c credentials/alex.json -d charging,measurements
Available domains: charging, climatisation, measurements, readiness, parking, access
Credentials file format (JSON)
-------------------------------
{
"credentials": {
"username": "me@example.com",
"password": "secret"
},
"vin": "WVWZZZE1ZMP123456",
"domains": ["charging", "measurements", "readiness"],
"interval_s": 300,
"host": "0.0.0.0",
"port": 9999,
"log_dir": "./logs"
}
Push server
-----------
The collector listens on host/port from the credentials file (default 0.0.0.0:9999).
Each connected TCP client receives every new snapshot as a single
line of JSON (newline-delimited) the moment it is collected.
Connect with: nc localhost 9999 or any TCP client that reads lines.
"""
import argparse
import json
import logging
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
from server import log_config, network, we_connect
from server.data_model import ALL_DOMAINS, collect_snapshot
from server.storage_helpers import auto_out_path, load_last_24h_records, load_store, save_store
log = logging.getLogger(__name__)
# ── main loop ────────────────────────────────────────────────────────────────
def run(args: argparse.Namespace) -> None:
log_config.setup(args.verbose)
raw = [d.strip() for d in args.domains.split(",")]
domains = list(ALL_DOMAINS) if raw == ["all"] else raw
unknown = [d for d in domains if d not in ALL_DOMAINS]
if unknown:
log.error("Unknown domain(s): %s. Available: %s", unknown, list(ALL_DOMAINS))
sys.exit(1)
if args.dry:
log.info("Dry-run mode — skipping WeConnect login")
wc = None
vehicle = None
vin = args.vin or "UNKNOWN"
else:
log.info("Connecting to WeConnect…")
wc = we_connect.connect(args.username, args.password)
vehicle, err = we_connect.select_vehicle(wc, args.vin)
if err:
log.error(err)
sys.exit(1)
vin = vehicle.vin.value
log.info("Vehicle: %s", vin)
logs_dir = Path(args.log_dir)
logs_dir.mkdir(parents=True, exist_ok=True)
out = auto_out_path(logs_dir, vin)
push_clients, push_lock, push_store_ref = network.start_push_server(args.host, args.port)
push_store_ref["preloaded"] = load_last_24h_records(logs_dir, vin)
if push_store_ref["preloaded"]:
log.info("Pre-loaded %d record(s) from the last 24 h", len(push_store_ref["preloaded"]))
current_day = datetime.now(timezone.utc).date()
store = load_store(out, vin, args.interval)
push_store_ref["current"] = store
if args.dry:
log.info("Push server running on %s:%d — no live collection (Ctrl-C to stop)", args.host, args.port)
else:
log.info(
"Collecting [%s] every %ds → %s (Ctrl-C to stop)",
", ".join(domains),
args.interval,
out,
)
try:
while True:
if not args.dry:
today = datetime.now(timezone.utc).date()
if today != current_day:
out = auto_out_path(logs_dir, vin)
store = load_store(out, vin, args.interval)
push_store_ref["current"] = store
current_day = today
log.info("Midnight UTC rotation → %s", out)
try:
wc.update()
snapshot = collect_snapshot(vehicle, domains)
store["records"].append(snapshot)
save_store(out, store, args.max_records)
network.broadcast(snapshot, push_clients, push_lock)
log.info("Record #%d saved at %s", len(store["records"]), snapshot["ts"])
except KeyboardInterrupt:
raise
except Exception:
log.exception("Collection failed — will retry at next interval")
time.sleep(args.interval)
except KeyboardInterrupt:
log.info("Stopped by user. %d records in %s", len(store["records"]), out)
# ── CLI ──────────────────────────────────────────────────────────────────────
def main() -> None:
p = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
creds = p.add_argument_group(
"credentials (also accepted via WC_USER / WC_PASS env vars or -c FILE)"
)
creds.add_argument(
"-c", "--credentials",
metavar="FILE",
help="JSON file with connection settings (see format above).",
)
creds.add_argument(
"-u", "--username",
default=os.environ.get("WC_USER"),
help="WeConnect / MyVolkswagen email (overrides credentials file)",
)
creds.add_argument(
"-p", "--password",
default=os.environ.get("WC_PASS"),
help="WeConnect password (overrides credentials file)",
)
p.add_argument(
"-d", "--domains",
default=None,
metavar="DOMAIN[,DOMAIN…]",
help=(
"Comma-separated domains to collect, or 'all' (overrides credentials file). "
f"Available: {', '.join(ALL_DOMAINS)}. "
"Default: charging,measurements,readiness"
),
)
p.add_argument(
"--max-records",
type=int,
default=None,
metavar="N",
help="Keep only the last N records per file (default: unlimited)",
)
p.add_argument("-v", "--verbose", action="store_true", help="Enable debug logging")
p.add_argument(
"--dry",
action="store_true",
help="Skip WeConnect login and data collection; run push server only (for GUI testing)",
)
p.add_argument(
"--list-domains",
action="store_true",
help="Print available domains and exit",
)
args = p.parse_args()
if args.list_domains:
print("\n".join(ALL_DOMAINS))
sys.exit(0)
creds_data: dict = {}
if args.credentials:
creds_path = Path(args.credentials)
if not creds_path.exists():
p.error(f"Credentials file not found: {args.credentials}")
try:
creds_data = json.loads(creds_path.read_text())
except json.JSONDecodeError as exc:
p.error(f"Invalid JSON in credentials file: {exc}")
# username / password: env / CLI flag > credentials.username > credentials file root
nested = creds_data.get("credentials", {})
if args.username is None:
args.username = nested.get("username") or creds_data.get("username")
if args.password is None:
args.password = nested.get("password") or creds_data.get("password")
args.vin = creds_data.get("vin")
if args.domains is None:
raw_domains = creds_data.get("domains")
if isinstance(raw_domains, list):
args.domains = ",".join(raw_domains)
else:
args.domains = raw_domains or "charging,measurements,readiness"
args.interval = int(creds_data.get("interval_s", 300))
args.host = creds_data.get("host", "0.0.0.0")
args.port = int(creds_data.get("port", 9999))
args.log_dir = creds_data.get("log_dir", "./logs")
if not args.dry and (not args.username or not args.password):
p.error(
"Username and password are required. "
"Provide them in the credentials file or via -u/-p / WC_USER / WC_PASS."
)
run(args)
if __name__ == "__main__":
main()