Sensor Reference

Source code reference for the collector sensor code.

collector.src.sensors.manager.SensorManager

Initializes and orchestrates the configured sensor drivers.

Attributes:

Name	Type	Description
collector_id	str	The configured collector ID.
drivers	list[AbstractSensorDriver]	The driver instances, or an empty list if they have yet to be initialized.

Source code in collector/src/sensors/manager.py

class SensorManager:
    """
    Initializes and orchestrates the configured
    sensor drivers.

    Attributes:
        collector_id: The configured collector ID.
        drivers: The driver instances, or an empty list
            if they have yet to be initialized.
    """

    collector_id: str
    drivers: list[AbstractSensorDriver] = []

    def __init__(self, config_manager: ConfigManager):
        """
        Initializes all SensorDrivers.

        Args:
            config_manager: ConfigManager instance
                used to retrieve the sensor configs.
        """
        self.collector_id = config_manager.config.collector_id

        for sensor_config in config_manager.config.sensors:
            # Retrieve the driver class for this sensor.
            DriverClass = get_sensor_driver_from_code(
                sensor_config.sensor_code, config_manager.micropython
            )

            driver = DriverClass(config=sensor_config)
            self.drivers.append(driver)

    def poll(self) -> tuple[list[CollectorRecord], list[CollectorError]]:
        """
        Polls all sensors.

        If the sensor returns data, uses it to construct a CollectorRecord.

        If the sensor raises an exception, uses it to construct a CollectorError.

        Returns:
            All recorded sensor data and caught sensor exceptions.
        """
        records: list[CollectorRecord] = []
        errors: list[CollectorError] = []

        for driver in self.drivers:
            # Timestamp in microseconds.
            timestamp = time.time() * 1000 * 1000
            try:
                data = driver.poll()
                record = CollectorRecord(
                    sensor_id=driver.config.sensor_id, data=data, timestamp=timestamp
                )
                records.append(record)
            except Exception as e:
                error = CollectorError(
                    sensor_id=driver.config.sensor_id,
                    error_message=str(e),
                    timestamp=timestamp,
                )
                errors.append(error)

        return records, errors

__init__(config_manager)

Initializes all SensorDrivers.

Parameters:

Name	Type	Description	Default
config_manager	ConfigManager	ConfigManager instance used to retrieve the sensor configs.	required

Source code in collector/src/sensors/manager.py

def __init__(self, config_manager: ConfigManager):
    """
    Initializes all SensorDrivers.

    Args:
        config_manager: ConfigManager instance
            used to retrieve the sensor configs.
    """
    self.collector_id = config_manager.config.collector_id

    for sensor_config in config_manager.config.sensors:
        # Retrieve the driver class for this sensor.
        DriverClass = get_sensor_driver_from_code(
            sensor_config.sensor_code, config_manager.micropython
        )

        driver = DriverClass(config=sensor_config)
        self.drivers.append(driver)

poll()

Polls all sensors.

If the sensor returns data, uses it to construct a CollectorRecord.

If the sensor raises an exception, uses it to construct a CollectorError.

Returns:

Type	Description
tuple[list[CollectorRecord], list[CollectorError]]	All recorded sensor data and caught sensor exceptions.

Source code in collector/src/sensors/manager.py

def poll(self) -> tuple[list[CollectorRecord], list[CollectorError]]:
    """
    Polls all sensors.

    If the sensor returns data, uses it to construct a CollectorRecord.

    If the sensor raises an exception, uses it to construct a CollectorError.

    Returns:
        All recorded sensor data and caught sensor exceptions.
    """
    records: list[CollectorRecord] = []
    errors: list[CollectorError] = []

    for driver in self.drivers:
        # Timestamp in microseconds.
        timestamp = time.time() * 1000 * 1000
        try:
            data = driver.poll()
            record = CollectorRecord(
                sensor_id=driver.config.sensor_id, data=data, timestamp=timestamp
            )
            records.append(record)
        except Exception as e:
            error = CollectorError(
                sensor_id=driver.config.sensor_id,
                error_message=str(e),
                timestamp=timestamp,
            )
            errors.append(error)

    return records, errors

collector.src.sensors.driver.AbstractSensorDriver

Bases: Protocol

An interface used to describe a type of sensor.

Attributes:

Name	Type	Description
config	SensorConfig	The parsed config object for this sensor.

Source code in collector/src/sensors/driver.py

class AbstractSensorDriver(Protocol):
    """
    An interface used to describe a type of sensor.

    Attributes:
        config: The parsed config object for this sensor.
    """

    config: SensorConfig

    def __init__(self, config: SensorConfig):
        """
        Initialize the sensor driver.

        Args:
            config: The parsed config object for this sensor.
        """
        self.config = config

    def poll(self) -> SensorData:
        """
        Utilizes the underlying device driver to collect sensor data.

        Returns:
            The data returned by the sensor.

        Raises:
            Exception: Raised if the sensor fails to return data.
        """
        ...

__init__(config)

Initialize the sensor driver.

Parameters:

Name	Type	Description	Default
config	SensorConfig	The parsed config object for this sensor.	required

Source code in collector/src/sensors/driver.py

def __init__(self, config: SensorConfig):
    """
    Initialize the sensor driver.

    Args:
        config: The parsed config object for this sensor.
    """
    self.config = config

poll()

Utilizes the underlying device driver to collect sensor data.

Returns:

Type	Description
SensorData	The data returned by the sensor.

Raises:

Type	Description
Exception	Raised if the sensor fails to return data.

Source code in collector/src/sensors/driver.py

def poll(self) -> SensorData:
    """
    Utilizes the underlying device driver to collect sensor data.

    Returns:
        The data returned by the sensor.

    Raises:
        Exception: Raised if the sensor fails to return data.
    """
    ...