sensors_channel.py

# Ce travail est publié sous licence MIT.
# Copyright (c) 2026 OpenMV LLC. Tous droits réservés.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Exemple de canal de capteurs (ToF + détection de visages)
#
# données de capteurs et cartes de profondeur ToF via des canaux de protocole CBOR.
# Utilise CBORChannel de la bibliothèque de protocole pour une configuration facile.

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

# Initialiser la caméra.
csi0 = csi.CSI()
csi0.reset()
csi0.pixformat(csi.RGB565)
csi0.framesize(csi.QVGA)

# Charger le modèle intégré de détection de visages.
model = ml.Model(
    "/rom/blazeface_front_128.tflite",
    postprocess=BlazeFace(threshold=0.5),
)
print(model)

# Initialiser le capteur ToF.
tof.init()
tof_w = tof.width()
tof_h = tof.height()

# Enregistrer le canal du capteur.
sensors = CBORChannel()
sensors.add("temp", type="label", unit="Cel")
sensors.add("humidity", type="label", unit="%RH")
sensors_ch = protocol.register(name="sensors", backend=sensors)

# Enregistrer le canal de profondeur.
depth = CBORChannel()
depth.add("depth", type="depth", width=tof_w, height=tof_h)
depth_ch = protocol.register(name="ToF depth", backend=depth)

# État simulé du capteur.
temp = 25.0
humidity = 50.0

while True:
    # Capturer une image et exécuter la détection de visage.
    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))

    # Simuler les lectures du capteur avec une dérive aléatoire.
    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))

    # Lire la carte de profondeur ToF.
    try:
        depth_data, dmin, dmax = tof.read_depth(vflip=True, hmirror=True)
    except RuntimeError:
        continue

    # Empaqueter les flottants de profondeur en binaire (4 octets par flottant).
    depth_bytes = struct.pack("<%df" % len(depth_data), *depth_data)

    # Mettre à jour les canaux.
    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))
    )