Skip to content
Fondazione ITS INCOM

Examples · Internet of Things

A weekly health check of your devices

The last heartbeat of twenty devices goes in: battery, signal, last seen, firmware. Your code finds the problems, a model turns them into a to-do list for the caretaker, and your code checks that no device is invented, missed or left for later.

Beginner · 25 min

The idea

Twenty sensors at the two sites of a school send a heartbeat every few minutes: “I am alive, my battery is at 64 %, I hear the Wi-Fi at −70 dBm, I run firmware 2.3.1”. Nobody reads twenty heartbeats every day. Once a week, someone should get a short list: what to fix, where, and what first.

The work is split into three parts:

  1. your code finds the facts: offline for more than 24 hours, battery under 20 %, weak signal, old firmware. These are rules, and a rule gives the same answer every week;
  2. a model turns the facts into a to-do list a person can follow, as JSON;
  3. your code checks the list: no device that was not flagged, no flagged device missed, and nothing urgent left for next month.

What you need

A key and the openai library, as in Your first API call in Python.

The heartbeats

The last heartbeat of every device, as heartbeats.csv. RSSI is the strength of the Wi-Fi signal in dBm: −60 is good; below about −85, the connection often drops. Before you run anything: which devices need a visit?

id,site,room,battery,rssi,last_seen,firmware
bus-co2-01,Busto Arsizio,Lab 1,87,-61,2026-10-09 07:55,2.3.1
bus-co2-02,Busto Arsizio,Lab 2,64,-70,2026-10-09 07:58,2.3.1
bus-co2-03,Busto Arsizio,Room 5,15,-72,2026-10-09 07:50,2.3.1
bus-co2-04,Busto Arsizio,Room 6,91,-88,2026-10-09 07:57,2.3.1
bus-co2-05,Busto Arsizio,Library,78,-66,2026-10-07 14:12,2.3.1
bus-tmp-01,Busto Arsizio,Gym,55,-79,2026-10-09 07:59,2.2.0
bus-tmp-02,Busto Arsizio,Server room,99,-58,2026-10-09 07:59,2.3.1
bus-door-01,Busto Arsizio,Main entrance,42,-74,2026-10-09 07:41,2.3.1
bus-door-02,Busto Arsizio,Back door,8,-91,2026-10-09 06:03,2.1.4
bus-leak-01,Busto Arsizio,Boiler room,73,-83,2026-10-09 07:30,2.3.1
leg-co2-01,Legnano,Lab A,82,-64,2026-10-09 07:56,2.3.1
leg-co2-02,Legnano,Lab B,19,-69,2026-10-09 07:52,2.3.1
leg-co2-03,Legnano,Room 12,70,-77,2026-10-09 07:58,2.3.0
leg-tmp-01,Legnano,Greenhouse,61,-86,2026-10-09 07:44,2.3.1
leg-soil-01,Legnano,Garden tomatoes,33,-80,2026-10-09 07:48,2.3.1
leg-soil-02,Legnano,Garden herbs,24,-84,2026-10-08 06:10,2.3.1
leg-soil-03,Legnano,Garden strawberries,0,-95,2026-10-02 18:20,2.2.0
leg-door-01,Legnano,Main entrance,58,-71,2026-10-09 07:59,2.3.1
leg-pwr-01,Legnano,Electrical room,100,-67,2026-10-09 07:59,2.3.1
leg-leak-01,Legnano,Toilets 1st floor,47,-75,2026-10-08 23:40,2.3.1

The program

Save it as health_check.py, next to the CSV:

import csv
import json
import os
import sys
from datetime import datetime

from openai import OpenAI

client = OpenAI(
    base_url="https://api.ai.itsincom.org/v1",
    api_key=os.environ["API_KEY"],
)
MODEL = sys.argv[1] if len(sys.argv) > 1 else "gemma-4-26b"

NOW = datetime(2026, 10, 9, 8, 0)  # in real life: datetime.now()
LATEST = "2.3.1"


def version(text):
    return tuple(int(part) for part in text.split("."))  # "2.10.0" > "2.9.0", as numbers


# 1. Your code finds the facts: the same rules, the same answer, every week
facts, urgent = {}, set()
with open("heartbeats.csv", newline="") as f:
    for d in csv.DictReader(f):
        problems = []
        hours = (NOW - datetime.fromisoformat(d["last_seen"])).total_seconds() / 3600
        if hours > 24:
            problems.append(f"offline for {hours:.0f} hours")
            urgent.add(d["id"])
        if int(d["battery"]) < 20:
            problems.append(f"battery {d['battery']} %")
            urgent.add(d["id"])
        if int(d["rssi"]) < -85:
            problems.append(f"weak signal {d['rssi']} dBm")
        if version(d["firmware"]) < version(LATEST):
            problems.append(f"firmware {d['firmware']}, latest is {LATEST}")
        if problems:
            facts[d["id"]] = f"{d['site']}, {d['room']}: " + "; ".join(problems)

print(f"{len(facts)} devices need attention:")
for device, text in facts.items():
    print(f"  {device:12} {text}")

# 2. The model turns the facts into a to-do list, and can name only the devices you flagged
schema = {
    "type": "object",
    "properties": {
        "tasks": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "priority": {"type": "string", "enum": ["today", "this week", "next month"]},
                    "devices": {"type": "array", "items": {"type": "string", "enum": list(facts)}},
                    "action": {"type": "string"},
                },
                "required": ["priority", "devices", "action"],
            },
        },
    },
    "required": ["tasks"],
}

answer = client.chat.completions.create(
    model=MODEL,
    temperature=0,
    max_tokens=4000,  # a stop, in case an answer runs away
    messages=[
        {"role": "system", "content": "You write the weekly to-do list for the caretaker of a school with two "
                                      "sites, Busto Arsizio and Legnano. Group the work into short tasks a person "
                                      "can do on one visit, and give each a priority. A device that is offline "
                                      "sends nothing: it comes first. A battery under 20 % will soon go offline. "
                                      "Firmware can wait for the next month."},
        {"role": "user", "content": "\n".join(f"{device}: {text}" for device, text in facts.items())},
    ],
    response_format={"type": "json_schema", "json_schema": {"name": "todo", "schema": schema}},
)
if answer.choices[0].finish_reason != "stop":
    sys.exit(f"The answer was cut off ({answer.usage.completion_tokens} tokens): run it again.")
tasks = json.loads(answer.choices[0].message.content)["tasks"]

print(f"\nTo do ({MODEL}):")
for priority in ["today", "this week", "next month"]:
    for task in tasks:
        if task["priority"] == priority:
            print(f"  [{priority}] {task['action']}  ({', '.join(task['devices'])})")

# 3. Your code checks the list: nothing invented, nothing missed, nothing urgent left for later
listed = {device for task in tasks for device in task["devices"]}
soon = {device for task in tasks if task["priority"] != "next month" for device in task["devices"]}
problems = [f"NOT FLAGGED: {device}" for device in sorted(listed - set(facts))]
for device, text in facts.items():
    if device not in listed:
        problems.append(f"MISSED:   {device:12} {text}")
    elif device in urgent and device not in soon:
        problems.append(f"TOO LATE: {device:12} {text}")
if problems:
    print("\n" + "\n".join(problems))
else:
    print(f"\nAll {len(facts)} flagged devices are in the list, and the urgent ones are not left for later.")

Run it with python3 health_check.py, or with another model: python3 health_check.py qwen3.6-35b.

What you should see

10 devices need attention:
  bus-co2-03   Busto Arsizio, Room 5: battery 15 %
  bus-co2-04   Busto Arsizio, Room 6: weak signal -88 dBm
  bus-co2-05   Busto Arsizio, Library: offline for 42 hours
  bus-tmp-01   Busto Arsizio, Gym: firmware 2.2.0, latest is 2.3.1
  bus-door-02  Busto Arsizio, Back door: battery 8 %; weak signal -91 dBm; firmware 2.1.4, latest is 2.3.1
  leg-co2-02   Legnano, Lab B: battery 19 %
  leg-co2-03   Legnano, Room 12: firmware 2.3.0, latest is 2.3.1
  leg-tmp-01   Legnano, Greenhouse: weak signal -86 dBm
  leg-soil-02  Legnano, Garden herbs: offline for 26 hours
  leg-soil-03  Legnano, Garden strawberries: offline for 158 hours; battery 0 %; weak signal -95 dBm; firmware 2.2.0, latest is 2.3.1

To do (gemma-4-26b):
  [today] Investigate and restore offline devices (Check connectivity/power).  (bus-co2-05, leg-soil-02, leg-soil-03)
  [next month] Replace batteries (Current levels < 20%).  (bus-co2-03, bus-door-02, leg-co2-02)
  [next month] Address weak signal/connectivity issues (Relocate or adjust antenna).  (bus-co2-04, bus-door-02, leg-tmp-01, leg-soil-03)
  [next month] Perform firmware updates.  (bus-tmp-01, bus-door-02, leg-co2-03, leg-soil-03)

TOO LATE: bus-co2-03   Busto Arsizio, Room 5: battery 15 %
TOO LATE: bus-door-02  Busto Arsizio, Back door: battery 8 %; weak signal -91 dBm; firmware 2.1.4, latest is 2.3.1
TOO LATE: leg-co2-02   Legnano, Lab B: battery 19 %

Ten devices out of twenty need something. Some are close calls on purpose: leg-soil-02 was last seen 26 hours ago, leg-leak-01 only 8; bus-leak-01 hears the Wi-Fi at −83 dBm, just above the limit. A rule decides them the same way every week.

What the checks found

The list looks reasonable at first sight: offline devices today, everything else grouped. But Gemma put the batteries at 8, 15 and 19 % into “next month”, in all seven of our runs. By next month those three devices will be offline too. The system message says that a battery under 20 % will soon go offline, but it does not say when to change it, and Gemma did not work it out.

The other models made other mistakes:

A model knows only what you tell it

Change the battery sentence to “A battery under 20 % will soon go offline: change it this week at the latest.” and run again. In our runs, three for each model:

A clearer prompt fixed one mistake and made room for another. That is why the checks stay in the program, whatever the prompt says.

Why the device ids are an enum

"enum": list(facts) builds the schema from the devices your code flagged. The model can name only those ten: not a healthy device, not a misspelled id, not “all the CO2 sensors”. So the NOT FLAGGED check never fired in our runs. Keep it anyway: it costs one line, and it protects you on the day someone removes the enum.

The checks count devices, not problems. bus-door-02 has three problems: a low battery, a weak signal and old firmware. A list that says only “replace the battery” passes. Qwen3.6 did exactly that with the signal of bus-door-02. A stricter check would look at every problem of every device.

Good to know

Exercise

All examples