Initial commit: WeConnect data collector
Adds collect.py with CLI flags for credentials (-u/-p), credential JSON file (-c), VIN, domains (-d), interval, output file, and --list-domains. Credentials file supports domains as array or string. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 4.6
+541
@@ -0,0 +1,541 @@
|
||||
#!/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 from env vars, all defaults:
|
||||
export WC_USER=me@example.com WC_PASS=secret
|
||||
python collect.py
|
||||
|
||||
# Credentials from a JSON file:
|
||||
python collect.py -c creds.json -i 600 -o id3.json
|
||||
|
||||
# Custom domains, 10-minute interval:
|
||||
python collect.py -u me@example.com -p secret \\
|
||||
-d charging,measurements,readiness -i 600 -o id3.json
|
||||
|
||||
# Everything, verbose, keep last 1 000 records:
|
||||
python collect.py -d all -v --max-records 1000
|
||||
|
||||
Available domains: charging, climatisation, measurements, readiness, parking, access
|
||||
|
||||
Credentials file format (JSON)
|
||||
-------------------------------
|
||||
{
|
||||
"username": "me@example.com",
|
||||
"password": "secret",
|
||||
"vin": "WVWZZZE1ZMP123456",
|
||||
"domains": ["charging", "measurements", "readiness"]
|
||||
}
|
||||
All four keys are optional; domains may also be a comma-separated string.
|
||||
Explicit CLI flags and env vars take precedence.
|
||||
|
||||
Sample output record
|
||||
--------------------
|
||||
{
|
||||
"ts": "2026-05-25T10:05:00+00:00",
|
||||
"charging": {
|
||||
"chargingStatus": { "chargingState": "readyForCharging", ... },
|
||||
"batteryStatus": { "currentSOC_pct": 80, "cruisingRangeElectric_km": 295 },
|
||||
"chargingSettings": { "targetSOC_pct": 80, ... },
|
||||
"plugStatus": { "plugConnectionState": "disconnected", ... }
|
||||
},
|
||||
"measurements": {
|
||||
"odometerStatus": { "odometer": 12345 },
|
||||
"rangeStatus": { "electricRange": 295, "totalRange_km": 295 },
|
||||
"batteryStatus": { "temperatureBatteryMax_K": 295.15,
|
||||
"temperatureBatteryMin_K": 293.15 },
|
||||
"temperatureOutsideStatus": { "temperatureOutside_C": 18.0 }
|
||||
},
|
||||
"readiness": {
|
||||
"readinessStatus": {
|
||||
"connectionState": { "isOnline": true, "isActive": true },
|
||||
"connectionWarning": { "insufficientBatteryLevelWarning": false }
|
||||
}
|
||||
},
|
||||
"parking": {
|
||||
"parkingPosition": { "lat": 52.123, "lon": 13.456 }
|
||||
},
|
||||
"access": {
|
||||
"accessStatus": {
|
||||
"overallStatus": "safe",
|
||||
"doors": { "frontLeft": { "lockState": "locked", "openState": "closed" } },
|
||||
"windows": { "frontLeft": { "openState": "closed" } }
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
from weconnect import weconnect
|
||||
except ImportError:
|
||||
print("weconnect is not installed. Run: pip install weconnect", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ── helpers ──────────────────────────────────────────────────────────────────
|
||||
|
||||
def _str(v) -> str:
|
||||
"""Convert an enum-like WeConnect value to a plain string."""
|
||||
return str(v.value) if hasattr(v, "value") else str(v)
|
||||
|
||||
|
||||
# ── domain extractors ────────────────────────────────────────────────────────
|
||||
# Each function returns a dict whose keys are the WeConnect status-object names
|
||||
# and whose values are dicts of field-name → value — mirroring the API layout.
|
||||
# Returns None when the domain is absent from the API response.
|
||||
|
||||
def extract_charging(vehicle) -> dict | None:
|
||||
try:
|
||||
d = vehicle.domains["charging"]
|
||||
except KeyError:
|
||||
return None
|
||||
|
||||
result: dict = {}
|
||||
|
||||
try:
|
||||
cs = d["chargingStatus"]
|
||||
result["chargingStatus"] = {
|
||||
"chargingState": _str(cs.chargingState),
|
||||
"chargePower_kW": cs.chargePower_kW.value,
|
||||
"remainingChargingTimeToComplete_min": cs.remainingChargingTimeToComplete_min.value,
|
||||
"chargeType": _str(cs.chargeType),
|
||||
}
|
||||
except Exception:
|
||||
log.debug("chargingStatus unavailable", exc_info=True)
|
||||
|
||||
try:
|
||||
bs = d["batteryStatus"]
|
||||
result["batteryStatus"] = {
|
||||
"currentSOC_pct": bs.currentSOC_pct.value,
|
||||
"cruisingRangeElectric_km": bs.cruisingRangeElectric_km.value,
|
||||
}
|
||||
except Exception:
|
||||
log.debug("charging/batteryStatus unavailable", exc_info=True)
|
||||
|
||||
try:
|
||||
cfg = d["chargingSettings"]
|
||||
result["chargingSettings"] = {
|
||||
"targetSOC_pct": cfg.targetSOC_pct.value,
|
||||
"maxChargeCurrentAC": _str(cfg.maxChargeCurrentAC),
|
||||
"autoUnlockPlugWhenCharged": _str(cfg.autoUnlockPlugWhenCharged),
|
||||
}
|
||||
except Exception:
|
||||
log.debug("chargingSettings unavailable", exc_info=True)
|
||||
|
||||
try:
|
||||
ps = d["plugStatus"]
|
||||
result["plugStatus"] = {
|
||||
"plugConnectionState": _str(ps.plugConnectionState),
|
||||
"plugLockState": _str(ps.plugLockState),
|
||||
"externalPower": _str(ps.externalPower),
|
||||
}
|
||||
except Exception:
|
||||
log.debug("plugStatus unavailable", exc_info=True)
|
||||
|
||||
return result or None
|
||||
|
||||
|
||||
def extract_climatisation(vehicle) -> dict | None:
|
||||
try:
|
||||
d = vehicle.domains["climatisation"]
|
||||
except KeyError:
|
||||
return None
|
||||
|
||||
result: dict = {}
|
||||
|
||||
try:
|
||||
cl = d["climatisationStatus"]
|
||||
result["climatisationStatus"] = {
|
||||
"climatisationState": _str(cl.climatisationState),
|
||||
"remainingClimatisationTime_min": cl.remainingClimatisationTime_min.value,
|
||||
}
|
||||
except Exception:
|
||||
log.debug("climatisationStatus unavailable", exc_info=True)
|
||||
|
||||
try:
|
||||
cs = d["climatisationSettings"]
|
||||
result["climatisationSettings"] = {
|
||||
"targetTemperature_C": cs.targetTemperature_C.value,
|
||||
"unitInCar": _str(cs.unitInCar),
|
||||
}
|
||||
except Exception:
|
||||
log.debug("climatisationSettings unavailable", exc_info=True)
|
||||
|
||||
try:
|
||||
wh = d["windowHeatingStatus"]
|
||||
result["windowHeatingStatus"] = {
|
||||
name: _str(win.windowHeatingState)
|
||||
for name, win in wh.windows.items()
|
||||
}
|
||||
except Exception:
|
||||
log.debug("windowHeatingStatus unavailable", exc_info=True)
|
||||
|
||||
return result or None
|
||||
|
||||
|
||||
def extract_measurements(vehicle) -> dict | None:
|
||||
try:
|
||||
d = vehicle.domains["measurements"]
|
||||
except KeyError:
|
||||
return None
|
||||
|
||||
result: dict = {}
|
||||
|
||||
try:
|
||||
od = d["odometerStatus"]
|
||||
result["odometerStatus"] = {"odometer": od.odometer.value}
|
||||
except Exception:
|
||||
log.debug("odometerStatus unavailable", exc_info=True)
|
||||
|
||||
try:
|
||||
rs = d["rangeStatus"]
|
||||
result["rangeStatus"] = {
|
||||
"electricRange": rs.electricRange.value,
|
||||
"totalRange_km": rs.totalRange_km.value,
|
||||
}
|
||||
except Exception:
|
||||
log.debug("rangeStatus unavailable", exc_info=True)
|
||||
|
||||
# Battery temperature — reported in Kelvin by the API
|
||||
try:
|
||||
bs = d["temperatureBatteryStatus"]
|
||||
result["temperatureBatteryStatus"] = {
|
||||
"temperatureHvBatteryMax_K": float(bs.temperatureHvBatteryMax_K.value - 273.15),
|
||||
"temperatureHvBatteryMin_K": float(bs.temperatureHvBatteryMin_K.value - 273.15),
|
||||
}
|
||||
except Exception:
|
||||
log.debug("measurements/temperatureBatteryStatus (temperature) unavailable", exc_info=True)
|
||||
|
||||
try:
|
||||
temp = d["temperatureOutsideStatus"]
|
||||
result["temperatureOutsideStatus"] = {
|
||||
"temperatureOutside_C": temp.temperatureOutside_C.value,
|
||||
}
|
||||
except Exception:
|
||||
log.debug("temperatureOutsideStatus unavailable", exc_info=True)
|
||||
|
||||
return result or None
|
||||
|
||||
|
||||
def extract_readiness(vehicle) -> dict | None:
|
||||
try:
|
||||
d = vehicle.domains["readiness"]
|
||||
except KeyError:
|
||||
return None
|
||||
|
||||
result: dict = {}
|
||||
|
||||
try:
|
||||
rs = d["readinessStatus"]
|
||||
conn = rs.connectionState
|
||||
warn = rs.connectionWarning
|
||||
result["readinessStatus"] = {
|
||||
"connectionState": {
|
||||
"isOnline": conn.isOnline.value,
|
||||
"isActive": conn.isActive.value,
|
||||
},
|
||||
"connectionWarning": {
|
||||
"insufficientBatteryLevelWarning": warn.insufficientBatteryLevelWarning.value,
|
||||
"insufficientBatteryLevelError": warn.insufficientBatteryLevelError.value,
|
||||
},
|
||||
}
|
||||
except Exception:
|
||||
log.debug("readinessStatus unavailable", exc_info=True)
|
||||
|
||||
return result or None
|
||||
|
||||
|
||||
def extract_parking(vehicle) -> dict | None:
|
||||
try:
|
||||
d = vehicle.domains["parking"]
|
||||
except KeyError:
|
||||
return None
|
||||
|
||||
result: dict = {}
|
||||
|
||||
try:
|
||||
pos = d["parkingPosition"]
|
||||
result["parkingPosition"] = {
|
||||
"lat": pos.lat.value,
|
||||
"lon": pos.lon.value,
|
||||
}
|
||||
except Exception:
|
||||
log.debug("parkingPosition unavailable", exc_info=True)
|
||||
|
||||
return result or None
|
||||
|
||||
|
||||
def extract_access(vehicle) -> dict | None:
|
||||
try:
|
||||
d = vehicle.domains["access"]
|
||||
except KeyError:
|
||||
return None
|
||||
|
||||
result: dict = {}
|
||||
|
||||
try:
|
||||
acc = d["accessStatus"]
|
||||
result["accessStatus"] = {
|
||||
"overallStatus": _str(acc.overallStatus),
|
||||
"doors": {
|
||||
name: {
|
||||
"lockState": _str(door.lockState),
|
||||
"openState": _str(door.openState),
|
||||
}
|
||||
for name, door in acc.doors.items()
|
||||
},
|
||||
"windows": {
|
||||
name: {"openState": _str(win.openState)}
|
||||
for name, win in acc.windows.items()
|
||||
},
|
||||
}
|
||||
except Exception:
|
||||
log.debug("accessStatus unavailable", exc_info=True)
|
||||
|
||||
return result or None
|
||||
|
||||
|
||||
ALL_DOMAINS: dict[str, callable] = {
|
||||
"charging": extract_charging,
|
||||
"climatisation": extract_climatisation,
|
||||
"measurements": extract_measurements,
|
||||
"readiness": extract_readiness,
|
||||
"parking": extract_parking,
|
||||
"access": extract_access,
|
||||
}
|
||||
|
||||
# ── snapshot / store helpers ─────────────────────────────────────────────────
|
||||
|
||||
def collect_snapshot(vehicle, domains: list[str]) -> dict:
|
||||
snapshot: dict = {"ts": datetime.now(timezone.utc).isoformat()}
|
||||
for domain in domains:
|
||||
data = ALL_DOMAINS[domain](vehicle)
|
||||
if data is not None:
|
||||
snapshot[domain] = data
|
||||
return snapshot
|
||||
|
||||
|
||||
def load_store(path: Path, vin: str, interval: int) -> dict:
|
||||
if path.exists():
|
||||
try:
|
||||
data = json.loads(path.read_text())
|
||||
if isinstance(data, dict) and "records" in data:
|
||||
return data
|
||||
except json.JSONDecodeError:
|
||||
log.warning("Output file is corrupt — starting fresh")
|
||||
return {
|
||||
"meta": {
|
||||
"vin": vin,
|
||||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||||
"interval_seconds": interval,
|
||||
},
|
||||
"records": [],
|
||||
}
|
||||
|
||||
|
||||
def save_store(path: Path, store: dict, max_records: int | None) -> None:
|
||||
if max_records and len(store["records"]) > max_records:
|
||||
store["records"] = store["records"][-max_records:]
|
||||
path.write_text(json.dumps(store, indent=2, default=str))
|
||||
|
||||
|
||||
# ── main loop ────────────────────────────────────────────────────────────────
|
||||
|
||||
def run(args: argparse.Namespace) -> None:
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG if args.verbose else logging.INFO,
|
||||
format="%(asctime)s %(levelname)s %(message)s",
|
||||
)
|
||||
|
||||
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)
|
||||
|
||||
out = Path(args.output)
|
||||
|
||||
log.info("Connecting to WeConnect…")
|
||||
wc = weconnect.WeConnect(
|
||||
username=args.username,
|
||||
password=args.password,
|
||||
updateAfterLogin=False,
|
||||
loginOnInit=True,
|
||||
)
|
||||
wc.login()
|
||||
wc.update()
|
||||
|
||||
vehicles = list(wc.vehicles.values())
|
||||
if not vehicles:
|
||||
log.error("No vehicles found in this WeConnect account")
|
||||
sys.exit(1)
|
||||
|
||||
if args.vin:
|
||||
vehicle = next((v for v in vehicles if v.vin.value == args.vin), None)
|
||||
if not vehicle:
|
||||
log.error(
|
||||
"VIN %s not found. Available: %s",
|
||||
args.vin,
|
||||
[v.vin.value for v in vehicles],
|
||||
)
|
||||
sys.exit(1)
|
||||
else:
|
||||
vehicle = vehicles[0]
|
||||
|
||||
vin = vehicle.vin.value
|
||||
log.info("Vehicle: %s", vin)
|
||||
|
||||
store = load_store(out, vin, args.interval)
|
||||
log.info(
|
||||
"Collecting [%s] every %ds → %s (Ctrl-C to stop)",
|
||||
", ".join(domains),
|
||||
args.interval,
|
||||
out,
|
||||
)
|
||||
|
||||
try:
|
||||
while True:
|
||||
try:
|
||||
wc.update()
|
||||
snapshot = collect_snapshot(vehicle, domains)
|
||||
store["records"].append(snapshot)
|
||||
save_store(out, store, args.max_records)
|
||||
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 keys: username, password, vin, domains. "
|
||||
"Values are overridden by explicit CLI flags and env vars."
|
||||
),
|
||||
)
|
||||
creds.add_argument(
|
||||
"-u", "--username",
|
||||
default=os.environ.get("WC_USER"),
|
||||
help="WeConnect / MyVolkswagen email",
|
||||
)
|
||||
creds.add_argument(
|
||||
"-p", "--password",
|
||||
default=os.environ.get("WC_PASS"),
|
||||
help="WeConnect password",
|
||||
)
|
||||
|
||||
p.add_argument("--vin", help="Vehicle VIN (uses first vehicle when omitted)")
|
||||
p.add_argument(
|
||||
"-d", "--domains",
|
||||
default=None,
|
||||
metavar="DOMAIN[,DOMAIN…]",
|
||||
help=(
|
||||
"Comma-separated domains to collect, or 'all'. "
|
||||
f"Available: {', '.join(ALL_DOMAINS)}. "
|
||||
"Default: charging,measurements,readiness"
|
||||
),
|
||||
)
|
||||
p.add_argument(
|
||||
"-i", "--interval",
|
||||
type=int,
|
||||
default=300,
|
||||
metavar="SECONDS",
|
||||
help="Collection interval in seconds (default: 300)",
|
||||
)
|
||||
p.add_argument(
|
||||
"-o", "--output",
|
||||
default="vehicle_data.json",
|
||||
metavar="FILE",
|
||||
help="Output JSON file (default: vehicle_data.json)",
|
||||
)
|
||||
p.add_argument(
|
||||
"--max-records",
|
||||
type=int,
|
||||
default=None,
|
||||
metavar="N",
|
||||
help="Keep only the last N records in the file (default: unlimited)",
|
||||
)
|
||||
p.add_argument("-v", "--verbose", action="store_true", help="Enable debug logging")
|
||||
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)
|
||||
|
||||
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}")
|
||||
if args.username is None:
|
||||
args.username = creds_data.get("username")
|
||||
if args.password is None:
|
||||
args.password = creds_data.get("password")
|
||||
if args.vin is None:
|
||||
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
|
||||
|
||||
if args.domains is None:
|
||||
args.domains = "charging,measurements,readiness"
|
||||
|
||||
if not args.username or not args.password:
|
||||
p.error(
|
||||
"Username and password are required. "
|
||||
"Pass them with -u/-p or set WC_USER / WC_PASS environment variables."
|
||||
)
|
||||
|
||||
run(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user