Reading Sensors

Reading a sensor is more than just calling a function. Real sensors produce noisy, sometimes erroneous data. This topic covers sensor reading patterns, noise filtering, calibration, and building a reliable sensor abstraction layer.

The Sensor Reading Pattern

Good sensor code follows this consistent pattern:

python

# MicroPython — clean sensor reading pattern
import utime

class SensorBase:
    """Base class for all sensor drivers."""

    def __init__(self, name, read_interval_ms=1000):
        self.name             = name
        self.read_interval_ms = read_interval_ms
        self._last_read_ms    = 0
        self._last_value      = None
        self._error_count     = 0
        self.MAX_ERRORS       = 5

    def _raw_read(self):
        """Override in subclass. Return raw value or raise."""
        raise NotImplementedError

    def read(self):
        """
        Read with rate-limiting, error handling, and caching.
        Returns (value, is_fresh) or (None, False) on error.
        """
        now = utime.ticks_ms()
        if utime.ticks_diff(now, self._last_read_ms) < self.read_interval_ms:
            return self._last_value, False  # Return cached value

        try:
            value = self._raw_read()
            self._last_value   = value
            self._last_read_ms = now
            self._error_count  = 0
            return value, True
        except Exception as e:
            self._error_count += 1
            print(f"[{self.name}] Error #{self._error_count}: {e}")
            if self._error_count >= self.MAX_ERRORS:
                print(f"[{self.name}] MAX ERRORS reached — check wiring!")
            return self._last_value, False  # Return last known good value

DHT22 — Temperature and Humidity

python

import dht
from machine import Pin

class DHT22Sensor(SensorBase):
    def __init__(self, pin_num):
        super().__init__("DHT22", read_interval_ms=2100)  # Min 2 s between reads
        self._sensor = dht.DHT22(Pin(pin_num))

    def _raw_read(self):
        self._sensor.measure()
        temp = self._sensor.temperature()
        hum  = self._sensor.humidity()

        # Sanity checks
        if not (-40 <= temp <= 80):
            raise ValueError(f"Temperature out of range: {temp}")
        if not (0 <= hum <= 100):
            raise ValueError(f"Humidity out of range: {hum}")

        return {"temperature": round(temp, 1), "humidity": round(hum, 1)}

# Usage
sensor = DHT22Sensor(pin_num=4)

for _ in range(10):
    data, is_fresh = sensor.read()
    if data:
        status = "fresh" if is_fresh else "cached"
        print(f"[{status}] T={data['temperature']}°C H={data['humidity']}%")
    import utime
    utime.sleep(1)

BME280 — Temperature, Humidity, Pressure

python

# BME280 over I2C — more accurate than DHT22
from machine import I2C, Pin
import bme280
import struct

class BME280Sensor(SensorBase):
    def __init__(self, sda_pin=21, scl_pin=22, addr=0x76):
        super().__init__("BME280", read_interval_ms=500)
        self._i2c   = I2C(0, sda=Pin(sda_pin), scl=Pin(scl_pin), freq=400_000)
        self._sensor = bme280.BME280(i2c=self._i2c, address=addr)
        self._sea_level_pa = 101325  # Pa (standard atmosphere)

    def _raw_read(self):
        temp, pres, hum = self._sensor.read_compensated_data()
        temp_c   = temp / 100
        pres_hpa = pres / 25600
        hum_pct  = hum / 1024

        # Calculate altitude from pressure (barometric formula)
        altitude_m = 44330 * (1 - (pres_hpa * 100 / self._sea_level_pa) ** (1 / 5.255))

        return {
            "temperature": round(temp_c, 2),
            "humidity":    round(hum_pct, 2),
            "pressure":    round(pres_hpa, 2),
            "altitude":    round(altitude_m, 1)
        }

sensor = BME280Sensor()
data, _ = sensor.read()
print(f"T={data['temperature']}°C P={data['pressure']}hPa Alt={data['altitude']}m")

ADC Sensors with Noise Filtering

ADC readings are inherently noisy. Use averaging or a moving average filter.

python

from machine import ADC, Pin
import utime

class ADCSensor(SensorBase):
    def __init__(self, pin_num, samples=10, sample_delay_ms=2):
        super().__init__("ADC", read_interval_ms=100)
        self._adc           = ADC(Pin(pin_num))
        self._adc.atten(ADC.ATTN_11DB)
        self._adc.width(ADC.WIDTH_12BIT)
        self._samples       = samples
        self._sample_delay  = sample_delay_ms

    def _raw_read(self):
        # Oversampling — average multiple readings
        readings = []
        for _ in range(self._samples):
            readings.append(self._adc.read())
            utime.sleep_ms(self._sample_delay)

        # Remove outliers (discard top and bottom 10%)
        readings.sort()
        trim = max(1, len(readings) // 10)
        trimmed = readings[trim:-trim] if trim else readings

        avg = sum(trimmed) / len(trimmed)
        return round(avg, 1)

    def read_voltage(self):
        raw, fresh = self.read()
        return raw * 3.3 / 4095 if raw is not None else None

# Moving Average Filter
class MovingAverage:
    def __init__(self, window_size=5):
        self._window = []
        self._size   = window_size

    def update(self, value):
        self._window.append(value)
        if len(self._window) > self._size:
            self._window.pop(0)
        return sum(self._window) / len(self._window)

    @property
    def is_ready(self):
        return len(self._window) == self._size

# Usage
ldr = ADCSensor(pin_num=34, samples=16)
ma  = MovingAverage(window_size=5)

for i in range(20):
    raw, fresh = ldr.read()
    if raw:
        smoothed = ma.update(raw)
        brightness = 100 - (raw / 4095 * 100)
        print(f"Raw: {raw:4.0f}  Smoothed: {smoothed:4.1f}  Bright: {brightness:.1f}%")
    utime.sleep_ms(200)

Ultrasonic Distance (HC-SR04) with Median Filter

python

from machine import Pin, time_pulse_us
import utime

class UltrasonicSensor(SensorBase):
    def __init__(self, trig_pin, echo_pin):
        super().__init__("HC-SR04", read_interval_ms=60)  # 60 ms min between pulses
        self._trig = Pin(trig_pin, Pin.OUT)
        self._echo = Pin(echo_pin, Pin.IN)
        self._trig.off()

    def _single_reading(self):
        self._trig.off()
        utime.sleep_us(2)
        self._trig.on()
        utime.sleep_us(10)
        self._trig.off()

        duration = time_pulse_us(self._echo, 1, 23200)  # Max ~4m at 343m/s
        if duration <= 0:
            raise ValueError("No echo — out of range or wiring error")

        return duration * 0.0343 / 2  # cm

    def _raw_read(self):
        """Median of 5 readings to remove spikes."""
        readings = []
        for _ in range(5):
            try:
                readings.append(self._single_reading())
            except ValueError:
                pass
            utime.sleep_ms(12)

        if not readings:
            raise ValueError("All 5 readings failed")

        readings.sort()
        return round(readings[len(readings) // 2], 1)  # Median

sensor = UltrasonicSensor(trig_pin=12, echo_pin=14)

for _ in range(10):
    dist, fresh = sensor.read()
    if dist is not None:
        print(f"Distance: {dist} cm")
        if dist < 20:
            print("  ⚠️  Object nearby!")
    utime.sleep(0.1)

Calibration

Raw sensor values often need calibration to be accurate.

python

class CalibratedSensor:
    """Linear calibration: y = mx + b"""

    def __init__(self, raw_sensor, slope=1.0, offset=0.0):
        self._sensor = raw_sensor
        self._slope  = slope
        self._offset = offset

    @classmethod
    def from_two_points(cls, raw_sensor, raw1, known1, raw2, known2):
        """
        Calculate calibration from two known reference points.

        Example: thermometer reads 1850 at 0°C and 3050 at 100°C
        """
        slope  = (known2 - known1) / (raw2 - raw1)
        offset = known1 - slope * raw1
        print(f"Calibration: slope={slope:.4f}  offset={offset:.2f}")
        return cls(raw_sensor, slope, offset)

    def read(self):
        raw, fresh = self._sensor.read()
        if raw is None:
            return None, False
        calibrated = self._slope * raw + self._offset
        return round(calibrated, 2), fresh

# Two-point calibration for MQ-135 air quality sensor
mq_raw = ADCSensor(pin_num=34, samples=32)
# Measured 500 raw in clean air (expected 400 ppm CO2 baseline)
# Measured 2000 raw with 1000 ppm reference gas
mq_calibrated = CalibratedSensor.from_two_points(
    mq_raw,
    raw1=500,   known1=400,   # Clean air reference
    raw2=2000,  known2=1000   # 1000 ppm reference
)

ppm, _ = mq_calibrated.read()
print(f"Air quality: {ppm} ppm CO2-equivalent")

Multi-Sensor Reading with Async

python

# Read multiple sensors concurrently using uasyncio
import uasyncio

temperature_data = {}

async def read_dht22(sensor):
    while True:
        data, fresh = sensor.read()
        if data and fresh:
            temperature_data["dht22"] = data
        await uasyncio.sleep_ms(2100)

async def read_bme280(sensor):
    while True:
        data, fresh = sensor.read()
        if data and fresh:
            temperature_data["bme280"] = data
        await uasyncio.sleep_ms(1000)

async def log_data():
    while True:
        if temperature_data:
            print(f"Readings: {temperature_data}")
        await uasyncio.sleep(5)

async def main():
    dht = DHT22Sensor(pin_num=4)
    bme = BME280Sensor()

    await uasyncio.gather(
        read_dht22(dht),
        read_bme280(bme),
        log_data()
    )

uasyncio.run(main())

Summary

Wrap sensor code in classes with error handling and rate limiting — raw dht.measure() calls in a loop will eventually crash
Oversampling (reading multiple times and averaging) reduces ADC noise significantly
Median filter is better than mean for removing spike outliers from ultrasonic sensors
Moving average provides smooth readings for slowly changing quantities like temperature
Two-point calibration (y = mx + b) corrects for sensor offset and scale errors
Use uasyncio to read multiple sensors concurrently without blocking

Next, we'll learn how to control actuators — LEDs, servos, motors, relays, and displays.