Skip to content
Fondazione ITS INCOM

Examples · Internet of Things

MQTT topics for a smart classroom

A model designs the topics and the JSON payloads of a classroom with sensors and lights. Python checks the design with JSON Schema and a few broken messages, finds the holes, and three more sentences in the request close them.

Advanced · 60 min

The idea

With MQTT, devices do not talk to each other: they publish messages on topics, and whoever is interested subscribes. The topic names and the payloads are a contract between the devices and the dashboard. If the sensor writes temp and the dashboard reads temperature, nothing breaks loudly: the chart just stays empty.

A model is good at drafting the contract, and Python is good at checking it:

  1. the model designs the topics and, for each one, a JSON Schema of the payload, with an example;
  2. your code checks the design: the topic names, the retained flags, the schemas, and a few broken messages that a device with a bug really sends;
  3. you read what is left, and decide.

What you need

A key, as in Your first API call in Python, and two libraries:

pip install openai jsonschema

1. The design

Save it as design.py:

import json
import os

from openai import OpenAI

client = OpenAI(
    base_url="https://api.ai.itsincom.org/v1",
    api_key=os.environ["API_KEY"],
)

DEVICES = ["air_sensor", "presence_sensor", "light_1", "light_2", "window_contact"]

schema = {
    "type": "object",
    "properties": {
        "topics": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "topic": {"type": "string"},
                    "device": {"type": "string", "enum": DEVICES},
                    "kind": {"type": "string", "enum": ["reading", "state", "command", "online"]},
                    "publisher": {"type": "string"},
                    "subscribers": {"type": "array", "items": {"type": "string"}},
                    "qos": {"type": "integer", "enum": [0, 1, 2]},
                    "retained": {"type": "boolean"},
                    # {"type": "object"} with nothing else: any JSON object. Here the model writes
                    # a JSON Schema of its own, and an example message.
                    "payload_schema": {"type": "object"},
                    "example": {"type": "object"},
                },
                "required": ["topic", "device", "kind", "publisher", "subscribers", "qos", "retained",
                             "payload_schema", "example"],
            },
        },
    },
    "required": ["topics"],
}

answer = client.chat.completions.create(
    model="gemma-4-26b",
    temperature=0,
    messages=[
        {"role": "system", "content": "You design MQTT topics and JSON payloads for IoT devices. Topics are "
                                      "lowercase, with levels from the general to the specific, no leading slash "
                                      "and no spaces. Kinds of topic: reading (a measurement, sent again and "
                                      "again), state (what a device is doing now), command (an order to a "
                                      "device), online (whether a device is connected). Every payload is a JSON "
                                      "object, described by a JSON Schema (draft 2020-12), with an example "
                                      "message that follows it."},
        {"role": "user", "content": "A smart classroom, room 12 of a school. air_sensor: temperature in °C, "
                                    "humidity in %, CO2 in ppm, every minute. presence_sensor. light_1 and "
                                    "light_2: the dashboard switches them on and off, and they report their "
                                    "state. window_contact. Every device says whether it is online. Design the "
                                    "topics: who publishes, who subscribes, QoS, retained, the payload schema "
                                    "and an example."},
    ],
    response_format={"type": "json_schema", "json_schema": {"name": "mqtt_design", "schema": schema}},
)
# "length" means the answer was cut off: its JSON is not complete.
if answer.choices[0].finish_reason != "stop":
    raise SystemExit("The answer was cut off: ask for fewer devices.")
design = json.loads(answer.choices[0].message.content)
design["devices"] = DEVICES  # saved with the design, for the checks

with open("design.json", "w") as f:
    json.dump(design, f, indent=2)

for t in design["topics"]:
    print(f"{t['topic']:42} {t['kind']:8} QoS {t['qos']}  retained: {t['retained']}")
print("Saved in design.json")

Run it with python3 design.py. It takes about ten seconds: the design is long.

school/room12/air_sensor/online            online   QoS 1  retained: True
school/room12/air_sensor/reading           reading  QoS 0  retained: False
school/room12/presence_sensor/reading      reading  QoS 1  retained: True
school/room12/light_1/command              command  QoS 1  retained: False
school/room12/light_1/state                state    QoS 1  retained: True
school/room12/window_contact/state         state    QoS 1  retained: True
Saved in design.json

Two lines of the schema do the work:

Open design.json: it is the contract, and the next program reads it.

2. The checks

Save it as check_design.py:

import json

from jsonschema import Draft202012Validator
from jsonschema.exceptions import SchemaError


def problems(t):
    found = []
    topic = t["topic"]
    if topic != topic.lower() or topic.startswith("/") or " " in topic or "//" in topic:
        found.append("the topic breaks the naming rules")
    if "+" in topic or "#" in topic:
        found.append("wildcards are for subscribing, never in a topic you publish to")
    if "{" in topic or "<" in topic:
        found.append("a placeholder: the device needs the real name")
    if t["kind"] == "command" and t["retained"]:
        found.append("a retained command runs again on every device that reconnects")
    if t["kind"] in ("state", "online") and not t["retained"]:
        found.append("not retained: a dashboard that connects now sees nothing until the next change")

    # The schema written by the model must itself be a valid JSON Schema.
    try:
        Draft202012Validator.check_schema(t["payload_schema"])
    except SchemaError as e:
        return found + [f"the payload schema is not valid: {e.message}"]
    schema = Draft202012Validator(t["payload_schema"])
    example = t["example"]
    if not schema.is_valid(example):
        found.append("its own example does not pass")

    # Messages that a device with a bug really sends. A good schema refuses every one of them.
    if schema.is_valid({}):
        found.append("accepts an empty message")
    for field, value in example.items():
        # bool is a kind of int in Python: True must not count as a number here
        if isinstance(value, (int, float)) and not isinstance(value, bool):
            if schema.is_valid({**example, field: str(value)}):
                found.append(f"accepts {field} as text: {json.dumps(str(value))}")
    if schema.is_valid({**example, "temprature": 21.5}):
        found.append("accepts fields it does not know, such as a misspelt «temprature»")
    return found


with open("design.json") as f:
    design = json.load(f)

for t in design["topics"]:
    found = problems(t)
    print(f"{t['topic']}: {'OK' if not found else ''}")
    for p in found:
        print(f"    - {p}")

# Every device must say whether it is online: a dashboard cannot tell «off» from «broken» otherwise.
for device in design["devices"]:
    kinds = sorted({t["kind"] for t in design["topics"] if t["device"] == device})
    if "online" not in kinds:
        print(f"{device}: no online topic (it has: {', '.join(kinds) or 'nothing'})")

Run it with python3 check_design.py. No model here: the same file gives the same answer every time.

school/room12/air_sensor/online: 
    - accepts fields it does not know, such as a misspelt «temprature»
school/room12/air_sensor/reading: 
    - accepts fields it does not know, such as a misspelt «temprature»
school/room12/presence_sensor/reading: 
    - accepts fields it does not know, such as a misspelt «temprature»
school/room12/light_1/command: 
    - accepts fields it does not know, such as a misspelt «temprature»
school/room12/light_1/state: 
    - accepts fields it does not know, such as a misspelt «temprature»
school/room12/window_contact/state: 
    - accepts fields it does not know, such as a misspelt «temprature»
presence_sensor: no online topic (it has: reading)
light_1: no online topic (it has: command, state)
light_2: no online topic (it has: nothing)
window_contact: no online topic (it has: state)

Two holes:

The schemas did well on the rest: every one was valid, every example passed, and none of them accepted an empty message or a number written as text.

3. Three more sentences

Add them at the end of the user message in design.py, after “and an example.”:

Write every topic of every device in full, with its real name. Every schema lists its required fields and refuses fields it does not know. Every device has its own online topic, the lights too.

Run both programs again. Now the design has twelve topics, three for each light and two for every other device, and every line of the check says OK:

school/room12/air_sensor/online: OK
school/room12/air_sensor/reading: OK
school/room12/presence_sensor/online: OK
school/room12/presence_sensor/reading: OK
school/room12/light_1/command: OK
school/room12/light_1/state: OK
school/room12/light_1/online: OK
school/room12/light_2/command: OK
school/room12/light_2/state: OK
school/room12/light_2/online: OK
school/room12/window_contact/online: OK
school/room12/window_contact/state: OK

We ran it four times, and all twelve passed every time. The order of the topics can change from one run to the next, and so can a kind: the window contact was a state in three runs and a reading in one.

The third sentence comes from our first tries. With only the first two, all the schemas passed, but in two runs out of three the lights still had no online topic. The checks said so, and one more sentence closed the hole: that is what the checks are for.

What the checks cannot see

All OK means that the design passed your checks. Now read design.json yourself. In ours:

Good to know

Exercise

All examples