MongoDB Database and Raspberry Pi Pico W Walkthrough P2

Learn how to use basic CRUD operations with MongoDB and the Raspberry Pi Pico W using their Data API. Before getting started make sure you watch Part 1. In this tutorial, we use the BME280 sensor to get temperature, sensor, and humidity data to work within our database. In reality, you can use any sensor you like!

You can purchase the BME280 from here:

BME280 Pre-Soldered

Also, be sure to subscribe and support the channel if you have not!

Subscribe:

Youtube

Support:

https://www.buymeacoffee.com/mmshilleh

Raspberry Pi Pico W

from machine import Pin, I2C, RTC
import json
import urequests as requests
import network
import ntptime
import time
import utime

import bme280

import constants


i2c = I2C(0,sda=Pin(0), scl=Pin(1), freq=400000)
URL = "<url>"
API_KEY = "<api key>"

def connect_to_wifi(ssid, psk):
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    wlan.connect(ssid, psk)

    while not wlan.isconnected() and wlan.status() >= 0:
        print("Waiting to Connect")
        time.sleep(10)
    if not wlan.isconnected():
        raise Exception("Wifi not available")
    print("Connected to WiFi")


def findOne(filter_dictionary):
    try:
        headers = { "api-key": API_KEY }
        searchPayload = {
            "dataSource": "Cluster0",
            "database": "BME280",
            "collection": "Readings",
            "filter": filter_dictionary,
        }
        response = requests.post(URL + "findOne", headers=headers, json=searchPayload)
        print("Response: (" + str(response.status_code) + "), msg = " + str(response.text))
        if response.status_code >= 200 and response.status_code < 300:
            print("Success Response")
        else:
            print(response.status_code)
            print("Error")
        response.close()
    except Exception as e:
        print(e)
        
        
def find(filter_dictionary):
    try:
        headers = { "api-key": API_KEY }
        searchPayload = {
            "dataSource": "Cluster0",
            "database": "BME280",
            "collection": "Readings",
            "filter": filter_dictionary,
        }
        response = requests.post(URL + "find", headers=headers, json=searchPayload)
        print("Response: (" + str(response.status_code) + "), msg = " + str(response.text))
        if response.status_code >= 200 and response.status_code < 300:
            print("Success Response")
        else:
            print(response.status_code)
            print("Error")
        response.close()
    except Exception as e:
        print(e)
        
        
def insertOne(temp, pressure, humidity, time):
    try:
        headers = { "api-key": API_KEY }
        documentToAdd = {"Device": "BME280",
                         "Temperature (C)": temp,
                         "Pressure": pressure,
                         "Humidity": humidity,
                         "Time": time}
        insertPayload = {
            "dataSource": "Cluster0",
            "database": "BME280",
            "collection": "Readings",
            "document": documentToAdd,
        }
        response = requests.post(URL + "insertOne", headers=headers, json=insertPayload)
        print(response)
        print("Response: (" + str(response.status_code) + "), msg = " + str(response.text))
        if response.status_code >= 200 and response.status_code < 300:
            print("Success Response")
        else:
            print(response.status_code)
            print("Error")
        response.close()
    except Exception as e:
        print(e)
        
        
def insertMany(document_list):
    try:
        headers = { "api-key": API_KEY }
        insertPayload = {
            "dataSource": "Cluster0",
            "database": "BME280",
            "collection": "Readings",
            "documents": document_list,
        }
        response = requests.post(URL + "insertMany", headers=headers, json=insertPayload)
        print("Response: (" + str(response.status_code) + "), msg = " + str(response.text))
        if response.status_code >= 200 and response.status_code < 300:
            print("Success Response")
        else:
            print(response.status_code)
            print("Error")
        response.close()
    except Exception as e:
        print(e)
        
        
def updateOne(filter_dictionary, update_dict):
    try:
        headers = { "api-key": API_KEY }
        update = {"set": update_dict}
        searchPayload = {
            "dataSource": "Cluster0",
            "database": "BME280",
            "collection": "Readings",
            "filter": filter_dictionary,
            "update": update_dict,
        }
        response = requests.post(URL + "updateOne", headers=headers, json=searchPayload)
        print("Response: (" + str(response.status_code) + "), msg = " + str(response.text))
        if response.status_code >= 200 and response.status_code < 300:
            print("Success Response")
        else:
            print(response.status_code)
            print("Error")
        response.close()
    except Exception as e:
        print(e)
        
        
def deleteOne(filter_dictionary):
    try:
        headers = { "api-key": API_KEY }
        searchPayload = {
            "dataSource": "Cluster0",
            "database": "BME280",
            "collection": "Readings",
            "filter": filter_dictionary,
        }
        response = requests.post(URL + "delete", headers=headers, json=searchPayload)
        print("Response: (" + str(response.status_code) + "), msg = " + str(response.text))
        if response.status_code >= 200 and response.status_code < 300:
            print("Success Response")
        else:
            print(response.status_code)
            print("Error")
        response.close()
    except Exception as e:
        print(e)


def main():
    connect_to_wifi(constants.INTERNET_NAME, constants.INTERNET_PASSWORD)
    #document_list = []
    while True:
        bme = bme280.BME280(i2c=i2c)
        temp, pressure, humidity = bme.values
        print(temp, pressure, humidity)
        rtc_time_tuple = RTC().datetime()
        formatted_time = "{:04}-{:02}-{:02} {:02}:{:02}:{:02}".format(
            rtc_time_tuple[0], rtc_time_tuple[1], rtc_time_tuple[2], 
            rtc_time_tuple[4], rtc_time_tuple[5], rtc_time_tuple[6]
        )
        print(formatted_time)
        
        #insertOne(temp, pressure, humidity, formatted_time)
        
        #document_list.append({"Device": "BME280",
            #"Temperature (C)": temp,
            #"Pressure": pressure,
            #"Humidity": humidity,
            #"Time": formatted_time}
        #)
        #if len(document_list) == 10:
           #print(json.dumps(document_list))
           #insertMany(document_list)
           #document_list = []
        #findOne({"Temperature (C)": "23.26C", "Humidity": "53.69%"})
        #find({"Temperature (C)": "24.65C"})
        #updateOne({"Temperature (C)": "23.26C"}, {"Temperature (C)": "24.26C"})
        deleteOne({"Temperature (C)": "24.26C"})

main()

This is the code we use in this tutorial, we comment and uncomment parts of the main function to activate different functions; this is more clear in the Youtube Video. Also, note that the code can be cleaned up considerably, there is a lot of repetition, mainly for demonstration purposes.

The functions we define are as follows:

• insertOne(params): In this function, we pass in values from our sensor. We also pass in a DateTime from the RTC() which is a real-time clock object to know when we recorded our data points. Although we don’t use that in this tutorial, it is good to have a timestamp on our BME280 data. We utilize the action insertOne appended to the URL and pass in the headers and the payload as defined in the function.
• insertMany(params): In this function, we pass in a list of values. This is a list of dictionaries in this case. This allows us to insert the datapoints in batch which is much quicker than doing one at a time as we do in the first function. We set the limit as 10 for the batch size in our code, but you can expand the size of the data points simply by changing the number in the if-statement.
• findOne(params): In this function, we pass in a dictionary that contains the properties of a datapoint we are looking for. In this case, we are searching for a datapoint that we added that has a temperature of 23.26C along with a humidity of 53.69%, if there is more than one datapoint with such values then it will return the first value it finds! You can define the search parameters as you like. We can see that we use the findOne extension along with passing a filter dictionary in the payload.
• findMany(params): In this function, we pass in a dictionary that contains the properties of the data points we are looking for once again. In this case, we can get a list of data points that match the corresponding value as opposed to a single point. The only difference is the action, which is “find” as opposed to “findOne”. This is useful for finding a series of data points.
• updateOne(params): In this function, we pass in a filter dictionary in addition to an update dictionary. MongoDB searches for the datapoint based on the filter and it replaces it accordingly depending on what we define in the update. For more info on this, see the video above. Of course, like the other functions, we can “updateMany” as well for batch-based operations.
• deleteOne(params): In this function, we pass in a filter as we do in the other functions. MongoDB searches for the data point and deletes it permanently based on the filter. It only deletes one point, if you want to delete more, use “deleteMany”. Be careful in such operations as you cannot reverse the operation.

Overall the operations with their API are very similar, just a difference in URL extension along with slightly different values in the payload. They have much more information in their documentation here. Nonetheless, this should give you the basis to do most things you have to do when using MongoDB with Data API.

Conclusion

Hope you enjoyed the tutorial and the series thus far. MongoDB makes it incredibly easy to use their cloud storage for free! This can be incredibly useful for cases where you want to store and access slow-moving data. Do not forget to subscribe to the channel above and stay tuned for more coding and microcontroller-related content.