sensors_channel.py

# Questo lavoro è concesso in licenza MIT.
# Copyright (c) 2026 OpenMV LLC. Tutti i diritti riservati.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Esempio di canale sensori (ToF + rilevamento volti)
#
# dati dei sensori e mappe di profondità ToF su canali del protocollo CBOR.
# Usa CBORChannel dalla libreria del protocollo per una configurazione semplice.

import csi
import tof
import struct
import random
from protocol import CBORChannel
import protocol
import ml
from ml.postprocessing.mediapipe import BlazeFace

# Inizializzare la fotocamera.
csi0 = csi.CSI()
csi0.reset()
csi0.pixformat(csi.RGB565)
csi0.framesize(csi.QVGA)

# Caricare il modello di rilevamento volti integrato.
model = ml.Model(
    "/rom/blazeface_front_128.tflite",
    postprocess=BlazeFace(threshold=0.5),
)
print(model)

# Inizializzare il sensore ToF.
tof.init()
tof_w = tof.width()
tof_h = tof.height()

# Registrare il canale del sensore.
sensors = CBORChannel()
sensors.add("temp", type="label", unit="Cel")
sensors.add("humidity", type="label", unit="%RH")
sensors_ch = protocol.register(name="sensors", backend=sensors)

# Registrare il canale di profondità.
depth = CBORChannel()
depth.add("depth", type="depth", width=tof_w, height=tof_h)
depth_ch = protocol.register(name="ToF depth", backend=depth)

# Stato del sensore simulato.
temp = 25.0
humidity = 50.0

while True:
    # Catturare il fotogramma ed eseguire il rilevamento dei volti.
    img = csi0.snapshot()

    for r, score, keypoints in model.predict([img]):
        ml.utils.draw_predictions(img, [r], ("face",), ((0, 0, 255),), format=None)
        ml.utils.draw_keypoints(img, keypoints, color=(255, 0, 0))

    # Simulare le letture del sensore con una deriva casuale.
    temp += random.uniform(-0.3, 0.3)
    temp = max(15.0, min(35.0, temp))
    humidity += random.uniform(-0.5, 0.5)
    humidity = max(20.0, min(80.0, humidity))

    # Leggere la mappa di profondità ToF.
    try:
        depth_data, dmin, dmax = tof.read_depth(vflip=True, hmirror=True)
    except RuntimeError:
        continue

    # Impacchettare i valori float di profondità in binario (4 byte per float).
    depth_bytes = struct.pack("<%df" % len(depth_data), *depth_data)

    # Aggiornare i canali.
    sensors["temp"] = round(temp, 1)
    sensors["humidity"] = round(humidity, 1)
    depth["depth"] = depth_bytes

    depth_ch.send_event(0xFFFF)
    sensors_ch.send_event(0xFFFF)

    print(
        "temp=%.1f | humidity=%.1f | depth=%d bytes"
        % (temp, humidity, len(depth_bytes))
    )