10.3. การสตรีมสด -- ผู้ชมคนเดียว

เบราว์เซอร์สามารถเรนเดอร์สตรีม Motion JPEG (MJPEG) แบบหลายส่วนได้โดยตรงภายในแท็ก <img> ส่งการตอบสนอง HTTP หนึ่งครั้งที่ไม่มีวันเสร็จสิ้นให้เบราว์เซอร์ เขียน JPEG ที่คั่นด้วยขอบเขต multipart และเบราว์เซอร์จะแสดงแต่ละเฟรมเมื่อมาถึง

สายสัญญาณนั้นตรงไปตรงมา: ส่วนหัวการตอบสนองหนึ่งชุด Content-Type: multipart/x-mixed-replace; boundary=frame จากนั้นบรรทัด --frame Content-Type: image/jpeg บรรทัดว่าง ไบต์ JPEG \r\n และทำซ้ำ เบราว์เซอร์ปิดการเชื่อมต่อเมื่อ <img> ถูกลบหรือปิดแท็บ

10.3.1. การจับภาพโดยไม่บล็อก

csi0.snapshot() แบบบล็อกที่ใช้มาจนถึงตอนนี้จะหยุด event loop ทั้งหมดจนกว่าเซนเซอร์จะส่งเฟรม ซึ่งใช้ได้ดีเมื่อคำขอหนึ่งครั้งจับภาพหนึ่งครั้งและไม่มีสิ่งอื่นทำงานอยู่ เมื่อเปิดสตรีมแล้ว เซิร์ฟเวอร์ต้องจัดการคำขออื่น ๆ ต่อไปในขณะที่จับเฟรมถัดไป -- การเรียกจับภาพต้องการ ยอมสละ ให้กับ event loop ในขณะที่รอเซนเซอร์

รูปแบบคือ wrapper AsyncCSI แบบบาง ๆ ที่ polling csi.CSI.snapshot() ในโหมดไม่บล็อก และ sleep coroutine ระหว่างการ polling บทเกี่ยวกับ asyncio ได้อธิบายรูปแบบนี้ใน AsyncCSI; ฝังไว้ในสคริปต์สำหรับตอนนี้:

import asyncio

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)

เมธอด CSI อื่น ๆ ทุกตัว (reset(), pixformat(), framesize(), gain_db(), ...) ถูกส่งต่อผ่าน __getattr__; มีเพียง snapshot() เท่านั้นที่ถูกแทนที่ด้วยเวอร์ชันที่รอได้ซึ่งให้ event loop กำหนดเวลา coroutine อื่น ๆ ระหว่างการ polling

แทนที่ csi.CSI() แบบเดิมจาก snapshot route ด้วย AsyncCSI():

csi0 = AsyncCSI()
csi0.reset()
csi0.pixformat(csi.RGB565)
csi0.framesize(csi.QVGA)

10.3.2. เนื้อหาการสตรีมเป็น iterator แบบคลาส

เนื้อหาการตอบสนองแบบสตรีมเป็นเพียงอ็อบเจกต์ที่ microdot วนซ้ำด้วย async for โดยส่งแต่ละ chunk ที่ yield ลง socket ใน CPython ปกติจะเป็น ฟังก์ชัน async generator -- async def พร้อม yield MicroPython ไม่รองรับสิ่งนั้น:

Note

asyncio ของ MicroPython ไม่รองรับฟังก์ชัน async-generator (async def name(): ... yield ...) เนื้อหาการตอบสนองแบบสตรีมต้องเป็น async iterator แบบคลาส ที่มี __aiter__ ส่งคืน self และ __anext__ นิยามเป็น async def

สำหรับสตรีม MJPEG หมายถึงคลาสที่ __anext__ รอเฟรมหนึ่งเฟรมและส่งคืนมันในกรอบ wrapper multipart:

BOUNDARY = b'frame'

class FrameStream:
    def __aiter__(self):
        return self

    async def __anext__(self):
        img = await csi0.snapshot()
        jpeg = bytes(img.compress(quality=85).bytearray())
        return (b'--' + BOUNDARY + b'\r\n'
                b'Content-Type: image/jpeg\r\n\r\n'
                + jpeg + b'\r\n')

@app.get('/stream.jpg')
async def stream(request):
    return Response(
        body=FrameStream(),
        headers={
            'Content-Type':
                b'multipart/x-mixed-replace; boundary=' + BOUNDARY,
        },
    )

อินสแตนซ์ถูกสร้างใหม่ต่อคำขอ ดังนั้นแต่ละ client ที่เชื่อมต่อจะได้รับ iterator ของตัวเอง เมื่อเบราว์เซอร์ตัดการเชื่อมต่อ microdot จะหยุดรอ __anext__ และ iterator ถูกเก็บขยะ

Note

การ wrap bytes(...) รอบ JPEG เป็นการป้องกัน bytearray() ส่งคืน view เข้าสู่บัฟเฟอร์ภาพของกล้อง และการเรียก snapshot() ครั้งถัดไปจะเขียนทับบัฟเฟอร์นั้นในที่เดิม การ wrap ด้วย bytes จะคัดลอก JPEG ออกมาเพื่อให้ chunk ที่ microdot กำลังเขียนอยู่มีความเสถียรแม้ว่าการ flush ของ writer ยังไม่เสร็จสิ้นในเวลาที่ __anext__ รันอีกครั้ง

10.3.3. การรันเซิร์ฟเวอร์ภายใน asyncio

การเรียก app.run(host=..., port=...) ก่อนหน้านี้เป็นการบล็อก ตัวจัดการ MJPEG ต้องการแบ่งปัน loop กับการ polling snapshot ของ AsyncCSI ดังนั้นให้แทนที่ app.run ด้วย start_server() ภายใน asyncio.run():

async def main():
    await app.start_server(host='0.0.0.0', port=80)

asyncio.run(main())

wrapper asyncio.run() ให้เซิร์ฟเวอร์เป็นหนึ่ง task ในหลาย task -- coroutine main จึงเป็นสถานที่ตามธรรมชาติในการ spawn การจับภาพ การตรวจจับความเคลื่อนไหว และทุกอย่างอื่นที่ต้องแบ่งปัน loop กับ HTTP server

10.3.4. ผู้ชมคนเดียวในแต่ละครั้ง

client แต่ละตัวที่เชื่อมต่อรัน iterator FrameStream ของตัวเอง ซึ่งหมายความว่าทุก client กระตุ้นการเรียก csi0.snapshot() ของตัวเอง สองเบราว์เซอร์หมายถึงการอ่านเซนเซอร์สองครั้งต่อช่วงเฟรม สามหมายถึงสาม และต่อไป เซนเซอร์ไม่สามารถส่งเฟรมได้เร็วกว่าอัตราเฟรมของตัวเอง ดังนั้นคำขอจะเข้าคิวรอกันและสตรีมของทุกคนจะช้าลง

วิธีแก้ไขคือ capture loop ที่ใช้ร่วมกันซึ่งเผยแพร่เฟรมหนึ่งไปยังผู้อ่านหลายคน