Reading Data from Weather Sensors with Python

Source: Dev.to

Apart from building a weather station, another thing I had been wanting to do for a long time was to learn Python.
With this Weather Station project I satisfied both :-)

Check the source code of this project on Codeberg.

Learning the Pimoroni’s WeatherHAT library

The first thing I did, after having everything installed, was to learn how the Pimoroni’s WeatherHAT library works in order to:

• Read the weather data from the sensors
• Display the weather data on the WeatherHAT’s screen

So I opened an SSH connection and started playing!

What I learned about the library

• It is necessary to call the update method of the sensor before getting the weather readings.
• The first reading is “crap”; an initial update and a short wait are required before obtaining reliable values.
• There are two temperature attributes: device_temperature and temperature.
  • temperature is calculated by applying a default offset of 7.5 °C to the device_temperature reading.

Calibrating the Temperature

The first thing I focused on was to properly calibrate the temperature. For this I spent several days gathering temperature data from the WeatherHAT sensor and from a nearby weather station, taking readings every hour.

What I noticed was that the difference between both sensors wasn’t a fixed value. The WeatherHAT readings were always 12 °C–16 °C higher than those from the nearby station.

Instead of applying a fixed offset I tried a linear regression. I created a class to calculate the offset based on a given device_temperature value. In the future I could find a better way to calculate the offset, which is why I introduced a TemperatureOffsetInterface.

class LinearRegressionOffset(TemperatureOffsetInterface):
    def __init__(self, config):
        self.a = 0
        self.b = 0
        self.config = config
        self.doLinearRegression()

    def getOffset(self, x):
        """Return the offset for a given device temperature."""
        offset = (self.b * x) + self.a
        return offset

    def doLinearRegression(self):
        """Calculate the linear regression coefficients a (intercept) and b (slope)."""
        x = self.config['data']['x'].split(",")
        y = self.config['data']['y'].split(",")

        x = list(map(float, x))
        y = list(map(float, y))

        medianX = statistics.median(x)
        medianY = statistics.median(y)
        numSum = 0
        denSum = 0

        for index, xValue in enumerate(x):
            numSum += (xValue - medianX) * (y[index] - medianY)
            denSum += (xValue - medianX) ** 2

        self.b = numSum / denSum
        self.a = medianY - (self.b * medianX)

I could have used a Python library to perform the regression, but remember—the goal of this project was to learn Python.

Decoupling From The Hardware

Having solved the temperature‑calibration problem it was time to start coding the reading system.

The process should run periodically (e.g., every 5 minutes). On each iteration the system:

1. Reads the data from the sensor.
2. Applies the temperature offset.
3. Displays the data on the screen.
4. Sends the data to an external service.

I develop on my own PC because it’s more comfortable than coding directly on the Pi. To make this possible I created mocks that simulate access to the WeatherHAT hardware.

# Example of using the abstract factory to obtain concrete implementations
sensor   = factory.createSensor()
display  = factory.createDisplay()

Once the main algorithm was done I simply had to implement the hardware‑specific parts. Using the examples provided by Pimoroni it was straightforward to get data from the sensor and show it on the WeatherHAT screen.

Sending The Data To An External Service

I wanted to be able to view the current temperature from anywhere and also store the readings for later statistics.

The next step was to send the data to an external service. Since my own Weather Data Webapp wasn’t ready yet, I started by sending the data to Adafruit IO. I created a dashboard that shows the current data and a graph that lets me compare the WeatherHAT readings with those from the nearby station—useful for checking the quality of the linear regression.

I defined a ConnectorInterface so that I could later replace the Adafruit connector with an implementation that sends data to my own webapp.

class ConnectorInterface:
    def send(self, weather_data):
        """Send a WeatherData instance to the configured endpoint."""
        pass

A MockConnector was also created to avoid sending bogus data while I was still working on the main flow of the algorithm.

Implementing An Abstract Factory

Everything was working fine, but switching between implementations (mock vs. production) became messy. To simplify the selection of components I applied the Abstract Factory Pattern. With a single configuration setting I can now load the appropriate set of components (Sensor, Display, and Connector).

class SystemFactoryInterface:
    def createDisplay(self) -> DisplayInterface:
        pass

    def createConnector(self, config) -> ConnectorInterface:
        pass

    def createSensor(self) -> SensorInterface:
        pass

Now the rest of the code can stay completely agnostic of whether it’s running on a development machine with mocks or on the Raspberry Pi with real hardware.

sensor   = factory.createSensor()
display  = factory.createDisplay()
connector = factory.createConnector(config)

Conclusion

I enjoyed this part a lot. Learn Python, read the data from the sensor, calibrate the temperature, apply some design patterns… It brought back the feeling I had when I first started learning to program.

Maybe because it is a project that is really different from the ones I’m more used to develop, maybe because it mixes different things, or maybe both :-).

In the next article I’ll try to explain the implementation of the Weather Data Webapp. This time with PHP and Symfony, a language and a framework that I’m more used to work with.

Example Interface Definitions (Python)

class ConnectorInterface:
    def connect(self) -> bool:
        ...

    def disconnect(self) -> bool:
        ...

class SensorInterface:
    def read(self) -> float:
        ...

class Config:
    ...

class MyFactory:
    def createConnector(self, config: Config) -> ConnectorInterface:
        pass

    def createSensor(self, config: Config) -> SensorInterface:
        pass