queue

app.py

@@ -3,7 +3,7 @@ from fastapi import FastAPI
 from config import args
 from device import device, torch_dtype
 from app_init import init_app
-from user_queue import user_data_events
+from user_queue import user_data
 from util import get_pipeline_class
 
 
@@ -11,4 +11,4 @@ app = FastAPI()
 
 pipeline_class = get_pipeline_class(args.pipeline)
 pipeline = pipeline_class(args, device, torch_dtype)
-init_app(app, user_data_events, args, pipeline)
+init_app(app, user_data, args, pipeline)

app_init.py

@@ -7,16 +7,15 @@ from fastapi import Request
 import logging
 import traceback
 from config import Args
-from user_queue import UserDataEventMap, UserDataEvent
+from user_queue import UserData
 import uuid
-from asyncio import Event, sleep
 import time
-from PIL import Image
 from types import SimpleNamespace
-from util import pil_to_frame, is_firefox
+from util import pil_to_frame, bytes_to_pil, is_firefox
+import asyncio
 
 
-def init_app(app: FastAPI, user_data_events: UserDataEventMap, args: Args, pipeline):
+def init_app(app: FastAPI, user_data: UserData, args: Args, pipeline):
     app.add_middleware(
         CORSMiddleware,
         allow_origins=["*"],
@@ -28,44 +27,42 @@ def init_app(app: FastAPI, user_data_events: UserDataEventMap, args: Args, pipel
     @app.websocket("/ws")
     async def websocket_endpoint(websocket: WebSocket):
         await websocket.accept()
-        if args.max_queue_size > 0 and len(user_data_events) >= args.max_queue_size:
+        user_count = user_data.get_user_count()
+        if args.max_queue_size > 0 and user_count >= args.max_queue_size:
             print("Server is full")
             await websocket.send_json({"status": "error", "message": "Server is full"})
             await websocket.close()
             return
-
         try:
-            uid = str(uuid.uuid4())
-            print(f"New user connected: {uid}")
-            await websocket.send_json(
-                {"status": "success", "message": "Connected", "userId": uid}
-            )
-            user_data_events[uid] = UserDataEvent()
+            user_id = uuid.uuid4()
+            print(f"New user connected: {user_id}")
+            await user_data.create_user(user_id, websocket)
             await websocket.send_json(
-                {"status": "start", "message": "Start Streaming", "userId": uid}
+                {"status": "connected", "message": "Connected", "userId": str(user_id)}
             )
-            await handle_websocket_data(websocket, uid)
+            await handle_websocket_data(user_id, websocket)
         except WebSocketDisconnect as e:
-            logging.error(f"WebSocket Error: {e}, {uid}")
+            logging.error(f"WebSocket Error: {e}, {user_id}")
             traceback.print_exc()
         finally:
-            print(f"User disconnected: {uid}")
-            del user_data_events[uid]
+            print(f"User disconnected: {user_id}")
+            user_data.delete_user(user_id)
 
     @app.get("/queue_size")
     async def get_queue_size():
-        queue_size = len(user_data_events)
+        queue_size = user_data.get_user_count()
         return JSONResponse({"queue_size": queue_size})
 
     @app.get("/stream/{user_id}")
     async def stream(user_id: uuid.UUID, request: Request):
-        uid = str(user_id)
         try:
+            print(f"New stream request: {user_id}")
 
             async def generate():
                 while True:
-                    data = await user_data_events[uid].wait_for_data()
-                    params = data["params"]
+                    params = await user_data.get_latest_data(user_id)
+                    if not params:
+                        continue
                     image = pipeline.predict(params)
                     if image is None:
                         continue
@@ -81,13 +78,12 @@ def init_app(app: FastAPI, user_data_events: UserDataEventMap, args: Args, pipel
                 headers={"Cache-Control": "no-cache"},
             )
         except Exception as e:
-            logging.error(f"Streaming Error: {e}, {user_data_events}")
+            logging.error(f"Streaming Error: {e}, {user_id} ")
             traceback.print_exc()
             return HTTPException(status_code=404, detail="User not found")
 
-    async def handle_websocket_data(websocket: WebSocket, user_id: uuid.UUID):
-        uid = str(user_id)
-        if uid not in user_data_events:
+    async def handle_websocket_data(user_id: uuid.UUID, websocket: WebSocket):
+        if not user_data.check_user(user_id):
             return HTTPException(status_code=404, detail="User not found")
         last_time = time.time()
         try:
@@ -98,19 +94,20 @@ def init_app(app: FastAPI, user_data_events: UserDataEventMap, args: Args, pipel
                 params = SimpleNamespace(**params.dict())
                 if info.input_mode == "image":
                     image_data = await websocket.receive_bytes()
-                    pil_image = Image.open(io.BytesIO(image_data))
-                    params.image = pil_image
-                user_data_events[uid].update_data({"params": params})
+                    params.image = bytes_to_pil(image_data)
+
+                await user_data.update_data(user_id, params)
                 if args.timeout > 0 and time.time() - last_time > args.timeout:
                     await websocket.send_json(
                         {
                             "status": "timeout",
                             "message": "Your session has ended",
-                            "userId": uid,
+                            "userId": user_id,
                         }
                     )
                     await websocket.close()
                     return
+                await asyncio.sleep(1.0 / 24)
 
         except Exception as e:
             logging.error(f"Error: {e}")

frontend/src/lib/components/Button.svelte

@@ -7,7 +7,7 @@
   <slot />
 </button>
 
-<style lang="postcss">
+<style lang="postcss" scoped>
   .button {
     @apply rounded bg-gray-700 p-2 font-normal text-white hover:bg-gray-800 disabled:cursor-not-allowed disabled:bg-gray-300 dark:disabled:bg-gray-700 dark:disabled:text-black;
   }

frontend/src/lib/components/ImagePlayer.svelte

@@ -1,18 +1,19 @@
 <script lang="ts">
-  import { isLCMRunning, lcmLiveState, lcmLiveActions } from '$lib/lcmLive';
+  import { lcmLiveStatus, LCMLiveStatus, streamId } from '$lib/lcmLive';
   import { onFrameChangeStore } from '$lib/mediaStream';
   import { PUBLIC_BASE_URL } from '$env/static/public';
 
-  $: streamId = $lcmLiveState?.streamId;
   $: {
-    console.log('streamId', streamId);
+    console.log('streamId', $streamId);
   }
+  $: isLCMRunning = $lcmLiveStatus !== LCMLiveStatus.DISCONNECTED;
+  $: console.log('isLCMRunning', isLCMRunning);
 </script>
 
 <div class="relative overflow-hidden rounded-lg border border-slate-300">
   <!-- svelte-ignore a11y-missing-attribute -->
-  {#if $isLCMRunning}
-    <img class="aspect-square w-full rounded-lg" src={PUBLIC_BASE_URL + '/stream/' + streamId} />
+  {#if isLCMRunning}
+    <img class="aspect-square w-full rounded-lg" src={PUBLIC_BASE_URL + '/stream/' + $streamId} />
   {:else}
     <div class="aspect-square w-full rounded-lg" />
   {/if}

frontend/src/lib/components/PipelineOptions.svelte

@@ -17,13 +17,13 @@
 <div>
   {#if featuredOptions}
     {#each featuredOptions as params}
-      {#if params.field === FieldType.range}
+      {#if params.field === FieldType.RANGE}
         <InputRange {params} bind:value={$pipelineValues[params.id]}></InputRange>
-      {:else if params.field === FieldType.seed}
+      {:else if params.field === FieldType.SEED}
         <SeedInput bind:value={$pipelineValues[params.id]}></SeedInput>
-      {:else if params.field === FieldType.textarea}
+      {:else if params.field === FieldType.TEXTAREA}
         <TextArea {params} bind:value={$pipelineValues[params.id]}></TextArea>
-      {:else if params.field === FieldType.checkbox}
+      {:else if params.field === FieldType.CHECKBOX}
         <Checkbox {params} bind:value={$pipelineValues[params.id]}></Checkbox>
       {/if}
     {/each}
@@ -33,17 +33,17 @@
 <details open>
   <summary class="cursor-pointer font-medium">Advanced Options</summary>
   <div
-    class="grid grid-cols-1 items-center gap-3 {pipelineValues.length > 5 ? 'sm:grid-cols-2' : ''}"
+    class="grid grid-cols-1 items-center gap-3 {pipelineParams.length > 5 ? 'sm:grid-cols-2' : ''}"
   >
     {#if advanceOptions}
       {#each advanceOptions as params}
-        {#if params.field === FieldType.range}
+        {#if params.field === FieldType.RANGE}
           <InputRange {params} bind:value={$pipelineValues[params.id]}></InputRange>
-        {:else if params.field === FieldType.seed}
+        {:else if params.field === FieldType.SEED}
           <SeedInput bind:value={$pipelineValues[params.id]}></SeedInput>
-        {:else if params.field === FieldType.textarea}
+        {:else if params.field === FieldType.TEXTAREA}
           <TextArea {params} bind:value={$pipelineValues[params.id]}></TextArea>
-        {:else if params.field === FieldType.checkbox}
+        {:else if params.field === FieldType.CHECKBOX}
           <Checkbox {params} bind:value={$pipelineValues[params.id]}></Checkbox>
         {/if}
       {/each}

frontend/src/lib/components/VideoInput.svelte

@@ -1,42 +1,38 @@
 <script lang="ts">
   import 'rvfc-polyfill';
-  import { onMount, onDestroy } from 'svelte';
+  import { onDestroy } from 'svelte';
   import {
-    mediaStreamState,
-    mediaStreamActions,
-    isMediaStreaming,
-    MediaStreamStatus,
-    onFrameChangeStore
+    mediaStreamStatus,
+    MediaStreamStatusEnum,
+    onFrameChangeStore,
+    mediaStream
   } from '$lib/mediaStream';
 
-  $: mediaStream = $mediaStreamState.mediaStream;
-
   let videoEl: HTMLVideoElement;
   let videoFrameCallbackId: number;
   const WIDTH = 512;
   const HEIGHT = 512;
+  const THROTTLE_FPS = 10;
 
   onDestroy(() => {
     if (videoFrameCallbackId) videoEl.cancelVideoFrameCallback(videoFrameCallbackId);
   });
 
-  function srcObject(node: HTMLVideoElement, stream: MediaStream) {
-    node.srcObject = stream;
-    return {
-      update(newStream: MediaStream) {
-        if (node.srcObject != newStream) {
-          node.srcObject = newStream;
-        }
-      }
-    };
+  $: if (videoEl) {
+    videoEl.srcObject = $mediaStream;
   }
+
+  let last_millis = 0;
   async function onFrameChange(now: DOMHighResTimeStamp, metadata: VideoFrameCallbackMetadata) {
-    const blob = await grapBlobImg();
-    onFrameChangeStore.set({ now, metadata, blob });
+    if (now - last_millis > 1000 / THROTTLE_FPS) {
+      const blob = await grapBlobImg();
+      onFrameChangeStore.set({ now, metadata, blob });
+      last_millis = now;
+    }
     videoFrameCallbackId = videoEl.requestVideoFrameCallback(onFrameChange);
   }
 
-  $: if ($isMediaStreaming == MediaStreamStatus.CONNECTED) {
+  $: if ($mediaStreamStatus == MediaStreamStatusEnum.CONNECTED) {
     videoFrameCallbackId = videoEl.requestVideoFrameCallback(onFrameChange);
   }
   async function grapBlobImg() {
@@ -70,7 +66,6 @@
     autoplay
     muted
     loop
-    use:srcObject={mediaStream}
   ></video>
 </div>
 <svg

frontend/src/lib/lcmLive.ts

@@ -1,27 +1,17 @@
 import { writable } from 'svelte/store';
 import { PUBLIC_WSS_URL } from '$env/static/public';
 
-export const isStreaming = writable(false);
-export const isLCMRunning = writable(false);
-
 
 export enum LCMLiveStatus {
-    INIT = "init",
     CONNECTED = "connected",
     DISCONNECTED = "disconnected",
+    WAIT = "wait",
 }
 
-interface lcmLive {
-    streamId: string | null;
-    status: LCMLiveStatus
-}
-
-const initialState: lcmLive = {
-    streamId: null,
-    status: LCMLiveStatus.INIT
-};
+const initStatus: LCMLiveStatus = LCMLiveStatus.DISCONNECTED;
 
-export const lcmLiveState = writable(initialState);
+export const lcmLiveStatus = writable<LCMLiveStatus>(initStatus);
+export const streamId = writable<string | null>(null);
 
 let websocket: WebSocket | null = null;
 export const lcmLiveActions = {
@@ -37,12 +27,8 @@ export const lcmLiveActions = {
                     console.log("Connected to websocket");
                 };
                 websocket.onclose = () => {
-                    lcmLiveState.update((state) => ({
-                        ...state,
-                        status: LCMLiveStatus.DISCONNECTED
-                    }));
+                    lcmLiveStatus.set(LCMLiveStatus.DISCONNECTED);
                     console.log("Disconnected from websocket");
-                    isLCMRunning.set(false);
                 };
                 websocket.onerror = (err) => {
                     console.error(err);
@@ -51,47 +37,29 @@ export const lcmLiveActions = {
                     const data = JSON.parse(event.data);
                     console.log("WS: ", data);
                     switch (data.status) {
-                        case "success":
-                            break;
-                        case "start":
-                            const streamId = data.userId;
-                            lcmLiveState.update((state) => ({
-                                ...state,
-                                status: LCMLiveStatus.CONNECTED,
-                                streamId: streamId,
-                            }));
-                            isLCMRunning.set(true);
-                            resolve(streamId);
+                        case "connected":
+                            const userId = data.userId;
+                            lcmLiveStatus.set(LCMLiveStatus.CONNECTED);
+                            streamId.set(userId);
                             break;
                         case "timeout":
                             console.log("timeout");
-                            isLCMRunning.set(false);
-                            lcmLiveState.update((state) => ({
-                                ...state,
-                                status: LCMLiveStatus.DISCONNECTED,
-                                streamId: null,
-                            }));
+                            lcmLiveStatus.set(LCMLiveStatus.DISCONNECTED);
+                            streamId.set(null);
                             reject("timeout");
                         case "error":
                             console.log(data.message);
-                            isLCMRunning.set(false);
-                            lcmLiveState.update((state) => ({
-                                ...state,
-                                status: LCMLiveStatus.DISCONNECTED,
-                                streamId: null,
-                            }));
+                            lcmLiveStatus.set(LCMLiveStatus.DISCONNECTED);
+                            streamId.set(null);
                             reject(data.message);
                     }
                 };
 
             } catch (err) {
                 console.error(err);
-                isLCMRunning.set(false);
-                lcmLiveState.update((state) => ({
-                    ...state,
-                    status: LCMLiveStatus.DISCONNECTED,
-                    streamId: null,
-                }));
+                lcmLiveStatus.set(LCMLiveStatus.DISCONNECTED);
+                streamId.set(null);
+
                 reject(err);
             }
         });
@@ -113,7 +81,7 @@ export const lcmLiveActions = {
             websocket.close();
         }
         websocket = null;
-        lcmLiveState.set({ status: LCMLiveStatus.DISCONNECTED, streamId: null });
-        isLCMRunning.set(false)
+        lcmLiveStatus.set(LCMLiveStatus.DISCONNECTED);
+        streamId.set(null);
     },
 };

frontend/src/lib/mediaStream.ts

@@ -1,38 +1,23 @@
 import { writable, type Writable } from 'svelte/store';
 
-export enum MediaStreamStatus {
+export enum MediaStreamStatusEnum {
     INIT = "init",
     CONNECTED = "connected",
     DISCONNECTED = "disconnected",
 }
 export const onFrameChangeStore: Writable<{ now: Number, metadata: VideoFrameCallbackMetadata, blob: Blob }> = writable();
-export const isMediaStreaming = writable(MediaStreamStatus.INIT);
 
-interface mediaStream {
-    mediaStream: MediaStream | null;
-    status: MediaStreamStatus
-    devices: MediaDeviceInfo[];
-}
-
-const initialState: mediaStream = {
-    mediaStream: null,
-    status: MediaStreamStatus.INIT,
-    devices: [],
-};
-
-export const mediaStreamState = writable(initialState);
+export const mediaDevices = writable<MediaDeviceInfo[]>([]);
+export const mediaStreamStatus = writable(MediaStreamStatusEnum.INIT);
+export const mediaStream = writable<MediaStream | null>(null);
 
 export const mediaStreamActions = {
     async enumerateDevices() {
-        console.log("Enumerating devices");
+        // console.log("Enumerating devices");
         await navigator.mediaDevices.enumerateDevices()
             .then(devices => {
                 const cameras = devices.filter(device => device.kind === 'videoinput');
-                console.log("Cameras: ", cameras);
-                mediaStreamState.update((state) => ({
-                    ...state,
-                    devices: cameras,
-                }));
+                mediaDevices.set(cameras);
             })
             .catch(err => {
                 console.error(err);
@@ -48,17 +33,14 @@ export const mediaStreamActions = {
 
         await navigator.mediaDevices
             .getUserMedia(constraints)
-            .then((mediaStream) => {
-                mediaStreamState.update((state) => ({
-                    ...state,
-                    mediaStream: mediaStream,
-                    status: MediaStreamStatus.CONNECTED,
-                }));
-                isMediaStreaming.set(MediaStreamStatus.CONNECTED);
+            .then((stream) => {
+                mediaStreamStatus.set(MediaStreamStatusEnum.CONNECTED);
+                mediaStream.set(stream);
             })
             .catch((err) => {
                 console.error(`${err.name}: ${err.message}`);
-                isMediaStreaming.set(MediaStreamStatus.DISCONNECTED);
+                mediaStreamStatus.set(MediaStreamStatusEnum.DISCONNECTED);
+                mediaStream.set(null);
             });
     },
     async switchCamera(mediaDevicedID: string) {
@@ -68,26 +50,19 @@ export const mediaStreamActions = {
         };
         await navigator.mediaDevices
             .getUserMedia(constraints)
-            .then((mediaStream) => {
-                mediaStreamState.update((state) => ({
-                    ...state,
-                    mediaStream: mediaStream,
-                    status: MediaStreamStatus.CONNECTED,
-                }));
+            .then((stream) => {
+                mediaStreamStatus.set(MediaStreamStatusEnum.CONNECTED);
+                mediaStream.set(stream)
             })
             .catch((err) => {
                 console.error(`${err.name}: ${err.message}`);
             });
     },
     async stop() {
-        navigator.mediaDevices.getUserMedia({ video: true }).then((mediaStream) => {
-            mediaStream.getTracks().forEach((track) => track.stop());
+        navigator.mediaDevices.getUserMedia({ video: true }).then((stream) => {
+            stream.getTracks().forEach((track) => track.stop());
         });
-        mediaStreamState.update((state) => ({
-            ...state,
-            mediaStream: null,
-            status: MediaStreamStatus.DISCONNECTED,
-        }));
-        isMediaStreaming.set(MediaStreamStatus.DISCONNECTED);
+        mediaStreamStatus.set(MediaStreamStatusEnum.DISCONNECTED);
+        mediaStream.set(null);
     },
 };

frontend/src/lib/store.ts

@@ -1,4 +1,4 @@
 
 import { writable, type Writable } from 'svelte/store';
 
-export const pipelineValues = writable({});
+export const pipelineValues = writable({} as Record<string, any>);

frontend/src/lib/types.ts

@@ -1,13 +1,13 @@
 export const enum FieldType {
-    range = "range",
-    seed = "seed",
-    textarea = "textarea",
-    checkbox = "checkbox",
+    RANGE = "range",
+    SEED = "seed",
+    TEXTAREA = "textarea",
+    CHECKBOX = "checkbox",
 }
 export const enum PipelineMode {
-    image = "image",
-    video = "video",
-    text = "text",
+    IMAGE = "image",
+    VIDEO = "video",
+    TEXT = "text",
 }
 
 export interface FieldProps {

frontend/src/routes/+page.svelte

@@ -8,12 +8,12 @@
   import Button from '$lib/components/Button.svelte';
   import PipelineOptions from '$lib/components/PipelineOptions.svelte';
   import Spinner from '$lib/icons/spinner.svelte';
-  import { isLCMRunning, lcmLiveState, lcmLiveActions, LCMLiveStatus } from '$lib/lcmLive';
+  import { lcmLiveStatus, lcmLiveActions, LCMLiveStatus } from '$lib/lcmLive';
   import {
-    mediaStreamState,
     mediaStreamActions,
-    isMediaStreaming,
-    onFrameChangeStore
+    mediaStreamStatus,
+    onFrameChangeStore,
+    MediaStreamStatusEnum
   } from '$lib/mediaStream';
   import { pipelineValues } from '$lib/store';
 
@@ -30,7 +30,7 @@
     const settings = await fetch(`${PUBLIC_BASE_URL}/settings`).then((r) => r.json());
     pipelineParams = Object.values(settings.input_params.properties);
     pipelineInfo = settings.info.properties;
-    isImageMode = pipelineInfo.input_mode.default === PipelineMode.image;
+    isImageMode = pipelineInfo.input_mode.default === PipelineMode.IMAGE;
     maxQueueSize = settings.max_queue_size;
     pipelineParams = pipelineParams.filter((e) => e?.disabled !== true);
     console.log('PARAMS', pipelineParams);
@@ -38,22 +38,37 @@
   }
   console.log('isImageMode', isImageMode);
 
+  $: {
+    console.log('lcmLiveState', $lcmLiveStatus);
+  }
+  $: {
+    console.log('mediaStreamState', $mediaStreamStatus);
+  }
   // send Webcam stream to LCM if image mode
   $: {
-    if (isImageMode && $lcmLiveState.status === LCMLiveStatus.CONNECTED) {
+    if (
+      isImageMode &&
+      $lcmLiveStatus === LCMLiveStatus.CONNECTED &&
+      $mediaStreamStatus === MediaStreamStatusEnum.CONNECTED
+    ) {
       lcmLiveActions.send($pipelineValues);
       lcmLiveActions.send($onFrameChangeStore.blob);
     }
   }
 
-  // send Webcam stream to LCM
-  $: {
-    if ($lcmLiveState.status === LCMLiveStatus.CONNECTED) {
-      lcmLiveActions.send($pipelineValues);
-    }
-  }
+  $: isLCMRunning = $lcmLiveStatus !== LCMLiveStatus.DISCONNECTED;
+  // $: {
+  //   console.log('onFrameChangeStore', $onFrameChangeStore);
+  // }
+
+  // // send Webcam stream to LCM
+  // $: {
+  //   if ($lcmLiveState.status === LCMLiveStatus.CONNECTED) {
+  //     lcmLiveActions.send($pipelineValues);
+  //   }
+  // }
   async function toggleLcmLive() {
-    if (!$isLCMRunning) {
+    if (!isLCMRunning) {
       if (isImageMode) {
         await mediaStreamActions.enumerateDevices();
         await mediaStreamActions.start();
@@ -112,13 +127,13 @@
     <PipelineOptions {pipelineParams}></PipelineOptions>
     <div class="flex gap-3">
       <Button on:click={toggleLcmLive}>
-        {#if $isLCMRunning}
+        {#if isLCMRunning}
           Stop
         {:else}
           Start
         {/if}
       </Button>
-      <Button disabled={$isLCMRunning} classList={'ml-auto'}>Snapshot</Button>
+      <Button disabled={isLCMRunning} classList={'ml-auto'}>Snapshot</Button>
     </div>
 
     <ImagePlayer>

frontend/svelte.config.js

@@ -1,10 +1,8 @@
 import adapter from '@sveltejs/adapter-static';
 import { vitePreprocess } from '@sveltejs/kit/vite';
-
 /** @type {import('@sveltejs/kit').Config} */
 const config = {
-  preprocess: vitePreprocess(),
-
+  preprocess: vitePreprocess({ postcss: true }),
   kit: {
     adapter: adapter({
       pages: '../public',

user_queue.py

@@ -1,29 +1,52 @@
-from typing import Dict, Union
+from typing import Dict
 from uuid import UUID
 import asyncio
-from PIL import Image
-from typing import Dict, Union
-from PIL import Image
+from fastapi import WebSocket
+from types import SimpleNamespace
+from typing import Dict
+from typing import Union
 
-InputParams = dict
-UserId = UUID
-EventDataContent = Dict[str, InputParams]
+UserDataContent = Dict[UUID, Dict[str, Union[WebSocket, asyncio.Queue]]]
 
 
-class UserDataEvent:
+class UserData:
     def __init__(self):
-        self.data_event = asyncio.Event()
-        self.data_content: EventDataContent = {}
+        self.data_content: Dict[UUID, UserDataContent] = {}
 
-    def update_data(self, new_data: EventDataContent):
-        self.data_content = new_data
-        self.data_event.set()
+    async def create_user(self, user_id: UUID, websocket: WebSocket):
+        self.data_content[user_id] = {
+            "websocket": websocket,
+            "queue": asyncio.Queue(),
+        }
+        await asyncio.sleep(1)
 
-    async def wait_for_data(self) -> EventDataContent:
-        await self.data_event.wait()
-        self.data_event.clear()
-        return self.data_content
+    def check_user(self, user_id: UUID) -> bool:
+        return user_id in self.data_content
 
+    async def update_data(self, user_id: UUID, new_data: SimpleNamespace):
+        user_session = self.data_content[user_id]
+        queue = user_session["queue"]
+        while not queue.empty():
+            try:
+                queue.get_nowait()
+            except asyncio.QueueEmpty:
+                continue
+        await queue.put(new_data)
 
-UserDataEventMap = Dict[UserId, UserDataEvent]
-user_data_events: UserDataEventMap = {}
+    async def get_latest_data(self, user_id: UUID) -> SimpleNamespace:
+        user_session = self.data_content[user_id]
+        queue = user_session["queue"]
+        try:
+            return await queue.get()
+        except asyncio.QueueEmpty:
+            return None
+
+    def delete_user(self, user_id: UUID):
+        if user_id in self.data_content:
+            del self.data_content[user_id]
+
+    def get_user_count(self) -> int:
+        return len(self.data_content)
+
+
+user_data = UserData()

util.py

@@ -20,6 +20,11 @@ def get_pipeline_class(pipeline_name: str) -> ModuleType:
     return pipeline_class
 
 
+def bytes_to_pil(image_bytes: bytes) -> Image.Image:
+    image = Image.open(io.BytesIO(image_bytes))
+    return image
+
+
 def pil_to_frame(image: Image.Image) -> bytes:
     frame_data = io.BytesIO()
     image.save(frame_data, format="JPEG")