Move server modules into server/ package
data_model, log_config, network, storage_helpers, we_connect are now under server/. collect.py stays in the project root and imports via the server package. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 4.6
@@ -0,0 +1,245 @@
|
||||
import logging
|
||||
from datetime import datetime, timezone
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ── value helpers ─────────────────────────────────────────────────────────────
|
||||
|
||||
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}
|
||||
|
||||
|
||||
# ── 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,
|
||||
}
|
||||
|
||||
|
||||
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
|
||||
@@ -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",
|
||||
)
|
||||
@@ -0,0 +1,64 @@
|
||||
import json
|
||||
import logging
|
||||
import socket
|
||||
import threading
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _accept_loop(
|
||||
server_sock: socket.socket,
|
||||
clients: list,
|
||||
lock: threading.Lock,
|
||||
last_snapshot: dict,
|
||||
) -> None:
|
||||
while True:
|
||||
try:
|
||||
conn, addr = server_sock.accept()
|
||||
log.info("Push client connected from %s", addr)
|
||||
with lock:
|
||||
if last_snapshot["data"] is not None:
|
||||
try:
|
||||
line = (json.dumps(last_snapshot["data"], default=str) + "\n").encode()
|
||||
conn.sendall(line)
|
||||
except OSError:
|
||||
log.debug("Failed to send initial snapshot to %s", addr)
|
||||
conn.close()
|
||||
continue
|
||||
clients.append(conn)
|
||||
except OSError:
|
||||
break
|
||||
|
||||
|
||||
def broadcast(snapshot: dict, clients: list, lock: threading.Lock, last_snapshot: dict) -> None:
|
||||
line = (json.dumps(snapshot, default=str) + "\n").encode()
|
||||
with lock:
|
||||
last_snapshot["data"] = snapshot
|
||||
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()
|
||||
last_snapshot: dict = {"data": None}
|
||||
t = threading.Thread(
|
||||
target=_accept_loop,
|
||||
args=(server_sock, clients, lock, last_snapshot),
|
||||
daemon=True,
|
||||
)
|
||||
t.start()
|
||||
log.info("Push server listening on %s:%d", host, port)
|
||||
return clients, lock, last_snapshot
|
||||
@@ -0,0 +1,35 @@
|
||||
import json
|
||||
import logging
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
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))
|
||||
@@ -0,0 +1,36 @@
|
||||
import logging
|
||||
import sys
|
||||
|
||||
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__)
|
||||
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user