12.8. ווידג’טים וטלמטריה עם CBORChannel#

הקצוות האחוריים בשני העמודים הקודמים מעבירים בייטים גולמיים, וערוץ בייטים גולמי זקוק לתוכנית מארח שיודעת לפענח אותם. רוב הזמן מה שמצלמה רוצה לפרסם פשוט מזה: כמה קריאות בעלות שם, כמה פקדים בעלי שם, צורת גל, מפת עומק. למקרה זה חבילת protocol מספקת את protocol.CBORChannel, קצה אחורי מוכן שמחזיק שדות בעלי שם, מסדר אותם כרשומות CBOR בפריסת SenML, ומפענח כתיבות של המארח בחזרה לערכי שדות. היתרון שלו הוא ש-OpenMV IDE כבר מבין אותו: רשום CBORChannel ותצוגת Channels של ה-IDE מציגה כל שדה כווידג’ט חי – תוויות, מתגים, מחוונים, גרפים, מפות – ללא קוד מארח כלל. אותן רשומות הן CBOR רגיל, כך שמארח מותאם מפענח אותן עם כל ספריית CBOR כשתגדל מעבר ל-IDE.

The Channels view showing a thermal camera script's controls channel: FPA and AUX temperature readouts with their unit, Measurement and High Temp toggles, Temp Min and Temp Max sliders, and H-Mirror and V-Flip toggles

הערוץ של הדוגמה controls_channel.py כפי שה-IDE מציג אותו: קריאות, מתגים ומחוונים עבור FLIR Lepton. כל פקד כותב בחזרה לסקריפט.#

12.8.1. קריאות ופקדים#

CBORChannel הוא מילון של שדות בעלי טיפוס. add מצהיר על כל אחד עם שם, type של ווידג’ט, והארגומנטים שאותו סוג צריך; השמה ל-ch["name"] מעדכנת שדה; ה-callback on_write מקבל את השינויים של המארח:

import time
import protocol
from protocol import CBORChannel

def on_write(ch, name, value):
    # Called for every control the host changes.
    print(name, "=", value)
    if name == "Reset":
        ch["Count"] = 0

ch = CBORChannel(on_write=on_write)
ch.add("Status", type="label", value="starting")
ch.add("Count", type="label", value=0)
ch.add("Enable", type="toggle", value=True)
ch.add("Threshold", type="slider", min=0, max=100, step=1,
       value=50, unit="%")
ch.add("Gap", type="spinbox", min=0.0, max=15.0, step=0.1,
       value=1.5, unit="mm")
ch.add("Mode", type="radio", options=["Idle", "Track", "Record"],
       value="Idle")
ch.add("Quality", type="select", options=["Low", "Medium", "High"],
       value="Medium")
ch.add("Name", type="lineedit", value="openmv-cam")
ch.add("Reset", type="pushbutton")
protocol.register(name="controls", backend=ch)

count = 0
while True:
    if ch["Enable"]:
        count += 1
        ch["Count"] = count
    ch["Status"] = "threshold %d%%" % ch["Threshold"]
    time.sleep_ms(100)

הרץ אותו ב-IDE, החלף את החלונית שמתחת לחוצץ הפריימים (frame buffer) ל-Channels, והווידג’טים מופיעים: label הוא ערך לקריאה בלבד (עם ה-unit שלו לצדו), text הוא גוש של טקסט עשיר סטטי, toggle הוא מתג, slider ו-spinbox קובעים מספר בתוך min / max / step (תיבת הספין להזנה מדויקת), radio ו-select בוחרים אחת מ-options, lineedit הוא שדה טקסט חופשי, ו-pushbutton הוא פעולה רגעית שקוראת ל-on_write עם True. הסקריפט קורא את הפקדים שלו בחזרה עם ch["Threshold"] מתי שירצה – הערוץ מחזיק את הערך האחרון – ו-on_write הוא הנקודה לאלה שזקוקים לתגובה מיידית. tuple (min, max, value) שמושם למחוון או לתיבת ספין מזיז את הטווח שלהם יחד עם הערך, וכך פקד עוקב אחר חיישן שגבולותיו תלויים בהגדרה אחרת.

ה-callback האופציונלי on_read(channel) רץ ממש לפני שהערוץ מסודר עבור המארח – המקום לדגום חיישן לתוך label כך שקריאה נלקחת רק כשמישהו מסתכל:

def on_read(ch):
    ch["FPA Temp"] = round(csi0.ioctl(csi.IOCTL_LEPTON_GET_FPA_TEMP), 1)

הדוגמה controls_channel.py תחת File → Examples → 12-Protocol משתמשת בדיוק בצורה זו כדי לחשוף את מצב המדידה, טווח הטמפרטורה והגדרות השיקוף / ההיפוך של FLIR Lepton בזמן שהמצלמה מזרימה.

12.8.2. צורות גל#

שדה waveform נושא בלוק של דגימות עם קצב דגימה. הדגימות הן הבייטים הגולמיים של array ב-typecode של השדה (16 סיביות ללא סימן כברירת מחדל); לכמה אותות בבת אחת – שלושת הצירים של מד-תאוצה – שלב אותם ואמור כמה series יש ואיך לקרוא להן:

import math
import time
import protocol
from array import array
from protocol import CBORChannel

RATE = 1000           # samples per second
CHUNK = 100           # samples per update (100 ms)

ch = CBORChannel()
ch.add("Mic", type="waveform", sample_rate=RATE,
       min=0, max=65535)                           # uint16
ch.add("IMU", type="waveform", sample_rate=RATE, series=3,
       typecode="f", min=-2.0, max=2.0,
       options=["X", "Y", "Z"], unit="g")          # float32 x 3
protocol.register(name="signals", backend=ch)

t = 0
while True:
    mic = array("H", (int(32768 + 20000 * math.sin(2 * math.pi * 50 * (t + i) / RATE))
                      for i in range(CHUNK)))
    imu = array("f")
    for i in range(CHUNK):
        phase = 2 * math.pi * 2 * (t + i) / RATE
        imu.extend((math.sin(phase), math.cos(phase), 1.0))
    ch["Mic"] = bytes(mic)    # the raw sample bytes
    ch["IMU"] = bytes(imu)
    t += CHUNK
    time.sleep_ms(100)

ה-IDE מצייר כל צורת גל כגרף נגלל עם עקבה אחת לכל סדרה, מחתים כל מקטע בזמן כך שפערים בין עדכונים נראים כפערים, ומציע תצוגת ספקטרום, טריגרים, סמנים והקלטה לקובצי CSV, WAV, NumPy או Edge Impulse. החלף את הסינוס הסינתטי בדגימות audio או בקריאות של מנהל התקן IMU והסקריפט הוא כלי לאיסוף נתונים.

12.8.3. מפות עומק#

שדה depth הוא רשת width x height של floats בני 32 סיביות, מרחק אחד לכל תא – מה שחיישן זמן-מעוף (time-of-flight) מפיק. min / max מצהירים על הטווח שהמארח צריך למפות לצבע; ה-IDE יכול גם לקבוע את טווח הצבעים אוטומטית מהנתונים:

import struct
import time
import tof
import protocol
from protocol import CBORChannel

tof.init()
ch = CBORChannel()
ch.add("depth", type="depth", width=tof.width(), height=tof.height(),
       min=0, max=1000)
reg = protocol.register(name="ToF", backend=ch)

while True:
    try:
        d, dmin, dmax = tof.read_depth(vflip=True, hmirror=True)
    except RuntimeError:
        continue
    ch["depth"] = struct.pack("<%df" % len(d), *d)
    reg.send_event(0xFFFF)
    time.sleep_ms(50)

הדוגמה sensors_channel.py משלבת ערוץ עומק כזה עם ערוץ קריאות ומזהה פנים על אותה מצלמה. הקריאה ל-send_event() אופציונלית – ה-IDE דוגם ערוצים בעצמו – אך היא מודיעה למארח מונחה-אירועים שפריים חדש מוכן מבלי שיצטרך לשאול.

12.8.4. פענוח במארח מותאם#

הרשומות הן CBOR תקני: כל קריאת ערוץ מחזירה מערך של מפות עם מפתחות המספרים השלמים של SenML (0 שם, 1 יחידה, 2 ערך מספרי, 3 ערך מחרוזת, 8 ערך נתונים, ומפתחות שליליים לסוג הווידג’ט, האפשרויות, הטווח והממדים). מארח שגדל מעבר ל-IDE קורא את הערוץ עם channel_read() ומפענח אותו עם כל ספריית CBOR (pip install cbor2); כתיבת פקד בחזרה היא מערך CBOR של מפות {0: name, 2: value} הנשלח עם channel_write(). העיון של protocol.CBORChannel מפרט כל מפתח. ברוב הפרויקטים, עם זאת, תצוגת Channels של ה-IDE היא המארח, והקוד היחיד הוא סקריפט צד-המצלמה שלמעלה.