How to interact with inator devices with LabRat#
It is possible to publish through LabRat into the connections that have been set up
Requirements
A computer wwith a working install of LabRat working, follow the quick start guide
A microprocessor (we are using a Pico W which has built in Wifi)
A DHT11 temperature and humidity sensor (we used the version from Seeed Studios)
We are going to use the same code and secrets.toml file that we used in the quickstart, this includes the database we are going to log to (if we want to)
import os
import sys
import microcontroller
import time
import json
import board
import adafruit_dht
import supervisor
dht = adafruit_dht.DHT11(board.GP0)
def hola(inatorname:str):
print(f"Hello from {inatorname}")
def reading():
try:
temp = dht.temperature
humidity = dht.humidity
if temp is None:
print("⚠️ DHT read failed")
else:
#Create dictionary
json_dict = {"inator":inatorname,"temperature":temp,"humidity":humidity} # Send data with JSON syntax
json_data = json.dumps(json_dict)
#print(f"{json_data}")
return json_data
except RuntimeError as e:
print(f"Sensor read error: {e}")
return None
#Retrieve variables from settings.toml
inatorname = os.getenv("INATORNAME")
accTime = os.getenv('ACQUIRETIME') #how often to acquire and transmit data
recTime = os.getenv('RECONTIME') #how oftern to reconnect in seconds
last_publish = 0
input_buffer = ""
while True:
now = time.monotonic()
if supervisor.runtime.serial_bytes_available:
char = sys.stdin.read(1)
if char in ("\n", "\r"):
if input_buffer:
print(f"Received command: {input_buffer}")
if input_buffer == "hola":
hola(inatorname)
if input_buffer == "read":
json_data = reading()
if json_data:
print(f"{json_data}")
input_buffer = ""
else:
input_buffer += char
if now - last_publish >= accTime:
json_data = reading()
if json_data:
print(f"{json_data}")
last_publish = now
time.sleep(0.01)
VERSION = 1.0
[CONFIG]
"AUTOBROW" = false # whether to automatically open the browser or not
"ALLOW_SETUP" = false
"COMMAND_QUEUE" = true # whether to allow a command COMMAND_QUEUE to run
[LOGGING]
[LOGGING.SQLITE]
[LOGGING.SQLITE.MYDB]
"FILE" = "path to the database as set up in quick start"
"CREATE" = false
[CONNECTIONS]
[CONNECTIONS.MQTT]
[CONNECTIONS.MQTT.HIVE]
"VERB" = True
"USERNAME" = ""
"KEY" = ""
"BROKER" = ""
"PORT" = 8883
"TOPIC" = "/intatorlog/#"
"TLS" = true
[CONNECTIONS.SERIAL] # Serial connection details
[CONNECTIONS.SERIAL.LOG] # Details of a Serial connection
"VERB" = True
"PORT" = "FIND OUT" # The name of the serial port
"BAUD" = "115200" # The baud rate for the port
[CONNECTIONS.TTY] # Run and monitor a terminal
[CONNECTIONS.TTY.NAME] # Name for the terminal connection NAME should be changed and be unique
[CONNECTIONS.FLASK]
"KEY" = "TO GENERATE SEE LABRAT README"
"HOST" = "127.0.0.1"
"PORT" = 8090
"LOCAL_ON" = true
[CONNECTIONS.FLASK.HTTP]
"ROUTE" = "/datafind"
[CONNECTIONS.FLASK.DASH]
"ROUTE" = "/dash"
[CONNECTIONS.FLASK.SETUP]
"ROUTE" = "/setup"
[CONNECTIONS.FLASK.DEVICES]
"ROUTE" = "/api/devices"
[CONNECTIONS.FLASK.UPDATE_CONNS]
"ROUTE" = "api/conns"
[CONNECTIONS.FLASK.PUBLISH_API]
"ROUTE" = "api/publish"
Note that we have added an extra Flask route, the PUBLISH_API.
Now start your logging
python labrat_log.py -secrets <Path to the toml file>
Once the logging is started, in your terminal type
help
Something like the following should be printed
help
Commands:
list
add mqtt <name> <broker/host> <port> <topic> [--username U] [--key K] [--tls] [--ca-cert PATH]
add serial <name> <port> <baud>
remove <type> <name>
publish <name> <topic> <json-payload>
get <logger-name> <device-name> [--start <datetime>][--end <datetime>] [--other flags]
quit
These are commands that are available to you inside LabRat. Try the ‘list’ command This will ist what connections you have and their status
> Command is :list
── Logging connections ──────────────────
🟢 active sqlite
────────────────────────────────────────
── Active connections ──────────────────
🟢 alive HIVE mqtt
🟢 alive LOG serial
🟢 alive flask flask
────────────────────────────────────────
in this case we have three connections. A MQTT connection called HIVE, a serial connection called LOG and a flask connection We are going to publish to the serial connection. We can type ‘help’ to remind us of the command
publish <name> <topic> <json-payload>
This is telling us we need to type publish the name of the connection to use (exactly as the connection name is in the list), followed by the name of the a topic (this is important for MQTT, but not for serial connections) followed by whatever we wish to send. So lets type the following
publish LOG test hola
You should get printed back
print(f"Hola from {inatorname}")
where {inatorname} is replaced with the name of your device.
Try
publish LOG test read
We can also do this via the Flask API we created by making a POST request
If you have it installed you can run the following curl command (Make sure to replace the <> with your actual Flask key)
curl --location -X POST 'http://127.0.0.1:5000/api/publish' \
--header 'X-API-KEY: <REPLACE WITH YOUR FLASK KEY>' \
--header 'Content-Type: application/json' \
--data '{
"name": "LOG",
"target": "test",
"payload": "{hola}"
}'
or from within Python (Make sure to replace the <> with your actual Flask key)
import requests
url = "http://127.0.0.1:5000/api/publish"
headers = {
"X-API-KEY": "<REPLACE WITH YOUR FLASK KEY>",
"Content-Type": "application/json"
}
payload = {
"name": "LOG",
"target": "test",
"payload": "{hola}"
}
response = requests.post(url, json=payload, headers=headers)
print(response.status_code)
print(response.text)
Therefore, we can make the microprocessors respond to commands. It is also possible to publish via a MQTT connection using the same syntax. In which case the topic must matched something you are subscribed to.