Salut à tous !
Si vous êtes client OCEA Smart Building, voici un script Python pour récupérer ses données d’eau froide automatiquement pour les injecter dans le tableau de bord Énergie de Home Assistant.
Ce tutoriel explique comment automatiser la récupération de l’index cumulé réel (la valeur totale de votre compteur) en simulant un export Excel depuis leur portail.
Points forts de cette méthode :
-
Fiabilité maximale : On utilise l’index réel du compteur.
-
Dashboard Énergie compatible : Utilise la classe
total_increasing. -
Léger : Optimisé pour tourner sur un vieux Raspberry Pi 2B (32 bits).
-
Dynamique : Le script trouve tout seul vos identifiants de logement et de compteur.
1. Prérequis
-
Un Raspberry Pi sous Raspberry Pi OS Lite (Bookworm ou Trixie recommandé).
-
Un compte Espace Résident OCEA fonctionnel.
-
Un Jeton d’accès de longue durée Home Assistant (à créer dans votre profil HA > Sécurité).
2. Préparation du système
Sur Raspberry Pi, surtout en 32 bits, certaines bibliothèques doivent être compilées. Installez les dépendances nécessaires :
sudo apt update
sudo apt install -y chromium-driver python3-full python3-dev libffi-dev build-essential libzstd-dev
3. Installation de l’environnement Python
On crée un dossier dédié et un environnement virtuel pour ne pas polluer le système :
mkdir ~/ocea && cd ~/ocea
python3 -m venv venv
source venv/bin/activate
# Installation des bibliothèques nécessaires
pip install requests selenium-wire openpyxl blinker==1.7.0
Note : openpyxl sert à lire le fichier Excel, selenium-wire à intercepter le jeton de connexion.
4. Configuration
Créez un fichier config.json dans ~/ocea :
{
"ocea_email": "votre@email.com",
"ocea_password": "votre_mot_de_passe",
"ha_base_url": "http://192.168.1.XX:8123",
"ha_token": "VOTRE_TOKEN_LONGUE_DUREE_HA"
}
5. Le Script
Créez le fichier ocea_collector.py. Ce script va :
-
Se connecter en « Headless » (sans écran) pour attraper le jeton de sécurité.
-
Chercher dynamiquement vos IDs de logement et de compteur.
-
Demander un export Excel des 7 derniers jours.
-
Extraire la toute dernière valeur de la colonne « Index ».
-
L’envoyer à Home Assistant avec les bons attributs.
import time
import json
import requests
import io
import os
from datetime import datetime, timedelta, timezone
from openpyxl import load_workbook
from seleniumwire import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
# ==========================================
# CONFIGURATION ET CONSTANTES
# ==========================================
SENSOR_ID = "sensor.eau_ocea_index_total"
FRIENDLY_NAME = "Eau OCEA Index Total"
CONFIG_FILE = "config.json"
BASE_URL = "https://espace-resident-api.ocea-sb.com/api/v1"
# ==========================================
def load_config():
path = os.path.join(os.path.dirname(__file__), CONFIG_FILE)
with open(path, 'r') as f:
return json.load(f)
CONFIG = load_config()
def get_token():
"""Récupère le jeton Bearer via Selenium (Headless)"""
chrome_options = Options()
chrome_options.add_argument("--headless")
chrome_options.add_argument("--no-sandbox")
chrome_options.add_argument("--disable-dev-shm-usage")
chrome_options.add_argument("--blink-settings=imagesEnabled=false")
service = Service(executable_path="/usr/bin/chromedriver")
driver = webdriver.Chrome(service=service, options=chrome_options)
token = None
try:
print("🔑 Connexion au portail OCEA...")
driver.get("https://espace-resident.ocea-sb.com/login")
time.sleep(8)
driver.find_element(By.ID, "email").send_keys(CONFIG['ocea_email'])
driver.find_element(By.ID, "password").send_keys(CONFIG['ocea_password'])
driver.find_element(By.ID, "next").click()
for _ in range(45):
for request in driver.requests:
if "espace-resident-api.ocea-sb.com" in request.url:
auth = request.headers.get('Authorization')
if auth and "Bearer" in auth:
token = auth
break
if token: break
time.sleep(1)
finally:
driver.quit()
return token
def get_dynamic_ids(token):
"""Recherche dynamiquement le Logement ID et le PDS ID (Compteur)"""
headers = {"Authorization": token, "Content-Type": "application/json"}
# 1. Récupération du Logement ID
print("🔍 Recherche du Logement ID...")
res_res = requests.get(f"{BASE_URL}/resident", headers=headers)
if res_res.status_code != 200:
return None, None
logement_id = res_res.json()['occupations'][0]['logementId']
print(f"🏠 Logement ID trouvé : {logement_id}")
# 2. Récupération du PDS ID (Compteur ID)
print("🔍 Recherche du Compteur ID (PDS ID)...")
res_app = requests.get(f"{BASE_URL}/local/{logement_id}/appareils", headers=headers)
if res_app.status_code != 200 or not res_app.json():
return logement_id, None
# On prend le pdsId du premier appareil trouvé
pds_id = res_app.json()[0]['pdsId']
print(f"📟 Compteur ID trouvé : {pds_id}")
return logement_id, pds_id
def get_last_index_from_xlsx(token, logement_id, pds_id):
"""Télécharge l'export et extrait le dernier index cumulé"""
headers = {"Authorization": token, "Content-Type": "application/json"}
url_export = f"{BASE_URL}/local/demande/export"
fin = datetime.now(timezone.utc)
debut = fin - timedelta(days=7)
payload = {
"localId": logement_id,
"periode": {
"dateDebut": debut.strftime('%Y-%m-%dT%H:%M:%S.000Z'),
"dateFin": fin.strftime('%Y-%m-%dT%H:%M:%S.000Z')
},
"pdsIds": [pds_id]
}
print("📥 Téléchargement de l'export Excel...")
res = requests.post(url_export, headers=headers, json=payload)
if res.status_code == 200:
file_content = io.BytesIO(res.content)
wb = load_workbook(file_content, read_only=True, data_only=True)
sheet = wb.active
rows = list(sheet.iter_rows(min_row=3, values_only=True))
valid_rows = [r for r in rows if len(r) > 3 and r[3] is not None]
if not valid_rows:
return None, None
derniere_ligne = valid_rows[-1]
raw_date = derniere_ligne[2]
raw_index = derniere_ligne[3]
index_value = float(str(raw_index).replace(',', '.'))
print(f"✅ Dernier relevé : {index_value} m³ ({raw_date})")
return index_value, raw_date
return None, None
def send_to_ha(value, date_str):
"""Envoie l'index à Home Assistant"""
ha_url = f"{CONFIG['ha_base_url']}/api/states/{SENSOR_ID}"
headers = {"Authorization": f"Bearer {CONFIG['ha_token']}", "Content-Type": "application/json"}
payload = {
"state": value,
"attributes": {
"unit_of_measurement": "m³",
"friendly_name": FRIENDLY_NAME,
"device_class": "water",
"state_class": "total_increasing",
"date_du_releve": date_str
}
}
requests.post(ha_url, headers=headers, json=payload)
print(f"🚀 Mis à jour dans HA : {value} m³")
if __name__ == "__main__":
t = get_token()
if t:
lid, pid = get_dynamic_ids(t)
if lid and pid:
index, date_releve = get_last_index_from_xlsx(t, lid, pid)
if index is not None:
send_to_ha(index, date_releve)
6. Intégration dans Home Assistant
Une fois le script lancé manuellement une première fois (python ocea_collector.py), une nouvelle entité va apparaître dans HA : sensor.eau_ocea_index_total.
Pour l’ajouter au Dashboard Énergie :
-
Allez dans Paramètres > Tableaux de bord > Énergie.
-
Dans Consommation d’eau, cliquez sur Ajouter une source d’eau.
-
Choisissez
Eau OCEA Index Total. -
Puisque c’est un index réel, HA s’occupe de tout pour les graphiques journaliers/mensuels.
7. Automatisation (Cron)
Pour que la relève se fasse toute seule chaque après-midi à 15h :
-
Ouvrez le planificateur :
crontab -e. -
Ajoutez cette ligne à la fin :
Bash
00 15 * * * /home/pi/ocea/venv/bin/python /home/pi/ocea/ocea_collector.py >> /home/pi/ocea/ocea.log 2>&1
