Mise à jour en masse à l'aide d'une carte Raspberry Pi
Cet exemple montre comment collecter des données à l'aide d'une carte Raspberry Pi connectée au Wi-Fi exécutant Python 2.7. Vous pouvez collecter en continu la température du CPU et l'utilisation du CPU toutes les 15 secondes et mettre à jour en bloc un canal ThingSpeak toutes les 2 minutes. Cet exemple utilise l'API Bulk-Write JSON Data
pour collecter des données par lots et les envoyer aux canaux ThingSpeak . En utilisant la mise à jour en bloc, vous pouvez réduire la consommation d'énergie de vos appareils. Étant donné que la carte Raspberry Pi n'est pas livrée avec une horloge en temps réel, vous pouvez utiliser l'horodatage relatif pour les messages de mise à jour en masse.
Installation
Créez un canal comme indiqué dans Collect Data in a New Channel.
Code
1) Importez les bibliothèques nécessaires pour le script.
import json import time import os import psutil import requests
2) Définissez des variables globales qui suivent l'heure de la dernière connexion et l'heure de la dernière mise à jour. Définissez des intervalles de temps pour mettre à jour les données et publiez les données sur ThingSpeak.
last_connection_time = time.time() # Track the last connection time last_update_time = time.time() # Track the last update time posting_interval = 120 # Post data once every 2 minutes update_interval = 15 # Update once every 15 seconds
3) Définissez vos paramètres de clé d'API en écriture ThingSpeak et d'ID du canal , ainsi que les paramètres du serveur ThingSpeak .
write_api_key = "YOUR-CHANNEL-WRITEAPIKEY" # Replace YOUR-CHANNEL-write_api_key with your channel write API key channel_ID = "YOUR-CHANNELID" # Replace YOUR-channel_ID with your channel ID url = "https://api.thingspeak.com/channels/" + channel_ID + "/bulk_update.json" # ThingSpeak server settings message_buffer = []
4) Définissez la fonction httpRequest
qui envoie des données à ThingSpeak et imprime le code de réponse du serveur. Le code de réponse 202
indique que le serveur a accepté la demande et va la traiter.
def httpRequest(): # Function to send the POST request to ThingSpeak channel for bulk update. global message_buffer bulk_data = json.dumps({'write_api_key':write_api_key,'updates':message_buffer}) # Format the json data buffer request_headers = {"User-Agent":"mw.doc.bulk-update (Raspberry Pi)","Content-Type":"application/json","Content-Length":str(len(bulk_data))} # Make the request to ThingSpeak try: print(request_headers) response = requests.post(url,headers=request_headers,data=bulk_data) print (response) # A 202 indicates that the server has accepted the request except e: print(e.code) # Print the error code message_buffer = [] # Reinitialize the message buffer global last_connection_time last_connection_time = time.time() # Update the connection time
5) Définissez la fonction getData
qui renvoie la température du CPU en Celsius ainsi que l'utilisation du CPU en pourcentage.
def getData(): # Function that returns the CPU temperature and percentage of CPU utilization cmd = '/opt/vc/bin/vcgencmd measure_temp' process = os.popen(cmd).readline().strip() cpu_temp = process.split('=')[1].split("'")[0] cpu_usage = psutil.cpu_percent(interval=2) return cpu_temp,cpu_usage
6) Définissez la fonction updatesJson
pour mettre à jour en continu le tampon des messages toutes les 15 secondes.
def updatesJson(): # Function to update the message buffer every 15 seconds with data. # And then call the httpRequest function every 2 minutes. # This examples uses the relative timestamp as it uses the "delta_t" parameter. # If your device has a real-time clock, you can also provide the absolute timestamp # using the "created_at" parameter. global last_update_time message = {} message['delta_t'] = int(round(time.time() - last_update_time)) Temp,Usage = getData() message['field1'] = Temp message['field2'] = Usage global message_buffer message_buffer.append(message) # If posting interval time has crossed 2 minutes update the ThingSpeak channel with your data if time.time() - last_connection_time >= posting_interval: httpRequest() last_update_time = time.time()
7) Exécutez une boucle infinie pour appeler en continu la fonction updatesJson
toutes les 15 secondes.
if __name__ == "__main__": # To ensure that this is run directly and does not run when imported while True: # If update interval time has crossed 15 seconds update the message buffer with data if time.time() - last_update_time >= update_interval: updatesJson()
Exemples associés
- Mise à jour en masse à l'aide d'un Arduino ou d'un ESP8266
- Mise à jour en bloc d'un canal ThingSpeak à l'aide d'un panneau de photons à particules