Utilisation de l'API **SURVEILLANCE STATION**== de Synology

Bonjour,

Ayant mis plusieurs semaines à solutionner mon problème, je vous propose de partager mon expérience sur l’utilisation de l’API SURVEILLANCE STATION dans HOMME ASSISTANT.

Situation

Sur mon NAS Synology, j’ai Surveillance Station qui gère l’enregistrement de mes flux vidéos de mes 5 caméras extérieures.
J’ai également mon Home Assistant en VM sur mon NAS.

Ma problématique:

J’ai des araignées qui tissent leur toile dans le champs de mes caméras ce qui me déclenchent des enregistrements continus de 30 minutes :rage:
Je perds le bénéfice du déclenchement de l’enregistrement sur évènement et j’utilise de la place sur mes disques dur pour rien.

Mon projet:

Dans HA, récupérer les durées des enregistrements de mes caméra via l’API de SURVEILLANCE STATION et m’envoyer une notification en cas d’enregistrement supérieur à 30 minutes.

Mon retour d’expérience:

J’ai exploré plusieurs pistes pour solutionner ma problématique mais seule l’importation des données fournies de l’API de SURVEILLANCE STATION à travers HA m’a permis d’obtenir le résultat attendu. L’API permet d’échanger des informations avec SURVEILLANCE STATION, il est donc possible de « télécommander » SURVEILLANCE STATION, avec 563 pages la liste des commandes est longue et les possibilités sont grandes. La commande principale que j’ai utilisé est SYNO.SurveillanceStation.Recording.
Malheureusement, je n’ai pas trouvé d’informations claires et facilement utilisables d’où mon intention de partager mon expérience dans ce sujet.
Ne cherchez pas à contacter l’assistance SYNOLOGY comme je l’ai fait, c’est en dehors de leur périmètre m’a t’on répondu ! :thinking:

Les prérequis:

  1. Installer Pyscript ce qui permet d’utiliser des scripts PYTHON (il existe peut être d’autre alternative, moi j’ai installé Pyscript et cela fonctionne), j’ai suivi les instructions du lien ci dessous:
    Chaine youtube de TechWithDavid: Youtube : Home Assistant - Python Scripts
    Et ces instructions là:
    Pyscript Installation Instructions: https://hacs-pyscript.readthedocs.io/…
  2. paramétrer HA pour l’envoie des notifications par mail en rajoutant ce code dans « configuration.yaml », dans mon cas j’utilise une adresse GMAIL dédié à mon home assistant:
# ligne rajoutée pour envoyer mails
notify:
  - name: "envoi_email"
    platform: smtp
    server: "smtp.gmail.com"
    port: 587
    sender: "adresse mail de ha@gmail.com"
    encryption: starttls
    username: "adresse mail de ha@gmail.com"
    password: "mot de passe"  # mot de passe d'application généré dans le compte Gmail
    recipient:
      - "adresse mail du destinataire@gmail.com"
    sender_name: "Home Assistant"

Les aides:

  1. L’IA :innocent: qui m’a générée mon code en python [YIAH] (https://www.yiaho.com/ia-generateur-de-code/) Cela m’a été d’une aide formidable pour moi qui ne voulais pas apprendre le python C’est une aide et pas une baguette magique !

  2. La documentation de l’API SURVEILLANCE STATION ici chez Synology malheureusement qui contient des erreurs qui peuvent être bloquantes :rage:

L’algorithme général:
C’est avoir un automatisme dans HA qui exécute un script en python qui va interroger l’API de SURVEILLANCE STATION; en cas de dépassement de la durée de l’enregistrement HA m’envoie une notification par mail.

Étapes du script:
Les différentes étapes du script consistent à échanger avec l’API en utilisant des URL, les informations en retour sont transmises en JSON.

  1. URL pour se connecter à l’API SURVEILLANCE STATION et obtenir un SID (une clé authentification qui sera utilisée pour tous les échanges avec l’API :
    http://IP_du_NAS:5000/webapi/auth.cgi?api=SYNO.API.Auth&version=6&method=login&account=LOGIN du NAS&passwd=Mot de passe du NAS&session=SurveillanceStation&format=sid

Ci dessous le retour des informations brutes sous firefox:

Ci dessous le retour des informations formaté en JSON sous firefox:

Dans cette exemple, la valeur du SID est :
_CDA7pMTgjb7hNtdxrZIiLK7qpc981hA0cDLiBUpP3YNh34WePSf77TLY1xajzEOKa-Gj1oaD0g5NU_WfrScX0
Cette clé sera valide pendant un certain temps (plusieurs heures mais je n’ai pas la valeur exacte.

  1. URL pour avoir des infos sur les enregistrements en utilisant la clé SID obtenue précédemment:
    http://IP_du_NAS:5000/webapi/entry.cgi?version=5&cameraIds={camera_id}&limit={Nb_des_derniers_evenements}&api=SYNO.SurveillanceStation.Recording&method=List&_sid=_CDA7pMTgjb7hNtdxrZIiLK7qpc981hA0cDLiBUpP3YNh34WePSf77TLY1xajzEOKa-Gj1oaD0g5NU_WfrScX0
    Le SID ci dessus n’est plus valable c’est à titre d’exemple.
    Je scrute les {Nb_des_derniers_evenements} enregistrements des caméras en incrémentant le {camera_id}, si la réponse est vide alors pas de caméra sur ce CaméraID sinon j’analyse les données JSON pour en extraire: Nom de la caméra, début de l’enregistrement, fin de l’enregistrement.
    En faisant la différence entre la fin et le début de l’enregistrement, j’obtiens la durée de mon enregistrement que je compare une variable « seuil_durée » :slight_smile:
    Si la durée de l’enregistrement est supérieur à seuil_durée alors je serais notifié par mail que sur cette caméra à telle heure il y a eu un enregistrement d’une durée de xx minutes

Ci dessous le retour des informations brutes sous firefox:


La quantité d’information est importante et illisible en version brute !

Ci dessous le retour des informations formaté en JSON sous firefox:

Informations complémentaires:

  • Je limite mes recherches sur les 5 derniers enregistrement par caméra &limit=5, comme je cherche des enregistrements long (30 minutes) et que le script sera exécuté toutes les heures, normalement je pourrait me limiter à 2 mais j’ai préféré avoir un peu de marge.
  • Pour ceux qui liront la documentation Synology, vous trouverez que la version à utiliser dans l’URL doit être la 6 ! Je n’ai jamais réussi à faire fonctionner l’URL en version 6 mais par hasard j’ai essayé la 5 et miracle tout fonctionne.
  • Il existe une autre URL pour obtenir la liste des caméras déclarées dans Surveillance Station et leurs informations correspondantes (CaméraID et son nom). :
    http://IP_du_NAS:5000/webapi/auth.cgi?api=SYNO.SurveillanceStation.Camera&version=2&method=List&_sid={ton_SID}
    Malheureusement la réponse de cette commande est trop volumineuse et génère une erreur sous HA. Dans un script python exécuté en dehors de HA cela fonctionne très bien.
    L’utilisation de cette URL éviterai la boucle sur des CaméraID inexistant.
  1. URL pour se déconnecter proprement http://IP_du_NAS:5000/webapi/auth.cgi?api=SYNO.API.Auth&version=6&method=logout&session=SurveillanceStation&_sid={ton_SID}

Ci dessous le script complet en python:

import subprocess  
import json  # Pour manipuler les objets JSON  
from datetime import datetime  # Importer datetime pour la conversion de timestamp

@service  
async def check_enregistrement():
    """Récupère le SID de l'API de connexion et les enregistrements des caméras en utilisant curl."""

    # Paramètres de configuration  
    log_surveillance_station = "XXXXX"  # Remplacez par votre log  
    mdp_surveillance_station = "YYYYYYY"  # Mot de passe pour l'API  
    seuil_durée = 900  # Seuil de durée en secondes pour le traitement  
    Numero_max_camera = 25  # Fixe la limite de la profondeur de recherche des caméras  
    Nb_des_derniers_evenements = 5  # Nombre des derniers événements par caméra  

    # Construire l'URL pour obtenir le SID avec le log  
    url_login = f"http://192.168.1.201:5000/webapi/auth.cgi?api=SYNO.API.Auth&version=6&method=login&account={log_surveillance_station}&passwd={mdp_surveillance_station}&session=SurveillanceStation&format=sid"

    command_login = ['curl', '-X', 'GET', url_login]

    try:
        # Obtenir le SID  
        response_login = await hass.async_add_executor_job(subprocess.check_output, command_login)
        response_text_login = response_login.decode()
        
        # Charger la réponse JSON  
        json_data_login = json.loads(response_text_login)
        
        # Extraire le SID  
        sid = json_data_login.get("data", {}).get("sid")
        
        if sid:
            log.info(f"SID extrait : {sid}")

            evenement = {}  # Nouvelle variable pour stocker les événements  

            # Boucle de 0 à Numero_max_camera pour rechercher les caméras enregistrées  
            for camera_id in range(Numero_max_camera + 1):  # On ajoute 1 pour inclure la caméra 25  
                # Construire l'URL avec limit et l'ID de la caméra courant  
                url_recordings = f"http://192.168.1.201:5000/webapi/entry.cgi?version=5&cameraIds={camera_id}&limit={Nb_des_derniers_evenements}&api=SYNO.SurveillanceStation.Recording&method=List&_sid={sid}"
                command_recordings = ['curl', '-X', 'GET', url_recordings]

                # Obtenir la réponse brute des enregistrements  
                response_recordings = await hass.async_add_executor_job(subprocess.check_output, command_recordings)
                response_text_recordings = response_recordings.decode()

                # Charger la réponse JSON  
                json_data_recordings = json.loads(response_text_recordings)

                # Vérifier si des événements sont présents  
                events = json_data_recordings.get("data", {}).get("events", [])
                
                # Initialiser une liste pour stocker les événements de cette caméra  
                if camera_id not in evenement:
                    evenement[camera_id] = []

                camera_name = ""  # Initialiser le nom de la caméra

                for event in events:  # Traiter chaque événement  
                    camera_name = event.get("camera_name")  # On peut récupérer le nom de la caméra  
                    
                    # Ignorer cette caméra si elle n'a pas de nom  
                    if not camera_name:
                        continue
                    
                    start_time = event.get("startTime")  # Extraire startTime  
                    stop_time = event.get("stopTime")  # Extraire stopTime  
                    name = event.get("name")  # Extraire le nom de l'enregistrement
                    
                    # Afficher le champ name pour le débogage  
                    log.info(f"Nom de l'événement : {name}")

                    # Calculer la durée  
                    duration_seconds = stop_time - start_time  # Durée en secondes  

                    # Convertir la durée en minutes et secondes  
                    minutes = duration_seconds // 60  
                    seconds = duration_seconds % 60  

                    # Modification de la mise en forme de formatted_duration  
                    if seconds > 0:
                        formatted_duration = f"{minutes}min {seconds}s"  # Format xxmin yys  
                    else:
                        formatted_duration = f"{minutes}min"  # Format xxmin

                    # Traitement de la date à partir du nom de l'enregistrement  
                    if name and '-' in name:
                        try:
                            date_str = name.split('/')[0]  # Extrait la partie avant le '/' 
                            
                            # Extraire jour, mois, et année selon les indices spécifiés  
                            day = date_str[6:8]   # 7e et 8e caractères  
                            month = date_str[4:6]  # 5e et 6e caractères  
                            year = date_str[:4]     # 4 premiers caractères  
                            
                            # Formatage de la date  
                            formatted_date = f"{day}/{month}/{year}"

                            # Convertir le timestamp UNIX en heure  
                            event_time = datetime.fromtimestamp(start_time).strftime('%H:%M:%S')  # Extraire l'heure  
                            
                            # Stockage des événements uniquement pour cette caméra  
                            evenement[camera_id].append({
                                "Nom de la Cam": camera_name,
                                "Durée": duration_seconds,  # Stockage de la durée brute  
                                "Formatted Durée": formatted_duration,  # Stockage de la durée formatée  
                                "Date": f"{formatted_date} à {event_time}"  # Stockage de la date et l'heure formatées  
                            })

                        except Exception as e:
                            log.error(f"Erreur lors de l'extraction de la date : {e}, nom : {name}")

            # Afficher le SID et toutes les caméras stockées 
            log.info(f"SID : {sid}")
            log.info(f"Évènements stockés : {evenement}")

            # Envoi des événements par e-mail  
            if evenement:
                email_content = "Événements récupérés :\n\n"  # Utiliser \n pour les retours à la ligne  
                any_event_detected = False  # Variable pour détecter s'il y a des événements

                for cam_id, ev_list in evenement.items():
                    if ev_list:  # S'il y a des événements pour cette caméra  
                        # Filtrer les événements qui dépassent le seuil  
                        valid_events = [ev for ev in ev_list if ev['Durée'] > seuil_durée]
                        if valid_events:
                            any_event_detected = True  # Au moins un événement valide trouvé  
                            for ev in valid_events:
                                email_content += f"Caméra : {ev['Nom de la Cam']}, Durée : {ev['Formatted Durée']}, Date : {ev['Date']}\n"
                        else:
                            camera_name = ev_list[0]["Nom de la Cam"]  # Nom de la caméra  
                            min_duration = seuil_durée // 60  
                            sec_duration = seuil_durée % 60  
                            if sec_duration > 0:
                                email_content += f"😊 La caméra {camera_name} n'a pas d'enregistrement d'une durée supérieure à {min_duration}min {sec_duration}s. \n"
                            else:
                                email_content += f"😊 La caméra {camera_name} n'a pas d'enregistrement d'une durée supérieure à {min_duration}min.\n"
                    else:  # Si aucune donnée pour cette caméra  
                        continue  # Ne rien faire si la caméra n'a pas d'enregistrements

                # Envoyer l'e-mail seulement si des événements ont été détectés  
                if any_event_detected and email_content.strip() != "Événements récupérés :\n\n":
                    await hass.services.async_call('notify', 'envoi_email', {
                        'message': email_content,
                        'title': 'Rapport d\'évènements des caméras'  # Changement du titre ici  
                    })

        else:
            log.error("Le SID n'a pas été trouvé dans la réponse.")

    except Exception as e:
        log.error(f"Erreur lors de l'appel à curl : {e}")

    finally:
        # Déconnexion  
        if sid:
            url_logout = f"http://192.168.1.201:5000/webapi/auth.cgi?api=SYNO.API.Auth&version=6&method=logout&session=SurveillanceStation&_sid={sid}"
            command_logout = ['curl', '-X', 'GET', url_logout]
            try:
                await hass.async_add_executor_job(subprocess.check_output, command_logout)
                log.info("Déconnexion réussie.")
            except Exception as e:
                log.error(f"Erreur lors de la déconnexion : {e}")       

Voici mon automatisation (version graphique):

Et la voici en version YAML:

alias: Vérifier l'enregistrement des caméra
description: Mesure la durée des enregistrements des cameras et envoie un mail
trigger:
  - platform: time_pattern
    hours: "*"
condition: []
action:
  - service: pyscript.check_enregistrement
    data: {}
mode: single

Voici le mail de notification que je reçois:

Conclusion:
Mon script est fonctionnel chez moi et convient à mon besoin, il est surement possible d’améliorer le code puisqu’il a été généré par IA mais j’ai déjà passé beaucoup de temps sur ce projet (sans parler de rédaction de ce sujet).
Le code a été développé avec les variables importantes accessibles au début du code (voir # Paramètres de configuration) afin que le code soit facilement adaptable.
J’ai rédigé ce sujet en essayant d’être le plus synthétique mais sans trop l’édulcorer.

En vous souhaitant une bonne lecture et que cela servira à d’autres…