Creating Your Own Sensor

This guide walks you through creating a custom sensor implementation for AtmosPyre.

What You'll Build

A sensor driver that integrates a Modbus RTU sensor with AtmosPyre's logging system.

Prerequisites

• Understanding of Sensor API and ReadTag System
• Access to your sensor's Modbus register map / communication manual
• Python 3.10+
• Sensor connected to a serial port

Implement the Following Interfaces

Attributes

YOUR_MEASUREMENT = YourMeasurementTag(ReadTagMetadata(unit='[unit]', description='[Measurement description]', precision=2, data_type='float', min_interval=1, source='[YourSensor] User Manual, Register [X]')) module-attribute

[Measurement name] measurement tag.

This tag instance provides access to [measurement description] from the [YourSensor] sensor.

Classes

YourMeasurementTag

Bases: ReadTag

Tag for [measurement name] measurement.

This tag class represents [measurement description] from the [YourSensor] sensor. It is used with the dispatch mechanism to route read operations to the appropriate register reading function.

Source code in atmospyre/sensors/_template/sensor_template.py

class YourMeasurementTag(ReadTag):
    """Tag for `[measurement name]` measurement.

    This tag class represents `[measurement description]` from the
    `[YourSensor]` sensor. It is used with the dispatch mechanism to
    route read operations to the appropriate register reading function.
    """

    pass

YourSensor

Bases: Sensor

[Manufacturer] [Model] [sensor type] with Modbus RTU communication.

The [YourSensor] is a [description of sensor capabilities and use cases]. It communicates via Modbus RTU protocol and measures [list of measurements].

Parameters:

Name	Type	Description	Default
port	str	Serial port name (e.g., 'COM3' on Windows, '/dev/ttyUSB0' on Linux)	required
slave_address	int	Modbus slave address, by default 1	1
baudrate	int	Serial communication speed in baud, by default 19200	19200
stopbits	int	Number of stop bits, by default 1	1
bytesize	int	Number of data bits per byte, by default 8	8
parity	str	Parity checking method ('N' for None, 'E' for Even, 'O' for Odd), by default 'N'	'N'
timeout	float	Read timeout in seconds, by default 0.5	0.5

Raises:

Type	Description
SerialException	If the serial port cannot be opened or configured
ModbusException	If Modbus communication fails

Source code in atmospyre/sensors/_template/sensor_template.py

class YourSensor(Sensor):
    """`[Manufacturer]` `[Model]` `[sensor type]` with Modbus RTU communication.

    The `[YourSensor]` is a `[description of sensor capabilities and use cases]`.
    It communicates via Modbus RTU protocol and measures `[list of measurements]`.

    Parameters
    ----------
    port : str
        Serial port name (e.g., 'COM3' on Windows, '/dev/ttyUSB0' on Linux)
    slave_address : int, optional
        Modbus slave address, by default 1
    baudrate : int, optional
        Serial communication speed in baud, by default 19200
    stopbits : int, optional
        Number of stop bits, by default 1
    bytesize : int, optional
        Number of data bits per byte, by default 8
    parity : str, optional
        Parity checking method ('N' for None, 'E' for Even, 'O' for Odd),
        by default 'N'
    timeout : float, optional
        Read timeout in seconds, by default 0.5

    Attributes
    ----------
    port : str
        Serial port name
    slave_address : int
        Modbus slave address
    valid_tags : list of ReadTag
        List of valid measurement tags for this sensor

    Raises
    ------
    SerialException
        If the serial port cannot be opened or configured
    ModbusException
        If Modbus communication fails

    Examples
    --------
    Basic initialization and single tag reading:

    >>> from atmospyre.sensors.implementations.yourtype import yoursensor
    >>> sensor = yoursensor.YourSensor(port='/dev/ttyUSB0', slave_address=1)
    >>> result = sensor.read([yoursensor.YOUR_MEASUREMENT])
    >>> print(result[yoursensor.YOUR_MEASUREMENT])
    [example value]

    Read multiple measurements:

    >>> from atmospyre.sensors.implementations.yourtype.yoursensor import (
    ...     YourSensor, YOUR_MEASUREMENT, ANOTHER_MEASUREMENT
    ... )
    >>> sensor = YourSensor(port='/dev/ttyUSB0')
    >>> result = sensor.read([YOUR_MEASUREMENT, ANOTHER_MEASUREMENT])
    >>> print(f"Measurement: {result[YOUR_MEASUREMENT]} [unit]")
    Measurement: [example] [unit]

    Custom serial configuration:

    >>> sensor = YourSensor(
    ...     port='COM3',
    ...     slave_address=2,
    ...     baudrate=9600,
    ...     timeout=1.0
    ... )
    """

    def __init__(
        self,
        port: str,
        slave_address: int = 1,
        baudrate: int = 19200,
        stopbits: int = 1,
        bytesize: int = 8,
        parity: str = "N",
        timeout: float = 0.5,
    ):
        """Initialize `[YourSensor]` sensor with serial communication parameters.

        Parameters
        ----------
        port : str
            Serial port name (e.g., 'COM3' on Windows, '/dev/ttyUSB0' on Linux)
        slave_address : int, optional
            Modbus slave address, by default 1
        baudrate : int, optional
            Serial communication speed in baud, by default 19200
        stopbits : int, optional
            Number of stop bits, by default 1
        bytesize : int, optional
            Number of data bits per byte, by default 8
        parity : str, optional
            Parity checking method ('N', 'E', 'O'), by default 'N'
        timeout : float, optional
            Read timeout in seconds, by default 0.5
        """
        # TODO: Update valid_tags list with your tag instances
        super().__init__(
            port=port,
            valid_tags=[YOUR_MEASUREMENT],  # Add all your tag instances here
            namespace=yoursensor_namespace,
            slave_address=slave_address,
            baudrate=baudrate,
            stopbits=stopbits,
            bytesize=bytesize,
            parity=parity,
            timeout=timeout,
        )

Functions

__init__(port, slave_address=1, baudrate=19200, stopbits=1, bytesize=8, parity='N', timeout=0.5)

Initialize [YourSensor] sensor with serial communication parameters.

Parameters:

Name	Type	Description	Default
port	str	Serial port name (e.g., 'COM3' on Windows, '/dev/ttyUSB0' on Linux)	required
slave_address	int	Modbus slave address, by default 1	1
baudrate	int	Serial communication speed in baud, by default 19200	19200
stopbits	int	Number of stop bits, by default 1	1
bytesize	int	Number of data bits per byte, by default 8	8
parity	str	Parity checking method ('N', 'E', 'O'), by default 'N'	'N'
timeout	float	Read timeout in seconds, by default 0.5	0.5

Source code in atmospyre/sensors/_template/sensor_template.py

def __init__(
    self,
    port: str,
    slave_address: int = 1,
    baudrate: int = 19200,
    stopbits: int = 1,
    bytesize: int = 8,
    parity: str = "N",
    timeout: float = 0.5,
):
    """Initialize `[YourSensor]` sensor with serial communication parameters.

    Parameters
    ----------
    port : str
        Serial port name (e.g., 'COM3' on Windows, '/dev/ttyUSB0' on Linux)
    slave_address : int, optional
        Modbus slave address, by default 1
    baudrate : int, optional
        Serial communication speed in baud, by default 19200
    stopbits : int, optional
        Number of stop bits, by default 1
    bytesize : int, optional
        Number of data bits per byte, by default 8
    parity : str, optional
        Parity checking method ('N', 'E', 'O'), by default 'N'
    timeout : float, optional
        Read timeout in seconds, by default 0.5
    """
    # TODO: Update valid_tags list with your tag instances
    super().__init__(
        port=port,
        valid_tags=[YOUR_MEASUREMENT],  # Add all your tag instances here
        namespace=yoursensor_namespace,
        slave_address=slave_address,
        baudrate=baudrate,
        stopbits=stopbits,
        bytesize=bytesize,
        parity=parity,
        timeout=timeout,
    )

Functions

_read(instrument, tag)

Read [measurement name] from register [X].

This function reads [measurement description] from Modbus register [X] using [format description, e.g., 32-bit floating point].

Parameters:

Name	Type	Description	Default
instrument	Instrument	minimalmodbus Instrument instance for Modbus communication	required
tag	YourMeasurementTag	Tag instance identifying this measurement type	required

Returns:

Type	Description
float	[Measurement description] in [unit]

Source code in atmospyre/sensors/_template/sensor_template.py

@dispatch(object, YourMeasurementTag, namespace=yoursensor_namespace)
def _read(instrument: Instrument, tag: YourMeasurementTag) -> float:
    """Read `[measurement name]` from register `[X]`.

    This function reads `[measurement description]` from Modbus register `[X]`
    using `[format description, e.g., 32-bit floating point]`.

    Parameters
    ----------
    instrument : Instrument
        minimalmodbus Instrument instance for Modbus communication
    tag : YourMeasurementTag
        Tag instance identifying this measurement type

    Returns
    -------
    float
        `[Measurement description]` in `[unit]`
    """
    pass