Activité pratique

"""Simulation of a temperature sensor."""
import random
import time


class OutofBoundsException(Exception):
    """Raised when a generated value is outside [0, 150]."""


class NotRespondingException(Exception):
    """Raised when the sensor simulates a failure to respond."""


class Sensor:
    """A sensor generating Gaussian-distributed temperature values."""

    def __init__(self, id: str = "sensor", mean: float = 90.0, std: float = 5.0,
                 sleep_time: float = 3.0, storage_file_name: str = "data.txt") -> None:
        """Initialize the sensor.

        Args:
            id (str): Sensor identifier.
            mean (float): Mean of the Gaussian distribution.
            std (float): Standard deviation of the Gaussian distribution.
            sleep_time (float): Time in seconds between two produced values.
            storage_file_name (str): Name of the file where values are stored.
        """
        self.__id = id
        self.__mean = mean
        self.__std = std
        self.__sleep_time = sleep_time
        self.__storage_file_name = storage_file_name

    def get_id(self) -> str:
        """Returns: str: the sensor identifier."""
        return self.__id

    def get_mean(self) -> float:
        """Returns: float: the Gaussian mean."""
        return self.__mean

    def get_std(self) -> float:
        """Returns: float: the Gaussian standard deviation."""
        return self.__std

    def get_sleep_time(self) -> float:
        """Returns: float: the time in seconds between two produced values."""
        return self.__sleep_time

    def get_storage_file_name(self) -> str:
        """Returns: str: the name of the storage file."""
        return self.__storage_file_name

    def get_value(self) -> float:
        """Generate a new temperature value.

        Returns:
            float: a value in [0, 150].

        Raises:
            NotRespondingException: simulated failure (~10% of calls).
            OutofBoundsException: the generated value is outside [0, 150].
        """
        if random.choice(range(10)) == 0:
            raise NotRespondingException(f"{self.__id} is not responding")
        value = random.gauss(self.__mean, self.__std)
        if value < 0 or value > 150:
            raise OutofBoundsException(f"{self.__id} produced an out-of-bounds value: {value}")
        return value

    def run(self) -> None:
        """Continuously generate values and store them in the storage file."""
        try:
            while True:
                with open(self.__storage_file_name, 'a') as file:
                    try:
                        file.write(f"{self.get_value()}\n")
                    except NotRespondingException:
                        file.write("NR\n")
                    except OutofBoundsException:
                        file.write("OoB\n")
                time.sleep(self.__sleep_time)
        except KeyboardInterrupt:
            print(f"{self.__id} stopped.")
        except IOError as e:
            print(f"Error writing to file {self.__storage_file_name}: {e}")
        except Exception as e:
            print(f"Unexpected error in {self.__id}: {e}")

"""Analysis of temperature values received from a sensor."""
import logging


class Analyzer:
    """Computes statistics on the latest values received from a sensor."""

    def __init__(self, buffer_size: int = 10, log_file: str = "analyzer.log") -> None:
        """Initialize the analyzer.

        Args:
            buffer_size (int): Number of values collected between two analyses.
            log_file (str): Name of the file receiving the logs.
        """
        self.__buffer_size = buffer_size
        self.__log_file = log_file
        self.__values: list[float] = []
        self.__errors: list[str] = []
        self.__logger = logging.getLogger(f"{__name__}.{id(self)}")
        self.__logger.setLevel(logging.DEBUG)
        self.__logger.addHandler(logging.FileHandler(self.__log_file))

    def get_values(self) -> list[float]:
        """Returns: list[float]: the values received so far."""
        return self.__values

    def get_errors(self) -> list[str]:
        """Returns: list[str]: the invalid values received so far."""
        return self.__errors

    def get_buffer_size(self) -> int:
        """Returns: int: the number of values between two analyses."""
        return self.__buffer_size

    def get_logger(self) -> logging.Logger:
        """Returns: logging.Logger: the logger used by the analyzer."""
        return self.__logger

    def add_data(self, value: str) -> None:
        """Add a new value coming from the sensor.

        Args:
            value (str): The raw value read from the data file.

        Note:
            A value that cannot be converted to float is stored as an
            error instead of raising an exception to the caller.
        """
        try:
            converted = float(value)
            self.__values.append(converted)
            self.__logger.debug(f"Added value:{converted}")
        except ValueError:
            self.__errors.append(value)
            self.__logger.error(f"Error adding value:{value}")
            return
        if len(self.__values) % self.__buffer_size == 0:
            self.log_analysis()

    def log_analysis(self) -> dict | None:
        """Compute and log statistics on the latest values.

        Returns:
            dict | None: {"average", "min", "max"} or None if no value
            has been received yet.
        """
        if not self.__values:
            self.__logger.warning("No valid values available for analysis.")
            return None
        window = self.__values[-self.__buffer_size:]
        stats = {"average": sum(window) / len(window), "min": min(window), "max": max(window)}
        self.__logger.info(f"Analysis results - {stats}")
        return stats

"""Reads temperature data written by a Sensor and forwards it to an Analyzer."""
import time
from analyzer import Analyzer


class Reader:
    """Continuously reads a data file and forwards each line to an Analyzer."""

    def __init__(self, analyzer: Analyzer, sleep_time: float = 1.0, file_name: str = "data.txt") -> None:
        """Initialize the reader.

        Args:
            analyzer (Analyzer): The analyzer receiving the read values.
            sleep_time (float): Time in seconds to wait when no new line is available.
            file_name (str): Name of the file to read from.
        """
        self.__analyzer = analyzer
        self.__sleep_time = sleep_time
        self.__file_name = file_name

    def get_analyzer(self) -> Analyzer:
        """Returns: Analyzer: the associated analyzer."""
        return self.__analyzer

    def get_sleep_time(self) -> float:
        """Returns: float: the wait time between two read attempts."""
        return self.__sleep_time

    def get_file_name(self) -> str:
        """Returns: str: the name of the file being read."""
        return self.__file_name

    def process_line(self, line: str) -> None:
        """Forward a data line to the analyzer, ignoring comment lines.

        Args:
            line (str): A line read from the data file.
        """
        if line.startswith('#'):
            return
        self.__analyzer.add_data(line.strip())

    def run(self) -> None:
        """Continuously read the data file and process new lines."""
        while True:
            try:
                with open(self.__file_name, 'r') as file:
                    while True:
                        line = file.readline()
                        if line:
                            self.process_line(line)
                        else:
                            time.sleep(self.__sleep_time)
            except KeyboardInterrupt:
                print("Reader stopped.")
                break
            except FileNotFoundError:
                print(f"File '{self.__file_name}' not found, retrying...")
                time.sleep(2 * self.__sleep_time)
            except IOError as e:
                print(f"Error reading '{self.__file_name}': {e}")
                time.sleep(2 * self.__sleep_time)
            except Exception as e:
                print(f"Unexpected error: {e}")
                time.sleep(2 * self.__sleep_time)