Session pratique

Consignes globales

Dans cette session, vous allez concevoir, un petit système d’analyse de données issues d’un capteur de température. Votre application ne doit jamais s’arrêter sur une erreur, même en cas :

• de données corrompues,
• de valeurs aberrantes,
• de panne du capteur.

Pour cela, vous allez appliquer trois principes tout au long de la session :

• la programmation défensive : chaque méthode gère explicitement les exceptions qui peuvent survenir,
• le développement guidé par les tests : les tests fournis décrivent le comportement attendu avant que vous n’écriviez le code,
• la documentation : chaque méthode publique est documentée (paramètres, valeur de retour, exceptions levées) avec des docstrings.

Contenu de l’activité

Préparation de l’environnement

Avant de commencer, créez un projet uv dédié à cette séance. Dans un terminal, placez-vous dans le dossier de la séance (par exemple imt/s5/info/prog/session2), puis lancez la commande suivante :

uv init --no-package

uv init --no-package

uv init --no-package

uv init --no-package

Vous écrirez tous les fichiers de cette séance dans ce projet, en ajoutant les bibliothèques dont vous avez besoin avec uv add et en exécutant vos programmes dans l’environnement virtuel associé. Au besoin, revoyez la fiche Gestion moderne de projet Python avec uv.

Organisation

Vous allez écrire trois fichiers Python, tous placés dans le même répertoire de travail :

• sensor.py : une classe Sensor qui génère des mesures de température et les écrit dans un fichier,
• analyzer.py : une classe Analyzer qui calcule des statistiques sur les mesures reçues et les trace dans un fichier de log,
• reader.py : une classe Reader qui lit le fichier de mesures et transmet chaque ligne à un Analyzer.

Pour chaque classe, un fichier de tests unitaires est fourni ci-dessous : copiez-collez-le tel quel dans un fichier tests_<nom>.py du même répertoire, puis complétez la classe correspondante jusqu’à ce que tous ces tests passent.

with open("data.txt", 'r') as file:
    line = file.readline()
    # 'r' : lecture, 'w' : écriture (écrase), 'a' : ajout en fin de fichier

import time
time.sleep(.7)

import random
random.choice(range(10))       # entier entre 0 et 9
random.gauss(mean, std)        # flottant selon une loi gaussienne

import logging
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
logger.addHandler(logging.FileHandler("analyzer.log"))
logger.info("message")   # aussi : .debug, .warning, .error

Exercice 1 : la classe Sensor

Implémentez une classe Sensor qui simule un capteur de données suivant une loi gaussienne (moyenne, écart-type) et qui tombe en panne une fois sur dix en moyenne.

Attributs (avec leurs valeurs par défaut) :

__id : str = "sensor"
__mean : float = 90.0
__std : float = 5.0
__sleep_time : float = 3.0
__storage_file_name : str = "data.txt"

Méthodes à implémenter :

OutofBoundsException et NotRespondingException sont deux classes d’exception à définir vous-même dans sensor.py (héritant de Exception).

Copiez-collez le fichier de tests suivant dans tests_sensor.py :

import unittest
import random
from sensor import Sensor, OutofBoundsException, NotRespondingException


class TestSensor(unittest.TestCase):

    def test_encapsulation_default_values(self):
        s = Sensor()
        self.assertEqual(s.get_id(), "sensor")
        self.assertEqual(s.get_mean(), 90.0)
        self.assertEqual(s.get_std(), 5.0)
        self.assertEqual(s.get_sleep_time(), 3.0)
        self.assertEqual(s.get_storage_file_name(), "data.txt")

    def test_encapsulation_custom_values(self):
        s = Sensor(id="s1", mean=50.0, std=2.0, sleep_time=1.0, storage_file_name="out.txt")
        self.assertEqual(s.get_id(), "s1")
        self.assertEqual(s.get_mean(), 50.0)
        self.assertEqual(s.get_std(), 2.0)
        self.assertEqual(s.get_sleep_time(), 1.0)
        self.assertEqual(s.get_storage_file_name(), "out.txt")

    def test_get_value(self):
        # Test de la fréquence de non-réponse : environ 10 %
        random.seed(1)
        sensor = Sensor()

        number_of_calls = 10
        seen_ok = 0
        seen_nr = 0

        for _ in range(number_of_calls):
            try:
                value = sensor.get_value()
                self.assertIsInstance(value, float)
                self.assertTrue(0 <= value <= 150)
                seen_ok += 1
            except NotRespondingException:
                seen_nr += 1
            except OutofBoundsException:
                self.fail("Une valeur par défaut ne devrait presque jamais être hors limites")

        self.assertGreater(seen_ok, 0)
        self.assertGreater(seen_nr, 0)

        # Test d'une valeur supérieure à 150
        random.seed(1)
        sensor = Sensor(mean=151.0)

        try:
            sensor.get_value()
            self.fail("OutofBoundsException aurait dû être levée")
        except OutofBoundsException:
            pass

        # Test d'une valeur inférieure à 0
        random.seed(1)
        sensor = Sensor(mean=-10.0)

        try:
            sensor.get_value()
            self.fail("OutofBoundsException aurait dû être levée")
        except OutofBoundsException:
            pass


if __name__ == '__main__':
    _ = unittest.main(verbosity=2)

Exécutez uv run python tests_sensor.py jusqu’à ce que tous les tests passent.

Avant de continuer, vérifiez :

• OutofBoundsException et NotRespondingException sont bien levées et interceptées aux bons endroits
• uv run python tests_sensor.py passe sans échec
• chaque méthode publique de Sensor a une docstring précisant paramètres, valeur de retour et exceptions levées

Exercice 2 : la classe Analyzer

Implémentez une classe Analyzer qui reçoit des valeurs (sous forme de texte) une par une, les convertit en float, et calcule des statistiques toutes les __buffer_size valeurs reçues.

Attributs :

__buffer_size : int = 10      # nombre de valeurs entre deux analyses
__values : list[float] = []
__errors : list[str] = []
__logger : logging.Logger
__log_file : str = "analyzer.log"

Méthodes à implémenter :

Copiez-collez le fichier de tests suivant dans tests_analyzer.py :

import os
import unittest
from analyzer import Analyzer


class TestAnalyzer(unittest.TestCase):

    def setUp(self):
        self.__log_file = "test_analyzer.log"

    def tearDown(self):
        if os.path.exists(self.__log_file):
            os.remove(self.__log_file)

    def test_encapsulation_default_values(self):
        a = Analyzer(log_file=self.__log_file)
        self.assertEqual(a.get_buffer_size(), 10)
        self.assertEqual(a.get_values(), [])
        self.assertEqual(a.get_errors(), [])

    def test_encapsulation_custom_values(self):
        a = Analyzer(buffer_size=3, log_file=self.__log_file)
        self.assertEqual(a.get_buffer_size(), 3)

    def test_add_data_valid_value(self):
        a = Analyzer(log_file=self.__log_file)
        a.add_data("42.5")
        self.assertEqual(a.get_values(), [42.5])
        self.assertEqual(a.get_errors(), [])

    def test_add_data_invalid_value(self):
        a = Analyzer(log_file=self.__log_file)
        a.add_data("not_a_number")
        self.assertEqual(a.get_errors(), ["not_a_number"])
        self.assertEqual(a.get_values(), [])

    def test_log_analysis_statistics(self):
        a = Analyzer(buffer_size=3, log_file=self.__log_file)
        a.add_data("10")
        a.add_data("20")
        # a.add_data("30")  # déclenche automatiquement l'analyse
        stats = a.log_analysis()
        if stats is not None:
            self.assertEqual(stats["average"], 15.0)
            self.assertEqual(stats["min"], 10.0)
            self.assertEqual(stats["max"], 20.0)
        else:
            self.fail("log_analysis returned None when it should have returned statistics.")

    def test_log_analysis_no_values(self):
        a = Analyzer(log_file=self.__log_file)
        self.assertIsNone(a.log_analysis())


if __name__ == '__main__':
    _ = unittest.main(verbosity=2)

Exécutez uv run python tests_analyzer.py jusqu’à ce que tous les tests passent.

Avant de continuer, vérifiez :

• la conversion invalide (ValueError) est interceptée et n’interrompt jamais add_data
• uv run python tests_analyzer.py passe sans échec
• chaque méthode publique de Analyzer a une docstring précisant paramètres, valeur de retour et cas particuliers gérés

Exercice 3 : la classe Reader

Implémentez une classe Reader qui lit en continu le fichier écrit par le capteur et transmet chaque ligne à un Analyzer.

Attributs :

__analyzer : Analyzer
__sleep_time : float = 1.0
__file_name : str = "data.txt"

Méthodes à implémenter :

Copiez-collez le fichier de tests suivant dans tests_reader.py :

import unittest
from reader import Reader
from analyzer import Analyzer


class TestReader(unittest.TestCase):

    def test_encapsulation_default_values(self):
        r = Reader(Analyzer())
        self.assertEqual(r.get_sleep_time(), 1.0)
        self.assertEqual(r.get_file_name(), "data.txt")

    def test_encapsulation_custom_values(self):
        r = Reader(Analyzer(), sleep_time=0.5, file_name="in.txt")
        self.assertEqual(r.get_sleep_time(), 0.5)
        self.assertEqual(r.get_file_name(), "in.txt")

    def test_process_line_ignores_comments(self):
        a = Analyzer()
        r = Reader(a)
        r.process_line("#Sensor ID: sensor\n")
        self.assertEqual(a.get_values(), [])
        self.assertEqual(a.get_errors(), [])

    def test_process_line_forwards_value(self):
        a = Analyzer()
        r = Reader(a)
        r.process_line("42.0\n")
        self.assertEqual(a.get_values(), [42.0])


if __name__ == '__main__':
    _ = unittest.main(verbosity=2)

Exécutez uv run python tests_reader.py jusqu’à ce que tous les tests passent.

Avant de continuer, vérifiez :

• les quatre exceptions de run sont interceptées séparément, avec le bon comportement pour chacune
• uv run python tests_reader.py passe sans échec
• chaque méthode publique de Reader a une docstring précisant paramètres, valeur de retour et exceptions gérées

Test de bout en bout

Ajoutez à la fin de sensor.py :

if __name__ == "__main__":
    Sensor().run()

et à la fin de reader.py :

if __name__ == "__main__":
    Reader(Analyzer()).run()

Ouvrez deux terminaux dans votre répertoire de travail et lancez-y respectivement python sensor.py et python reader.py, dans n’importe quel ordre. Le lecteur doit patienter tant que le fichier n’existe pas ou est vide, puis afficher/journaliser les analyses au fur et à mesure. Arrêtez et relancez le capteur (Ctrl-C puis nouvelle exécution) : le lecteur ne doit pas s’arrêter pour autant.

Avant de conclure, vérifiez :

• le lecteur survit à l’absence, puis à l’arrêt, du capteur sans planter
• pylint sensor.py analyzer.py reader.py ne signale plus de violation majeure de la PEP8
• la documentation générée par pydoc -w sensor analyzer reader est complète et compréhensible sans lire le code

Pour aller plus loin

Si le temps le permet :

• Plusieurs types de capteurs. Ajoutez un second type de capteur qui retourne un booléen (test de bon fonctionnement) au lieu d’un réel, en préfixant chaque écriture par le nom et le type du capteur (ex. sensor1:BinarySensor:1). Documentez et vérifiez la PEP8 ; les tests fournis ne couvrent pas cette extension, à vous d’en écrire de nouveaux si vous le souhaitez.
• Plusieurs analyseurs. Faites en sorte qu’un Reader puisse envoyer les données de chaque capteur à un analyseur dédié (statistiques par capteur, et un analyseur d’erreurs séparé déclenché à l’arrêt du programme plutôt qu’à intervalle régulier).
• Parallélisme. Plutôt que deux terminaux séparés, utilisez le module threading pour exécuter le capteur et le lecteur en parallèle dans un même programme. Vous découvrirez en S6 des mécanismes de communication inter-processus (IPC) plus adaptés qu’un fichier partagé pour ce genre d’échange.