diff --git a/collect.py b/collect.py index 10af9de..d5577d1 100644 --- a/collect.py +++ b/collect.py @@ -44,394 +44,30 @@ Credentials file format (JSON) } 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", - "chargePower": {"value": 9.5, "unit": "kW"}, ... }, - "batteryStatus": { "currentSOC": {"value": 80, "unit": "%"}, - "cruisingRangeElectric": {"value": 295, "unit": "km"} }, - "chargingSettings": { "targetSOC": {"value": 80, "unit": "%"}, ... }, - "plugStatus": { "plugConnectionState": "disconnected", ... } - }, - "measurements": { - "odometerStatus": { "odometer": {"value": 12345, "unit": "km"} }, - "rangeStatus": { "electricRange": {"value": 295, "unit": "km"}, - "totalRange": {"value": 295, "unit": "km"} }, - "temperatureBatteryStatus": { "temperatureHvBatteryMax": {"value": 22.0, "unit": "degC"}, - "temperatureHvBatteryMin": {"value": 20.0, "unit": "degC"} }, - "temperatureOutsideStatus": { "temperatureOutside": {"value": 18.0, "unit": "degC"} } - }, - "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 socket import sys -import threading 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) +import log_config +import network +import we_connect +from data_model import auto_out_path, load_store, save_store +from we_connect import ALL_DOMAINS 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) - - -def _phys(value, unit: str) -> dict: - return {"value": value, "unit": unit} - - -# ── 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": _phys(cs.chargePower_kW.value, "kW"), - "remainingChargingTimeToComplete": _phys(cs.remainingChargingTimeToComplete_min.value, "min"), - "chargeType": _str(cs.chargeType), - } - except Exception: - log.debug("chargingStatus unavailable", exc_info=True) - - try: - bs = d["batteryStatus"] - result["batteryStatus"] = { - "currentSOC": _phys(bs.currentSOC_pct.value, "%"), - "cruisingRangeElectric": _phys(bs.cruisingRangeElectric_km.value, "km"), - } - except Exception: - log.debug("charging/batteryStatus unavailable", exc_info=True) - - try: - cfg = d["chargingSettings"] - result["chargingSettings"] = { - "targetSOC": _phys(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": _phys(cl.remainingClimatisationTime_min.value, "min"), - } - except Exception: - log.debug("climatisationStatus unavailable", exc_info=True) - - try: - cs = d["climatisationSettings"] - result["climatisationSettings"] = { - "targetTemperature": _phys(cs.targetTemperature_C.value, "degC"), - "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": _phys(od.odometer.value, "km")} - except Exception: - log.debug("odometerStatus unavailable", exc_info=True) - - try: - rs = d["rangeStatus"] - result["rangeStatus"] = { - "electricRange": _phys(rs.electricRange.value, "km"), - "totalRange": _phys(rs.totalRange_km.value, "km"), - } - except Exception: - log.debug("rangeStatus unavailable", exc_info=True) - - # Battery temperature — reported in Kelvin by the API - try: - bs = d["temperatureBatteryStatus"] - result["temperatureBatteryStatus"] = { - "temperatureHvBatteryMax": _phys(float(bs.temperatureHvBatteryMax_K.value - 273.15), "degC"), - "temperatureHvBatteryMin": _phys(float(bs.temperatureHvBatteryMin_K.value - 273.15), "degC"), - } - except Exception: - log.debug("measurements/temperatureBatteryStatus (temperature) unavailable", exc_info=True) - - try: - temp = d["temperatureOutsideStatus"] - result["temperatureOutsideStatus"] = { - "temperatureOutside": _phys(temp.temperatureOutside_C.value, "degC"), - } - 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)) - - -# ── push server ────────────────────────────────────────────────────────────── - -def _accept_loop( - server_sock: socket.socket, - clients: list, - lock: threading.Lock, -) -> None: - while True: - try: - conn, addr = server_sock.accept() - log.info("Push client connected from %s", addr) - with lock: - clients.append(conn) - except OSError: - break - - -def _broadcast(snapshot: dict, clients: list, lock: threading.Lock) -> None: - line = (json.dumps(snapshot, default=str) + "\n").encode() - with lock: - dead = [] - for conn in clients: - try: - conn.sendall(line) - except OSError: - dead.append(conn) - for conn in dead: - clients.remove(conn) - conn.close() - log.info("Push client disconnected") - - -def _start_push_server(host: str, port: int) -> tuple: - server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - server_sock.bind((host, port)) - server_sock.listen() - clients: list = [] - lock = threading.Lock() - t = threading.Thread(target=_accept_loop, args=(server_sock, clients, lock), daemon=True) - t.start() - log.info("Push server listening on %s:%d", host, port) - return clients, lock - - # ── main loop ──────────────────────────────────────────────────────────────── -def _auto_out_path(logs_dir: Path, vin: str) -> Path: - stamp = datetime.now(timezone.utc).strftime("%Y_%m_%d_%H_%M_%S") - return logs_dir / f"{stamp}_{vin}.json" - - def run(args: argparse.Namespace) -> None: - logging.basicConfig( - level=logging.DEBUG if args.verbose else logging.INFO, - format="%(asctime)s %(levelname)s %(message)s", - ) + log_config.setup(args.verbose) raw = [d.strip() for d in args.domains.split(",")] domains = list(ALL_DOMAINS) if raw == ["all"] else raw @@ -441,32 +77,13 @@ def run(args: argparse.Namespace) -> None: sys.exit(1) log.info("Connecting to WeConnect…") - wc = weconnect.WeConnect( - username=args.username, - password=args.password, - updateAfterLogin=False, - loginOnInit=True, - ) - wc.login() - wc.update() + wc = we_connect.connect(args.username, args.password) - vehicles = list(wc.vehicles.values()) - if not vehicles: - log.error("No vehicles found in this WeConnect account") + vehicle, err = we_connect.select_vehicle(wc, args.vin) + if err: + log.error(err) 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) @@ -476,9 +93,9 @@ def run(args: argparse.Namespace) -> None: else: logs_dir = Path("logs") logs_dir.mkdir(exist_ok=True) - out = _auto_out_path(logs_dir, vin) + out = auto_out_path(logs_dir, vin) - push_clients, push_lock = _start_push_server(args.host, args.port) + push_clients, push_lock = network.start_push_server(args.host, args.port) current_day = datetime.now(timezone.utc).date() store = load_store(out, vin, args.interval) @@ -494,17 +111,17 @@ def run(args: argparse.Namespace) -> None: if logs_dir is not None: today = datetime.now(timezone.utc).date() if today != current_day: - out = _auto_out_path(logs_dir, vin) + out = auto_out_path(logs_dir, vin) store = load_store(out, vin, args.interval) current_day = today log.info("Midnight UTC rotation → %s", out) try: wc.update() - snapshot = collect_snapshot(vehicle, domains) + snapshot = we_connect.collect_snapshot(vehicle, domains) store["records"].append(snapshot) save_store(out, store, args.max_records) - _broadcast(snapshot, push_clients, push_lock) + network.broadcast(snapshot, push_clients, push_lock) log.info("Record #%d saved at %s", len(store["records"]), snapshot["ts"]) except KeyboardInterrupt: raise @@ -578,6 +195,7 @@ def main() -> None: metavar="N", help="Keep only the last N records in the file (default: unlimited)", ) + srv = p.add_argument_group("push server") srv.add_argument( "--host", @@ -638,4 +256,4 @@ def main() -> None: if __name__ == "__main__": - main() + main() \ No newline at end of file diff --git a/data_model.py b/data_model.py new file mode 100644 index 0000000..c557783 --- /dev/null +++ b/data_model.py @@ -0,0 +1,43 @@ +import json +import logging +from datetime import datetime, timezone +from pathlib import Path + +log = logging.getLogger(__name__) + + +def _str(v) -> str: + return str(v.value) if hasattr(v, "value") else str(v) + + +def _phys(value, unit: str) -> dict: + return {"value": value, "unit": unit} + + +def auto_out_path(logs_dir: Path, vin: str) -> Path: + stamp = datetime.now(timezone.utc).strftime("%Y_%m_%d_%H_%M_%S") + return logs_dir / f"{stamp}_{vin}.json" + + +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)) \ No newline at end of file diff --git a/log_config.py b/log_config.py new file mode 100644 index 0000000..754c92e --- /dev/null +++ b/log_config.py @@ -0,0 +1,8 @@ +import logging + + +def setup(verbose: bool) -> None: + logging.basicConfig( + level=logging.DEBUG if verbose else logging.INFO, + format="%(asctime)s %(levelname)s %(message)s", + ) \ No newline at end of file diff --git a/network.py b/network.py new file mode 100644 index 0000000..c0309c9 --- /dev/null +++ b/network.py @@ -0,0 +1,49 @@ +import json +import logging +import socket +import threading + +log = logging.getLogger(__name__) + + +def _accept_loop( + server_sock: socket.socket, + clients: list, + lock: threading.Lock, +) -> None: + while True: + try: + conn, addr = server_sock.accept() + log.info("Push client connected from %s", addr) + with lock: + clients.append(conn) + except OSError: + break + + +def broadcast(snapshot: dict, clients: list, lock: threading.Lock) -> None: + line = (json.dumps(snapshot, default=str) + "\n").encode() + with lock: + dead = [] + for conn in clients: + try: + conn.sendall(line) + except OSError: + dead.append(conn) + for conn in dead: + clients.remove(conn) + conn.close() + log.info("Push client disconnected") + + +def start_push_server(host: str, port: int) -> tuple: + server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server_sock.bind((host, port)) + server_sock.listen() + clients: list = [] + lock = threading.Lock() + t = threading.Thread(target=_accept_loop, args=(server_sock, clients, lock), daemon=True) + t.start() + log.info("Push server listening on %s:%d", host, port) + return clients, lock \ No newline at end of file diff --git a/we_connect.py b/we_connect.py new file mode 100644 index 0000000..29e30f8 --- /dev/null +++ b/we_connect.py @@ -0,0 +1,272 @@ +import logging +import sys +from datetime import datetime, timezone + +from data_model import _phys, _str + +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__) + + +# ── domain extractors ──────────────────────────────────────────────────────── + +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": _phys(cs.chargePower_kW.value, "kW"), + "remainingChargingTimeToComplete": _phys(cs.remainingChargingTimeToComplete_min.value, "min"), + "chargeType": _str(cs.chargeType), + } + except Exception: + log.debug("chargingStatus unavailable", exc_info=True) + + try: + bs = d["batteryStatus"] + result["batteryStatus"] = { + "currentSOC": _phys(bs.currentSOC_pct.value, "%"), + "cruisingRangeElectric": _phys(bs.cruisingRangeElectric_km.value, "km"), + } + except Exception: + log.debug("charging/batteryStatus unavailable", exc_info=True) + + try: + cfg = d["chargingSettings"] + result["chargingSettings"] = { + "targetSOC": _phys(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": _phys(cl.remainingClimatisationTime_min.value, "min"), + } + except Exception: + log.debug("climatisationStatus unavailable", exc_info=True) + + try: + cs = d["climatisationSettings"] + result["climatisationSettings"] = { + "targetTemperature": _phys(cs.targetTemperature_C.value, "degC"), + "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": _phys(od.odometer.value, "km")} + except Exception: + log.debug("odometerStatus unavailable", exc_info=True) + + try: + rs = d["rangeStatus"] + result["rangeStatus"] = { + "electricRange": _phys(rs.electricRange.value, "km"), + "totalRange": _phys(rs.totalRange_km.value, "km"), + } + except Exception: + log.debug("rangeStatus unavailable", exc_info=True) + + try: + bs = d["temperatureBatteryStatus"] + result["temperatureBatteryStatus"] = { + "temperatureHvBatteryMax": _phys(float(bs.temperatureHvBatteryMax_K.value - 273.15), "degC"), + "temperatureHvBatteryMin": _phys(float(bs.temperatureHvBatteryMin_K.value - 273.15), "degC"), + } + except Exception: + log.debug("measurements/temperatureBatteryStatus unavailable", exc_info=True) + + try: + temp = d["temperatureOutsideStatus"] + result["temperatureOutsideStatus"] = { + "temperatureOutside": _phys(temp.temperatureOutside_C.value, "degC"), + } + 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, +} + + +# ── connection helpers ─────────────────────────────────────────────────────── + +def connect(username: str, password: str): + wc = weconnect.WeConnect( + username=username, + password=password, + updateAfterLogin=False, + loginOnInit=True, + ) + wc.login() + wc.update() + return wc + + +def select_vehicle(wc, vin: str | None): + """Return (vehicle, error_message). error_message is None on success.""" + vehicles = list(wc.vehicles.values()) + if not vehicles: + return None, "No vehicles found in this WeConnect account" + if vin: + vehicle = next((v for v in vehicles if v.vin.value == vin), None) + if not vehicle: + available = [v.vin.value for v in vehicles] + return None, f"VIN {vin} not found. Available: {available}" + return vehicle, None + return vehicles[0], None + + +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 \ No newline at end of file