117 lines
7.6 KiB
Python
117 lines
7.6 KiB
Python
#!/usr/bin/env python3
|
|
# lecture-tableau.py — lit le kilométrage et le niveau de batterie sur une PHOTO de tableau de bord ou une
|
|
# CAPTURE de l'application constructeur, envoyée par Thierry sur WhatsApp (04/10/2026, « GO photo Jeep »).
|
|
# Le modèle de la Spark (vision) lit ; ce script met la lecture EN ATTENTE ; rien n'est écrit tant que
|
|
# Thierry n'a pas confirmé. La confirmation passe le texte au concierge, seul chemin d'écriture.
|
|
# python3 lecture-tableau.py lire <chemin_image> [JEEP|VW|ZERO] → lit, imprime une ligne, met en attente
|
|
# python3 lecture-tableau.py confirmer → envoie la lecture en attente au concierge
|
|
# python3 lecture-tableau.py annuler → efface la lecture en attente
|
|
# Le véhicule vient de Thierry (légende de la photo), jamais de l'image.
|
|
import base64, json, os, re, subprocess, sys, time, urllib.request
|
|
|
|
WS = "/home/node/.openclaw/workspace"
|
|
PENDING = os.path.join(WS, "releve-photo-pending.json")
|
|
SPARK = os.environ.get("SPARK_URL", "http://10.10.10.8:8000/v1/chat/completions")
|
|
HA = os.environ.get("HA_BASE", "http://10.10.10.17:8123")
|
|
|
|
def minutes_branchee():
|
|
"""Minutes depuis le branchement si une voiture est branchée, sinon None (lecture HA seule)."""
|
|
try:
|
|
tok = open(os.path.join(WS, ".ha-token")).read().strip()
|
|
r = urllib.request.urlopen(urllib.request.Request(
|
|
f"{HA}/api/states/binary_sensor.victron_evcs_vehicle_connected",
|
|
headers={"Authorization": f"Bearer {tok}"}), timeout=10)
|
|
st = json.load(r)
|
|
if st.get("state") != "on": return None
|
|
from datetime import datetime, timezone
|
|
return (datetime.now(timezone.utc) - datetime.fromisoformat(st["last_changed"])).total_seconds() / 60
|
|
except Exception:
|
|
return None
|
|
|
|
VEHICULES = {"JEEP": "JEEP", "VW": "VW", "ZERO": "ZERO", "MOTO": "ZERO", "ZÉRO": "ZERO"}
|
|
# 12/10/2026 (propriétaire) : la VW n'affiche pas de % mais l'autonomie électrique. Le % est alors DÉDUIT ici
|
|
# (règle de trois sur l'autonomie pleine lue dans HA, recalée par EVCS), jamais par l'agent ; la ligne dit
|
|
# « estimé » et le relevé porte la source « photo_autonomie:<km>/<réf> » (EVCS : la capacité peut l'écarter).
|
|
AUTONOMIE_PLEINE = {"VW": "input_number.victron_evcs_autonomie_vw_km"}
|
|
|
|
def autonomie_pleine(veh):
|
|
"""Autonomie à 100 % lue dans HA ; None si absente ou illisible (alors pas d'estimation)."""
|
|
if veh not in AUTONOMIE_PLEINE: return None
|
|
try:
|
|
tok = open(os.path.join(WS, ".ha-token")).read().strip()
|
|
st = json.load(urllib.request.urlopen(urllib.request.Request(f"{HA}/api/states/{AUTONOMIE_PLEINE[veh]}",
|
|
headers={"Authorization": f"Bearer {tok}"}), timeout=10))
|
|
v = round(float(st["state"]))
|
|
return v if 10 <= v <= 1000 else None
|
|
except Exception:
|
|
return None
|
|
PROMPT = ("Photo d'un tableau de bord de véhicule, ou capture d'une application constructeur. Réponds en JSON "
|
|
"strict, sans texte autour : {\"kilometrage_km\": entier ou null, \"batterie_pct\": entier ou null, "
|
|
"\"autonomie_electrique_km\": entier ou null}. kilometrage_km = l'ODOMÈTRE (kilométrage total du "
|
|
"véhicule, souvent 5 ou 6 chiffres), JAMAIS une autonomie ni un trajet. batterie_pct = le pourcentage "
|
|
"de la batterie de traction s'il est écrit en chiffres, sinon null (ne pas l'estimer d'après une jauge). "
|
|
"Valeur illisible ou absente = null.")
|
|
|
|
def lire(img, veh):
|
|
if not os.path.isfile(img): print(f"Lecture impossible : image introuvable ({img})."); return
|
|
ext = os.path.splitext(img)[1].lower().lstrip(".") or "jpeg"
|
|
data = base64.b64encode(open(img, "rb").read()).decode()
|
|
body = {"model": "qwen3-coder", "temperature": 0, "max_tokens": 200, "messages": [{"role": "user", "content": [
|
|
{"type": "text", "text": PROMPT},
|
|
{"type": "image_url", "image_url": {"url": f"data:image/{'jpeg' if ext == 'jpg' else ext};base64,{data}"}}]}]}
|
|
try:
|
|
r = urllib.request.urlopen(urllib.request.Request(SPARK, data=json.dumps(body).encode(),
|
|
headers={"Content-Type": "application/json"}), timeout=240)
|
|
txt = json.load(r)["choices"][0]["message"]["content"]
|
|
v = json.loads(re.search(r"\{.*\}", txt, re.S).group(0))
|
|
except Exception as e:
|
|
print(f"Lecture impossible : le modèle n'a pas répondu correctement ({e})."); return
|
|
km, pct, aut = v.get("kilometrage_km"), v.get("batterie_pct"), v.get("autonomie_electrique_km")
|
|
km = km if isinstance(km, int) and 1000 <= km <= 2_000_000 else None
|
|
pct = pct if isinstance(pct, int) and 0 <= pct <= 100 else None
|
|
if km is None:
|
|
print("📷 Kilométrage illisible sur l'image : rien à noter. Donne-le par écrit (« JEEP 112716 km 46 % »)."); return
|
|
estime, source = "", None
|
|
ref = autonomie_pleine(veh) if pct is None and isinstance(aut, int) else None
|
|
if ref and 0 <= aut <= ref * 1.2:
|
|
pct = min(100, round(aut * 100 / ref))
|
|
estime = f" (estimé : {aut} km d'autonomie sur {ref} km à 100 %)"
|
|
source = f"photo_autonomie:{aut}/{ref}"
|
|
lu = f"{km:,} km".replace(",", " ") + (f", batterie {pct} %{estime}" if pct is not None else ", batterie non lisible")
|
|
if isinstance(aut, int) and not estime: lu += f", autonomie électrique {aut} km"
|
|
if not veh:
|
|
print(f"📷 Lu sur l'image : {lu}. Quel véhicule est-ce (JEEP, VW, ZERO) ? Rien n'est noté."); return
|
|
# 04/10 (EVCS) : le relevé est daté au branchement ; un % lu sur une photo prise après plus de 10 min de
|
|
# charge n'est plus celui du branchement et fausserait km/kWh et capacité → on ne garde que le kilométrage.
|
|
mb = minutes_branchee()
|
|
note = ""
|
|
if pct is not None and mb is not None and mb > 10:
|
|
note = (f" Niveau NON retenu : la voiture charge depuis {int(mb)} min, le % de la photo n'est plus celui "
|
|
f"du branchement (photo valable pour le niveau seulement au retour, avant de brancher).")
|
|
pct, source = None, None
|
|
lu = f"{km:,} km".replace(",", " ") + (f", autonomie électrique {aut} km" if isinstance(aut, int) else "")
|
|
texte = f"{veh} {km} km" + (f" {pct} %" if pct is not None else "")
|
|
json.dump({"texte": texte, "lu": lu, "vehicule": veh, "image": img, "a": int(time.time()), "source": source}, open(PENDING, "w"))
|
|
print(f"📷 Lu sur l'image : {veh}, {lu}.{note} Je note ce relevé ? Réponds « oui » ou « non ». Rien n'est encore noté.")
|
|
|
|
def confirmer():
|
|
if not os.path.exists(PENDING): print("Aucune lecture de photo en attente : rien à noter."); return
|
|
p = json.load(open(PENDING))
|
|
if time.time() - p["a"] > 3600:
|
|
os.remove(PENDING); print("La lecture de photo date de plus d'une heure : annulée, renvoie la photo."); return
|
|
env = dict(os.environ)
|
|
if p.get("source"): env["CONCIERGE_RELEVE_SOURCE"] = p["source"] # niveau estimé → 5e champ du relevé
|
|
res = subprocess.run(["node", os.path.join(WS, "concierge-live.js"), "answer", p["texte"]],
|
|
capture_output=True, text=True, timeout=120, env=env)
|
|
os.remove(PENDING)
|
|
print((res.stdout.strip() or f"Rien n'est noté : erreur du concierge ({res.stderr.strip()[:200]})."))
|
|
|
|
cmd = sys.argv[1] if len(sys.argv) > 1 else ""
|
|
if cmd == "lire" and len(sys.argv) > 2:
|
|
lire(sys.argv[2], VEHICULES.get(sys.argv[3].upper()) if len(sys.argv) > 3 else None)
|
|
elif cmd == "confirmer": confirmer()
|
|
elif cmd == "annuler":
|
|
if os.path.exists(PENDING): os.remove(PENDING)
|
|
print("Lecture de photo annulée : rien n'est noté.")
|
|
else: print("usage : lecture-tableau.py lire <image> [JEEP|VW|ZERO] | confirmer | annuler")
|