Index

class Sensor(ABC):
    """Abstract base class for all sensor implementations.

    Subclasses must also inherit from a bus interface mixin
    (``I2CInterface`` or ``MODBUSInterface``) so that the dynamic loader
    can determine which bus to pass during initialization.

    Attributes:
        parameter: Name of the physical parameter being measured.
        bus: The communication bus instance assigned to this sensor.
        confidence: Measurement confidence/uncertainty specification.
        units: Unit string for the measured values (e.g. "°C", "%").
        time_between_reads: Interval in seconds between consecutive reads.
        db_save_function: Callable that persists Datapoint(s).
    """

    def __init__(
        self,
        parameter: str,
        bus: Bus,
        confidence: Confidence,
        units: str,
        time_between_reads: float,
        db_save_function: Callable[[Datapoint | list[Datapoint]], None],
        config_save_function: Callable | None = None
    ):
        """Initialize the sensor.

        Args:
            parameter: Name of the measured parameter (e.g. 'temperature').
            bus: Communication bus instance matching this sensor's bus_type().
            confidence: Measurement confidence interval specification.
            units: Unit of measurement (e.g. '°C', '%').
            time_between_reads: Seconds between automatic readings.
            db_save_function: Function to persist measurements to the DB.
            config_save_function: Callable for saving the passed arguments in the sensors config.
        """
        self.parameter = parameter
        self.bus = bus
        self.confidence = confidence
        self.units = units
        self.time_between_reads = time_between_reads
        self.db_save_function = db_save_function
        if config_save_function is not None:
            self.config_save_function = config_save_function

    @abstractmethod
    async def read(self):
        """Take a measurement and save it via db_save_function.

        Implementations should read from the hardware and call
        ``await self.db_save_function(datapoint)`` with a Measurement
        or list of Measurements.
        """
        pass

    async def reading_loop(self):
        """Run an infinite loop that calls read() at the configured interval.

        This method is started as a task by the Unit's sensing task group.
        Override only if non-uniform timing is needed.
        """
        _logger.debug(f"Started reading from sensor for parameter {self.parameter}")
        while True:
            await self.read()
            await anyio.sleep(self.time_between_reads)

    def get_capabilities(self):
        """Return a dict describing this sensor's capabilities.

        The default implementation returns a single entry keyed by
        ``self.parameter``. Override this method if the sensor produces
        multiple parameters (see GreenhouseAS7341 for an example).

        Returns:
            Dict mapping parameter names to capability info dicts.
        """
        return {
            self.parameter: {
                "units": self.units,
                "confidence": str(self.confidence),
                "time between reads": str(self.time_between_reads) + " seconds"
            }
        }