OP#9015: Migrate to async across the codebase by converting observers, methods, buffers, writers, and test cases to support AsyncIterator and AsyncSubject.

Author: randy-seng

OTVision/abstraction/observer.py

@@ -1,7 +1,9 @@
-from typing import Callable, TypeVar
+import asyncio
+from typing import Awaitable, Callable, TypeVar
 
 VALUE = TypeVar("VALUE")
 type Observer[T] = Callable[[T], None]
+type AsyncObserver[T] = Callable[[T], Awaitable[None]]
 
 
 class Subject[T]:
@@ -41,3 +43,46 @@ def __init__(self, subject: Subject[T]) -> None:
 
     def register(self, observer: Observer[T]) -> None:
         self._subject.register(observer)
+
+
+class AsyncSubject[T]:
+    """Generic async subject class to handle and notify async observers.
+
+    This class ensures that no duplicate observers can be registered.
+    The order that registered observers are notified is dictated by the order they have
+    been registered. Meaning, first to be registered is first to be notified.
+    All observers are awaited concurrently using asyncio.gather().
+    """
+
+    def __init__(self) -> None:
+        self._observers: list[AsyncObserver[T]] = []
+
+    def register(self, observer: AsyncObserver[T]) -> None:
+        """Listen to changes of subject.
+
+        Args:
+            observer (AsyncObserver[T]): the observer to be registered. This must be an
+                async `Callable` that returns an `Awaitable`.
+        """
+        new_observers = self._observers.copy()
+        new_observers.append(observer)
+        self._observers = list(dict.fromkeys(new_observers))
+
+    async def notify(self, value: T) -> None:
+        """Notifies observers about the value asynchronously.
+
+        All observers are notified concurrently using asyncio.gather().
+
+        Args:
+            value (T): value to notify the observer with.
+        """
+        if self._observers:
+            await asyncio.gather(*[observer(value) for observer in self._observers])
+
+
+class AsyncObservable[T]:
+    def __init__(self, subject: AsyncSubject[T]) -> None:
+        self._subject = subject
+
+    def register(self, observer: AsyncObserver[T]) -> None:
+        self._subject.register(observer)

OTVision/application/buffer.py

@@ -10,10 +10,10 @@ def __init__(self) -> None:
 
     async def filter(self, pipe: AsyncIterator[T]) -> AsyncIterator[T]:
         async for element in pipe:
-            self.buffer(element)
+            await self.buffer(element)
             yield element
 
-    def buffer(self, to_buffer: T) -> None:
+    async def buffer(self, to_buffer: T) -> None:
         self._buffer.append(to_buffer)
 
     def _get_buffered_elements(self) -> list[T]:
@@ -24,5 +24,5 @@ def _reset_buffer(self) -> None:
         self._buffer = list()
 
     @abstractmethod
-    def on_flush(self, event: OBSERVING_TYPE) -> None:
+    async def on_flush(self, event: OBSERVING_TYPE) -> None:
         raise NotImplementedError

OTVision/detect/builder.py

@@ -2,7 +2,7 @@
 from argparse import ArgumentParser
 from functools import cached_property
 
-from OTVision.abstraction.observer import Subject
+from OTVision.abstraction.observer import AsyncSubject
 from OTVision.application.config import Config, DetectConfig
 from OTVision.application.config_parser import ConfigParser
 from OTVision.application.configure_logger import ConfigureLogger
@@ -127,7 +127,7 @@ def frame_count_provider(self) -> FrameCountProvider:
     @cached_property
     def otdet_file_writer(self) -> OtdetFileWriter:
         return OtdetFileWriter(
-            subject=Subject[OtdetFileWrittenEvent](),
+            subject=AsyncSubject[OtdetFileWrittenEvent](),
             builder=self.otdet_builder,
             get_current_config=self.get_current_config,
             current_object_detector_metadata=self.current_object_detector_metadata,
@@ -147,7 +147,7 @@ def current_object_detector(self) -> CurrentObjectDetector:
 
     @cached_property
     def detected_frame_buffer(self) -> DetectedFrameBuffer:
-        return DetectedFrameBuffer(subject=Subject[DetectedFrameBufferEvent]())
+        return DetectedFrameBuffer(subject=AsyncSubject[DetectedFrameBufferEvent]())
 
     @cached_property
     def detected_frame_producer(self) -> DetectedFrameProducer:

OTVision/detect/detected_frame_buffer.py

@@ -1,7 +1,7 @@
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 
-from OTVision.abstraction.observer import Observable, Subject
+from OTVision.abstraction.observer import AsyncObservable, AsyncSubject
 from OTVision.application.buffer import Buffer
 from OTVision.domain.frame import DetectedFrame
 
@@ -51,25 +51,25 @@ class DetectedFrameBufferEvent:
 
 
 class DetectedFrameBuffer(
-    Buffer[DetectedFrame, FlushEvent], Observable[DetectedFrameBufferEvent]
+    Buffer[DetectedFrame, FlushEvent], AsyncObservable[DetectedFrameBufferEvent]
 ):
-    def __init__(self, subject: Subject[DetectedFrameBufferEvent]) -> None:
+    def __init__(self, subject: AsyncSubject[DetectedFrameBufferEvent]) -> None:
         Buffer.__init__(self)
-        Observable.__init__(self, subject)
+        AsyncObservable.__init__(self, subject)
 
-    def on_flush(self, event: FlushEvent) -> None:
+    async def on_flush(self, event: FlushEvent) -> None:
         buffered_elements = self._get_buffered_elements()
-        self._notify_observers(buffered_elements, event)
+        await self._notify_observers(buffered_elements, event)
         self._reset_buffer()
 
-    def _notify_observers(
+    async def _notify_observers(
         self, elements: list[DetectedFrame], event: FlushEvent
     ) -> None:
-        self._subject.notify(
+        await self._subject.notify(
             DetectedFrameBufferEvent(
                 source_metadata=event.source_metadata, frames=elements
             )
         )
 
-    def buffer(self, to_buffer: DetectedFrame) -> None:
+    async def buffer(self, to_buffer: DetectedFrame) -> None:
         self._buffer.append(to_buffer.without_image())

OTVision/detect/file_based_detect_builder.py

@@ -1,6 +1,6 @@
 from functools import cached_property
 
-from OTVision.abstraction.observer import Subject
+from OTVision.abstraction.observer import AsyncSubject, Subject
 from OTVision.application.event.new_video_start import NewVideoStartEvent
 from OTVision.detect.builder import DetectBuilder
 from OTVision.detect.detected_frame_buffer import FlushEvent
@@ -19,7 +19,7 @@ class FileBasedDetectBuilder(DetectBuilder):
     @cached_property
     def input_source(self) -> VideoSource:
         return VideoSource(
-            subject_flush=Subject[FlushEvent](),
+            subject_flush=AsyncSubject[FlushEvent](),
             subject_new_video_start=Subject[NewVideoStartEvent](),
             get_current_config=self.get_current_config,
             frame_rotator=self.frame_rotator,

OTVision/detect/otdet_file_writer.py

@@ -2,7 +2,7 @@
 from dataclasses import dataclass
 from pathlib import Path
 
-from OTVision.abstraction.observer import Observer, Subject
+from OTVision.abstraction.observer import AsyncObserver, AsyncSubject
 from OTVision.application.detect.current_object_detector_metadata import (
     CurrentObjectDetectorMetadata,
 )
@@ -45,7 +45,7 @@ class OtdetFileWriter:
 
     def __init__(
         self,
-        subject: Subject[OtdetFileWrittenEvent],
+        subject: AsyncSubject[OtdetFileWrittenEvent],
         builder: OtdetBuilder,
         get_current_config: GetCurrentConfig,
         current_object_detector_metadata: CurrentObjectDetectorMetadata,
@@ -57,7 +57,7 @@ def __init__(
         self._current_object_detector_metadata = current_object_detector_metadata
         self._save_path_provider = save_path_provider
 
-    def write(self, event: DetectedFrameBufferEvent) -> None:
+    async def write(self, event: DetectedFrameBufferEvent) -> None:
         """Writes detection results to a file in OTDET format.
 
         Processes the detected frames and associated metadata, builds the OTDET
@@ -118,23 +118,23 @@ def write(self, event: DetectedFrameBufferEvent) -> None:
 
         finished_msg = "Finished detection"
         log.info(finished_msg)
-        self.__notify(
+        await self.__notify(
             num_frames=actual_frames,
             builder_config=builder_config,
             save_location=detections_file,
         )
 
-    def __notify(
+    async def __notify(
         self, num_frames: int, builder_config: OtdetBuilderConfig, save_location: Path
     ) -> None:
-        self._subject.notify(
+        await self._subject.notify(
             OtdetFileWrittenEvent(
                 number_of_frames=num_frames,
                 otdet_builder_config=builder_config,
                 save_location=save_location,
             )
         )
 
-    def register_observer(self, observer: Observer[OtdetFileWrittenEvent]) -> None:
+    def register_observer(self, observer: AsyncObserver[OtdetFileWrittenEvent]) -> None:
         """Register an observer to receive notifications about otdet file writes.."""
         self._subject.register(observer)

OTVision/detect/rtsp_based_detect_builder.py

@@ -1,6 +1,6 @@
 from functools import cached_property
 
-from OTVision.abstraction.observer import Subject
+from OTVision.abstraction.observer import AsyncSubject, Subject
 from OTVision.application.config import StreamConfig
 from OTVision.application.event.new_video_start import NewVideoStartEvent
 from OTVision.detect.builder import DetectBuilder
@@ -33,7 +33,7 @@ def stream_config(self) -> StreamConfig:
     @cached_property
     def input_source(self) -> RtspInputSource:
         return RtspInputSource(
-            subject_flush=Subject[FlushEvent](),
+            subject_flush=AsyncSubject[FlushEvent](),
             subject_new_video_start=Subject[NewVideoStartEvent](),
             datetime_provider=self.datetime_provider,
             frame_counter=Counter(),

OTVision/detect/rtsp_input_source.py

@@ -1,3 +1,4 @@
+import asyncio
 import socket
 from datetime import datetime, timedelta
 from time import sleep
@@ -13,7 +14,7 @@
 )
 from numpy import ndarray
 
-from OTVision.abstraction.observer import Subject
+from OTVision.abstraction.observer import AsyncSubject, Subject
 from OTVision.application.config import (
     DATETIME_FORMAT,
     Config,
@@ -87,7 +88,7 @@ def fps(self) -> float:
 
     def __init__(
         self,
-        subject_flush: Subject[FlushEvent],
+        subject_flush: AsyncSubject[FlushEvent],
         subject_new_video_start: Subject[NewVideoStartEvent],
         datetime_provider: DatetimeProvider,
         frame_counter: Counter,
@@ -153,10 +154,10 @@ async def produce(self) -> AsyncIterator[Frame]:
                         occurrence=occurrence,
                     )
                     if self.flush_condition_met():
-                        self._notify_flush_observers()
+                        await self._notify_flush_observers()
                         self._outdated = True
                         self._frame_counter.reset()
-            self._notify_flush_observers()
+            await self._notify_flush_observers()
         except InvalidRtspUrlError as cause:
             logger().error(cause)
 
@@ -209,7 +210,7 @@ def start(self) -> None:
     def flush_condition_met(self) -> bool:
         return self.current_frame_number % self.flush_buffer_size == 0
 
-    def _notify_flush_observers(self) -> None:
+    async def _notify_flush_observers(self) -> None:
         frame_width = self._get_width()
         frame_height = self._get_height()
         frames = (
@@ -219,7 +220,7 @@ def _notify_flush_observers(self) -> None:
         )
         duration = timedelta(seconds=round(frames / self.fps))
         output = self.create_output()
-        self.subject_flush.notify(
+        await self.subject_flush.notify(
             FlushEvent.create(
                 source=self.rtsp_url,
                 output=output,
@@ -256,7 +257,9 @@ def create_output(self) -> str:
     def notify_new_config(self, config: NewOtvisionConfigEvent) -> None:
         try:
             logger().debug("New OTVision config detected. Flushing buffers...")
-            self._notify_flush_observers()
+
+            # Create task to handle async flush notification
+            asyncio.create_task(self._notify_flush_observers())
         except NoConfigurationFoundError:
             logger().info("No configuration found for RTSP stream. Skipping flushing.")
 

OTVision/detect/video_input_source.py

@@ -7,7 +7,7 @@
 from av.container.input import InputContainer
 from tqdm.asyncio import tqdm
 
-from OTVision.abstraction.observer import Subject
+from OTVision.abstraction.observer import AsyncSubject, Subject
 from OTVision.application.config import DATETIME_FORMAT, Config
 from OTVision.application.detect.timestamper import Timestamper
 from OTVision.application.event.new_video_start import NewVideoStartEvent
@@ -59,7 +59,7 @@ def _start_time(self) -> datetime | None:
 
     def __init__(
         self,
-        subject_flush: Subject[FlushEvent],
+        subject_flush: AsyncSubject[FlushEvent],
         subject_new_video_start: Subject[NewVideoStartEvent],
         get_current_config: GetCurrentConfig,
         frame_rotator: AvVideoFrameRotator,
@@ -137,7 +137,7 @@ async def produce(self) -> AsyncIterator[Frame]:
                                 }
                             )
                         counter += 1
-                self.notify_flush_event_observers(video_file, video_fps)
+                await self.notify_flush_event_observers(video_file, video_fps)
                 self._on_video_finished(video_file)
             except Exception as e:
                 log.error(f"Error processing {video_file}", exc_info=e)
@@ -192,7 +192,7 @@ def __overwrite_existing_detection_file(self, detections_file: Path) -> bool:
             return False
         return True
 
-    def notify_flush_event_observers(
+    async def notify_flush_event_observers(
         self, current_video_file: Path, video_fps: float
     ) -> None:
         if expected_duration := self._current_config.detect.expected_duration:
@@ -205,7 +205,7 @@ def notify_flush_event_observers(
             current_video_file, start_time=self._start_time
         )
 
-        self.subject_flush.notify(
+        await self.subject_flush.notify(
             FlushEvent.create(
                 source=str(current_video_file),
                 output=str(current_video_file),

OTVision/domain/video_writer.py

@@ -22,7 +22,7 @@ def close(self) -> None:
         raise NotImplementedError
 
     @abstractmethod
-    def notify_on_flush_event(self, event: FlushEvent) -> None:
+    async def notify_on_flush_event(self, event: FlushEvent) -> None:
         raise NotImplementedError
 
     @abstractmethod