sensors_channel.py

# Esta obra está licenciada bajo la licencia MIT.
# Copyright (c) 2026 OpenMV LLC. Todos los derechos reservados.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# Ejemplo de canal de sensores (ToF + detección de rostros)
#
# datos de sensores y mapas de profundidad ToF a través de canales de protocolo CBOR.
# Usa CBORChannel de la biblioteca de protocolo para una configuración sencilla.

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

# Inicializa la cámara.
csi0 = csi.CSI()
csi0.reset()
csi0.pixformat(csi.RGB565)
csi0.framesize(csi.QVGA)

# Carga el modelo integrado de detección de rostros.
model = ml.Model(
    "/rom/blazeface_front_128.tflite",
    postprocess=BlazeFace(threshold=0.5),
)
print(model)

# Inicializa el sensor ToF.
tof.init()
tof_w = tof.width()
tof_h = tof.height()

# Registra el canal de sensor.
sensors = CBORChannel()
sensors.add("temp", type="label", unit="Cel")
sensors.add("humidity", type="label", unit="%RH")
sensors_ch = protocol.register(name="sensors", backend=sensors)

# Registra el canal de profundidad.
depth = CBORChannel()
depth.add("depth", type="depth", width=tof_w, height=tof_h)
depth_ch = protocol.register(name="ToF depth", backend=depth)

# Estado simulado del sensor.
temp = 25.0
humidity = 50.0

while True:
    # Captura un fotograma y ejecuta la detección de rostros.
    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))

    # Simula lecturas del sensor con desviación aleatoria.
    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))

    # Lee el mapa de profundidad ToF.
    try:
        depth_data, dmin, dmax = tof.read_depth(vflip=True, hmirror=True)
    except RuntimeError:
        continue

    # Empaqueta los valores flotantes de profundidad en binario (4 bytes por flotante).
    depth_bytes = struct.pack("<%df" % len(depth_data), *depth_data)

    # Actualiza los canales.
    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))
    )