10.3. Canlı yayın – tek izleyici

Tarayıcılar çok parçalı Motion JPEG (MJPEG) yayınlarını doğrudan bir <img> etiketi içinde işleyebilir. Tarayıcıya asla bitmeyen tek bir HTTP yanıtı verin, çok parçalı bir sınırlayıcıyla ayrılmış JPEG’ler yazın ve tarayıcı her çerçeveyi geldiği anda görüntüler.

Hat basittir: bir yanıt başlığı, Content-Type: multipart/x-mixed-replace; boundary=frame, ardından bir --frame satırı, Content-Type: image/jpeg, boş bir satır, JPEG baytları, \r\n ve tekrar. <img> kaldırıldığında veya sekme kapatıldığında tarayıcı bağlantıyı kapatır.

10.3.1. Engellemeden yakalama

Şimdiye kadar kullanılan engelleyici csi0.snapshot(), sensör bir çerçeve teslim edene kadar tüm olay döngüsünü durdurur. Bir istek bir anlık görüntü tetikleyip başka hiçbir şey çalışmadığında bu uygundu. Bir yayın açıldığında ise, sunucu sonraki çerçeve yakalanırken diğer istekleri işlemeye devam etmek zorundadır – yakalama çağrısının sensörü beklerken olay döngüsüne yer vermesi (yield) gerekir.

Desen, csi.CSI.snapshot() yöntemini engellemeyen modda yoklayan ve yoklamalar arasında eşyordamı uyutan ince bir AsyncCSI sarmalayıcısıdır. asyncio bölümü bu deseni AsyncCSI belgesinde adım adım anlattı; şimdilik onu betiğin içine satır içi koyun:

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)

Diğer her CSI yöntemi (reset(), pixformat(), framesize(), gain_db(), …) __getattr__ üzerinden iletilir; yalnızca snapshot(), olay döngüsünün yoklamalar arasında diğer eşyordamları zamanlamasına izin veren beklenebilir (awaitable) bir sürümle değiştirilir.

Anlık görüntü rotasındaki çıplak csi.CSI() örneğini bir AsyncCSI() ile değiştirin:

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

10.3.2. Yayın gövdeleri sınıf tabanlı yineleyicilerdir

Bir yayın yanıt gövdesi, microdot’un async for ile yinelediği ve her üretilen parçayı sokete gönderdiği bir nesneden ibarettir. CPython’da bu normalde bir asenkron üreteç işlevidir – yield içeren async def. MicroPython bunu desteklemez:

Not

MicroPython’un asyncio modülü asenkron üreteç işlevlerini (async def name(): ... yield ...) desteklemez. Yayın yanıt gövdeleri, __aiter__ metodu self döndüren ve __anext__ metodu async def olarak tanımlanan sınıf tabanlı asenkron yineleyiciler olmalıdır.

Bir MJPEG yayını için bu, __anext__ metodu bir çerçeveyi bekleyen (await) ve onu çok parçalı sarmalayıcı içinde çerçevelenmiş olarak döndüren bir sınıf demektir:

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,
        },
    )

Örnek istek başına yeniden oluşturulur, böylece bağlanan her istemci kendi yineleyicisini alır. Tarayıcı bağlantıyı kestiğinde microdot __anext__ beklemeyi bırakır ve yineleyici çöp toplama ile temizlenir.

Not

JPEG’in etrafındaki bytes(...) sarmalaması savunma amaçlıdır. bytearray(), kameranın görüntü arabelleğine bir görünüm döndürür ve sonraki snapshot() çağrısı o arabelleği yerinde yeniden yazar. bytes içine sarmak JPEG’i dışarı kopyalar, böylece microdot’un yazmakta olduğu parça, yazıcının boşaltma işlemi __anext__ tekrar çalıştığında henüz bitmemiş olsa bile kararlı kalır.

10.3.3. Sunucuyu asyncio içinde çalıştırma

Önceki app.run(host=..., port=...) çağrısı engelleyicidir. MJPEG işleyicisinin döngüyü AsyncCSI anlık görüntü yoklamalarıyla paylaşması gerekir, bu yüzden app.run yerine bir asyncio.run() içinde start_server() kullanın:

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

asyncio.run(main())

asyncio.run() sarmalayıcısı, sunucunun birçok görev arasında bir görev olmasına izin verir – main eşyordamı, yakalama, hareket tespiti ve HTTP sunucusuyla döngüyü paylaşması gereken her şeyi başlatmak için doğal yerdir.

10.3.4. Bir seferde tek izleyici

Bağlanan her istemci kendi FrameStream yineleyicisini çalıştırır, bu da her istemcinin kendi csi0.snapshot() çağrısını tetiklemesi anlamına gelir. İki tarayıcı, çerçeve aralığı başına iki sensör okuması demektir, üçü üç, vesaire. Sensör çerçeveleri kendi çerçeve hızından daha hızlı teslim edemez, bu yüzden istekler arka arkaya sıraya girer ve herkesin yayını yavaşlar.

Çözüm, bir çerçeveyi birçok okuyucuya yayınlayan tek bir paylaşılan yakalama döngüsüdür.