Version V15.01 du routeur - Version imprimable

Code :
import requests

import time

import csv

from requests.exceptions import ConnectTimeout, ReadTimeout, RequestException



SHELLY_URL = "http://192.168.1.34/rpc/Shelly.GetStatus"

WAIT_AFTER_RESPONSE = 0.1  # 50ms après la réponse

MAX_RETRIES = 3

CONNECT_TIMEOUT = 3

READ_TIMEOUT = 3

DURATION = 180  # secondes



def get_status_with_retry(request_number, timeout_counter, start_time):

    """

    Effectue la requête avec retries, affiche le détail complet en cas de timeout ou erreur

    + indique le temps écoulé depuis le début du test.

    """

    for attempt in range(1, MAX_RETRIES + 1):

        try:

            r = requests.get(SHELLY_URL, timeout=(CONNECT_TIMEOUT, READ_TIMEOUT))

            r.raise_for_status()

            return r.json(), timeout_counter, False

        except ConnectTimeout:

            timeout_counter += 1

            elapsed = time.time() - start_time

            print(f"[Requête #{request_number}] ConnectTimeout {timeout_counter} "

                  f"(tentative {attempt}/{MAX_RETRIES}, à {elapsed:.2f}s depuis le début, "

                  f"timeout={CONNECT_TIMEOUT}s)")

            time.sleep(0.05)

        except ReadTimeout:

            timeout_counter += 1

            elapsed = time.time() - start_time

            print(f"[Requête #{request_number}] ReadTimeout {timeout_counter} "

                  f"(tentative {attempt}/{MAX_RETRIES}, à {elapsed:.2f}s depuis le début, "

                  f"timeout={READ_TIMEOUT}s)")

            time.sleep(0.05)

        except RequestException as e:

            elapsed = time.time() - start_time

            print(f"[Requête #{request_number}] Erreur: {e} "

                  f"(à {elapsed:.2f}s depuis le début)")

            time.sleep(0.05)

    return None, timeout_counter, True  # échec complet



def main():

    response_times = []

    refresh_intervals = []

    timeout_counter = 0

    request_number = 0

    success_count = 0

    last_values = None

    last_change_time = None

    start_time = time.time()



    # Préparer le CSV

    with open("shelly_log.csv", "w", newline="", encoding="utf-8") as csvfile:

        writer = csv.writer(csvfile, delimiter=";")

        writer.writerow([

            "req_num", "timestamp_s", "resp_time_ms", "timeout_flag",

            "U1", "I1", "P1", "U2", "I2", "P2", "U3", "I3", "P3"

        ])



        while time.time() - start_time < DURATION:

            request_number += 1

            t0 = time.perf_counter()

            data, timeout_counter, failed = get_status_with_retry(request_number, timeout_counter, start_time)

            t1 = time.perf_counter()



            resp_time = (t1 - t0) * 1000 if not failed else None

            timestamp_s = time.time() - start_time



            if data is None:

                writer.writerow([request_number, f"{timestamp_s:.3f}", "", 1, "", "", "", "", "", "", "", "", ""])

                time.sleep(WAIT_AFTER_RESPONSE)

                continue



            try:

                values = tuple(

                    (

                        data[f"em1:{i}"]["voltage"],

                        data[f"em1:{i}"]["current"],

                        data[f"em1:{i}"]["act_power"]

                    )

                    for i in [0, 1, 2]

                )

            except KeyError:

                print(f"[Requête #{request_number}] Structure inattendue")

                writer.writerow([request_number, f"{timestamp_s:.3f}", f"{resp_time:.2f}", 0, "", "", "", "", "", "", "", "", ""])

                time.sleep(WAIT_AFTER_RESPONSE)

                continue



            # Détection changement global

            if values != last_values:

                now = time.perf_counter()

                if last_change_time is not None:

                    refresh_intervals.append((now - last_change_time) * 1000)

                last_change_time = now

                last_values = values



            response_times.append(resp_time)

            success_count += 1



            # Écriture CSV

            writer.writerow([

                request_number,

                f"{timestamp_s:.3f}",

                f"{resp_time:.2f}",

                0,

                values[0][0], values[0][1], values[0][2],

                values[1][0], values[1][1], values[1][2],

                values[2][0], values[2][1], values[2][2],

            ])



            time.sleep(WAIT_AFTER_RESPONSE)



    # Ignorer le premier intervalle

    filtered_intervals = refresh_intervals[1:] if len(refresh_intervals) > 1 else []



    # Résultats

    if response_times:

        print("\n=== Temps de réponse API (réussites uniquement) ===")

        print(f"Min: {min(response_times):.2f} ms")

        print(f"Max: {max(response_times):.2f} ms")

        print(f"Moy: {sum(response_times)/len(response_times):.2f} ms")



    if filtered_intervals:

        print("\n=== Intervalle de mise à jour global (3 phases) ===")

        print(f"(Première valeur ignorée, {len(filtered_intervals)} intervalles utilisés)")

        print(f"Min: {min(filtered_intervals):.2f} ms")

        print(f"Max: {max(filtered_intervals):.2f} ms")

        print(f"Moy: {sum(filtered_intervals)/len(filtered_intervals):.2f} ms")



    total_time = time.time() - start_time

    print("\n=== Bilan final ===")

    print(f"Durée totale du test : {total_time:.2f} s")

    print(f"Nombre total de requêtes tentées : {request_number}")

    print(f"Nombre de requêtes réussies : {success_count}")

    print(f"Nombre de changements détectés (hors 1er) : {len(filtered_intervals)}")

    print(f"Taux de réussite : {100 * success_count / request_number:.1f} %")

    print(f"Total timeouts : {timeout_counter}")

    print("\n? Détails enregistrés dans 'shelly_log.csv'")



if __name__ == "__main__":

    main()