hier soir, je me suis amusé avec Claude pour essayer de l’intégrer dans Jeedom. Je ne sais absolument pas faire un plugin mais je l’ai orienté vers docker. Peut-être que cela pour servir à un développeur de plugin.
C’est donc fonctionnel avec mon navimow H1500. Via MQTT discovery, je récupère les informations de la batterie, l’état de tondeuse, le le temps de tonte (calculé par le docker). Il y a également 3 commandes actions : démarrer , pause, et retour à la base.
## Étape 1 — Installer Docker (si pas déjà fait)
bash
curl -fsSL https://get.docker.com | sh
sudo usermod -aG docker $USER
newgrp docker
docker --version
## Étape 2 — Récupérer votre token Navimow
2.1 — Installer les dépendances Python sur votre PC Windows
pip install requests flask
2.2 — Créer le script get_navimow_token.py
Ouvrez le Bloc-notes, collez le code ci-dessous, et enregistrez sous get_navimow_token.py sur votre Bureau (choisissez « Tous les fichiers » dans le type).
python
import webbrowser
import threading
import json
from flask import Flask, request
import requests
app = Flask(__name__)
AUTH_URL = "https://navimow-h5-fra.willand.com/smartHome/login"
TOKEN_URL = "https://navimow-fra.ninebot.com/openapi/oauth/getAccessToken"
CLIENT_ID = "homeassistant"
CLIENT_SECRET = "57056e15-722e-42be-bbaa-b0cbfb208a52"
REDIRECT = "http://localhost:8765/callback"
@app.route("/callback")
def callback():
code = request.args.get("code")
if not code:
return f"Pas de code reçu : {dict(request.args)}"
r = requests.post(TOKEN_URL, data={
"grant_type": "authorization_code",
"code": code,
"redirect_uri": REDIRECT,
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET
})
data = r.json()
token = data.get("access_token")
refresh = data.get("refresh_token")
expires = data.get("expires_in")
with open("navimow_token_full.json", "w") as f:
json.dump(data, f, indent=2)
if token:
print(f"\n✅ ACCESS TOKEN :\n{token}")
print(f"\n🔄 REFRESH TOKEN :\n{refresh}")
print(f"\n⏱️ Expire dans : {expires}s")
return f"""
<h2>✅ Token récupéré !</h2>
<h3>Réponse complète :</h3><pre>{json.dumps(data, indent=2)}</pre>
<h3>Access Token :</h3><code style="word-break:break-all">{token}</code>
<h3>Refresh Token :</h3><code style="word-break:break-all">{refresh}</code>
"""
if __name__ == "__main__":
url = f"{AUTH_URL}?client_id={CLIENT_ID}&response_type=code&redirect_uri={REDIRECT}&channel=homeassistant"
threading.Timer(1, lambda: webbrowser.open(url)).start()
print("Ouverture du navigateur Navimow...")
app.run(port=8765)
2.3 — Lancer le script
cd Desktop
python get_navimow_token.py
Une page de login Navimow s’ouvre. Connectez-vous, puis notez l’access_token et le refresh_token affichés.
Ce script n’est nécessaire qu’une seule fois. Le bridge renouvelle ensuite les tokens automatiquement.
Étape 3 — Créer les fichiers du bridge
bash
mkdir -p ~/navimow-jeedom
cd ~/navimow-jeedom
3.1 — Dockerfile
bash
cat > ~/navimow-jeedom/Dockerfile << 'EOF'
FROM python:3.11-slim
WORKDIR /app
RUN apt-get update && apt-get install -y --no-install-recommends gcc && rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY navimow_bridge.py .
RUN mkdir -p /data && chmod 777 /data
RUN useradd -m appuser && chown -R appuser /app
USER appuser
CMD ["python", "-u", "navimow_bridge.py"]
EOF
3.2 — requirements.txt
bash
cat > ~/navimow-jeedom/requirements.txt << 'EOF'
navimow-sdk
aiohttp>=3.9.0
paho-mqtt>=1.6.1
EOF
3.3 — navimow_bridge.py
bash
cat > ~/navimow-jeedom/navimow_bridge.py << 'EOF'
#!/usr/bin/env python3
import asyncio, json, logging, os, signal, sys, time
import aiohttp
import paho.mqtt.client as mqtt
from mower_sdk.api import MowerAPI
from mower_sdk.models import MowerCommand
NAVIMOW_TOKEN = os.environ["NAVIMOW_TOKEN"]
NAVIMOW_REFRESH_TOKEN = os.environ.get("NAVIMOW_REFRESH_TOKEN", "")
NAVIMOW_API_URL = os.environ.get("NAVIMOW_API_URL", "https://navimow-fra.ninebot.com")
NAVIMOW_DEVICE_ID = os.environ.get("NAVIMOW_DEVICE_ID", "")
MQTT_HOST = os.environ["MQTT_HOST"]
MQTT_PORT = int(os.environ.get("MQTT_PORT", "1883"))
MQTT_USER = os.environ.get("MQTT_USER", "")
MQTT_PASS = os.environ.get("MQTT_PASS", "")
MQTT_BASE_TOPIC = os.environ.get("MQTT_BASE_TOPIC", "navimow")
POLL_INTERVAL = int(os.environ.get("POLL_INTERVAL", "30"))
POLL_INTERVAL_MOWING = int(os.environ.get("POLL_INTERVAL_MOWING", "15"))
DISCOVERY_PREFIX = os.environ.get("DISCOVERY_PREFIX", "homeassistant")
TOKEN_FILE = os.environ.get("TOKEN_FILE", "/data/tokens.json")
CLIENT_ID = "homeassistant"
CLIENT_SECRET = "57056e15-722e-42be-bbaa-b0cbfb208a52"
TOKEN_URL = "https://navimow-fra.ninebot.com/openapi/oauth/getAccessToken"
TOKEN_REFRESH_MARGIN = 300
MOWING_STATES = {"mowing", "isrunning", "running", "returning"}
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S")
log = logging.getLogger("navimow-bridge")
def topic(suffix): return f"{MQTT_BASE_TOPIC}/{suffix}"
def safe_value(val): return val.value if hasattr(val, 'value') else val
class TokenManager:
def __init__(self):
self.session = None
saved = self._load_from_file()
if saved:
self.access_token = saved["access_token"]; self.refresh_token = saved["refresh_token"]; self.expires_at = saved["expires_at"]
log.info(f"💾 Tokens chargés depuis {TOKEN_FILE} (expire dans {int(self.expires_at - time.time())}s)")
else:
self.access_token = NAVIMOW_TOKEN; self.refresh_token = NAVIMOW_REFRESH_TOKEN; self.expires_at = time.time() + 3600
log.info("🔑 Tokens chargés depuis les variables d'environnement"); self._save_to_file()
def set_session(self, s): self.session = s
def is_expired(self): return time.time() >= (self.expires_at - TOKEN_REFRESH_MARGIN)
def _load_from_file(self):
try:
if os.path.exists(TOKEN_FILE):
with open(TOKEN_FILE) as f: data = json.load(f)
if all(k in data for k in ("access_token","refresh_token","expires_at")): return data
except Exception as e: log.warning(f"⚠️ Impossible de lire {TOKEN_FILE} : {e}")
return None
def _save_to_file(self):
try:
os.makedirs(os.path.dirname(TOKEN_FILE), exist_ok=True)
with open(TOKEN_FILE,"w") as f: json.dump({"access_token":self.access_token,"refresh_token":self.refresh_token,"expires_at":self.expires_at},f,indent=2)
log.info(f"💾 Tokens sauvegardés dans {TOKEN_FILE}")
except Exception as e: log.warning(f"⚠️ Impossible de sauvegarder {TOKEN_FILE} : {e}")
async def refresh(self):
if not self.refresh_token: log.error("❌ Pas de refresh_token"); return False
try:
log.info("🔄 Renouvellement du token...")
async with self.session.post(TOKEN_URL, data={"grant_type":"refresh_token","refresh_token":self.refresh_token,"client_id":CLIENT_ID,"client_secret":CLIENT_SECRET}) as resp:
data = await resp.json()
if data.get("access_token"):
self.access_token = data["access_token"]; self.refresh_token = data.get("refresh_token", self.refresh_token)
self.expires_at = time.time() + int(data.get("expires_in",3600)); self._save_to_file()
log.info(f"✅ Token renouvelé, expire dans {int(data.get('expires_in',3600))}s"); return True
log.error(f"❌ Pas de access_token : {data}")
except Exception as e: log.error(f"❌ Exception refresh : {e}", exc_info=True)
return False
class JeedomMQTT:
def __init__(self, loop):
self.loop = loop
self.client = mqtt.Client(client_id="navimow-bridge", clean_session=True)
if MQTT_USER: self.client.username_pw_set(MQTT_USER, MQTT_PASS)
self.client.on_connect = self._on_connect; self.client.on_message = self._on_message; self.client.on_disconnect = self._on_disconnect
self.command_callback = None
def connect(self):
log.info(f"Connexion MQTT → {MQTT_HOST}:{MQTT_PORT}"); self.client.connect(MQTT_HOST, MQTT_PORT, keepalive=60); self.client.loop_start()
def _on_connect(self, client, userdata, flags, rc):
if rc==0: log.info("✅ MQTT connecté"); client.subscribe(topic("cmnd/#")); self.publish("bridge/status","online")
else: log.error(f"❌ MQTT erreur : code {rc}")
def _on_disconnect(self, client, userdata, rc): log.warning(f"⚠️ MQTT déconnecté ({rc})")
def _on_message(self, client, userdata, msg):
payload = msg.payload.decode("utf-8",errors="ignore").strip()
log.info(f"📨 MQTT reçu : {msg.topic} → {payload}")
if self.command_callback:
parts = msg.topic.split("/")
if len(parts)>=3: asyncio.run_coroutine_threadsafe(self.command_callback(parts[-1].lower(),payload),self.loop)
def publish(self, suffix, value, retain=False): self.client.publish(topic(suffix), str(value), retain=retain)
def publish_discovery(self, component, object_id, config):
t = f"{DISCOVERY_PREFIX}/{component}/{object_id}/config"; self.client.publish(t, json.dumps(config), retain=True); log.info(f"📣 Discovery : {t}")
class NavimowBridge:
def __init__(self, loop):
self.loop=loop; self.token_manager=TokenManager(); self.jeedom=JeedomMQTT(loop)
self.jeedom.command_callback=self.handle_command; self.api=None
self.device_id=NAVIMOW_DEVICE_ID or None; self.device_name="Navimow H1500"
self._running=True; self._is_mowing=False; self._mowing_start=None
def _current_poll_interval(self): return POLL_INTERVAL_MOWING if self._is_mowing else POLL_INTERVAL
async def start(self):
self.jeedom.connect(); await asyncio.sleep(2)
session = aiohttp.ClientSession(); self.token_manager.set_session(session)
self.api = MowerAPI(session=session, token=self.token_manager.access_token, base_url=NAVIMOW_API_URL)
if not self.device_id: await self.discover_device()
if not self.device_id: log.error("❌ Aucun appareil trouvé."); return
self.publish_discovery_messages(); await asyncio.sleep(2)
log.info(f"🤖 Robot : {self.device_id}")
while self._running:
await self._refresh_token_if_needed(); await self.refresh_status(); await asyncio.sleep(self._current_poll_interval())
async def _refresh_token_if_needed(self):
if self.token_manager.is_expired():
if await self.token_manager.refresh(): self.api.set_token(self.token_manager.access_token)
async def discover_device(self):
try:
devices = await self.api.async_get_devices()
if devices:
d=devices[0]; self.device_id=getattr(d,'id',None) or str(d); self.device_name=getattr(d,'name','Navimow')
log.info(f"🔍 Appareil : {self.device_id} ({self.device_name})")
except Exception as e: log.error(f"Erreur découverte : {e}", exc_info=True)
def publish_discovery_messages(self):
di={"identifiers":[self.device_id],"name":self.device_name,"model":"H1500","manufacturer":"Segway Navimow"}
self.jeedom.publish_discovery("sensor","navimow_state", {"name":"Etat tondeuse", "unique_id":f"{self.device_id}_state", "state_topic":topic("stat/state"), "icon":"mdi:robot-mower","device":di})
self.jeedom.publish_discovery("sensor","navimow_vehicle_state", {"name":"Etat détaillé", "unique_id":f"{self.device_id}_vehicle_state", "state_topic":topic("stat/vehicleState"), "icon":"mdi:robot-mower","device":di})
self.jeedom.publish_discovery("sensor","navimow_battery", {"name":"Batterie", "unique_id":f"{self.device_id}_battery", "state_topic":topic("stat/battery"), "unit_of_measurement":"%","device_class":"battery","icon":"mdi:battery","device":di})
self.jeedom.publish_discovery("sensor","navimow_battery_level", {"name":"Niveau batterie", "unique_id":f"{self.device_id}_battery_level", "state_topic":topic("stat/batteryLevel"), "icon":"mdi:battery-heart","device":di})
self.jeedom.publish_discovery("sensor","navimow_mowing_time_min",{"name":"Duree tonte (min)", "unique_id":f"{self.device_id}_mowing_time_min","state_topic":topic("stat/mowing_time_min"),"unit_of_measurement":"min","icon":"mdi:timer","device":di})
self.jeedom.publish_discovery("button","navimow_start", {"name":"Démarrer tonte", "unique_id":f"{self.device_id}_start", "command_topic":topic("cmnd/start"), "payload_press":"go","icon":"mdi:play","device":di})
self.jeedom.publish_discovery("button","navimow_pause", {"name":"Pause", "unique_id":f"{self.device_id}_pause", "command_topic":topic("cmnd/pause"), "payload_press":"pause","icon":"mdi:pause","device":di})
self.jeedom.publish_discovery("button","navimow_dock", {"name":"Retour au dock", "unique_id":f"{self.device_id}_dock", "command_topic":topic("cmnd/dock"), "payload_press":"dock","icon":"mdi:home","device":di})
log.info("📣 Messages MQTT Discovery publiés")
async def refresh_status(self):
try:
status = await self.api.async_get_device_status(self.device_id)
if status: self._publish_status(status)
except Exception as e: log.error(f"Erreur status : {e}", exc_info=True)
def _publish_status(self, status):
state=safe_value(getattr(status,'status','unknown')); battery=getattr(status,'battery',0)
error=safe_value(getattr(status,'error_code','none')); extra=getattr(status,'extra',{}) or {}
was_mowing=self._is_mowing; self._is_mowing=str(state).lower() in MOWING_STATES
if self._is_mowing and not was_mowing: self._mowing_start=time.time(); log.info("⏱️ Début de tonte enregistré")
elif not self._is_mowing and was_mowing: self._mowing_start=None; log.info("⏱️ Fin de tonte")
if self._is_mowing and self._mowing_start:
elapsed=int(time.time()-self._mowing_start)
self.jeedom.publish("stat/mowing_time_min", int(elapsed/60), retain=True)
self.jeedom.publish("stat/state", state, retain=True)
self.jeedom.publish("stat/battery", battery, retain=True)
self.jeedom.publish("stat/error", error, retain=True)
if 'vehicleState' in extra: self.jeedom.publish("stat/vehicleState", extra['vehicleState'], retain=True)
if 'descriptiveCapacityRemaining' in extra: self.jeedom.publish("stat/batteryLevel", extra['descriptiveCapacityRemaining'], retain=True)
log.info(f"📊 état={state} batterie={battery}% erreur={error} (poll={self._current_poll_interval()}s)")
async def handle_command(self, command, payload):
await self._refresh_token_if_needed()
try:
cmd_map={"start":MowerCommand.START,"pause":MowerCommand.PAUSE,"resume":MowerCommand.RESUME,"dock":MowerCommand.DOCK}
if command=="status": await self.refresh_status(); return
if command not in cmd_map: log.warning(f"Commande inconnue : {command}"); return
await self.api.async_send_command(self.device_id, cmd_map[command])
log.info(f"✅ Commande '{command}' envoyée"); self.jeedom.publish("stat/last_command", command)
await asyncio.sleep(3); await self.refresh_status()
except Exception as e: log.error(f"Erreur commande '{command}' : {e}", exc_info=True)
async def stop(self):
self._running=False; self.jeedom.publish("bridge/status","offline")
self.jeedom.client.loop_stop(); self.jeedom.client.disconnect()
async def main():
loop=asyncio.get_event_loop(); bridge=NavimowBridge(loop)
for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, lambda: asyncio.create_task(bridge.stop()))
try: await bridge.start()
except Exception as e: log.error(f"Erreur fatale : {e}", exc_info=True); sys.exit(1)
if __name__ == "__main__": asyncio.run(main())
EOF
3.4 — docker-compose.yml
Remplacez VOTRE_ACCESS_TOKEN et VOTRE_REFRESH_TOKEN par les valeurs obtenues à l’étape 2 :
bash
cat > ~/navimow-jeedom/docker-compose.yml << 'EOF'
services:
navimow-bridge:
build: .
container_name: navimow-bridge
restart: unless-stopped
environment:
NAVIMOW_TOKEN: "VOTRE_ACCESS_TOKEN"
NAVIMOW_REFRESH_TOKEN: "VOTRE_REFRESH_TOKEN"
MQTT_HOST: "127.0.0.1"
MQTT_PORT: "1883"
MQTT_USER: ""
MQTT_PASS: ""
MQTT_BASE_TOPIC: "navimow"
NAVIMOW_DEVICE_ID: ""
POLL_INTERVAL: "30"
POLL_INTERVAL_MOWING: "15"
NAVIMOW_API_URL: "https://navimow-fra.ninebot.com"
DISCOVERY_PREFIX: "homeassistant"
TOKEN_FILE: "/data/tokens.json"
volumes:
- navimow_data:/data
network_mode: host
logging:
driver: "json-file"
options:
max-size: "5m"
max-file: "3"
volumes:
navimow_data:
EOF
Étape 4 — Lancer le bridge
bash
cd ~/navimow-jeedom
docker compose build
docker compose up -d
docker compose logs -f
Les logs doivent afficher :
💾 Tokens chargés depuis les variables d'environnement
💾 Tokens sauvegardés dans /data/tokens.json
✅ MQTT connecté
🔍 Appareil : XXXXXXXX (Navimow H1500)
📣 Messages MQTT Discovery publiés
🤖 Robot : XXXXXXXX
📊 état=docked batterie=100% erreur=none (poll=30s)
Lors des redémarrages suivants :
💾 Tokens chargés depuis /data/tokens.json (expire dans XXXXs)
Étape 5 — Jeedom : vérifier les équipements
Dans Jeedom → Plugins → Protocole domotique → MQTT Manager. Les équipements suivants sont créés automatiquement via MQTT Discovery :
| Nom |
Type |
Topic |
| Etat tondeuse |
Info chaîne |
navimow/stat/state |
| Etat détaillé |
Info chaîne |
navimow/stat/vehicleState |
| Batterie |
Info numérique (%) |
navimow/stat/battery |
| Niveau batterie |
Info chaîne |
navimow/stat/batteryLevel |
| Durée tonte (min) |
Info numérique (min) |
navimow/stat/mowing_time_min |
| Démarrer tonte |
Action bouton |
navimow/cmnd/start |
| Pause |
Action bouton |
navimow/cmnd/pause |
| Retour au dock |
Action bouton |
navimow/cmnd/dock |
États possibles du robot
Valeur stat/state |
Valeur stat/vehicleState |
Signification |
docked |
isDocked |
Dans le dock |
charging |
— |
En charge |
mowing |
isRunning |
En tonte |
paused |
— |
En pause |
returning |
— |
Retour au dock |
error |
— |
Erreur |
Niveau batterie (stat/batteryLevel)
| Valeur |
Signification |
FULL |
100% |
HIGH |
> 50% |
MEDIUM |
25-50% |
LOW |
< 25% |
EMPTY |
Vide |
Gestion des tokens
Le bridge renouvelle automatiquement le token OAuth2 toutes les ~55 minutes. Les tokens sont persistés dans un volume Docker (/data/tokens.json) et survivent aux redémarrages.
En cas d’expiration du refresh_token (après plusieurs semaines d’inactivité), les logs afficheront CODE_OAUTH_INFO_ILLEGAL. Il faudra alors relancer get_navimow_token.py sur le PC Windows :
bash
nano ~/navimow-jeedom/docker-compose.yml
# Modifier NAVIMOW_TOKEN et NAVIMOW_REFRESH_TOKEN
docker compose restart
Commandes utiles
bash
# Logs en direct
docker compose -f ~/navimow-jeedom/docker-compose.yml logs -f
# Redémarrer
docker compose -f ~/navimow-jeedom/docker-compose.yml restart
# Arrêter
docker compose -f ~/navimow-jeedom/docker-compose.yml down
# Statut
docker compose -f ~/navimow-jeedom/docker-compose.yml ps
Crédits
Frédéric