8.14. AsyncCSI¶
Một tập lệnh OpenMV Cam điển hình kết thúc bằng while True: img = csi0.snapshot() -- một vòng lặp ảnh chụp chặn không cần asyncio chút nào. Ngay khi ứng dụng phải làm điều gì đó khác cùng với việc chụp ảnh -- lắng nghe một nút, gửi dữ liệu đến một thiết bị ngang hàng, chạy một task nền -- lệnh gọi chặn trở nên cản trở. Trong khi snapshot đang chờ khung hình tiếp theo, vòng lặp sự kiện không chạy, vì vậy mọi coroutine khác trong chương trình bị đóng băng cho đến khi khung hình đến.
Trang này xây dựng một lớp bọc nhỏ xung quanh CSI biến snapshot thành một coroutine thân thiện với await. Kết quả là một bộ thay thế trực tiếp cho phép vòng lặp chụp ảnh cùng tồn tại với phần còn lại của chương trình asyncio.
8.14.1. Các thành phần¶
Một phần của API CSI thực hiện phần lớn công việc -- snapshot() ở chế độ không chặn. Gọi snapshot(blocking=False) sẽ trả về khung hình tiếp theo (nếu có sẵn) hoặc None (nếu chưa có). Lần gọi không chặn đầu tiên cũng khởi động DMA capture của camera nếu nó chưa chạy, vì vậy lớp bọc không cần làm gì đặc biệt để khởi tạo.
Phần kia là asyncio.sleep_ms(). Lớp bọc thăm dò các snapshot không chặn trong một vòng lặp, nhường cho vòng lặp sự kiện bằng await asyncio.sleep_ms(0) giữa các lần kiểm tra để mọi coroutine sẵn sàng khác có cơ hội chạy trước lần thăm dò tiếp theo.
8.14.2. Lớp bọc¶
import asyncio
import csi
class AsyncCSI:
def __init__(self, *args, **kwargs):
self._csi = csi.CSI(*args, **kwargs)
def __getattr__(self, name):
return getattr(self._csi, name)
async def snapshot(self):
while True:
img = self._csi.snapshot(blocking=False)
if img is not None:
return img
await asyncio.sleep_ms(0)
Hàm khởi tạo bọc một thể hiện CSI. __getattr__ chuyển tiếp mọi thuộc tính mà lớp bọc không tự định nghĩa -- reset, pixformat, framesize, tất cả các điều chỉnh sensor -- đến CSI bên dưới, vì vậy lớp bọc trông giống hệt đối tượng không được bọc ngoại trừ một phương thức quan trọng.
async def snapshot là phần mới. Nó gọi snapshot(blocking=False); nếu lệnh gọi trả về một ảnh, coroutine trả về nó. Nếu không, nó nhường lại cho vòng lặp sự kiện bằng await asyncio.sleep_ms(0) để các coroutine khác có cơ hội chạy, rồi quay lại vòng lặp và thử lại. Lần lặp đầu tiên khởi động DMA; các lần lặp tiếp theo lấy khung hình khi chúng trở nên khả dụng.
8.14.3. Một vòng lặp chụp ảnh có bạn đồng hành¶
Với lớp bọc được đặt vào vị trí, một vòng lặp chụp ảnh phù hợp với chương trình asyncio lớn hơn theo cách tương tự như bất kỳ coroutine nào khác. Ví dụ dưới đây chạy ba coroutine đồng thời: vòng lặp chụp ảnh, một bộ nhấp nháy LED, và một heartbeat in hello mỗi giây:
import asyncio
import csi
from machine import LED
async def capture_loop(cam):
while True:
img = await cam.snapshot()
# process img here
async def blinker(led, period_ms):
while True:
led.on()
await asyncio.sleep_ms(period_ms)
led.off()
await asyncio.sleep_ms(period_ms)
async def hello(period_s):
while True:
print("hello")
await asyncio.sleep(period_s)
async def main():
cam = AsyncCSI()
cam.reset()
cam.pixformat(csi.RGB565)
cam.framesize(csi.QVGA)
asyncio.create_task(blinker(LED("LED_BLUE"), 200))
asyncio.create_task(hello(1))
await capture_loop(cam)
asyncio.run(main())
Cả ba coroutine đều tiến triển trên cùng một vòng lặp sự kiện. Trong khi capture_loop đang nhường giữa các lần thăm dò snapshot không chặn, blinker bật tắt LED và hello in ra. Trong khi blinker và hello đang ngủ, capture_loop thăm dò camera. Khoảng thời gian thăm dò ngắn -- một tick vòng lặp sự kiện duy nhất -- vì vậy nó thêm độ trễ không đáng kể vào thời điểm ứng dụng thấy một khung hình mới.
Vòng lặp chụp ảnh không chặn vòng lặp sự kiện. Thêm nhiều công việc đồng thời hơn -- ví dụ một client UART -- chỉ là thêm một lệnh gọi create_task() khác bên trong main.
Ghi chú
Cài đặt bộ đệm khung hình vẫn quan trọng ở hình thức này. Chế độ một bộ đệm khiến snapshot(blocking=False) trả về None cho đến khi khung hình tiếp theo được chụp; chế độ hai hoặc ba bộ đệm làm mượt điều đó để lớp bọc thường tìm thấy một khung hình đã đệm sẵn đang chờ ở lần thăm dò đầu tiên sau khi khung hình trước đã được xử lý.