sensors_channel.py

# هذا العمل مرخص بموجب ترخيص MIT.
# حقوق النشر (c) 2026 لشركة OpenMV LLC. جميع الحقوق محفوظة.
# https://github.com/openmv/openmv/blob/master/LICENSE
#
# مثال قناة أجهزة الاستشعار (ToF + كشف الوجوه)
#
# بيانات أجهزة الاستشعار وخرائط العمق ToF عبر قنوات بروتوكول CBOR.
# يستخدم CBORChannel من مكتبة البروتوكول لسهولة الإعداد.

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

# تهيئة الكاميرا.
csi0 = csi.CSI()
csi0.reset()
csi0.pixformat(csi.RGB565)
csi0.framesize(csi.QVGA)

# تحميل نموذج كشف الوجوه المدمج.
model = ml.Model(
    "/rom/blazeface_front_128.tflite",
    postprocess=BlazeFace(threshold=0.5),
)
print(model)

# تهيئة مستشعر ToF.
tof.init()
tof_w = tof.width()
tof_h = tof.height()

# تسجيل قناة المستشعر.
sensors = CBORChannel()
sensors.add("temp", type="label", unit="Cel")
sensors.add("humidity", type="label", unit="%RH")
sensors_ch = protocol.register(name="sensors", backend=sensors)

# تسجيل قناة العمق.
depth = CBORChannel()
depth.add("depth", type="depth", width=tof_w, height=tof_h)
depth_ch = protocol.register(name="ToF depth", backend=depth)

# حالة مستشعر محاكاة.
temp = 25.0
humidity = 50.0

while True:
    # التقاط إطار وتشغيل كشف الوجوه.
    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))

    # محاكاة قراءات المستشعر مع انحراف عشوائي.
    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))

    # قراءة خريطة العمق من ToF.
    try:
        depth_data, dmin, dmax = tof.read_depth(vflip=True, hmirror=True)
    except RuntimeError:
        continue

    # حزم قيم العمق العشرية (float) في صيغة ثنائية (4 بايت لكل قيمة).
    depth_bytes = struct.pack("<%df" % len(depth_data), *depth_data)

    # تحديث القنوات.
    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))
    )