BC Robotics
Raspberry Pi Pico Weather Station – Part 2 (MicroPython)
PRODUCT TUTORIAL
- Chris @ BCR
- April 13, 2023
- 10:34 pm
- 2 Comments
In Part 1 of this tutorial we attached the Pico to the Weather Board and looked at all of the sensors. If you have not yet completed this part of the tutorial, please complete this before proceeding further.
Part 2 of this tutorial is split into two versions. We will be walking through this in both Arduino and MicroPython with step by step example code. This is not claimed to be the best possible way of doing this, and we encourage anyone interested to further the code and improve the accuracy. Our goal here is to make this as easy to follow as possible. This specific version of the tutorial is for MicroPython, our Arduino version is available here.
MicroPython Installation and Configuration
If you have not already installed the Arduino IDE and installed the Raspberry Pi Pico library, we have created a tutorial specifically covering this. Jump over to our Programming the Raspberry Pi Pico with MicroPython tutorial and complete it before proceeding.
MicroPython Libraries
We are lucky with MicroPython, as most of the libraries we need are built right in for this project; however, there is one that we will need to add / create for everything to work correctly. Adding a library to the Pico is a little different than most systems, and is done by way of creating a new file and saving it directly to the Pico. We can then access this file from our code just like a normal library.
This BME library was created by Paul Cunnane and Peter Dahlebrg, based on the Adafruit BME280 Python library. In Thonny, create a new file and paste the following code into it:
# Authors: Paul Cunnane 2016, Peter Dahlebrg 2016
#
# This module borrows from the Adafruit BME280 Python library. Original
# Copyright notices are reproduced below.
#
# Those libraries were written for the Raspberry Pi. This modification is
# intended for the MicroPython and esp8266 boards.
#
# Copyright (c) 2014 Adafruit Industries
# Author: Tony DiCola
#
# Based on the BMP280 driver with BME280 changes provided by
# David J Taylor, Edinburgh (www.satsignal.eu)
#
# Based on Adafruit_I2C.py created by Kevin Townsend.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
import time
from ustruct import unpack, unpack_from
from array import array
# BME280 default address.
BME280_I2CADDR = 0x76
# Operating Modes
BME280_OSAMPLE_1 = 1
BME280_OSAMPLE_2 = 2
BME280_OSAMPLE_4 = 3
BME280_OSAMPLE_8 = 4
BME280_OSAMPLE_16 = 5
BME280_REGISTER_CONTROL_HUM = 0xF2
BME280_REGISTER_CONTROL = 0xF4
class BME280:
def __init__(self,
mode=BME280_OSAMPLE_1,
address=BME280_I2CADDR,
i2c=None,
**kwargs):
# Check that mode is valid.
if mode not in [BME280_OSAMPLE_1, BME280_OSAMPLE_2, BME280_OSAMPLE_4,
BME280_OSAMPLE_8, BME280_OSAMPLE_16]:
raise ValueError(
'Unexpected mode value {0}. Set mode to one of '
'BME280_ULTRALOWPOWER, BME280_STANDARD, BME280_HIGHRES, or '
'BME280_ULTRAHIGHRES'.format(mode))
self._mode = mode
self.address = address
if i2c is None:
raise ValueError('An I2C object is required.')
self.i2c = i2c
# load calibration data
dig_88_a1 = self.i2c.readfrom_mem(self.address, 0x88, 26)
dig_e1_e7 = self.i2c.readfrom_mem(self.address, 0xE1, 7)
self.dig_T1, self.dig_T2, self.dig_T3, self.dig_P1, \
self.dig_P2, self.dig_P3, self.dig_P4, self.dig_P5, \
self.dig_P6, self.dig_P7, self.dig_P8, self.dig_P9, \
_, self.dig_H1 = unpack("<HhhHhhhhhhhhBB", dig_88_a1)
self.dig_H2, self.dig_H3 = unpack("<hB", dig_e1_e7)
e4_sign = unpack_from("<b", dig_e1_e7, 3)[0]
self.dig_H4 = (e4_sign << 4) | (dig_e1_e7[4] & 0xF)
e6_sign = unpack_from("<b", dig_e1_e7, 5)[0]
self.dig_H5 = (e6_sign << 4) | (dig_e1_e7[4] >> 4)
self.dig_H6 = unpack_from("<b", dig_e1_e7, 6)[0]
self.i2c.writeto_mem(self.address, BME280_REGISTER_CONTROL,
bytearray([0x3F]))
self.t_fine = 0
# temporary data holders which stay allocated
self._l1_barray = bytearray(1)
self._l8_barray = bytearray(8)
self._l3_resultarray = array("i", [0, 0, 0])
def read_raw_data(self, result):
""" Reads the raw (uncompensated) data from the sensor.
Args:
result: array of length 3 or alike where the result will be
stored, in temperature, pressure, humidity order
Returns:
None
"""
self._l1_barray[0] = self._mode
self.i2c.writeto_mem(self.address, BME280_REGISTER_CONTROL_HUM,
self._l1_barray)
self._l1_barray[0] = self._mode << 5 | self._mode << 2 | 1
self.i2c.writeto_mem(self.address, BME280_REGISTER_CONTROL,
self._l1_barray)
sleep_time = 1250 + 2300 * (1 << self._mode)
sleep_time = sleep_time + 2300 * (1 << self._mode) + 575
sleep_time = sleep_time + 2300 * (1 << self._mode) + 575
time.sleep_us(sleep_time) # Wait the required time
# burst readout from 0xF7 to 0xFE, recommended by datasheet
self.i2c.readfrom_mem_into(self.address, 0xF7, self._l8_barray)
readout = self._l8_barray
# pressure(0xF7): ((msb << 16) | (lsb << 8) | xlsb) >> 4
raw_press = ((readout[0] << 16) | (readout[1] << 8) | readout[2]) >> 4
# temperature(0xFA): ((msb << 16) | (lsb << 8) | xlsb) >> 4
raw_temp = ((readout[3] << 16) | (readout[4] << 8) | readout[5]) >> 4
# humidity(0xFD): (msb << 8) | lsb
raw_hum = (readout[6] << 8) | readout[7]
result[0] = raw_temp
result[1] = raw_press
result[2] = raw_hum
def read_compensated_data(self, result=None):
""" Reads the data from the sensor and returns the compensated data.
Args:
result: array of length 3 or alike where the result will be
stored, in temperature, pressure, humidity order. You may use
this to read out the sensor without allocating heap memory
Returns:
array with temperature, pressure, humidity. Will be the one from
the result parameter if not None
"""
self.read_raw_data(self._l3_resultarray)
raw_temp, raw_press, raw_hum = self._l3_resultarray
# temperature
var1 = ((raw_temp >> 3) - (self.dig_T1 << 1)) * (self.dig_T2 >> 11)
var2 = (((((raw_temp >> 4) - self.dig_T1) *
((raw_temp >> 4) - self.dig_T1)) >> 12) * self.dig_T3) >> 14
self.t_fine = var1 + var2
temp = (self.t_fine * 5 + 128) >> 8
# pressure
var1 = self.t_fine - 128000
var2 = var1 * var1 * self.dig_P6
var2 = var2 + ((var1 * self.dig_P5) << 17)
var2 = var2 + (self.dig_P4 << 35)
var1 = (((var1 * var1 * self.dig_P3) >> 8) +
((var1 * self.dig_P2) << 12))
var1 = (((1 << 47) + var1) * self.dig_P1) >> 33
if var1 == 0:
pressure = 0
else:
p = 1048576 - raw_press
p = (((p << 31) - var2) * 3125) // var1
var1 = (self.dig_P9 * (p >> 13) * (p >> 13)) >> 25
var2 = (self.dig_P8 * p) >> 19
pressure = ((p + var1 + var2) >> 8) + (self.dig_P7 << 4)
# humidity
h = self.t_fine - 76800
h = (((((raw_hum << 14) - (self.dig_H4 << 20) -
(self.dig_H5 * h)) + 16384)
>> 15) * (((((((h * self.dig_H6) >> 10) *
(((h * self.dig_H3) >> 11) + 32768)) >> 10) +
2097152) * self.dig_H2 + 8192) >> 14))
h = h - (((((h >> 15) * (h >> 15)) >> 7) * self.dig_H1) >> 4)
h = 0 if h < 0 else h
h = 419430400 if h > 419430400 else h
humidity = h >> 12
if result:
result[0] = temp
result[1] = pressure
result[2] = humidity
return result
return array("i", (temp, pressure, humidity))
@property
def values(self):
""" human readable values """
t, p, h = self.read_compensated_data()
p = p // 256
pi = p // 100
pd = p - pi * 100
hi = h // 1024
hd = h * 100 // 1024 - hi * 100
return ("{}C".format(t / 100), "{}.{:02d}hPa".format(pi, pd),
"{}.{:02d}%".format(hi, hd))
Once this has been done, click “File” and “Save As”. Next, select “Raspberry Pi Pico” as your save location and name the file “bme280.py” . Once it has saved, the library has been added.
Step 1 - Starting The Code
Time to start writing some code! We are going to break this down into a step by step process. The program will collect data, process it into meaningful values, and finally, it will print all of the values out via the Serial so we can confirm everything is working correctly.
Create a new file in Thonny and start by adding our libraries in using the following code
import onewire, ds18x20, time, bme280, utime, _thread
from machine import Pin, I2C, ADC
14.3%
Step 2 - Core Structure
Next, we need to set up the base structure of the program. We will be running both Pico cores (similar to our Arduino version of this tutorial) so we need to set up a few things to ensure this works correctly.
We will be using a semaphore system to ensure there is no issues with passing values between our two cores. Since only one can access the global variables at a time, we need to limit access by way of a semaphore. The semaphore works just like a talking stick. Each of the two threads will request the semaphore as needed. If it is available, a thread will be able to request the semaphore and access the variable. If not, it will wait until the other thread is finished, and then proceed once it has possession of the semaphore.
So, first we will create the semaphore object:
import onewire, ds18x20, time, bme280, utime, _thread
from machine import Pin, I2C, ADC
spLock = _thread.allocate_lock() # creating semaphore
Next, we will define Core1 and put in a basic request and release of the semaphore. Since this loop is monitoring sensors, we want it to run very quickly so the delay is very short.
import onewire, ds18x20, time, bme280, utime, _thread
from machine import Pin, I2C, ADC
spLock = _thread.allocate_lock() # creating semaphore
#Core 1 handles our monitoring of the Rain Gauge and Anemometer and hosts the Core 1 sensors (Rain Gauge, Anemometer)
def core1_task():
while True:
spLock.acquire() # Acquire semaphore lock
utime.sleep(0.01) # 0.01 sec or 10us delay
spLock.release()
While Core0 always starts with a program, Core1 does not. Next, we need to call Core1 to start.
import onewire, ds18x20, time, bme280, utime, _thread
from machine import Pin, I2C, ADC
spLock = _thread.allocate_lock() # creating semaphore
#Core 1 handles our monitoring of the Rain Gauge and Anemometer and hosts the Core 1 sensors (Rain Gauge, Anemometer)
def core1_task():
while True:
spLock.acquire() # Acquire semaphore lock
utime.sleep(0.01) # 0.01 sec or 10us delay
spLock.release()
#Start Core 1
_thread.start_new_thread(core1_task, ())
Next, we need to create our regular loop in Core0 and add a basic request and release of the semaphore in there as well. This thread is going to collect and print data, and we only want to do this every 15 seconds, so this will only request the semaphore every 15 seconds to ensure the other thread has nearly unblocked access to the variables while it collects data.
import onewire, ds18x20, time, bme280, utime, _thread
from machine import Pin, I2C, ADC
spLock = _thread.allocate_lock() # creating semaphore
#Core 1 handles our monitoring of the Rain Gauge and Anemometer and hosts the Core 1 sensors (Rain Gauge, Anemometer)
def core1_task():
while True:
spLock.acquire() # Acquire semaphore lock
utime.sleep(0.01) # 0.01 sec or 10us delay
spLock.release()
#Start Core 1
_thread.start_new_thread(core1_task, ())
#Main Loop
while True:
utime.sleep(15) #Wait 15 Seconds
spLock.acquire() #Acquire semaphore lock
spLock.release()
Finally, we will create our two global variables. These will be used to collect the number of “ticks” or LOW transits created by the reed switches on our Anemometer and Tipping Rain Gauge. Core1 will watch these two sensors and increment our global variables by 1 each time it detects a transit. After 15 seconds, the Core0 will take this number and use it to calculate the wind speed and the amount of rain that has fallen during the period and reset the counts for the next 15 second period.
import onewire, ds18x20, time, bme280, utime, _thread
from machine import Pin, I2C, ADC
spLock = _thread.allocate_lock() # creating semaphore
#Global count variables
windCount = 0
rainCount = 0
#Core 1 handles our monitoring of the Rain Gauge and Anemometer and hosts the Core 1 sensors (Rain Gauge, Anemometer)
def core1_task():
while True:
spLock.acquire() # Acquire semaphore lock
utime.sleep(0.01) # 0.01 sec or 10us delay
spLock.release()
#Start Core 1
_thread.start_new_thread(core1_task, ())
#Main Loop
while True:
utime.sleep(15) #Wait 15 Seconds
spLock.acquire() #Acquire semaphore lock
spLock.release()
28.6%
Step 3 - Create Core 0 Sensors
For any sensor not requiring constant monitoring, we will let Core0 handle collecting the data. We just need a sample at the time of processing every 15 seconds, no need to involve the other core which is involved in time sensitive data collection. We will start with the DS18B20 temperature sensor which is attached to pin 2.
Each DS18B20 has a unique identifier and multiple can be used on a single bus. While we are only using a single sensor, polling the bus for sensors and creating an array of attached sensors is simpler than figuring out the exact address.
import onewire, ds18x20, time, bme280, utime, _thread
from machine import Pin, I2C, ADC
spLock = _thread.allocate_lock() # creating semaphore
#Global count variables
windCount = 0
rainCount = 0
#Create Core 0 Sensors
#DS18B20
tempSensor = Pin(2, Pin.IN) #Assign the sensor bus to pin 2
sensor = ds18x20.DS18X20(onewire.OneWire(tempSensor)) #create the sensor object
roms = sensor.scan() #scan the bus for sensors
#Core 1 handles our monitoring of the Rain Gauge and Anemometer and hosts the Core 1 sensors (Rain Gauge, Anemometer)
def core1_task():
while True:
spLock.acquire() # Acquire semaphore lock
utime.sleep(0.01) # 0.01 sec or 10us delay
spLock.release()
#Start Core 1
_thread.start_new_thread(core1_task, ())
#Main Loop
while True:
utime.sleep(15) #Wait 15 Seconds
spLock.acquire() #Acquire semaphore lock
spLock.release()
Next we will create the I2C connection and the BME280 sensor attached to it. We are using the Qwiic port, attached to pins 16 and 17.
Start by creating the i2c connection with SDA as pin 16 and SCL as pin 17. Next, add the BME sensor at address 0x77.
import onewire, ds18x20, time, bme280, utime, _thread
from machine import Pin, I2C, ADC
spLock = _thread.allocate_lock() # creating semaphore
#Global count variables
windCount = 0
rainCount = 0
#Create Core 0 Sensors
#DS18B20
tempSensor = Pin(2, Pin.IN) #Assign the sensor bus to pin 2
sensor = ds18x20.DS18X20(onewire.OneWire(tempSensor)) #create the sensor object
roms = sensor.scan() #scan the bus for sensors
#BME280
i2c=I2C(0,sda=Pin(16), scl=Pin(17), freq=400000) #assign the I2C bus to pins 16, 17 (Qwiic Connector)
bme = bme280.BME280(i2c=i2c, address=0x77) #Create the BME object
#Core 1 handles our monitoring of the Rain Gauge and Anemometer and hosts the Core 1 sensors (Rain Gauge, Anemometer)
def core1_task():
while True:
spLock.acquire() # Acquire semaphore lock
utime.sleep(0.01) # 0.01 sec or 10us delay
spLock.release()
#Start Core 1
_thread.start_new_thread(core1_task, ())
#Main Loop
while True:
utime.sleep(15) #Wait 15 Seconds
spLock.acquire() #Acquire semaphore lock
spLock.release()
Finally, we need to create the Wind Vane. This is the simplest of all the sensors. It is a basic analog input connected to A0 (Pin 26)
import onewire, ds18x20, time, bme280, utime, _thread
from machine import Pin, I2C, ADC
spLock = _thread.allocate_lock() # creating semaphore
#Global count variables
windCount = 0
rainCount = 0
#Create Core 0 Sensors
#DS18B20
tempSensor = Pin(2, Pin.IN) #Assign the sensor bus to pin 2
sensor = ds18x20.DS18X20(onewire.OneWire(tempSensor)) #create the sensor object
roms = sensor.scan() #scan the bus for sensors
#BME280
i2c=I2C(0,sda=Pin(16), scl=Pin(17), freq=400000) #assign the I2C bus to pins 16, 17 (Qwiic Connector)
bme = bme280.BME280(i2c=i2c, address=0x77) #Create the BME object
#Wind Vane
windDir = ADC(Pin(26)) #Assign the Wind Vane to ADC0 (Pin 26)
#Core 1 handles our monitoring of the Rain Gauge and Anemometer and hosts the Core 1 sensors (Rain Gauge, Anemometer)
def core1_task():
while True:
spLock.acquire() # Acquire semaphore lock
utime.sleep(0.01) # 0.01 sec or 10us delay
spLock.release()
#Start Core 1
_thread.start_new_thread(core1_task, ())
#Main Loop
while True:
utime.sleep(15) #Wait 15 Seconds
spLock.acquire() #Acquire semaphore lock
spLock.release()
42.9%
Step 4 - Add In Core 1 Sensors
The Rain Gauge and Anemometer are both simple reed switch sensors. Each time the tipping bucket tips, or the Anemometer rotates, a magnet passes a reed switch and causes a change in the electrical signal. These changes are tallied and used to calculate the total rainfall or windspeed. To catch the magnet passing the sensor, we need to have very quick code running all the time to ensure we never miss any passes; this is why we are running this on a separate core from everything else.
First we will create each input and a flag that will be used to help ensure an accurate measurement.
import onewire, ds18x20, time, bme280, utime, _thread
from machine import Pin, I2C, ADC
spLock = _thread.allocate_lock() # creating semaphore
#Global count variables
windCount = 0
rainCount = 0
#Create Core 0 Sensors
#DS18B20
tempSensor = Pin(2, Pin.IN) #Assign the sensor bus to pin 2
sensor = ds18x20.DS18X20(onewire.OneWire(tempSensor)) #create the sensor object
roms = sensor.scan() #scan the bus for sensors
#BME280
i2c=I2C(0,sda=Pin(16), scl=Pin(17), freq=400000) #assign the I2C bus to pins 16, 17 (Qwiic Connector)
bme = bme280.BME280(i2c=i2c, address=0x77) #Create the BME object
#Wind Vane
windDir = ADC(Pin(26)) #Assign the Wind Vane to ADC0 (Pin 26)
#Core 1 handles our monitoring of the Rain Gauge and Anemometer and hosts the Core 1 sensors (Rain Gauge, Anemometer)
def core1_task():
#Rain Gauge
rainInput = Pin(3, Pin.IN, Pin.PULL_UP)
rainFlag = 0
#Anemometer
windInput = Pin(4, Pin.IN, Pin.PULL_UP)
windFlag = 0
while True:
spLock.acquire() # Acquire semaphore lock
utime.sleep(0.01) # 0.01 sec or 10us delay
spLock.release()
#Start Core 1
_thread.start_new_thread(core1_task, ())
#Main Loop
while True:
utime.sleep(15) #Wait 15 Seconds
spLock.acquire() #Acquire semaphore lock
spLock.release()
Since our code is running very fast, it is likely going to read that sensor multiple times as the magnet is passing the reed switch, so we will need to ensure each occurrence is only counted once. This is going to be done with a “flag”. On the first read of the sensor, a flag will be set to true. This flag will stay true until the sensor changes again after the magnet passes.
Each sensor can be thought of as running the following logic:
If the current reading of this sensor is 0 (LOW) and the last reading was 1 (HIGH), then increment our global variable by 1 . Finally, set our flag to match our input so that this code will no longer run until the next occurrence of a LOW Transit.
Add the Rain Gauge and Anemometer code:
import onewire, ds18x20, time, bme280, utime, _thread
from machine import Pin, I2C, ADC
spLock = _thread.allocate_lock() # creating semaphore
#Global count variables
windCount = 0
rainCount = 0
#Create Core 0 Sensors
#DS18B20
tempSensor = Pin(2, Pin.IN) #Assign the sensor bus to pin 2
sensor = ds18x20.DS18X20(onewire.OneWire(tempSensor)) #create the sensor object
roms = sensor.scan() #scan the bus for sensors
#BME280
i2c=I2C(0,sda=Pin(16), scl=Pin(17), freq=400000) #assign the I2C bus to pins 16, 17 (Qwiic Connector)
bme = bme280.BME280(i2c=i2c, address=0x77) #Create the BME object
#Wind Vane
windDir = ADC(Pin(26)) #Assign the Wind Vane to ADC0 (Pin 26)
#Core 1 handles our monitoring of the Rain Gauge and Anemometer and hosts the Core 1 sensors (Rain Gauge, Anemometer)
def core1_task():
#Rain Gauge
rainInput = Pin(3, Pin.IN, Pin.PULL_UP)
rainFlag = 0
#Anemometer
windInput = Pin(4, Pin.IN, Pin.PULL_UP)
windFlag = 0
while True:
spLock.acquire() # Acquire semaphore lock
#Rain Gauge
if(rainInput.value() == 0 and rainFlag == 1): #Compare to our flag to look for a LOW transit
global rainCount #Ensure we write to the global count variable
rainCount += 1 #Since the sensor has transited low, increase the count by 1
rainFlag = rainInput.value() #Set our flag to match our input
#Anemometer
if(windInput.value() == 0 and windFlag == 1): #Compare to our flag to look for a LOW transit
global windCount #Ensure we write to the global count variable
windCount += 1 #Since the sensor has transited low, increase the count by 1
windFlag = windInput.value() #Set our flag to match our input
utime.sleep(0.01) # 0.01 sec or 10us delay
spLock.release()
#Start Core 1
_thread.start_new_thread(core1_task, ())
#Main Loop
while True:
utime.sleep(15) #Wait 15 Seconds
spLock.acquire() #Acquire semaphore lock
spLock.release()
57.1%
Step 5 - Wind Direction
Wind direction is not particularly complicated, our wind direction sensor outputs an analog value based on the direction it is pointing. We have gone through and manually measured each of the different directions already so all we need to figure out which set of values our analog reading falls between and we will know the direction. There is no elegant way to do this other than test each value to see if it is between the two limit values of each heading. Since this ends up being quite a lot of code we will create a new function to just return the direction anytime it is called. It will read the sensor and output the direction in the form of a heading. Create this function at line 25:
Note:
In Arduino, the default output for the analog readings is 10-bit (0-1023) , while MicroPython uses 16-bit (0-65535). Since we don't need that increased accuracy, we will divide the result by 64 to get the same 10-bit range as the Arduino and keep the testing values the same in both tutorials.
import onewire, ds18x20, time, bme280, utime, _thread
from machine import Pin, I2C, ADC
spLock = _thread.allocate_lock() # creating semaphore
#Global count variables
windCount = 0
rainCount = 0
#Create Core 0 Sensors
#DS18B20
tempSensor = Pin(2, Pin.IN) #Assign the sensor bus to pin 2
sensor = ds18x20.DS18X20(onewire.OneWire(tempSensor)) #create the sensor object
roms = sensor.scan() #scan the bus for sensors
#BME280
i2c=I2C(0,sda=Pin(16), scl=Pin(17), freq=400000) #assign the I2C bus to pins 16, 17 (Qwiic Connector)
bme = bme280.BME280(i2c=i2c, address=0x77) #Create the BME object
#Wind Vane
windDir = ADC(Pin(26)) #Assign the Wind Vane to ADC0 (Pin 26)
#Calculate Wind Direction and return as a a string
def calculate_wind_direction():
s = "N/A"
reading = windDir.read_u16() / 64 #Read A0, convert to 10-bit (0-1023)
if 250 <= reading <= 284:
s = "ESE"
elif 285 <= reading <= 304:
s = "ENE"
elif 305 <= reading <= 324:
s = "E"
elif 325 <= reading <= 374:
s = "SSE"
elif 375 <= reading <= 450:
s = "SE"
elif 451 <= reading <= 509:
s = "SSW"
elif 510 <= reading <= 549:
s = "S"
elif 550 <= reading <= 649:
s = "NNE"
elif 650 <= reading <= 724:
s = "NE"
elif 725 <= reading <= 797:
s = "WSW"
elif 798 <= reading <= 824:
s = "SW"
elif 825 <= reading <= 874:
s = "NNW"
elif 875 <= reading <= 909:
s = "N"
elif 910 <= reading <= 934:
s = "WNW"
elif 935 <= reading <= 974:
s = "NW"
elif 975 <= reading <= 1023:
s = "W"
else:
s = "N/A"
return s #Return our wind direction
#Core 1 handles our monitoring of the Rain Gauge and Anemometer and hosts the Core 1 sensors (Rain Gauge, Anemometer)
def core1_task():
#Rain Gauge
rainInput = Pin(3, Pin.IN, Pin.PULL_UP)
rainFlag = 0
#Anemometer
windInput = Pin(4, Pin.IN, Pin.PULL_UP)
windFlag = 0
while True:
spLock.acquire() # Acquire semaphore lock
#Rain Gauge
if(rainInput.value() == 0 and rainFlag == 1): #Compare to our flag to look for a LOW transit
global rainCount #Ensure we write to the global count variable
rainCount += 1 #Since the sensor has transited low, increase the count by 1
rainFlag = rainInput.value() #Set our flag to match our input
#Anemometer
if(windInput.value() == 0 and windFlag == 1): #Compare to our flag to look for a LOW transit
global windCount #Ensure we write to the global count variable
windCount += 1 #Since the sensor has transited low, increase the count by 1
windFlag = windInput.value() #Set our flag to match our input
utime.sleep(0.01) # 0.01 sec or 10us delay
spLock.release()
#Start Core 1
_thread.start_new_thread(core1_task, ())
#Main Loop
while True:
utime.sleep(15) #Wait 15 Seconds
spLock.acquire() #Acquire semaphore lock
spLock.release()
71.4%
Step 6 - Collect And Print The Data
Within our Core0 loop (that runs every 15 seconds) we are going to collect the raw data, make it into useable data, and print all of the values to serial so we can ensure everything is working correctly.
We will start with the DS18B20 sensor. As mentioned earlier, multiple DS18B20 can be connected to a single bus, but we are going to assume only one sensor is connected for now and read it directly from the array at address 0. We are also using the “round” function to round the value.
import onewire, ds18x20, time, bme280, utime, _thread
from machine import Pin, I2C, ADC
spLock = _thread.allocate_lock() # creating semaphore
#Global count variables
windCount = 0
rainCount = 0
#Create Core 0 Sensors
#DS18B20
tempSensor = Pin(2, Pin.IN) #Assign the sensor bus to pin 2
sensor = ds18x20.DS18X20(onewire.OneWire(tempSensor)) #create the sensor object
roms = sensor.scan() #scan the bus for sensors
#BME280
i2c=I2C(0,sda=Pin(16), scl=Pin(17), freq=400000) #assign the I2C bus to pins 16, 17 (Qwiic Connector)
bme = bme280.BME280(i2c=i2c, address=0x77) #Create the BME object
#Wind Vane
windDir = ADC(Pin(26)) #Assign the Wind Vane to ADC0 (Pin 26)
#Calculate Wind Direction and return as a a string
def calculate_wind_direction():
s = "N/A"
reading = windDir.read_u16() / 64 #Read A0, convert to 10-bit (0-1023)
if 250 <= reading <= 284:
s = "ESE"
elif 285 <= reading <= 304:
s = "ENE"
elif 305 <= reading <= 324:
s = "E"
elif 325 <= reading <= 374:
s = "SSE"
elif 375 <= reading <= 450:
s = "SE"
elif 451 <= reading <= 509:
s = "SSW"
elif 510 <= reading <= 549:
s = "S"
elif 550 <= reading <= 649:
s = "NNE"
elif 650 <= reading <= 724:
s = "NE"
elif 725 <= reading <= 797:
s = "WSW"
elif 798 <= reading <= 824:
s = "SW"
elif 825 <= reading <= 874:
s = "NNW"
elif 875 <= reading <= 909:
s = "N"
elif 910 <= reading <= 934:
s = "WNW"
elif 935 <= reading <= 974:
s = "NW"
elif 975 <= reading <= 1023:
s = "W"
else:
s = "N/A"
return s #Return our wind direction
#Core 1 handles our monitoring of the Rain Gauge and Anemometer and hosts the Core 1 sensors (Rain Gauge, Anemometer)
def core1_task():
#Rain Gauge
rainInput = Pin(3, Pin.IN, Pin.PULL_UP)
rainFlag = 0
#Anemometer
windInput = Pin(4, Pin.IN, Pin.PULL_UP)
windFlag = 0
while True:
spLock.acquire() # Acquire semaphore lock
#Rain Gauge
if(rainInput.value() == 0 and rainFlag == 1): #Compare to our flag to look for a LOW transit
global rainCount #Ensure we write to the global count variable
rainCount += 1 #Since the sensor has transited low, increase the count by 1
rainFlag = rainInput.value() #Set our flag to match our input
#Anemometer
if(windInput.value() == 0 and windFlag == 1): #Compare to our flag to look for a LOW transit
global windCount #Ensure we write to the global count variable
windCount += 1 #Since the sensor has transited low, increase the count by 1
windFlag = windInput.value() #Set our flag to match our input
utime.sleep(0.01) # 0.01 sec or 10us delay
spLock.release()
#Start Core 1
_thread.start_new_thread(core1_task, ())
#Main Loop
while True:
utime.sleep(15) #Wait 15 Seconds
spLock.acquire() #Acquire semaphore lock
#Get DS18B20 Temperature
sensor.convert_temp()
temperature = round(sensor.read_temp(roms[0]),1) #Read first sensor in array
spLock.release()
Next, we will read the BME280 sensor to get our case temperature, humidity, and barometric pressure.
The BME280 library (by default) outputs strings including units of measure for each measured parameter, which is not particularly helpful if you need to do any sort of math. Instead, we will pull the data out in a raw format, and convert it to useful units ourselves.
- Humidity needs to be divided by 1024 to provide a normal percentage measure.
- Pressure needs to be floor divided by 256, and then divided by 1000 to get kilopascals.
- Temperature needs to be divided by 100 to get degrees Celsius.
import onewire, ds18x20, time, bme280, utime, _thread
from machine import Pin, I2C, ADC
spLock = _thread.allocate_lock() # creating semaphore
#Global count variables
windCount = 0
rainCount = 0
#Create Core 0 Sensors
#DS18B20
tempSensor = Pin(2, Pin.IN) #Assign the sensor bus to pin 2
sensor = ds18x20.DS18X20(onewire.OneWire(tempSensor)) #create the sensor object
roms = sensor.scan() #scan the bus for sensors
#BME280
i2c=I2C(0,sda=Pin(16), scl=Pin(17), freq=400000) #assign the I2C bus to pins 16, 17 (Qwiic Connector)
bme = bme280.BME280(i2c=i2c, address=0x77) #Create the BME object
#Wind Vane
windDir = ADC(Pin(26)) #Assign the Wind Vane to ADC0 (Pin 26)
#Calculate Wind Direction and return as a a string
def calculate_wind_direction():
s = "N/A"
reading = windDir.read_u16() / 64 #Read A0, convert to 10-bit (0-1023)
if 250 <= reading <= 284:
s = "ESE"
elif 285 <= reading <= 304:
s = "ENE"
elif 305 <= reading <= 324:
s = "E"
elif 325 <= reading <= 374:
s = "SSE"
elif 375 <= reading <= 450:
s = "SE"
elif 451 <= reading <= 509:
s = "SSW"
elif 510 <= reading <= 549:
s = "S"
elif 550 <= reading <= 649:
s = "NNE"
elif 650 <= reading <= 724:
s = "NE"
elif 725 <= reading <= 797:
s = "WSW"
elif 798 <= reading <= 824:
s = "SW"
elif 825 <= reading <= 874:
s = "NNW"
elif 875 <= reading <= 909:
s = "N"
elif 910 <= reading <= 934:
s = "WNW"
elif 935 <= reading <= 974:
s = "NW"
elif 975 <= reading <= 1023:
s = "W"
else:
s = "N/A"
return s #Return our wind direction
#Core 1 handles our monitoring of the Rain Gauge and Anemometer and hosts the Core 1 sensors (Rain Gauge, Anemometer)
def core1_task():
#Rain Gauge
rainInput = Pin(3, Pin.IN, Pin.PULL_UP)
rainFlag = 0
#Anemometer
windInput = Pin(4, Pin.IN, Pin.PULL_UP)
windFlag = 0
while True:
spLock.acquire() # Acquire semaphore lock
#Rain Gauge
if(rainInput.value() == 0 and rainFlag == 1): #Compare to our flag to look for a LOW transit
global rainCount #Ensure we write to the global count variable
rainCount += 1 #Since the sensor has transited low, increase the count by 1
rainFlag = rainInput.value() #Set our flag to match our input
#Anemometer
if(windInput.value() == 0 and windFlag == 1): #Compare to our flag to look for a LOW transit
global windCount #Ensure we write to the global count variable
windCount += 1 #Since the sensor has transited low, increase the count by 1
windFlag = windInput.value() #Set our flag to match our input
utime.sleep(0.01) # 0.01 sec or 10us delay
spLock.release()
#Start Core 1
_thread.start_new_thread(core1_task, ())
#Main Loop
while True:
utime.sleep(15) #Wait 15 Seconds
spLock.acquire() #Acquire semaphore lock
#Get DS18B20 Temperature
sensor.convert_temp()
temperature = round(sensor.read_temp(roms[0]),1) #Read first sensor in array
#Read BME280 Values
humidity = bme.read_compensated_data()[2] / 1024
pressure = (bme.read_compensated_data()[1] // 256) / 1000 #Convert to kPa by dividing the result by 1000
caseTemperature = bme.read_compensated_data()[0] / 100
spLock.release()
Wind direction is very easy, we will just call our “calculate_wind_direction()” function that we created earlier and it will return a string with the wind direction
import onewire, ds18x20, time, bme280, utime, _thread
from machine import Pin, I2C, ADC
spLock = _thread.allocate_lock() # creating semaphore
#Global count variables
windCount = 0
rainCount = 0
#Create Core 0 Sensors
#DS18B20
tempSensor = Pin(2, Pin.IN) #Assign the sensor bus to pin 2
sensor = ds18x20.DS18X20(onewire.OneWire(tempSensor)) #create the sensor object
roms = sensor.scan() #scan the bus for sensors
#BME280
i2c=I2C(0,sda=Pin(16), scl=Pin(17), freq=400000) #assign the I2C bus to pins 16, 17 (Qwiic Connector)
bme = bme280.BME280(i2c=i2c, address=0x77) #Create the BME object
#Wind Vane
windDir = ADC(Pin(26)) #Assign the Wind Vane to ADC0 (Pin 26)
#Calculate Wind Direction and return as a a string
def calculate_wind_direction():
s = "N/A"
reading = windDir.read_u16() / 64 #Read A0, convert to 10-bit (0-1023)
if 250 <= reading <= 284:
s = "ESE"
elif 285 <= reading <= 304:
s = "ENE"
elif 305 <= reading <= 324:
s = "E"
elif 325 <= reading <= 374:
s = "SSE"
elif 375 <= reading <= 450:
s = "SE"
elif 451 <= reading <= 509:
s = "SSW"
elif 510 <= reading <= 549:
s = "S"
elif 550 <= reading <= 649:
s = "NNE"
elif 650 <= reading <= 724:
s = "NE"
elif 725 <= reading <= 797:
s = "WSW"
elif 798 <= reading <= 824:
s = "SW"
elif 825 <= reading <= 874:
s = "NNW"
elif 875 <= reading <= 909:
s = "N"
elif 910 <= reading <= 934:
s = "WNW"
elif 935 <= reading <= 974:
s = "NW"
elif 975 <= reading <= 1023:
s = "W"
else:
s = "N/A"
return s #Return our wind direction
#Core 1 handles our monitoring of the Rain Gauge and Anemometer and hosts the Core 1 sensors (Rain Gauge, Anemometer)
def core1_task():
#Rain Gauge
rainInput = Pin(3, Pin.IN, Pin.PULL_UP)
rainFlag = 0
#Anemometer
windInput = Pin(4, Pin.IN, Pin.PULL_UP)
windFlag = 0
while True:
spLock.acquire() # Acquire semaphore lock
#Rain Gauge
if(rainInput.value() == 0 and rainFlag == 1): #Compare to our flag to look for a LOW transit
global rainCount #Ensure we write to the global count variable
rainCount += 1 #Since the sensor has transited low, increase the count by 1
rainFlag = rainInput.value() #Set our flag to match our input
#Anemometer
if(windInput.value() == 0 and windFlag == 1): #Compare to our flag to look for a LOW transit
global windCount #Ensure we write to the global count variable
windCount += 1 #Since the sensor has transited low, increase the count by 1
windFlag = windInput.value() #Set our flag to match our input
utime.sleep(0.01) # 0.01 sec or 10us delay
spLock.release()
#Start Core 1
_thread.start_new_thread(core1_task, ())
#Main Loop
while True:
utime.sleep(15) #Wait 15 Seconds
spLock.acquire() #Acquire semaphore lock
#Get DS18B20 Temperature
sensor.convert_temp()
temperature = round(sensor.read_temp(roms[0]),1) #Read first sensor in array
#Read BME280 Values
humidity = bme.read_compensated_data()[2] / 1024
pressure = (bme.read_compensated_data()[1] // 256) / 1000 #Convert to kPa by dividing the result by 1000
caseTemperature = bme.read_compensated_data()[0] / 100
#Get Wind Direction
windHeading = calculate_wind_direction()
spLock.release()
The last two items to calculate are Wind Speed and Rainfall. Both sensor values are calculated based on their counts. Once we are done with each, we reset the count for next time.
Wind speed is measured as 1 increment per second = 2.4km/h. So we will divide the count by the total duration of 15 seconds to get the average number of increments per second over the period. From there, we multiply that by 2.4km/h to get the average wind speed over that 15 second period.
Rainfall is a lot simpler. Each increment equates to 0.2794mm of rain. Simply multiply the count by 0.2794 to get the total rainfall recorded across that 15 second period.
import onewire, ds18x20, time, bme280, utime, _thread
from machine import Pin, I2C, ADC
spLock = _thread.allocate_lock() # creating semaphore
#Global count variables
windCount = 0
rainCount = 0
#Create Core 0 Sensors
#DS18B20
tempSensor = Pin(2, Pin.IN) #Assign the sensor bus to pin 2
sensor = ds18x20.DS18X20(onewire.OneWire(tempSensor)) #create the sensor object
roms = sensor.scan() #scan the bus for sensors
#BME280
i2c=I2C(0,sda=Pin(16), scl=Pin(17), freq=400000) #assign the I2C bus to pins 16, 17 (Qwiic Connector)
bme = bme280.BME280(i2c=i2c, address=0x77) #Create the BME object
#Wind Vane
windDir = ADC(Pin(26)) #Assign the Wind Vane to ADC0 (Pin 26)
#Calculate Wind Direction and return as a a string
def calculate_wind_direction():
s = "N/A"
reading = windDir.read_u16() / 64 #Read A0, convert to 10-bit (0-1023)
if 250 <= reading <= 284:
s = "ESE"
elif 285 <= reading <= 304:
s = "ENE"
elif 305 <= reading <= 324:
s = "E"
elif 325 <= reading <= 374:
s = "SSE"
elif 375 <= reading <= 450:
s = "SE"
elif 451 <= reading <= 509:
s = "SSW"
elif 510 <= reading <= 549:
s = "S"
elif 550 <= reading <= 649:
s = "NNE"
elif 650 <= reading <= 724:
s = "NE"
elif 725 <= reading <= 797:
s = "WSW"
elif 798 <= reading <= 824:
s = "SW"
elif 825 <= reading <= 874:
s = "NNW"
elif 875 <= reading <= 909:
s = "N"
elif 910 <= reading <= 934:
s = "WNW"
elif 935 <= reading <= 974:
s = "NW"
elif 975 <= reading <= 1023:
s = "W"
else:
s = "N/A"
return s #Return our wind direction
#Core 1 handles our monitoring of the Rain Gauge and Anemometer and hosts the Core 1 sensors (Rain Gauge, Anemometer)
def core1_task():
#Rain Gauge
rainInput = Pin(3, Pin.IN, Pin.PULL_UP)
rainFlag = 0
#Anemometer
windInput = Pin(4, Pin.IN, Pin.PULL_UP)
windFlag = 0
while True:
spLock.acquire() # Acquire semaphore lock
#Rain Gauge
if(rainInput.value() == 0 and rainFlag == 1): #Compare to our flag to look for a LOW transit
global rainCount #Ensure we write to the global count variable
rainCount += 1 #Since the sensor has transited low, increase the count by 1
rainFlag = rainInput.value() #Set our flag to match our input
#Anemometer
if(windInput.value() == 0 and windFlag == 1): #Compare to our flag to look for a LOW transit
global windCount #Ensure we write to the global count variable
windCount += 1 #Since the sensor has transited low, increase the count by 1
windFlag = windInput.value() #Set our flag to match our input
utime.sleep(0.01) # 0.01 sec or 10us delay
spLock.release()
#Start Core 1
_thread.start_new_thread(core1_task, ())
#Main Loop
while True:
utime.sleep(15) #Wait 15 Seconds
spLock.acquire() #Acquire semaphore lock
#Get DS18B20 Temperature
sensor.convert_temp()
temperature = round(sensor.read_temp(roms[0]),1) #Read first sensor in array
#Read BME280 Values
humidity = bme.read_compensated_data()[2] / 1024
pressure = (bme.read_compensated_data()[1] // 256) / 1000 #Convert to kPa by dividing the result by 1000
caseTemperature = bme.read_compensated_data()[0] / 100
#Get Wind Direction
windHeading = calculate_wind_direction()
#Calculate Wind Speed
measuredWind = (windCount / 15) * 2.4
windCount = 0
#Calculate Rainfall
rainfall = rainCount * 0.2794
rainCount = 0
spLock.release()
Finally, we will print the results. This chunk of code will output the values in an easy to read format.
import onewire, ds18x20, time, bme280, utime, _thread
from machine import Pin, I2C, ADC
spLock = _thread.allocate_lock() # creating semaphore
#Global count variables
windCount = 0
rainCount = 0
#Create Core 0 Sensors
#DS18B20
tempSensor = Pin(2, Pin.IN) #Assign the sensor bus to pin 2
sensor = ds18x20.DS18X20(onewire.OneWire(tempSensor)) #create the sensor object
roms = sensor.scan() #scan the bus for sensors
#BME280
i2c=I2C(0,sda=Pin(16), scl=Pin(17), freq=400000) #assign the I2C bus to pins 16, 17 (Qwiic Connector)
bme = bme280.BME280(i2c=i2c, address=0x77) #Create the BME object
#Wind Vane
windDir = ADC(Pin(26)) #Assign the Wind Vane to ADC0 (Pin 26)
#Calculate Wind Direction and return as a a string
def calculate_wind_direction():
s = "N/A"
reading = windDir.read_u16() / 64 #Read A0, convert to 10-bit (0-1023)
if 250 <= reading <= 284:
s = "ESE"
elif 285 <= reading <= 304:
s = "ENE"
elif 305 <= reading <= 324:
s = "E"
elif 325 <= reading <= 374:
s = "SSE"
elif 375 <= reading <= 450:
s = "SE"
elif 451 <= reading <= 509:
s = "SSW"
elif 510 <= reading <= 549:
s = "S"
elif 550 <= reading <= 649:
s = "NNE"
elif 650 <= reading <= 724:
s = "NE"
elif 725 <= reading <= 797:
s = "WSW"
elif 798 <= reading <= 824:
s = "SW"
elif 825 <= reading <= 874:
s = "NNW"
elif 875 <= reading <= 909:
s = "N"
elif 910 <= reading <= 934:
s = "WNW"
elif 935 <= reading <= 974:
s = "NW"
elif 975 <= reading <= 1023:
s = "W"
else:
s = "N/A"
return s #Return our wind direction
#Core 1 handles our monitoring of the Rain Gauge and Anemometer and hosts the Core 1 sensors (Rain Gauge, Anemometer)
def core1_task():
#Rain Gauge
rainInput = Pin(3, Pin.IN, Pin.PULL_UP)
rainFlag = 0
#Anemometer
windInput = Pin(4, Pin.IN, Pin.PULL_UP)
windFlag = 0
while True:
spLock.acquire() # Acquire semaphore lock
#Rain Gauge
if(rainInput.value() == 0 and rainFlag == 1): #Compare to our flag to look for a LOW transit
global rainCount #Ensure we write to the global count variable
rainCount += 1 #Since the sensor has transited low, increase the count by 1
rainFlag = rainInput.value() #Set our flag to match our input
#Anemometer
if(windInput.value() == 0 and windFlag == 1): #Compare to our flag to look for a LOW transit
global windCount #Ensure we write to the global count variable
windCount += 1 #Since the sensor has transited low, increase the count by 1
windFlag = windInput.value() #Set our flag to match our input
utime.sleep(0.01) # 0.01 sec or 10us delay
spLock.release()
#Start Core 1
_thread.start_new_thread(core1_task, ())
#Main Loop
while True:
utime.sleep(15) #Wait 15 Seconds
spLock.acquire() #Acquire semaphore lock
#Get DS18B20 Temperature
sensor.convert_temp()
temperature = round(sensor.read_temp(roms[0]),1) #Read first sensor in array
#Read BME280 Values
humidity = bme.read_compensated_data()[2] / 1024
pressure = (bme.read_compensated_data()[1] // 256) / 1000 #Convert to kPa by dividing the result by 1000
caseTemperature = bme.read_compensated_data()[0] / 100
#Get Wind Direction
windHeading = calculate_wind_direction()
#Calculate Wind Speed
measuredWind = (windCount / 15) * 2.4
windCount = 0
#Calculate Rainfall
rainfall = rainCount * 0.2794
rainCount = 0
#Print Results
print("Temperature: " , temperature , "°C")
print("Humidity: ", humidity , "%")
print("Barometric Pressure: " , pressure, " kPa")
print("Case Temperature: " , caseTemperature , "°C")
print("Wind: " , windHeading , " " , measuredWind, " km/h")
print("Rainfall Measured: " , rainfall, "mm")
print(" ")
spLock.release()
85.7%
Step 7 - Upload And Test
Once the code is completed, hit “Play” and save it to your Pico. After a period of 15 seconds the first set of results should arrive and all sensors can be tested.
In the next part of this tutorial, we will look at connecting to WiFi and uploading the data to a data handling service.
100%
2 thoughts on “Raspberry Pi Pico Weather Station – Part 2 (MicroPython)”
Graham J
Many thanks for publishing the Micropython tutorial for this project.
Looking forward to the WiFi connection tutorial.
Jim Craig
Thankyou for publishing the Micropython tutorial for the weather station. Looking forward to the Wifi connection portion.