Communicating with Child Devices Using Python (Web Server IoT)

Collecting Temperature and Humidity Data from TWELITE ARIA

Let’s build a simple IoT system that receives temperature and humidity data from the sensor tag TWELITE ARIA, sends it to a web server, and displays it on a graph.

In the basic script for TWELITE DIP, we only performed simple operations on DI1 and DO1. However, in an actual IoT system, it is necessary to send acquired data to an upstream server using methods such as REST APIs.

About ThingSpeak

Here, we will use ThingSpeak, a service by MathWorks, in combination with the MWings library.

Creating an Account

Visit the ThingSpeak website and create a MathWorks account.

Creating a Channel

Create a “Channel” and configure it as shown below (the “Name” and “Description” can be arbitrary).

Obtaining API Keys

Go to the “API Keys” tab on the Channel page and note down the 16-character “Write API key”.

Configuring and Starting TWELITE ARIA

Changing Settings for TWELITE ARIA

Change the settings of TWELITE ARIA and set the Transmission Interval of TWELITE ARIA Mode to 20 seconds or more (to avoid overloading the server).

Starting TWELITE ARIA

Insert a CR2032 battery to power up TWELITE ARIA.

Writing and Running the Script

Installing Required Modules

Prepare Python 3.12 or later and install the mwings (or mwingslite) and requests modules.

pip install mwings requests

Creating the Script

Create the script stick_aria_thingspeak.py as shown below. Use mwings.parsers.app_aria to receive data and the requests module to send HTTP GET requests to ThingSpeak.

# -*- coding:utf-8 -*-

from zoneinfo import ZoneInfo
from time import perf_counter

import mwings as mw
import requests


API_KEY = "XXXXXXXXXXXXXXXX"  # Replace with your ThingSpeak API key
BASE_URL = "https://api.thingspeak.com/update"
SEND_MIN_INTERVAL = 20  # Minimum interval in seconds to send data to ThingSpeak


# Main function
def main() -> None:
    # Create a twelite object
    twelite = mw.Twelite(mw.utils.ask_user_for_port())

    # Use JST for received data
    twelite.set_timezone(ZoneInfo("Asia/Tokyo"))

    # Initialize last send time
    last_send_time = perf_counter() - SEND_MIN_INTERVAL

    # Initialize the target serial ID
    target_serial_id = -1

    # Register event handlers
    @twelite.on(mw.common.PacketType.APP_ARIA)
    def on_app_aria(packet: mw.parsers.app_aria.ParsedPacket) -> None:

        # Filter packets by serial ID
        if target_serial_id < 0:
            # Set the serial ID from the received packet
            target_serial_id = packet.source_serial_id
            print(f"Serial ID set to {target_serial_id:08X}")
        elif packet.source_serial_id != target_serial_id:
            # Ignore packets from other serial ID devices
            print(
                f"Ignoring packet from serial ID {packet.source_serial_id:08X}, expected {target_serial_id:08X}"
            )
            return

        # Throttle sending to ThingSpeak
        if perf_counter() - last_send_time < SEND_MIN_INTERVAL:
            print("Skipping send due to minimum interval")
            return  # Skip sending if within the minimum interval
        last_send_time = perf_counter()  # Update last send time

        # Send data to ThingSpeak
        payload = {
            "api_key": API_KEY,
            "field1": f"{packet.temp_100x / 100.0:.2f}",  # Temperature
            "field2": f"{packet.humid_100x / 100.0:.2f}",  # Humidity
            "field3": f"{packet.supply_voltage}",  # Supply voltage (mV)
            "field4": f"{packet.lqi}",  # Link Quality Indicator
        }
        response = requests.get(BASE_URL, params=payload)

        # Check the response status
        if response.status_code == 200:
            print(f"OK: entry ID = {response.text}")
        else:
            print(f"NG: status code = {response.status_code}")

    # Start receiving
    try:
        # Set as daemon thread
        twelite.daemon = True
        # Start the thread, Join to the main thread
        twelite.start()
        print("Started receiving")
        while True:
            twelite.join(0.5)  # Receive
    except KeyboardInterrupt:
        # Stop the thread
        print("Flushing...")
        twelite.stop()
        print("Completed")


if __name__ == "__main__":
    # Call the main function
    main()

Running the Script

Execute the script.

python stick_aria_thingspeak.py

If data is successfully received from TWELITE ARIA and sent correctly, you will see the following output showing entry IDs in sequence:

Started receiving
Serial ID set to 8201C2DC
OK: entry ID = 1
OK: entry ID = 2
OK: entry ID = 3
OK: entry ID = 4
OK: entry ID = 5
...

Click the “Private View” tab on ThingSpeak. The transmitted data should appear as a graph.

By default, the Y-axis range adjusts dynamically. You can configure the min and max values for each graph via the ✏️ icon.

Script Overview

The data sent to the server is constructed in the following section:

# Send data to ThingSpeak
payload = {
    "api_key": API_KEY,
    "field1": f"{packet.temp_100x / 100.0:.2f}",  # Temperature
    "field2": f"{packet.humid_100x / 100.0:.2f}",  # Humidity
    "field3": f"{packet.supply_voltage}",  # Supply voltage (mV)
    "field4": f"{packet.lqi}",  # Link Quality Indicator
}
response = requests.get(BASE_URL, params=payload)

From mwings.parsers.app_aria, we retrieve the temperature, humidity, coin cell voltage, and signal quality (LQI) expressed as a value between 0 and 255, then convert these to strings to construct the query for the GET request.