Je préviens tout de suite, ce Retex n’est qu’à vocation pédagogique, je le rédige pour partager une méthodologie, à savoir que j’ai réalisé cette fonctionnalité différemment depuis, donc je n’ai pas ce process en prod.
L’intégration pour exécuter le script Python
Pour une raison que je n’ai jamais compris, les fonctionnalités scripts python intégrées à HA sont archi-mini, aucun import n’est possible, je dois donc utiliser une intégration appelée pyscript.
La doc est ici.
Je passe les détails pour l’installer, c’est très simple avec HACS et tout est dans la doc.
Dans configuration.yaml, j’ai bien ces lignes :
pyscript:
allow_all_imports: true
hass_is_global: true
Je récupère un jeton
J’ouvre l’interface utilisateur de Home Assistant.
Dans le coin inférieur gauche, je clique sur mon nom d’utilisateur (ou l’icône du profil).
Je clique sur “Profil”.
Je fais défiler vers le bas jusqu’à la section “Jetons d’accès longue durée”.
Je clique sur “Créer un jeton” pour générer un nouveau jeton.
Je donne un nom à mon jeton pour m’y retrouver plus facilement.
Une fois le jeton créé, je vois son contenu. Je copie la valeur du jeton, c’est-à-dire la série de caractères alphanumériques, et je la colle dans ma configuration (paragraphe suivant)
Déclaration de la plateforme qui fait le lien entre Python et le sensor
Dans configuration.yaml, j’ajoute ces lignes :
rest_command:
send_data_to_api:
url: 'http://homeassistant.local:8123/api/states/sensor.wallbox_charges'
method: 'post'
content_type: 'application/json'
headers:
Authorization: 'Bearer TOKEN' # Remplacer TOKEN par le token cf paragraphe précédent
payload: '{{ data | tojson }}'
Il s’agit de la porte d’entrée qui sera approvisionné par le script Python (plus bas) et qui enverra les informations dans le sensor sensor.wallbox_charges.
Mon script Python
Je place le fichier wallbox_charges.py dans le dossier /config/pyscript
import datetime
import pytz
@service
def wallbox_charges(lastdata=None, action=None, prix="123", km="123", dateDebut="01/01/2000 12:00:00", valeur="123"):
fuseau_horaire_paris = pytz.timezone("Europe/Paris")
maintenant_timestamp = int(datetime.datetime.now().timestamp())
maintenant_heure_paris = datetime.datetime.fromtimestamp(maintenant_timestamp, fuseau_horaire_paris)
maintenant_formatee = maintenant_heure_paris.strftime("%d/%m/%Y %H:%M:%S")
maintenantMoins20min = datetime.datetime.now() - datetime.timedelta(minutes=20)
maintenantMoins20min_formatee = maintenantMoins20min.strftime("%H:%M")
dateDebutRecue = datetime.datetime.strptime(dateDebut, '%d/%m/%Y %H:%M:%S').replace(tzinfo=fuseau_horaire_paris)
debut_timestamp_formatee = dateDebutRecue.strftime("%d/%m/%Y %H:%M:%S")
debut_timestamp_formatee_SansSecondes = dateDebutRecue.strftime("%d/%m/%Y %H:%M")
# Transformez les chaînes en objets datetime
maintenant_dt = datetime.datetime.strptime(maintenant_formatee, "%d/%m/%Y %H:%M:%S")
debut_dt = datetime.datetime.strptime(debut_timestamp_formatee, "%d/%m/%Y %H:%M:%S")
# Calculez la différence entre les objets datetime
tempsDeCharge = maintenant_dt - debut_dt - datetime.timedelta(minutes=20)
# Formatez la durée en heures, minutes et secondes
heures, reste = divmod(tempsDeCharge.seconds, 3600)
minutes, secondes = divmod(reste, 60)
# Affichez la durée au format HH:MM:SS
tempsDeCharge_formatee = f"{heures:02}:{minutes:02}"
# Écrivez un message d'information dans le journal
#log.info(action)
if action == "enregistrer":
valeuraajouter = [{
"saved_at": maintenant_timestamp,
"début": debut_timestamp_formatee_SansSecondes,
"fin": maintenantMoins20min_formatee,
"durée": tempsDeCharge_formatee,
"consommation": valeur,
"km": km,
"coût": round(valeur * prix, 2)
}]
if lastdata is not None:
valeuraajouter = lastdata + valeuraajouter
data = {
"state": debut_timestamp_formatee_SansSecondes,
"attributes": {
"updated_at": maintenant_dt.strftime("%d/%m/%Y %H:%M:%S"),
"sessionscharge":
valeuraajouter
}
}
if action == "init":
data = {
"state": "on",
"attributes": {
"updated_at": "14/10/2023 16:42:39",
"sessionscharge": [
{
"consommation": 2.14,
"coût": 0.43,
"durée": "00:27",
"début": "19/09/2023 20:36",
"fin": "21:03",
"km": 17,
"saved_at": 1695150359
},
{
"consommation": 23.29,
"coût": 4.66,
"durée": "04:19",
"début": "21/09/2023 21:06",
"fin": "01:25",
"km": 192,
"saved_at": 1695323159
},
{
"consommation": 7.68,
"coût": 1.54,
"durée": "01:23",
"début": "22/09/2023 08:45",
"fin": "10:08",
"km": 63,
"saved_at": 1695409559
},
{
"consommation": 23.80,
"coût": 4.76,
"durée": "04:26",
"début": "23/09/2023 16:57",
"fin": "21:23",
"km": 196,
"saved_at": 1695495959
}
]
}
}
# Appelez le service REST pour envoyer les données
service_data = {
"token": "xxxxxx", # Remplacer par le token
"data": data
}
if action == "enregistrer" or action == "init":
# Appelez le service REST de manière asynchrone
await hass.services.async_call('rest_command', 'send_data_to_api', service_data)
# Écrivez un message d'erreur dans le journal
#log.error("Fin script")
Ajouter deux automatisations
Une pour initialiser le sensor, l’autre pour ajouter une donnée
Initialiser les données
alias: Initialise les sessions de charge
description: ""
trigger: []
condition: []
action:
- service: pyscript.wallbox_charges
data:
action: init
mode: single
Enregistrer les données
alias: Enregistre une session de charge
description: >-
Cette automatisation se déclenche lorsque le capteur n'a pas changé de valeur
depuis 20 minutes et n'est pas à 0
trigger:
- platform: state
entity_id:
- sensor.wallbox_portal_added_range
for:
hours: 0
minutes: 20
seconds: 0
enabled: true
condition:
- condition: numeric_state
entity_id: sensor.wallbox_portal_added_energy
above: 0
- condition: not
conditions:
- condition: state
entity_id: sensor.wallbox_charges
state: >-
{{ as_timestamp(states('sensor.wallbox_portal_added_range_min')) | int
| timestamp_custom('%d/%m/%Y %H:%M:%S', true)}}
enabled: false
action:
- service: pyscript.wallbox_charges
data:
action: enregistrer
valeur: "{{states('sensor.wallbox_portal_added_energy')}}"
prix: "{{states('sensor.wallbox_portal_energy_price')}}"
km: "{{states('sensor.wallbox_portal_added_range')}}"
lastdata: "{{state_attr('sensor.wallbox_charges', 'sessionscharge')}}"
dateDebut: >-
{{ as_timestamp(states('sensor.wallbox_portal_added_range_min')) | int |
timestamp_custom('%d/%m/%Y %H:%M:%S', true)}}
- service: variable.update_sensor
data:
replace_attributes: false
value: >-
{{ as_timestamp(states('sensor.wallbox_portal_added_range_min')) | int
| timestamp_custom('%d/%m/%Y %H:%M:%S', true)}}
target:
entity_id: sensor.wallbox_charges
enabled: false
mode: single
Pour avoir la valeur minimal de la charge, j’utilise statistic pour avoir la valeur min
Ce sensor est déclaré dans sensors.yaml
- platform: statistics
name: "Wallbox Portal Added Range Min"
entity_id: sensor.wallbox_portal_added_range
state_characteristic: datetime_value_min
max_age:
hours: 10
Résultat du stockage des infos
Carte d’affichage de toutes les sessions
<div>Sessions de charge</div>
Dernière mise à jour : {{ state_attr('sensor.wallbox_charges','updated_at')}} <br><br>
{% set items = state_attr('sensor.wallbox_charges', 'sessionscharge') %}
<table width='100%'>
<tbody>
{% for i in range(0, items | count, 1) %}
<tr>
<td>{{ items[i].début }} à {{ items[i].fin }} ({{ items[i].durée }})</td>
<td>{{ items[i].consommation }} kWh</td>
<td>+{{ items[i].km }} km</td>
<td>{{ items[i].coût }} €</td>
</tr>
{% endfor %}
</tbody>
</table>
Carte d’affichage avec des totaux par année, mois, semaine
type: markdown
content: >-
<div><h1>Sessions de charge</div> {%- set default_language = 'fr' %} Dernière
mise à jour : {{ state_attr('sensor.wallbox_charges','updated_at')}} <br><br>
{% set items = state_attr('sensor.wallbox_charges', 'sessionscharge') %}
{% set septjours = now() - timedelta(days=7) %}
{% set power_semaine = namespace(value=0) %}
{% set cout_semaine = namespace(value=0) %}
{% set unmois = now() - timedelta(days=30) %}
{% set power_mois = namespace(value=0) %}
{% set cout_mois = namespace(value=0) %}
{% set uneannee = now() - timedelta(days=365) %}
{% set power_annee = namespace(value=0) %}
{% set cout_annee = namespace(value=0) %}
{% for i in range(0, items | count, 1) %}
{% if items[i].saved_at | int >= as_timestamp(uneannee) %}
{% set power_annee.value = power_annee.value + items[i].consommation %}
{% set cout_annee.value = cout_annee.value + items[i].coût %}
{% if items[i].saved_at | int >= as_timestamp(unmois) %}
{% set power_mois.value = power_mois.value + items[i].consommation %}
{% set cout_mois.value = cout_mois.value + items[i].coût %}
{% if items[i].saved_at | int >= as_timestamp(septjours) %}
{% set power_semaine.value = power_semaine.value + items[i].consommation %}
{% set cout_semaine.value = cout_semaine.value + items[i].coût %}
{% endif %}
{% endif %}
{% endif %}
{% endfor %}
</ha-alert> <ha-alert alert-type="info" title="Dernier mois : {{
power_mois.value | int }} kWh - {{ cout_mois.value | float | round(2) }}
€"></ha-alert> <ha-alert alert-type="info" title="Dernière année : {{
power_annee.value | int }} kWh - {{ cout_annee.value | float | round(2) }}
€"></ha-alert>
<br>
{% set items = state_attr('sensor.wallbox_charges', 'sessionscharge') %}
{% set septjours = now() - timedelta(days=7) %}
<table width='100%'> <tbody> {% for i in range(0, items | count, 1) %} {% if
items[i].saved_at | int >= as_timestamp(septjours) %} {% set datetime_obj =
strptime(items[i].début, '%d/%m/%Y %H:%M') %}
<tr> <td><h5>{{items[i].début }} à {{ items[i].fin }} ({{ items[i].durée
}})</td> <td>{{
items[i].consommation }} kWh</td> <td>+{{ items[i].km }} km</td> <td>{{
items[i].coût }} €</td> </tr> {% endif %} {% endfor %} </tbody> </table>
<ha-alert alert-type="success" title="Dernière semaine : {{
power_semaine.value | int}} kWh - {{ cout_semaine.value | float | round(2)
}}€">
style:
.: |
ha-card {
margin: 0px 0px 0px 0px;
box-shadow: none;
}
ha-markdown:
$: |
h5 {
font-size: 15px !important;
font-weight: normal !important;
padding: 0px 0px 0px 8px !important;
border-left: 3px solid #429f46;
}
