#!/usr/bin/env python3 """ CarConnectivity vehicle data collector. Periodically fetches selected domains from a VW vehicle via CarConnectivity and appends timestamped records to a JSON file. The JSON structure mirrors the 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, electric_drive, connectivity, vehicle, position, doors 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 carconnectivity.errors import AuthenticationError, TemporaryAuthenticationError from server.data_model import ALL_DOMAINS, collect_snapshot, apply_procedural from server.storage_helpers import auto_out_path, load_last_24h_records, load_store, save_store from jaydiff.diff import diff_full as jay_diff_full 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 CarConnectivity login") wc = None vehicle = None vin = args.vin or "UNKNOWN" else: log.info("Connecting via CarConnectivity…") 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["diff_mode"] = args.diff push_store_ref["preloaded"] = load_last_24h_records(logs_dir, vin) for _r in push_store_ref["preloaded"]: if "procedural" not in _r: apply_procedural(_r) 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) for _r in store["records"]: if "procedural" not in _r: apply_procedural(_r) push_store_ref["current"] = store last_broadcast: dict = {} 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: we_connect.update(wc) snapshot = collect_snapshot(vehicle, domains) apply_procedural(snapshot) store["records"].append(snapshot) save_store(out, store, args.max_records) if args.diff and last_broadcast: payload = jay_diff_full(last_broadcast, snapshot, combine_upd_add=True) else: payload = snapshot network.broadcast(payload, push_clients, push_lock) last_broadcast = snapshot log.info("Record #%d saved at %s", len(store["records"]), snapshot["ts"]) except KeyboardInterrupt: raise except (AuthenticationError, TemporaryAuthenticationError): log.warning("Authentication error — will retry at next interval") 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="MyVolkswagen email (overrides credentials file)", ) creds.add_argument( "-p", "--password", default=os.environ.get("WC_PASS"), help="MyVolkswagen 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,electric_drive,connectivity" ), ) 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( "-i", "--interval", type=int, default=None, metavar="SECONDS", help="Collection interval in seconds (overrides credentials file, default: 300)", ) p.add_argument( "--host", default=None, help="Push server bind address (overrides credentials file, default: 0.0.0.0)", ) p.add_argument( "--port", type=int, default=None, help="Push server TCP port (overrides credentials file, default: 9999)", ) p.add_argument( "--log-dir", default=None, metavar="DIR", help="Directory for rotating log files (overrides credentials file, default: ./logs)", ) p.add_argument("-v", "--verbose", action="store_true", help="Enable debug logging") p.add_argument( "--diff", action="store_true", default=None, help="Send diffs instead of full snapshots over the push server (overrides credentials file, default: off)", ) p.add_argument( "--dry", action="store_true", help="Skip CarConnectivity 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,electric_drive,connectivity" if args.interval is None: args.interval = int(creds_data.get("interval_s", 300)) if args.host is None: args.host = creds_data.get("host", "0.0.0.0") if args.port is None: args.port = int(creds_data.get("port", 9999)) if args.log_dir is None: args.log_dir = creds_data.get("log_dir", "./logs") if args.diff is None: args.diff = bool(creds_data.get("diff", False)) 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()