Salut à tous,
J’ai écris un script en python, pour récupérer et traiter les datas depuis l’api de MF, et renvoyer vers mon broker MQTT les infos dont j’ai besoin.
En gros selon les datas MF et pour mon département.
ce que je fais :
-
Avoir un seul niveau d’alerte pour aujourd’hui.
-
Je compare les « phenomenon_id » par rapport au tableau décrit dans la doc de l’api, afin de les classer dans le bonne ordre et les nommer correctement.
-
Je peux commenter dans les variables les phénomènes qui ne m’intéressent pas selon le lieu où j’habite.
-
Je peux choisir d’ignorer ou de mettre à l’état « unavailable » les sensors qui m’intéressent mais que MF ne fournie pas en ce moment (ex: « Grand-froid » pour le 27 aujourd’hui).
ce que je ne fais pas :
-
Afficher les débuts et fins de période.
-
Récupérer les images d’illustration.
-
Pas de niveau d’alerte pour demain.
Après réception des datas par le broker à l’aide d’un sensor MQTT je créé un sensor « weather_alert » quasi identique à celui que fournissait l’intégration Météo-France avant que celle-ci ne cesse de fonctionner.
Bref quelques screen pour illustrer :
Le script
#!/usr/bin/python3
# -*- coding: UTF8 -*-
from datetime import datetime
import requests
import json
import paho.mqtt.client as mqtt
##########################################################
### qq variables pour la navigation dans le json de MF ###
##########################################################
# ici DOMAIN_IDS 16 pour le dpt de l'eure
DOMAIN_IDS = 16
# ici on peut commenter les vigilances qui ne concernent pas vraiment sa région (ex les avalanches pour la normandie)
PHENOMENON_IDS = {
'1': 'Vent',
'2': 'Pluie',
'3': 'Orages',
'4': 'Crues',
'5': 'Neige-verglas',
'6': 'Canicule',
'7': 'Grand-froid',
# '8': 'Avalanches',
# '9': 'Vagues-submersion'
}
# ici la liste des codes couleur
COLOR_CODE = {
1: 'Vert',
2: 'Jaune',
3: 'Orange',
4: 'Rouge'
}
################################################################
### La requête qui permet d'obtenir les data depuis l'api MF ###
################################################################
headers = {
'accept': '*/*',
'apikey': 'VOTRE_API_KEY',
}
data = requests.get('https://public-api.meteofrance.fr/public/DPVigilance/v1/cartevigilance/encours', headers=headers).json()
###############################################################################################################################
### ci-dessous requêtes sur les validity qui vont permettre plus tard de situer l'execution du script dans la bonne période ###
###############################################################################################################################
# Récupérer les dates de début et de fin pour les deux périodes
begin_period0 = int((datetime.strptime(data['product']['periods'][0]['begin_validity_time'], '%Y-%m-%dT%H:%M:%SZ')).timestamp())
end_period0 = int((datetime.strptime(data['product']['periods'][0]['end_validity_time'], '%Y-%m-%dT%H:%M:%SZ')).timestamp())
begin_period1 = int((datetime.strptime(data['product']['periods'][1]['begin_validity_time'], '%Y-%m-%dT%H:%M:%SZ')).timestamp())
end_period1 = int((datetime.strptime(data['product']['periods'][1]['end_validity_time'], '%Y-%m-%dT%H:%M:%SZ')).timestamp())
# Obtenir la date et l'heure actuelles
current_time = int((datetime.now()).timestamp())
#################################
### les différentes fonctions ###
#################################
# récupérer le niveau max de la période
def process_max_color(PERIOD_INDEX):
color_data = data['product']['periods'][PERIOD_INDEX]['timelaps']['domain_ids'][DOMAIN_IDS]['max_color_id']
color_name = COLOR_CODE[color_data]
return color_name
# récupérer le niveau de chaque vigilance
def process_att_color(PERIOD_INDEX):
color_data = {}
for phenomenon_id, phenomenon_name in PHENOMENON_IDS.items():
found = False
for phenomenon_item in data['product']['periods'][PERIOD_INDEX]['timelaps']['domain_ids'][DOMAIN_IDS]['phenomenon_items']:
if phenomenon_item['phenomenon_id'] == phenomenon_id:
color_id = phenomenon_item['phenomenon_max_color_id']
color_name = COLOR_CODE.get(color_id, 'unavailable')
color_data[phenomenon_name] = color_name
found = True
break
# les 2 lignes suivantes peuvent être commenter si on ne veut pas les vigilances au status "unavailable" (si non présentes dans les datas de MF)
# si elles sont commentées pas d'attributs créé dans le sensor côté HA, sinon il sera là à l'état "unavailable"
# if not found:
# color_data[phenomenon_name] = 'unavailable'
return color_data
# génération du json en fonction de la période
def gen_json(PERIOD_INDEX):
json_data = {
"today": process_max_color(PERIOD_INDEX),
"today_attributes": process_att_color(PERIOD_INDEX)
}
return json_data
####################################################################################
### appel des fonctions en fonction de la date des périodes pour générer le json ###
####################################################################################
if begin_period0 <= current_time <= end_period0:
json_data = gen_json(0)
print("Nous sommes dans la première période.")
elif begin_period1 <= current_time <= end_period1:
json_data = gen_json(1)
print("Nous sommes dans la deuxième période.")
else:
print("Nous ne sommes dans aucune des périodes définies.")
result_json = json.dumps(json_data)
###############################################
### MQTT configuration, et envoie des datas ###
###############################################
mqtt_broker = '192.168.XX.XX' # ip du broker
mqtt_port = 1883 # port du broker
mqtt_topic = 'weather/alerts' # topic à créer
mqtt_username = 'VOTRE_USER_MQTT' # user mqtt
mqtt_password = 'VOTRE_PASSWD_MQTT' # password mqtt
def on_connect(client, userdata, flags, rc):
print(f"Connected to MQTT broker with code {rc}")
client.publish(mqtt_topic, result_json, retain=True)
mqtt_client = mqtt.Client()
mqtt_client.on_connect = on_connect
mqtt_client.username_pw_set(mqtt_username, mqtt_password)
mqtt_client.connect(mqtt_broker, mqtt_port, 60)
mqtt_client.loop_start()
mqtt_client.loop_stop()
mqtt_client.disconnect()
le sensor mqtt
- name: "Eure Weather Alert"
unique_id: eure_weather_alert
state_topic: "weather/alerts"
value_template: "{{ value_json['today'] }}"
json_attributes_topic: "weather/alerts"
json_attributes_template: "{{ value_json['today_attributes'] | tojson }}"
icon: mdi:weather-cloudy-alert
Pour l’instant c’est toujours en mode « POC » à la maison, et oui je ne suis pas dev et mon python n’est sûrement pas le meilleur
Le script sera lancé sur une de mes machines, mais qui ne sera pas celle où tourne HA.
Me reste juste à comprendre un peu mieux « quand » météo france met à jour ses datas, et à me décider si je lance en crontab ou si je fais un service systemd timer.
Pour l’affichage je reste sur ce que je faisais avant, càd extraire les attributs du sensor weather_alert pour en faire des sensors à part entière, et jouer avec button-card pour afficher ça comme ça :
C’est une ancienne capture où j’avais forcé les états pour vérifier le bon fonctionnement de mes templates button-card.
Bref je sais pas si ça intéressa qqn, mais je voulais juste partager
[EDIT] L’objectif de base était de ne modifier/ajouter sur HA qu’un minimum de chose à l’existant en attendant une màj de l’intégration officielle