Build an AWS IoT Telemetry Station

Send Raspberry Pi sensor and RFID events securely to AWS IoT Core.

Introduction

30 Second Summary

A comfortable exam room depends on conditions people rarely notice until they drift. Shared equipment can also move between rooms without a clear central record.

In this project, you will turn a Raspberry Pi 5 into an exam-room telemetry station using parts from the Freenove Complete Starter Kit FNK0066. The station sends room readings plus privacy-conscious equipment events to AWS IoT Core through a certificate-authenticated MQTT5 connection.

What You'll Build

The finished station fills the AWS IoT MQTT test client with live room readings plus hashed equipment-movement events from your Raspberry Pi.

By the end of this project, you'll have:

  • Live DHT11 readings that let you watch temperature and humidity change in the terminal plus the cloud feed.
  • A privacy-conscious equipment scan that publishes a 12-character hash from the MFRC522 RFID reader.
  • A secure cloud feed where each publication returns a QoS 1 acknowledgement through an X.509 device identity with least-privilege permissions.
  • Secret Mission: Add an approved-equipment list that rejects unknown tags before publication.

Are there any prerequisites?

You need a Raspberry Pi 5 with internet access plus SSH and local access. You also need an AWS account with permission to create AWS IoT Core resources plus the FNK0066 DHT11, MFRC522, RFID tag, breadboard, ribbon cable, extension board, jumper wires, and 10 kΩ resistor.

Before We Start

This first checkpoint defines what your exam-room station demonstrates before any setup begins. You will focus on synthetic or equipment-only telemetry without collecting patient identifiers.

Prepare the Raspberry Pi

Your station can only trust later sensor readings when the Raspberry Pi has a known operating system and toolchain.

Raspberry Pi OS Trixie 64-bit uses Python 3.13 plus newer GPIO plumbing. This step pins Pi 5-compatible dependencies before any wiring begins.

In this step, get ready to:
  • Prepare an SSH-managed Python environment for the station.
  • Install WiringPi and compile the local DHT11 driver.
  • Enable SPI and add the pinned Freenove RFID module.
Connect and prepare the Python environment

SSH gives you a remote terminal on the Pi from Windows PowerShell. A virtual environment keeps the station's pinned packages separate from the operating system's Python packages.

  • Press the Windows key to open Windows search.
  • Type PowerShell into the search field.
  • Press Enter to open Windows PowerShell.
  • Record your Raspberry Pi username as your Raspberry Pi username.
  • Record the Pi hostname or IP address as your Pi hostname or IP address.
  • Connect to the Pi by running this command:
ssh [[PI_USERNAME="your Raspberry Pi username"]]@[[PI_HOST="your Pi hostname or IP address"]]

What does this command do?

The ssh command opens an encrypted terminal session on the Pi. The prompt changes to show your Raspberry Pi username and hostname after you enter its password.

  • Enter yes if SSH asks you to confirm the host fingerprint.
  • Enter your Raspberry Pi password when PowerShell requests it.
  • Check the operating system and processor architecture by running:
cat /etc/os-release
uname -m

What do these checks prove?

The first command reads the operating system release information. The second command reports the processor architecture used to select compatible packages.

You should see Raspberry Pi OS Trixie in the release information. You should also see aarch64 for the 64-bit ARM architecture.

The package manager may ask for confirmation before installing the tools. The installation can take a few minutes while the Pi downloads package indexes and build utilities.

  • Install GNU nano, Git, GCC, build tools, Python 3, pip, virtual-environment support, and Wget by running:
sudo apt update
sudo apt install nano git gcc build-essential python3 python3-full python3-pip wget

What does this install?

  • GNU nano gives you a terminal-based editor for the station files.
  • Git and Wget provide access to pinned source files.
  • GCC and the build tools compile the local DHT11 shared library.
  • Python 3 plus pip provide the isolated runtime for the station scripts.

The command finishes by returning you to the terminal prompt. Packages that were already present are kept at the repository-supported version.

Package installation failed?

Confirm the Pi has internet access before retrying the package commands. A repository connection failure usually means the Pi cannot reach its configured package server.

Need help with the output? Help me troubleshoot a Raspberry Pi OS package installation failure.

  • Create the exam-room-station directory in your home folder by running:
mkdir -p ~/exam-room-station
cd ~/exam-room-station
pwd

What does this command do?

These commands create the station directory before moving your terminal into it. The final command prints the directory path so you can confirm where future files are stored.

You should see a path ending in /exam-room-station.

  • Create requirements.txt with GNU nano by running:
nano requirements.txt

What should you see?

GNU nano opens an empty editing area named requirements.txt. This file pins the Python package versions used throughout the station.

  • Paste the following package list into requirements.txt.
adafruit-lgpio==0.2.2.0
awsiotsdk==1.31.0
gpiozero==2.0.1.post3
spidev==3.8

Why pin these packages?

  • The lgpio provider supports the Python version used by Raspberry Pi OS Trixie on 64-bit ARM.
  • The AWS IoT SDK supplies the MQTT5 client used in the cloud steps.
  • GPIO Zero handles the Pi 5 GPIO controller used by the RFID reset pin.
  • Spidev communicates with the RFID reader over SPI.
  • Save requirements.txt in GNU nano.
  • Exit GNU nano to return to the SSH terminal.
  • Confirm the saved package pins by running:
cat requirements.txt

What should the file contain?

The terminal should print the four pinned dependency lines exactly as you entered them. This check catches missing lines before the package installation begins.

The first installation may take a few minutes while pip downloads the AWS CRT wheel and the remaining packages for 64-bit ARM.

  • Create the .venv environment by running the commands below:
python3 -m venv .venv
.venv/bin/python -m pip install -r requirements.txt

Why use the environment directly?

The first command creates an isolated Python environment inside .venv. The second command installs every version recorded in requirements.txt through that environment's interpreter.

Using .venv/bin/python keeps later commands consistent even when the environment is not activated.

The install succeeds when pip reports the pinned packages as installed and returns you to the terminal prompt.

Python environment failed to install?

Confirm your prompt is inside ~/exam-room-station before recreating .venv. A missing virtual-environment module means the earlier system package installation did not finish.

Need a second pair of eyes? Help me diagnose my virtual-environment or pip installation output.

Install WiringPi and build the DHT driver

WiringPi provides the timing-sensitive GPIO functions used by Freenove's DHT11 driver. Version 3.18 supports the Raspberry Pi 5 on Trixie.

Your Pi may already have a WiringPi command from an earlier setup. This check separates the correct version from an older installation or a missing command.

  • Check the installed WiringPi version by running:
gpio -v

How to read this check

A working command prints the WiringPi version information. Choose the tab that matches the version or error shown in your terminal.

✔️ I see version 3.18

Great. WiringPi 3.18 is already installed for the next build.

ⓧ I see an older version

The pinned ARM64 package updates the existing WiringPi installation to version 3.18.

  • Download the official ARM64 package by running the commands below:
wget https://github.com/WiringPi/WiringPi/releases/download/3.18/wiringpi_3.18_arm64.deb
sudo apt install ./wiringpi_3.18_arm64.deb

What does this install?

The first command downloads the official WiringPi 3.18 ARM64 release package. The second command installs that package through Raspberry Pi OS package management.

ⓧ Command not found

WiringPi is missing from the Pi. The official ARM64 package adds the GPIO library plus its terminal utility.

  • Download the official ARM64 package by running the commands below:
wget https://github.com/WiringPi/WiringPi/releases/download/3.18/wiringpi_3.18_arm64.deb
sudo apt install ./wiringpi_3.18_arm64.deb

What does this install?

The download is pinned to the official WiringPi 3.18 release. Installing the local package adds the library headers needed by GCC and the gpio verification utility.

  • Confirm WiringPi can read the Pi 5 pin map by running:
gpio readall

What does this prove?

The command asks WiringPi to display its GPIO table. A populated table proves the installed utility recognizes the Raspberry Pi 5 pin layout.

You should see a table containing physical pin numbers and GPIO mappings. WiringPi is now ready to compile the DHT11 driver.

  • Download the pinned Freenove DHT source files into ~/exam-room-station by running:
wget -O DHT.c https://raw.githubusercontent.com/Freenove/Freenove_Complete_Starter_Kit_for_Raspberry_Pi/88d429c5db64d60854eb87bccf9ea3a64a736c78/Libs/Python-Libs/Freenove_DHT11/DHT.c
wget -O DHT.h https://raw.githubusercontent.com/Freenove/Freenove_Complete_Starter_Kit_for_Raspberry_Pi/88d429c5db64d60854eb87bccf9ea3a64a736c78/Libs/Python-Libs/Freenove_DHT11/DHT.h

Why pin the source revision?

Both URLs point to one Freenove commit. Pinning the revision keeps DHT.c and DHT.h aligned with the functions used by this project.

  • Compile the DHT source into libdht.so by running the official driver commands:
gcc -fPIC -c DHT.c -o DHT.o -lwiringPi
gcc -shared -fPIC DHT.o -o libdht.so -lwiringPi

What does this build?

The first command compiles DHT.c into a position-independent object file. The second command links that object with WiringPi to create the shared library loaded by Python later.

  • Confirm the DHT source and compiled library exist by running:
ls -l DHT.c DHT.h libdht.so

What should you see?

The output should list DHT.c, DHT.h, and libdht.so with non-zero file sizes. The shared library is the bridge between Python and the timing-sensitive C driver.

DHT driver failed to compile?

Confirm DHT.c and DHT.h are in ~/exam-room-station. A missing WiringPi header means the ARM64 package installation did not complete.

Still blocked? Help me troubleshoot the GCC output from the Freenove DHT11 build.

Enable SPI and add the RFID module

SPI provides the hardware link used by the MFRC522 reader. Raspberry Pi OS exposes that link as a device only after the interface is enabled.

  • Open the Raspberry Pi configuration utility by running:
sudo raspi-config

What happens next?

The configuration utility takes over the terminal with a keyboard-driven menu. Menu numbering can vary across Raspberry Pi OS releases, so follow the labels shown below.

  • Select Interfacing Options.
  • Select SPI.
  • Select Yes to enable SPI.
  • Select Finish.
  • Select Yes if the utility asks to restart the Pi.

The SSH session closes while the Pi restarts. That disconnect is expected because the operating system is applying the interface change.

  • Return to Windows PowerShell after the Pi finishes restarting.
  • Reconnect to the Pi by running:
ssh [[PI_USERNAME="your Raspberry Pi username"]]@[[PI_HOST="your Pi hostname or IP address"]]

Why reconnect?

The restart ended the earlier SSH process. This command starts a fresh remote terminal session with the same saved username and host.

  • Return to the station directory by running:
cd ~/exam-room-station

Where are your files?

The project files remain in ~/exam-room-station after the restart. This command returns the terminal to that directory.

  • Download only the pinned Freenove RFID module by running:
wget -O MFRC522.py https://raw.githubusercontent.com/Freenove/Freenove_Complete_Starter_Kit_for_Raspberry_Pi/88d429c5db64d60854eb87bccf9ea3a64a736c78/Code/Python_GPIOZero_Code/34.1.1_RFID/MFRC522.py

Why download one file?

The command saves the unchanged MFRC522.py module from the pinned Freenove commit. The station only needs this RFID adapter from the larger kit repository.

  • Confirm every setup artifact exists by running:
ls -l requirements.txt DHT.c DHT.h libdht.so MFRC522.py

What does this checkpoint prove?

The output should list the dependency file, both DHT source files, the compiled shared library, and the RFID module. All five files belong inside ~/exam-room-station.

Use the references below if you want to compare the project-authored files with their required final contents.

✔️ Awesome, I've got everything!

Your pinned dependency file and local DHT driver are ready. Keep the downloaded MFRC522.py file unchanged.

ⓧ I'd like to double check the full code

Your project-authored files should match the references below exactly. MFRC522.py remains the unchanged file from Freenove commit 88d429c5db64d60854eb87bccf9ea3a64a736c78.

adafruit-lgpio==0.2.2.0
awsiotsdk==1.31.0
gpiozero==2.0.1.post3
spidev==3.8

What belongs in this file?

These four pins define the Pi 5 GPIO provider, AWS IoT SDK, GPIO Zero release, and SPI package used by the station.

#ifndef DHT_H
#define DHT_H

#include <wiringPi.h>
#include <stdint.h>

#define DHTLIB_OK 0
#define DHTLIB_ERROR_CHECKSUM -1
#define DHTLIB_ERROR_TIMEOUT -2
#define DHTLIB_INVALID_VALUE -999
#define DHTLIB_DHT11_WAKEUP 18
#define DHTLIB_TIMEOUT 100

void setDHT11Pin(int pin);
int readSensor(int pin, int wakeupDelay);
int readDHT11(void);
double getHumidity(void);
double getTemperature(void);

#endif

What does the header define?

The header declares the DHT11 status values, timing constants, and functions that the compiled library exports.

#include "DHT.h"

uint8_t bits[5];
static int dhtPin;
static double humidity = 0;
static double temperature = 0;

void setDHT11Pin(int pin)
{
    dhtPin = pin;
    wiringPiSetupGpio();
}

int readSensor(int pin, int wakeupDelay)
{
    int mask = 0x80;
    int idx = 0;
    int32_t started;

    for (int i = 0; i < 5; i++) {
        bits[i] = 0;
    }

    pinMode(pin, OUTPUT);
    digitalWrite(pin, LOW);
    delay(wakeupDelay);
    digitalWrite(pin, HIGH);
    delayMicroseconds(40);
    pinMode(pin, INPUT);

    started = micros();
    while (digitalRead(pin) == LOW) {
        if ((micros() - started) > DHTLIB_TIMEOUT) {
            return DHTLIB_ERROR_TIMEOUT;
        }
    }

    started = micros();
    while (digitalRead(pin) == HIGH) {
        if ((micros() - started) > DHTLIB_TIMEOUT) {
            return DHTLIB_ERROR_TIMEOUT;
        }
    }

    for (int i = 0; i < 40; i++) {
        started = micros();
        while (digitalRead(pin) == LOW) {
            if ((micros() - started) > DHTLIB_TIMEOUT) {
                return DHTLIB_ERROR_TIMEOUT;
            }
        }

        started = micros();
        while (digitalRead(pin) == HIGH) {
            if ((micros() - started) > DHTLIB_TIMEOUT) {
                return DHTLIB_ERROR_TIMEOUT;
            }
        }

        if ((micros() - started) > 60) {
            bits[idx] |= mask;
        }
        mask >>= 1;
        if (mask == 0) {
            mask = 0x80;
            idx++;
        }
    }

    pinMode(pin, OUTPUT);
    digitalWrite(pin, HIGH);
    return DHTLIB_OK;
}

int readDHT11(void)
{
    int result = readSensor(dhtPin, DHTLIB_DHT11_WAKEUP);
    if (result != DHTLIB_OK) {
        humidity = DHTLIB_INVALID_VALUE;
        temperature = DHTLIB_INVALID_VALUE;
        return result;
    }

    humidity = bits[0];
    temperature = bits[2] + bits[3] * 0.1;
    uint8_t sum = bits[0] + bits[1] + bits[2] + bits[3];
    if (bits[4] != sum) {
        return DHTLIB_ERROR_CHECKSUM;
    }
    return DHTLIB_OK;
}

double getHumidity(void)
{
    return humidity;
}

double getTemperature(void)
{
    return temperature;
}

What does the implementation provide?

The implementation times the DHT11 signal through WiringPi. It validates the checksum before exposing temperature and humidity values to Python.

Before you run the final checks, what should each command prove about GPIO access, Python dependencies, and SPI?

  • Verify the complete Raspberry Pi setup by running:
gpio readall
.venv/bin/python -c "from awsiot import mqtt5_client_builder; import gpiozero, lgpio, spidev; print('Dependencies ready')"
ls /dev/sp*

What do the final checks prove?

  • The GPIO table proves WiringPi recognizes the Raspberry Pi 5 pin map.
  • The dependency message proves the virtual environment can import the AWS MQTT5 builder plus the GPIO and SPI packages.
  • The SPI device path proves Raspberry Pi OS exposed the enabled hardware interface.

You should see the GPIO table, Dependencies ready, and an SPI device such as /dev/spidev0.0. That confirms the software and hardware interfaces are ready for the first local sensor reading.

One of the final checks failed?

A missing GPIO table points back to the WiringPi installation. A failed Python import points back to requirements.txt or the .venv installation.

A missing SPI device means the interface change did not take effect. Return to the configuration utility and confirm SPI is enabled before restarting again.

Share the failing check without including your Pi password. Help me diagnose the Raspberry Pi setup verification.

Your Pi now has a pinned toolchain, a compiled DHT11 driver, and an enabled SPI interface. Next, you can wire the environmental sensor and turn this setup into live room readings.

Read the Exam-Room Environment

Your Raspberry Pi 5 now has its hardware toolchain. The local DHT driver is ready to talk to a sensor.

Cloud configuration can hide whether a bad reading comes from software or wiring. A visible local DHT11 reading gives you a clean hardware checkpoint before AWS enters the picture.

In this step, get ready to:
  • Wire the DHT11 data line to BCM GPIO17 with the FNK0066 pull-up arrangement.
  • Create a Python adapter that loads the compiled DHT driver.
  • Print five local JSON readings from the exam-room station.
Wire the DHT11 safely

The GPIO header carries live power while the Pi is running. A full shutdown protects the Pi while you build the sensor circuit.

This wiring is a little fiddly. Keep the Pi unplugged while you match each connection against the kit diagram.

  • Use the system power control in your local Raspberry Pi session to shut down the Pi.
  • Disconnect the Raspberry Pi power cable.
  • Use the official FNK0066 DHT11 wiring diagram to orient the DHT11 on the breadboard.
  • Connect the DHT11 power pin to the power rail shown in the diagram.
  • Connect the DHT11 ground pin to the ground rail shown in the diagram.
  • Connect the DHT11 data pin to BCM GPIO17 on physical pin 11.
  • Place the 10 kΩ resistor between the data line and the DHT11 power line.
  • Inspect every connection against the diagram before restoring power.

The careful part is complete. Your DHT11 circuit is ready for a software test.

Create the sensor adapter

The adapter uses ctypes to call the compiled functions inside libdht.so. Its retry loop absorbs the brief timing failures that can occur during a DHT11 read.

  • Reconnect the Raspberry Pi power cable.
  • Reconnect from Windows PowerShell using the same SSH command from the previous step.

Your remote shell prompt confirms that the Raspberry Pi has finished booting.

  • Create ~/exam-room-station/environment_sensor.py with GNU nano by pasting this sensor adapter code:
import ctypes
import json
import time
from pathlib import Path

LIBRARY_PATH = Path(__file__).with_name("libdht.so")
LIB = ctypes.CDLL(str(LIBRARY_PATH))
LIB.setDHT11Pin.argtypes = [ctypes.c_int]
LIB.readDHT11.restype = ctypes.c_int
LIB.getHumidity.restype = ctypes.c_double
LIB.getTemperature.restype = ctypes.c_double


class EnvironmentSensor:
    def __init__(self, pin=17):
        self.pin = pin
        LIB.setDHT11Pin(pin)
        time.sleep(1)

    def read(self):
        for _ in range(15):
            if LIB.readDHT11() == 0:
                return {
                    "temperatureC": round(LIB.getTemperature(), 1),
                    "humidityPercent": round(LIB.getHumidity(), 1),
                }
            time.sleep(0.1)
        raise RuntimeError("DHT11 did not return a valid checksum before retry limit")

What does this sensor adapter do?

  • The imports provide native-library access, JSON formatting, timing controls, and a reliable file path.
  • The LIBRARY_PATH value locates libdht.so beside the Python file.
  • The ctypes declarations tell Python which value type each compiled function returns.
  • The EnvironmentSensor class initializes BCM GPIO17.
  • The read() method attempts a reading up to 15 times before reporting a failure.
  • A successful read returns temperatureC plus humidityPercent in a Python dictionary.
  • Append this direct-run loop at the end of environment_sensor.py:
if __name__ == "__main__":
    sensor = EnvironmentSensor()
    for _ in range(5):
        print(json.dumps(sensor.read()))
        time.sleep(2)

What does the runner do?

  • The direct-run check starts this loop only when you execute environment_sensor.py as a script.
  • The loop prints five readings in JSON format.
  • The two-second pause gives the DHT11 time between samples.
  • Save environment_sensor.py in GNU nano.

✔️ Awesome, I've got everything!

Your sensor adapter is saved. Continue to the live hardware test.

ⓧ I'd like to double check the full code

Compare your saved file with this complete version. Every line should match.

import ctypes
import json
import time
from pathlib import Path

LIBRARY_PATH = Path(__file__).with_name("libdht.so")
LIB = ctypes.CDLL(str(LIBRARY_PATH))
LIB.setDHT11Pin.argtypes = [ctypes.c_int]
LIB.readDHT11.restype = ctypes.c_int
LIB.getHumidity.restype = ctypes.c_double
LIB.getTemperature.restype = ctypes.c_double


class EnvironmentSensor:
    def __init__(self, pin=17):
        self.pin = pin
        LIB.setDHT11Pin(pin)
        time.sleep(1)

    def read(self):
        for _ in range(15):
            if LIB.readDHT11() == 0:
                return {
                    "temperatureC": round(LIB.getTemperature(), 1),
                    "humidityPercent": round(LIB.getHumidity(), 1),
                }
            time.sleep(0.1)
        raise RuntimeError("DHT11 did not return a valid checksum before retry limit")


if __name__ == "__main__":
    sensor = EnvironmentSensor()
    for _ in range(5):
        print(json.dumps(sensor.read()))
        time.sleep(2)
Run the live sensor test

The compiled driver accesses the GPIO hardware directly. Running the script with sudo gives it the required hardware access.

Before you run the test, what two measurements do you expect each JSON object to contain?

  • Run the baseline sensor test with this command:
sudo .venv/bin/python environment_sensor.py

What should I see?

You should see five JSON lines appear about two seconds apart.

Each line should contain numeric temperatureC and humidityPercent values.

That result proves the circuit can produce local environmental telemetry.

  • Warm the plastic body of the DHT11 gently with your fingers for a few seconds.
  • Keep your fingers away from the exposed leads.

Before you run the test again, which reading do you think responds first?

  • Run a second sensor test with the same command:
sudo .venv/bin/python environment_sensor.py

What changed?

You should see at least one later reading shift after warming the sensor. The exact change depends on your room conditions.

Not seeing valid readings?

If the script cannot load the shared library, confirm that libdht.so sits beside environment_sensor.py in ~/exam-room-station.

If the reads reach the retry limit, compare the data wire, power wire, ground wire, and 10 kΩ resistor with the official FNK0066 diagram.

Still stuck? Help me troubleshoot my DHT11 readings.

You have turned a physical room condition into structured local telemetry. Next, you will add equipment movement events from an RFID tag.

Detect Equipment Movement

Your Raspberry Pi now turns DHT11 signals into five local JSON readings. You proved the room conditions can be measured at the source.

Equipment movement needs a separate physical signal. In this step, an RFID reader identifies each equipment tag.

Hashing converts the tag's UID bytes into a shorter equipment identifier. You will also test whether the resulting event can be seen beyond the active session.

In this step, get ready to:
  • Wire the MFRC522 reader to the Raspberry Pi.
  • Build a local adapter that detects separate tag scans.
  • Print the local UID plus its 12-character tag hash.
Wire the MFRC522 reader

The MFRC522 uses SPI to exchange tag data with the Pi. Its reset line uses BCM GPIO25.

The reader accepts 3.3 V power only. Powering down first protects both boards while you move jumper wires.

  • Shut down the Raspberry Pi from your SSH session.
  • Wait for the Pi's activity light to stop flashing.
  • Disconnect the Pi's power cable.

Why power down first?

Moving connections on a powered circuit can place voltage on the wrong pin. A complete shutdown gives you a safe wiring state.

  • Connect MFRC522 SDA/SS to physical pin 24.
  • Connect MFRC522 SCK to physical pin 23.

The reader now has its device-selection wire plus its clock wire.

  • Connect MFRC522 MOSI to physical pin 19.
  • Connect MFRC522 MISO to physical pin 21.

The four SPI signal lines now match the FNK0066 pin layout.

  • Connect MFRC522 RST to physical pin 22.
  • Connect MFRC522 GND to a ground pin.

The reader now has a reset path plus a common ground with the Pi.

  • Connect MFRC522 power to the 3.3 V physical pin 1.
  • Compare every connection with the FNK0066 RFID wiring layout before restoring power.

Unsure about a connection?

Read the physical pin numbers from the Raspberry Pi header diagram. Do not substitute the nearby 5 V pin for the reader's 3.3 V connection.

Need another pair of eyes? Help me check my MFRC522 wiring before I power the Raspberry Pi.

Build the equipment reader

The reset line prepares the reader before its SPI connection opens. GPIO Zero handles that reset through BCM GPIO25.

  • Reconnect the Pi's power cable.
  • Wait for the Raspberry Pi to finish starting.
  • Reconnect to the Pi over SSH from Windows PowerShell.
  • Use GNU nano to create ~/exam-room-station/equipment_reader.py.
  • Paste the reader imports plus its initialization code below:
import hashlib
import json
import time

from gpiozero import OutputDevice
import MFRC522


class EquipmentReader:
    def __init__(self):
        self.reset = OutputDevice(25, active_high=True, initial_value=True)
        self.reset.off()
        time.sleep(0.05)
        self.reset.on()
        time.sleep(0.05)
        self.reader = MFRC522.MFRC522(bus=0, dev=0, spd=1000000)
        self.last_uid = None
        self.last_seen = 0.0

What does this code do?

  • The reset pulse prepares the MFRC522 before communication starts.
  • The MFRC522.MFRC522() call opens SPI bus 0 with device 0 at the configured speed.
  • The last_uid plus last_seen values provide the state needed to suppress repeated detections.
  • Save equipment_reader.py in GNU nano.
  • Return to the SSH prompt.
  • Confirm the new file exists by running this command:
ls

What does this check prove?

The command lists the files in your current directory. You should see equipment_reader.py with the existing project files.

Don't see the new file?

Confirm the file was saved inside ~/exam-room-station. A file saved in your home directory will appear in a different listing.

Still stuck? Help me find where GNU nano saved equipment_reader.py.

A tag can remain close enough to trigger several rapid reads. The polling method keeps one event per scan by suppressing the same UID for three seconds.

  • Return to equipment_reader.py in GNU nano.
  • Place the cursor after self.last_seen = 0.0.
  • Paste the polling method below with its indentation preserved:
    def poll(self):
        status, _ = self.reader.MFRC522_Request(self.reader.PICC_REQIDL)
        if status != self.reader.MI_OK:
            return None

        status, uid = self.reader.MFRC522_Anticoll()
        if status != self.reader.MI_OK:
            return None

        uid_bytes = bytes(uid[:4])
        raw_uid = ":".join(f"{value:02X}" for value in uid_bytes)
        now = time.monotonic()
        if raw_uid == self.last_uid and now - self.last_seen < 3:
            return None

        self.last_uid = raw_uid
        self.last_seen = now
        return {
            "localUid": raw_uid,
            "tagHash": hashlib.sha256(uid_bytes).hexdigest()[:12],
        }

How does tag polling work?

  • The request checks whether a compatible RFID tag is present.
  • The anticollision call reads the UID when a tag responds successfully.
  • The first four UID bytes become the local colon-separated identifier.
  • The SHA-256 digest produces a 12-character tagHash for equipment events.
  • Save equipment_reader.py in GNU nano.
  • Confirm def poll(self): remains indented inside EquipmentReader.

Does the polling method look misaligned?

The poll() definition needs four leading spaces. Every line inside the method needs another indentation level.

Need help checking the structure? Help me fix the indentation in my EquipmentReader class.

The final block adds hardware cleanup plus the loop that prints each accepted scan. Cleanup releases both the SPI connection and reset pin when the process stops.

  • Place the cursor after the closing brace of the dictionary returned by poll().
  • Paste the cleanup method plus the script runner below:
    def close(self):
        self.reader.spi.close()
        self.reset.close()


if __name__ == "__main__":
    equipment = EquipmentReader()
    print("Scan a tag. Press Ctrl+C to stop.")
    try:
        while True:
            scan = equipment.poll()
            if scan:
                print(json.dumps(scan))
            time.sleep(0.2)
    except KeyboardInterrupt:
        pass
    finally:
        equipment.close()

What completes the reader?

  • The loop polls the reader every 0.2 seconds.
  • A successful scan is printed as JSON.
  • The finally block releases the reader even when you stop the script from the keyboard.
  • Save equipment_reader.py in GNU nano.
  • Return to the SSH prompt.
Scan a tag and expose the local limit

Before you run the reader, make a quick prediction about where the first equipment event will appear.

  • Start the equipment reader by running this command:
sudo .venv/bin/python equipment_reader.py

What does this command do?

The command runs the reader through the prepared virtual environment with hardware permissions. The process stays active while it waits for tag scans.

  • Present an FNK0066 RFID tag to the MFRC522 reader.
  • Remove the tag from the reader.
  • Wait at least three seconds.
  • Present the same tag again.
  • Stop the reader with Ctrl+C after the second result.

You should see one JSON line for each separate scan. Each line contains localUid plus a 12-character tagHash.

Tag scan not appearing?

Confirm the MFRC522 receives 3.3 V power. Check that SDA/SS reaches physical pin 24 plus RST reaches physical pin 22.

Hold the tag flat against the reader for a moment. Remove it fully before the next attempt.

Still missing the scan? Help me troubleshoot my MFRC522 reader on Raspberry Pi 5.

✔️ Awesome, I've got everything!

Your reader code is complete. Make sure equipment_reader.py is saved before you continue.

ⓧ I'd like to double check the full code

import hashlib
import json
import time

from gpiozero import OutputDevice
import MFRC522


class EquipmentReader:
    def __init__(self):
        self.reset = OutputDevice(25, active_high=True, initial_value=True)
        self.reset.off()
        time.sleep(0.05)
        self.reset.on()
        time.sleep(0.05)
        self.reader = MFRC522.MFRC522(bus=0, dev=0, spd=1000000)
        self.last_uid = None
        self.last_seen = 0.0

    def poll(self):
        status, _ = self.reader.MFRC522_Request(self.reader.PICC_REQIDL)
        if status != self.reader.MI_OK:
            return None

        status, uid = self.reader.MFRC522_Anticoll()
        if status != self.reader.MI_OK:
            return None

        uid_bytes = bytes(uid[:4])
        raw_uid = ":".join(f"{value:02X}" for value in uid_bytes)
        now = time.monotonic()
        if raw_uid == self.last_uid and now - self.last_seen < 3:
            return None

        self.last_uid = raw_uid
        self.last_seen = now
        return {
            "localUid": raw_uid,
            "tagHash": hashlib.sha256(uid_bytes).hexdigest()[:12],
        }

    def close(self):
        self.reader.spi.close()
        self.reset.close()


if __name__ == "__main__":
    equipment = EquipmentReader()
    print("Scan a tag. Press Ctrl+C to stop.")
    try:
        while True:
            scan = equipment.poll()
            if scan:
                print(json.dumps(scan))
            time.sleep(0.2)
    except KeyboardInterrupt:
        pass
    finally:
        equipment.close()

How to use this reference

Compare this file with your saved copy. Match the imports, indentation, method names, values, and script runner exactly.

That is a solid hardware win. Your Pi now distinguishes separate equipment scans while preparing a shorter identifier for later events.

The event still exists only in the active SSH terminal. This intended shortfall shows why the station needs a central path for telemetry.

Your two hardware adapters now produce local readings. Next, you will create a secure AWS IoT Core path that carries a test event beyond the SSH session.

Create the Secure AWS IoT Path

Your Raspberry Pi now reads the room environment and detects equipment tags. Those results still disappear inside the local SSH session when nobody is watching.

This step gives the station a managed message broker through AWS IoT Core. The Pi uses an X.509 certificate to authenticate before publishing over MQTT5.

In this step, get ready to:
  • Create a least-privilege policy for the station.
  • Register the Pi as an AWS IoT thing with its own certificate.
  • Publish a test event that appears in the MQTT test client.
Create the least-privilege policy

An AWS IoT policy decides which broker actions a certificate can authorize. This least-privilege policy permits this station to connect as its attached thing and publish to two specific topics.

Check your AWS costs

Creating an AWS account can require payment information. Account verification can also introduce a delay.

Eligible accounts receive an AWS IoT Core free tier for the first 12 months. It includes 2,250,000 connection minutes and 500,000 messages.

Usage becomes billable after the applicable allowance or credits expire. This project creates no storage rule or downstream database.

  • Sign in to the AWS Management Console in your browser.
  • Complete the AWS account creation flow if you do not have an account.
  • Choose one AWS Region from the Region selector.
  • Record the selected Region here: your selected AWS Region.
  • Open your AWS account menu in the upper-right corner.
  • Record the 12-digit account ID here: your 12-digit AWS account ID.
  • Use the console search bar to open AWS IoT Core.
  • Return to the existing SSH session.
  • Start GNU nano with the project policy filename by running:
nano iot-policy.json

What does this command do?

Nano opens iot-policy.json inside ~/exam-room-station. Nano creates the file when it does not already exist.

  • Paste this policy document into iot-policy.json.
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iot:Connect",
      "Resource": "arn:aws:iot:REGION:ACCOUNT_ID:client/${iot:Connection.Thing.ThingName}",
      "Condition": {
        "Bool": {
          "iot:Connection.Thing.IsAttached": "true"
        }
      }
    },
    {
      "Effect": "Allow",
      "Action": "iot:Publish",
      "Resource": [
        "arn:aws:iot:REGION:ACCOUNT_ID:topic/exam-room/${iot:Connection.Thing.ThingName}/environment",
        "arn:aws:iot:REGION:ACCOUNT_ID:topic/exam-room/${iot:Connection.Thing.ThingName}/equipment"
      ]
    }
  ]
}

What does this policy allow?

  • The iot:Connect statement permits a client ID derived from the attached thing name.
  • The attachment condition prevents an unrelated certificate from claiming that identity.
  • The iot:Publish statement limits publication to the environment topic and the equipment topic.
  • The thing-name policy variable keeps the same permission structure tied to the registered device.
  • Replace every REGION placeholder with your selected AWS Region.
  • Replace every ACCOUNT_ID placeholder with your 12-digit AWS account ID.

The client ARN and both topic ARNs now point to resources in your selected AWS account.

  • Save iot-policy.json in Nano.

✔️ Awesome, I've got everything!

Your policy file is saved with your AWS Region and account ID in place of the two template placeholders.

ⓧ I'd like to double check the full code

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "iot:Connect",
      "Resource": "arn:aws:iot:REGION:ACCOUNT_ID:client/${iot:Connection.Thing.ThingName}",
      "Condition": {
        "Bool": {
          "iot:Connection.Thing.IsAttached": "true"
        }
      }
    },
    {
      "Effect": "Allow",
      "Action": "iot:Publish",
      "Resource": [
        "arn:aws:iot:REGION:ACCOUNT_ID:topic/exam-room/${iot:Connection.Thing.ThingName}/environment",
        "arn:aws:iot:REGION:ACCOUNT_ID:topic/exam-room/${iot:Connection.Thing.ThingName}/equipment"
      ]
    }
  ]
}
  • Return to AWS IoT Core in your browser.
  • Choose Security from the left navigation pane.
  • Choose Policies.
  • Choose Create policy.
  • Enter ExamRoomStationPolicy in the Policy name field.
  • Paste the completed contents of iot-policy.json into the Policy document editor.
  • Choose Create.

The Policies page now lists ExamRoomStationPolicy. Your station has a permission boundary ready for its device identity.

Policy rejected?

Check that every REGION placeholder uses the selected Region. Check that every ACCOUNT_ID placeholder contains the 12-digit account ID.

Confirm that the policy document contains matching quotation marks and braces.

Still stuck? Help me troubleshoot my AWS IoT policy document.

Register the thing and protect its certificate

An AWS IoT thing represents the physical station in the device registry. Its certificate proves that the connecting Pi holds the matching private key.

  • Choose All devices from the AWS IoT Core navigation pane.
  • Choose Things.
  • Choose Create things.
  • Select Create a single thing.
  • Choose Next.
  • Enter exam-room-station-01 in the Thing name field.
  • Choose Next.
  • Select Auto-generate a new certificate (recommended).

Download the private key now

AWS shows the generated private key only during this creation flow. Store it securely because the device cannot authenticate without it.

Keep the private key out of screenshots and shared folders. You will restrict the copy on the Pi to its owner.

  • Choose Next.
  • Select the checkbox beside ExamRoomStationPolicy.
  • Choose Create thing.
  • Download the Device certificate.
  • Download the Private key file.
  • Download Amazon Root CA 1.
  • Rename the downloaded device certificate to device.pem.crt.
  • Rename the downloaded private key to private.pem.key.
  • Rename the downloaded root certificate to AmazonRootCA1.pem.
  • Choose Done.

The Things page now lists exam-room-station-01 with an active certificate and the attached policy. That device identity is ready for mutual TLS.

  • Return to the existing SSH session.
  • Create the certificate directory inside ~/exam-room-station by running:
mkdir -p certs

What does this command do?

The command creates the certs directory when it is missing. The existing directory remains unchanged when it already exists.

  • Open a second Windows PowerShell window.
  • Copy the three downloaded certificate files to the Pi by running:
scp "$HOME\Downloads\device.pem.crt" "$HOME\Downloads\private.pem.key" "$HOME\Downloads\AmazonRootCA1.pem" [[PI_SSH_USER="your Pi SSH username"]]@[[PI_SSH_HOST="your Pi hostname or IP"]]:exam-room-station/certs/

What does this transfer do?

The scp command transfers the three local files through the existing SSH service. The destination places them inside ~/exam-room-station/certs on the Pi.

  • Return to the existing SSH session.
  • Restrict the private key to its owner by running:
chmod 600 certs/private.pem.key
ls -l certs/private.pem.key

What should I see?

The first command grants read and write access only to the file owner. The second command displays the resulting permissions for certs/private.pem.key.

You should see owner read and write permissions with no permissions granted to the group or other users.

Certificate transfer failed?

Check that all three renamed files are in your Windows Downloads folder. Confirm that the Pi username and host match the values used for your working SSH connection.

If the destination is missing, confirm that certs exists inside ~/exam-room-station.

Need help? Help me transfer my AWS IoT certificate files to my Raspberry Pi.

Build and test the MQTT5 publisher

The device identity can now authenticate with AWS IoT Core. A small publisher turns that secure connection into a visible cloud event.

Why use the device SDK directly?

The AWS IoT Device SDK for Python v2 exposes the certificate connection and publication acknowledgement in a compact client. This keeps the focus on device identity and message delivery.

AWS IoT Greengrass adds component deployments and a local runtime. Those capabilities add setup without improving this station's core telemetry path.

  • Return to AWS IoT Core in your browser.
  • Choose Domain configurations from the navigation pane.
  • Copy the domain name for your selected Region.
  • Record the endpoint here: your AWS IoT endpoint.
  • Open the MQTT test client from the AWS IoT Core navigation pane.
  • Choose the option to subscribe to a topic.
  • Enter exam-room/exam-room-station-01/# as the topic filter.
  • Choose Subscribe.

The MQTT test client is now waiting for either station topic. The wildcard captures both environment and equipment messages.

  • Return to the existing SSH session.
  • Use Nano to create cloud_client.py inside ~/exam-room-station.
  • Paste the connection setup and lifecycle callbacks below:
import json
import threading

from awscrt import mqtt5
from awsiot import mqtt5_client_builder


class CloudPublisher:
    def __init__(self, endpoint, certificate, private_key, root_ca, client_id):
        self.connected = threading.Event()
        self.stopped = threading.Event()
        self.client = mqtt5_client_builder.mtls_from_path(
            endpoint=endpoint,
            cert_filepath=certificate,
            pri_key_filepath=private_key,
            ca_filepath=root_ca,
            client_id=client_id,
            on_lifecycle_connection_success=self._on_connection_success,
            on_lifecycle_connection_failure=self._on_connection_failure,
            on_lifecycle_stopped=self._on_stopped,
        )

    def _on_connection_success(self, lifecycle_data):
        print(f"MQTT5 connected: {lifecycle_data.connack_packet.reason_code!r}")
        self.connected.set()

    def _on_connection_failure(self, lifecycle_data):
        print(f"MQTT5 connection failed: {lifecycle_data.exception}")

    def _on_stopped(self, _lifecycle_data):
        self.stopped.set()

What does this code do?

  • The two event objects track successful connection and complete shutdown states.
  • The mutual TLS builder receives the endpoint and three certificate paths.
  • The client ID identifies the station during the MQTT5 connection.
  • The lifecycle callbacks print connection failures and release waiting code after success.
  • Save cloud_client.py.
  • Check that the first half imports successfully by running:
.venv/bin/python -c "from cloud_client import CloudPublisher; print('CloudPublisher loaded')"

What should I see?

You should see CloudPublisher loaded. This confirms that Python can import the SDK modules and parse the class definition.

CloudPublisher import failed?

Confirm that cloud_client.py is saved inside ~/exam-room-station.

Check the indentation inside CloudPublisher against the code block.

Still stuck? Help me fix the first half of cloud_client.py.

  • Return to cloud_client.py in Nano.
  • Add the connection and publication methods below _on_stopped().
    def start(self):
        self.client.start()
        if not self.connected.wait(20):
            raise TimeoutError("AWS IoT Core connection timed out")

    def publish(self, topic, payload):
        message = json.dumps(payload, separators=(",", ":"))
        future = self.client.publish(
            mqtt5.PublishPacket(
                topic=topic,
                payload=message,
                qos=mqtt5.QoS.AT_LEAST_ONCE,
                payload_format_indicator=mqtt5.PayloadFormatIndicator.AWS_MQTT5_PFI_UTF8,
                content_type="application/json",
            )
        )
        result = future.result(10)
        print(f"PubAck for {topic}: {result.puback.reason_code!r}")

    def stop(self):
        self.client.stop()
        self.stopped.wait(10)

How does publishing work?

  • The start() method waits up to 20 seconds for a successful connection callback.
  • The publish() method converts the payload into compact JSON.
  • The publish packet uses QoS 1 through AT_LEAST_ONCE.
  • The returned future waits for a broker acknowledgement before printing PubAck.
  • The stop() method requests a clean disconnect.
  • Save cloud_client.py.
  • Check the complete class import by running:
.venv/bin/python -c "from cloud_client import CloudPublisher; print('CloudPublisher loaded')"

What should I see now?

You should see CloudPublisher loaded again. This run parses the connection and publication methods you just added.

Seeing a syntax error?

Make sure start(), publish() and stop() align with the other methods inside CloudPublisher.

Confirm that every opening parenthesis has a matching closing parenthesis.

Need help? Help me debug the completed CloudPublisher class.

✔️ Awesome, I've got everything!

Your complete cloud_client.py file is saved and imports successfully.

ⓧ I'd like to double check the full code

import json
import threading

from awscrt import mqtt5
from awsiot import mqtt5_client_builder


class CloudPublisher:
    def __init__(self, endpoint, certificate, private_key, root_ca, client_id):
        self.connected = threading.Event()
        self.stopped = threading.Event()
        self.client = mqtt5_client_builder.mtls_from_path(
            endpoint=endpoint,
            cert_filepath=certificate,
            pri_key_filepath=private_key,
            ca_filepath=root_ca,
            client_id=client_id,
            on_lifecycle_connection_success=self._on_connection_success,
            on_lifecycle_connection_failure=self._on_connection_failure,
            on_lifecycle_stopped=self._on_stopped,
        )

    def _on_connection_success(self, lifecycle_data):
        print(f"MQTT5 connected: {lifecycle_data.connack_packet.reason_code!r}")
        self.connected.set()

    def _on_connection_failure(self, lifecycle_data):
        print(f"MQTT5 connection failed: {lifecycle_data.exception}")

    def _on_stopped(self, _lifecycle_data):
        self.stopped.set()

    def start(self):
        self.client.start()
        if not self.connected.wait(20):
            raise TimeoutError("AWS IoT Core connection timed out")

    def publish(self, topic, payload):
        message = json.dumps(payload, separators=(",", ":"))
        future = self.client.publish(
            mqtt5.PublishPacket(
                topic=topic,
                payload=message,
                qos=mqtt5.QoS.AT_LEAST_ONCE,
                payload_format_indicator=mqtt5.PayloadFormatIndicator.AWS_MQTT5_PFI_UTF8,
                content_type="application/json",
            )
        )
        result = future.result(10)
        print(f"PubAck for {topic}: {result.puback.reason_code!r}")

    def stop(self):
        self.client.stop()
        self.stopped.wait(10)
  • Use Nano to create cloud_test.py inside ~/exam-room-station.
  • Paste the argument parser below:
import argparse
from datetime import datetime, timezone

from cloud_client import CloudPublisher

THING_NAME = "exam-room-station-01"
TOPIC = f"exam-room/{THING_NAME}/environment"


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("--endpoint", required=True)
    parser.add_argument("--cert", required=True)
    parser.add_argument("--key", required=True)
    parser.add_argument("--ca", required=True)
    return parser.parse_args()

What does this code prepare?

  • The thing name matches the AWS IoT registry entry and the future MQTT client ID.
  • The topic resolves to the station's environment path permitted by the policy.
  • The four required arguments receive the endpoint and certificate paths at runtime.
  • Save cloud_test.py.
  • Check the argument parser by running:
.venv/bin/python cloud_test.py --help

What should I see?

You should see help output listing the endpoint and three certificate path arguments. This confirms that Python can import CloudPublisher and parse the test script.

Help output missing?

Confirm that cloud_test.py is saved beside cloud_client.py.

Check the four argument names against the code block.

Still stuck? Help me debug the cloud_test.py argument parser.

  • Return to cloud_test.py in Nano.
  • Add the connection test below parse_args().
def main():
    args = parse_args()
    cloud = CloudPublisher(
        args.endpoint,
        args.cert,
        args.key,
        args.ca,
        THING_NAME,
    )
    cloud.start()
    try:
        cloud.publish(
            TOPIC,
            {
                "deviceId": THING_NAME,
                "eventType": "connection_test",
                "timestamp": datetime.now(timezone.utc).isoformat(),
            },
        )
    finally:
        cloud.stop()


if __name__ == "__main__":
    main()

How does the connection test work?

  • The test passes the endpoint and certificate paths into CloudPublisher.
  • The certificate-authenticated client connects with exam-room-station-01 as its client ID.
  • The payload contains the device ID and a UTC timestamp.
  • The finally block stops the client after success or failure.
  • Save cloud_test.py.

✔️ Awesome, I've got everything!

Your connection test is ready to authenticate and publish one timestamped event.

ⓧ I'd like to double check the full code

import argparse
from datetime import datetime, timezone

from cloud_client import CloudPublisher

THING_NAME = "exam-room-station-01"
TOPIC = f"exam-room/{THING_NAME}/environment"


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("--endpoint", required=True)
    parser.add_argument("--cert", required=True)
    parser.add_argument("--key", required=True)
    parser.add_argument("--ca", required=True)
    return parser.parse_args()


def main():
    args = parse_args()
    cloud = CloudPublisher(
        args.endpoint,
        args.cert,
        args.key,
        args.ca,
        THING_NAME,
    )
    cloud.start()
    try:
        cloud.publish(
            TOPIC,
            {
                "deviceId": THING_NAME,
                "eventType": "connection_test",
                "timestamp": datetime.now(timezone.utc).isoformat(),
            },
        )
    finally:
        cloud.stop()


if __name__ == "__main__":
    main()
  • Replace YOUR_ENDPOINT in the command below with your AWS IoT endpoint.

Before you run the connection test, do you expect AWS IoT Core to accept this certificate under the topic policy?

  • Test the certificate-authenticated publication by running this command:
sudo .venv/bin/python cloud_test.py --endpoint YOUR_ENDPOINT --cert certs/device.pem.crt --key certs/private.pem.key --ca certs/AmazonRootCA1.pem

What should I see?

The terminal first reports MQTT5 connected. It then prints PubAck for the environment topic.

The MQTT test client displays a JSON payload with connection_test as the event type. That message proves the certificate authenticated and the policy authorized the publication.

Connection or publication failed?

Confirm that the endpoint comes from the same Region as exam-room-station-01.

Check that the certificate is active and attached to the thing. Confirm that ExamRoomStationPolicy is attached to the certificate.

A newly changed policy can take six to eight minutes to propagate. If every value matches, wait briefly before retrying.

Need help? Help me troubleshoot my AWS IoT MQTT5 connection.

Your station now has a certificate-authenticated cloud path with a visible QoS 1 acknowledgement. Next, you will connect both hardware adapters to this publisher and watch the live station feed.

Publish the Live Station Feed

Your certificate-authenticated MQTT5 path already reaches AWS IoT Core. The test message proved that your device identity can connect through mutual TLS.

Now your verified hardware adapters need to share that path. Combining both adapters with the cloud publisher turns separate terminal tests into one end-to-end exam-room station.

In this step, get ready to:
  • Create the station program with the device identity and two MQTT topics.
  • Publish an environment event every ten seconds.
  • Publish alternating equipment movement events after separate RFID scans.
Create the station entry point

The station uses exam-room-station-01 as its device ID and MQTT client ID. Its two topic constants match the environment and equipment paths allowed by your policy.

  • In the SSH terminal from earlier, use Nano to create exam_room_station.py inside ~/exam-room-station.
  • Build the station entry point by adding this first section:
import argparse
import time
from datetime import datetime, timezone

from cloud_client import CloudPublisher
from environment_sensor import EnvironmentSensor
from equipment_reader import EquipmentReader

THING_NAME = "exam-room-station-01"
ENVIRONMENT_TOPIC = f"exam-room/{THING_NAME}/environment"
EQUIPMENT_TOPIC = f"exam-room/{THING_NAME}/equipment"


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("--endpoint", required=True)
    parser.add_argument("--cert", required=True)
    parser.add_argument("--key", required=True)
    parser.add_argument("--ca", required=True)
    return parser.parse_args()


def timestamp():
    return datetime.now(timezone.utc).isoformat()


def main():
    args = parse_args()


if __name__ == "__main__":
    main()

What does this section prepare?

  • The imports reuse your verified environment sensor, equipment reader, and cloud publisher.
  • The topic constants keep both event streams under the registered thing name.
  • The argument parser accepts the endpoint and mutual TLS certificate paths used by your earlier connection test.
  • The timestamp() helper creates a UTC timestamp for every event.
  • Keep exam_room_station.py open in Nano.
  • Place your cursor directly below args = parse_args().
  • Add the verified hardware adapters and cloud connection by pasting this section:
    environment = EnvironmentSensor()
    equipment = EquipmentReader()
    cloud = CloudPublisher(
        args.endpoint,
        args.cert,
        args.key,
        args.ca,
        THING_NAME,
    )
    equipment_states = {}
    last_environment_publish = 0.0

    cloud.start()
    print("Station running. Press Ctrl+C to stop.")
    equipment.close()
    cloud.stop()

How does the station start?

  • The two hardware classes reuse the local adapters you tested earlier.
  • The CloudPublisher receives the thing name as its MQTT client ID.
  • The equipment state dictionary stores each tag's latest movement state.
  • The environment timer starts at zero so the first reading publishes immediately.
  • Save exam_room_station.py.
  • Exit Nano.
  • Use your known AWS IoT endpoint in place of YOUR_ENDPOINT when you run this connection check:
sudo .venv/bin/python exam_room_station.py --endpoint YOUR_ENDPOINT --cert certs/device.pem.crt --key certs/private.pem.key --ca certs/AmazonRootCA1.pem

What should you see?

You should see a successful MQTT5 connection followed by the station start message. The program then closes both clients and returns control to your terminal.

Station does not connect?

Use the same endpoint and certificate paths that worked with cloud_test.py.

Confirm the DHT11 and MFRC522 wiring still matches the local tests from earlier steps.

Need help isolating the failure? Help me troubleshoot why exam_room_station.py cannot start its hardware adapters or MQTT5 connection.

Publish environment readings

A ten-second interval gives the room sensor a steady reporting rhythm. Each event combines the device identity, event type, UTC timestamp, temperature, and humidity before CloudPublisher.publish() sends it with QoS 1.

  • Return to exam_room_station.py in Nano.
  • Insert this loop directly below print("Station running. Press Ctrl+C to stop.") and above equipment.close():
    try:
        while True:
            now = time.monotonic()
            if now - last_environment_publish >= 10:
                try:
                    reading = environment.read()
                    event = {
                        "deviceId": THING_NAME,
                        "eventType": "environment",
                        "timestamp": timestamp(),
                        **reading,
                    }
                    print(event)
                    cloud.publish(ENVIRONMENT_TOPIC, event)
                except RuntimeError as error:
                    print(f"Environment read skipped: {error}")
                last_environment_publish = now

            time.sleep(0.2)
    except KeyboardInterrupt:
        print("Stopping station")
    finally:

How does the environment loop work?

  • The monotonic timer triggers the first environment reading immediately.
  • The timer permits another environment reading after ten seconds.
  • The dictionary expansion adds temperatureC and humidityPercent from EnvironmentSensor.read().
  • The runtime error handler skips a transient invalid DHT11 reading while keeping the station active.
  • The cleanup path closes the hardware reader and MQTT5 client when you stop the station.
  • Save exam_room_station.py.
  • Run the environment publishing loop with your endpoint in place of YOUR_ENDPOINT using this command:
sudo .venv/bin/python exam_room_station.py --endpoint YOUR_ENDPOINT --cert certs/device.pem.crt --key certs/private.pem.key --ca certs/AmazonRootCA1.pem

What should you see?

You should see a local environment dictionary followed by PubAck output for the environment topic. A matching JSON message should arrive in the subscribed MQTT test client.

  • Wait for a second environment event to confirm the ten-second publishing rhythm.
  • Stop the station with Ctrl+C.

Your local DHT11 readings now reach the cloud as timestamped environment events. Each PubAck confirms that AWS IoT Core acknowledged the corresponding QoS 1 publication.

No environment event in the cloud?

Confirm the MQTT test client still subscribes to exam-room/exam-room-station-01/#.

If the terminal reports skipped sensor reads repeatedly, check the DHT11 data connection on GPIO17 and the 10 kΩ pull-up arrangement.

Still stuck? Help me trace a local DHT11 reading that does not appear in the AWS IoT MQTT test client.

Add equipment movement events

Each accepted scan uses the twelve-character tag hash from your local reader. The station stores the latest state for that hash so separate scans alternate between checked_out and checked_in.

  • Return to exam_room_station.py in Nano.
  • Find the existing time.sleep(0.2) line inside the while True: loop.
  • Insert the equipment publishing section directly above that line:
            scan = equipment.poll()
            if scan:
                tag_hash = scan["tagHash"]
                previous_state = equipment_states.get(tag_hash, "checked_in")
                next_state = "checked_out" if previous_state == "checked_in" else "checked_in"
                equipment_states[tag_hash] = next_state
                event = {
                    "deviceId": THING_NAME,
                    "eventType": "equipment_movement",
                    "timestamp": timestamp(),
                    "tagHash": tag_hash,
                    "state": next_state,
                }
                print(event)
                cloud.publish(EQUIPMENT_TOPIC, event)

How are movement events created?

  • The reader returns a scan only when it detects a tag outside its duplicate-suppression window.
  • Only the derived tagHash enters the cloud event.
  • The equipment state dictionary remembers the last state for each hash.
  • The equipment event goes to the dedicated equipment topic through the same QoS 1 publisher.
  • Save exam_room_station.py.

✔️ Awesome, I've got everything!

Your station file now contains both publishing paths. Keep it saved before the end-to-end test.

ⓧ I'd like to double check the full code

import argparse
import time
from datetime import datetime, timezone

from cloud_client import CloudPublisher
from environment_sensor import EnvironmentSensor
from equipment_reader import EquipmentReader

THING_NAME = "exam-room-station-01"
ENVIRONMENT_TOPIC = f"exam-room/{THING_NAME}/environment"
EQUIPMENT_TOPIC = f"exam-room/{THING_NAME}/equipment"


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument("--endpoint", required=True)
    parser.add_argument("--cert", required=True)
    parser.add_argument("--key", required=True)
    parser.add_argument("--ca", required=True)
    return parser.parse_args()


def timestamp():
    return datetime.now(timezone.utc).isoformat()


def main():
    args = parse_args()
    environment = EnvironmentSensor()
    equipment = EquipmentReader()
    cloud = CloudPublisher(
        args.endpoint,
        args.cert,
        args.key,
        args.ca,
        THING_NAME,
    )
    equipment_states = {}
    last_environment_publish = 0.0

    cloud.start()
    print("Station running. Press Ctrl+C to stop.")
    try:
        while True:
            now = time.monotonic()
            if now - last_environment_publish >= 10:
                try:
                    reading = environment.read()
                    event = {
                        "deviceId": THING_NAME,
                        "eventType": "environment",
                        "timestamp": timestamp(),
                        **reading,
                    }
                    print(event)
                    cloud.publish(ENVIRONMENT_TOPIC, event)
                except RuntimeError as error:
                    print(f"Environment read skipped: {error}")
                last_environment_publish = now

            scan = equipment.poll()
            if scan:
                tag_hash = scan["tagHash"]
                previous_state = equipment_states.get(tag_hash, "checked_in")
                next_state = "checked_out" if previous_state == "checked_in" else "checked_in"
                equipment_states[tag_hash] = next_state
                event = {
                    "deviceId": THING_NAME,
                    "eventType": "equipment_movement",
                    "timestamp": timestamp(),
                    "tagHash": tag_hash,
                    "state": next_state,
                }
                print(event)
                cloud.publish(EQUIPMENT_TOPIC, event)

            time.sleep(0.2)
    except KeyboardInterrupt:
        print("Stopping station")
    finally:
        equipment.close()
        cloud.stop()


if __name__ == "__main__":
    main()

Before you run the complete station, what do you expect to stay the same between two scans of one tag? What do you expect to alternate?

  • Switch back to the AWS IoT MQTT test client from earlier.
  • Confirm the subscription remains set to exam-room/exam-room-station-01/#.
  • Start the complete station with your endpoint in place of YOUR_ENDPOINT by running this command:
sudo .venv/bin/python exam_room_station.py --endpoint YOUR_ENDPOINT --cert certs/device.pem.crt --key certs/private.pem.key --ca certs/AmazonRootCA1.pem

What does this command start?

The command starts both hardware adapters under one certificate-authenticated MQTT5 connection. It keeps running until you stop it with Ctrl+C.

  • Wait for an environment event to appear in the MQTT test client.
  • Present the same FNK0066 RFID tag to the MFRC522 reader.
  • Remove the tag from the reader.
  • Present the same tag again as a separate scan.
  • Stop the station with Ctrl+C after both equipment events arrive.

You should see environment JSON arrive about every ten seconds. Each publication has matching PubAck output in the terminal.

You should also see two equipment messages with the same tagHash. Their state values alternate from checked_out to checked_in.

Missing an equipment event?

Remove the tag fully between scans so the reader can detect a separate presentation.

Confirm the MQTT test client subscription includes the equipment topic under exam-room/exam-room-station-01/#.

Need help comparing the terminal and cloud streams? Help me troubleshoot missing RFID movement events or PubAck output from my exam-room station.

You have completed the end-to-end station. One Raspberry Pi now publishes live room conditions and privacy-conscious equipment movement events through its least-privilege device identity.

Secret mission

Admit Only Approved Equipment

Turn the station into a controlled equipment feed. Add a local allowlist that publishes approved equipment movements while rejecting unknown tags on the Raspberry Pi.

Clean Up Your Resources

Clean Up Your Resources

Your Raspberry Pi files have no ongoing cloud cost. Choose whether to keep them while accepting AWS IoT Core usage charges, pause the live connections, or delete the project.

Cost warning

Eligible AWS accounts can use AWS IoT Core within its 12-month free tier. Connectivity plus messages become billable after the allowance or credits expire.

Disconnect the station when you finish a demonstration to stop its active MQTT connectivity usage.

Resources you used:

  • AWS IoT thing exam-room-station-01.
  • Active X.509 certificate attached to the AWS IoT thing.
  • AWS IoT policy ExamRoomStationPolicy attached to the certificate.
  • Local ~/exam-room-station folder containing .venv, drivers, certificates, compiled library, configuration, plus source files.

Keep everything running

No action needed. Choose this if you are actively building or demonstrating the station.

  • Keep the station process open while you need live telemetry.
  • Keep the MQTT test client subscription active while you need to watch approved equipment events.
  • Keep exam-room-station-01, its active certificate, plus ExamRoomStationPolicy attached.
  • Keep ~/exam-room-station on the Pi for future demonstrations.

Pause - I'll come back to this later

Stop the live processes to end active connectivity usage. Your cloud identity plus local files remain ready for another session.

  • Return to the PowerShell SSH session running exam_room_station.py.
  • Press Ctrl+C to stop the station process.
  • Confirm that the terminal prints Stopping station.
  • Close the MQTT test client browser tab to end its connection.
  • Shut down the Raspberry Pi using its local power controls.
  • Leave exam-room-station-01, its certificate, plus ExamRoomStationPolicy attached.

Your station is now paused without losing its device identity. Start it again whenever you want to continue the demonstration.

Delete - I don't want to use this again

Deleting the station can feel drastic. The steps below remove only this project's cloud identity plus Pi files.

Live connections to close:

  • Close the MQTT test client browser tab to end its connection.
  • Return to your PowerShell SSH session.
  • Press Ctrl+C to stop exam_room_station.py.

Certificate cleanup order:

  • Select All devices in the AWS IoT Core left navigation.
  • Select Things.
  • Open exam-room-station-01.
  • Open the active certificate attached to the thing.
  • Detach ExamRoomStationPolicy from the certificate.
  • Detach the certificate from exam-room-station-01.
  • Deactivate the certificate.
  • Delete the certificate.

Registry cleanup order:

  • Return to All devices in the AWS IoT Core left navigation.
  • Select Things.
  • Delete exam-room-station-01.
  • Open the AWS IoT Core policy list from the left navigation.
  • Select ExamRoomStationPolicy.
  • Delete ExamRoomStationPolicy.

Local files to remove:

  • Copy any source files you want to keep to another location on the Pi.
  • Remove the local project folder by running this command in your SSH session:
rm -rf ~/exam-room-station

What does this command remove?

The command permanently deletes the ~/exam-room-station folder from the Raspberry Pi. Its contents include the virtual environment plus every project file.

The owner-restricted private key is removed with the folder.

  • Confirm that exam-room-station no longer appears under the Pi home folder in the Raspberry Pi file browser.

Folder still present?

  • Check that the command path matches ~/exam-room-station exactly.
  • Confirm that the station process has stopped before repeating the deletion command.

Help me remove the remaining project folder.

That closes the loop. The station's cloud identity plus local project files are now removed.

Nice Work!

Nice Work!

That’s the full exam-room telemetry station working! Your Raspberry Pi 5 now delivers privacy-conscious telemetry to AWS IoT Core through certificate-authenticated MQTT5.

You've learned how to:

  • Read live temperature and humidity with the DHT11. Detected equipment tags with the MFRC522 before converting local UIDs into 12-character tag hashes.
  • Created a least-privilege device identity with an AWS IoT thing. Bound its X.509 certificate to a policy that permits only the station's own connection and publish topics.
  • Published privacy-conscious JSON events through MQTT5 with QoS 1. Matched terminal PubAck output to live environment and equipment streams in the AWS IoT MQTT test client.
  • Secret Mission: Added an approved-equipment allowlist that rejects unknown tags locally. Published approved movement events with synthetic equipment labels.

Ready to quiz yourself?