Bon,
Je suis en train de tester pour passer mon historique long terme sur influxDB.
Pour commencer, j’ai extrait mes données de HA via des requêtes de cette forme :
SELECT
sm.statistic_id,
sm.unit_of_measurement AS Unit,
s.start_ts AS start_ts,
s.state As state
FROM
statistics AS s
JOIN
statistics_meta AS sm
ON s.metadata_id = sm.id
WHERE
sm.statistic_id = 'sensor.zlinky_tic_easf01'
ORDER BY
s.start_ts ASC;
Via l’interface du module complémentaire SQLite Web j’ai exporté direcetement en csv.
A noter que pour certain capteur, il faut remplacer s.state par s.mean, en fonction des valeurs enregistrées dans la base de données.
Ensuite, j’ai créé ce script python pour transformer le format de ces csv en format compatible avec influxDB :
# --- Nom fichier .csv ---
CSV_FILE = "zlinky_tic_easf01.csv"
import pandas as pd
import math
import unicodedata
def escape_tag(s: str) -> str:
"""Échappe espaces, virgules, égal et backslash dans les tags."""
return (
str(s)
.replace("\\", "\\\\") # échapper backslash d'abord
.replace(" ", "\\ ")
.replace(",", "\\,")
.replace("=", "\\=")
)
def escape_measurement(s: str) -> str:
"""Échappe espaces, virgules et backslash dans le nom de mesure (Influx LP)."""
return (
str(s)
.replace("\\", "\\\\")
.replace(" ", "\\ ")
.replace(",", "\\,")
)
def sanitize_float(x):
"""Convertit en float (point décimal), retourne None si non numérique ou non-fini."""
if pd.isna(x):
return None
s = str(x).strip().replace(",", ".")
try:
val = float(s)
return val if math.isfinite(val) else None
except ValueError:
return None
def to_ns_from_epoch_seconds(ts):
"""
Convertit un timestamp (secondes Epoch, potentiel float/str) en nanosecondes (int).
Retourne None si invalide.
"""
if ts is None or (isinstance(ts, float) and math.isnan(ts)):
return None
s = str(ts).strip()
try:
seconds = float(s)
return int(seconds * 1e9)
except Exception:
# Fallback: si ce n'est pas un nombre, tenter une date ISO
try:
t = pd.Timestamp(s)
if pd.isna(t):
return None
return int(t.timestamp() * 1e9)
except Exception:
return None
# --- Paramétrage ---
DEFAULT_MEASUREMENT = "value" # si Unit est manquant, on met 'value'
SOURCE_TAG = "HA" # pour distinguer de l'historique brut
# Lecture CSV en UTF-8
df = pd.read_csv(CSV_FILE, encoding="utf-8")
# Colonnes attendues (format statistiques long terme)
required_cols = {"statistic_id", "Unit", "start_ts", "state"}
missing = required_cols - set(df.columns)
if missing:
raise ValueError(f"Colonnes manquantes dans le CSV: {missing}")
lines = []
ignored = 0
for _, row in df.iterrows():
# --- entity_id : enlever le prefixe 'sensor.' s'il est présent ---
raw_entity = str(row["statistic_id"])
if raw_entity.startswith("sensor."):
raw_entity = raw_entity[len("sensor."):]
# --- Unit / Measurement ---
raw_unit = str(row["Unit"]).strip() if not pd.isna(row["Unit"]) else ""
measurement_name = raw_unit if raw_unit else DEFAULT_MEASUREMENT
# Échappement + normalisation Unicode (NFC)
measurement = unicodedata.normalize("NFC", escape_measurement(measurement_name))
entity = unicodedata.normalize("NFC", escape_tag(raw_entity))
source = unicodedata.normalize("NFC", escape_tag(SOURCE_TAG))
domain = unicodedata.normalize("NFC", escape_tag("sensor"))
unit_tag = unicodedata.normalize("NFC", escape_tag(raw_unit)) if raw_unit else None
# Tags
tags = f"domain={domain},entity_id={entity},source={source}"
# Field (float)
value = sanitize_float(row["state"])
if value is None:
ignored += 1
continue
fields = f"value={value}"
# Timestamp en ns (à partir de start_ts en secondes Epoch)
ts_ns = to_ns_from_epoch_seconds(row["start_ts"])
if ts_ns is None:
ignored += 1
continue
# Ligne LP
line = f"{measurement},{tags} {fields} {ts_ns}"
lines.append(line)
# Écriture du fichier LP en UTF-8 (sans BOM)
with open("data_line_protocol.txt", "w", encoding="utf-8", newline="\n") as f:
f.write("\n".join(lines))
print(
f"{len(lines)} lignes LP écrites dans 'data_line_protocol.txt' (UTF-8, measurement: variable par Unit). "
f"Lignes ignorées: {ignored}"
)
J’ai un peu galéré avec les capteurs ayant pour unité °C, car influxDB ne voulait pas reconnaitre correctement le caractère °.
Une fois ce script lancé sur chacun des fichiers csv, je peux désormais l’importer dans le bucket voulu via Line Protocol.
Maintenant que j’ai mis des valeurs dans un bucket de test, j’en suis à regarder comment récupérer et afficher ça proprement dans HA.
), et je regarde comment afficher les valeurs dans HA. Quand j’aurai bien pris en main cette partie, je pourrais faire le grand saut (soit via l’implémentation toute simple depuis l’intégration officielle, soit en mode bourrin comme j’ai préparé).