This fixes #12 but only tried on Windows

Author: egeakman

mjpeg_streamer/__init__.py

@@ -1,5 +1,5 @@
 from .server import MjpegServer, Server
-from .stream import CustomStream, ManagedStream, Stream
+from .stream import ManagedStream, Stream, StreamBase
 
-__all__ = ["CustomStream", "ManagedStream", "MjpegServer", "Stream", "Server"]
+__all__ = ["StreamBase", "ManagedStream", "MjpegServer", "Stream", "Server"]
 __version__ = "2024.2.8"

mjpeg_streamer/server.py

@@ -23,7 +23,10 @@ async def __call__(self, request: web.Request) -> web.StreamResponse:
                 "Content-Type": "multipart/x-mixed-replace;boundary=image-boundary"
             },
         )
-        await response.prepare(request)
+        try:
+            await response.prepare(request)
+        except (ConnectionResetError, ConnectionAbortedError):
+            pass
         if not viewer_token:
             viewer_token = await self._stream._add_viewer()
             response.set_cookie("viewer_token", viewer_token)
@@ -125,7 +128,4 @@ def stop(self) -> None:
 
 class MjpegServer(Server):
     # Alias for Server, to maintain backwards compatibility
-    def __init__(
-        self, host: Union[str, List[str,]] = "localhost", port: int = 8080
-    ) -> None:
-        super().__init__(host, port)
+    pass

mjpeg_streamer/stream.py

@@ -2,7 +2,7 @@
 import time
 import uuid
 from collections import deque
-from typing import Deque, List, Optional, Set, Tuple, Union
+from typing import Deque, Dict, List, Optional, Set, Tuple, Union
 
 import cv2
 import numpy as np
@@ -14,29 +14,23 @@ def __init__(
         name: str,
         fps: int = 30,
     ) -> None:
-        if type(self) is StreamBase:
-            raise TypeError(
-                "StreamBase is an abstract class and cannot be instantiated."
-            )
         self.name = name.casefold().replace(" ", "_")
         self.fps = fps
         self._frame: np.ndarray = np.zeros((320, 240, 1), dtype=np.uint8)
         self._lock: asyncio.Lock = asyncio.Lock()
         self._frames_buffer: Deque[int] = deque(maxlen=fps)
         self._bandwidth_last_modified_time: float = time.time()
         self._active_viewers: Set[str] = set()
-        self._bandwidth_background_task: Optional[asyncio.Task] = None
+        self._tasks: Dict[str, asyncio.Task] = {"_clear_bandwidth": None}
 
     async def _ensure_background_tasks(self) -> None:
-        if (
-            self._bandwidth_background_task is None
-            or self._bandwidth_background_task.done()
-        ):
-            self._bandwidth_background_task = asyncio.create_task(
-                self.__clear_bandwidth()
-            )
+        for task_name, task in self._tasks.items():
+            if task is None or task.done():
+                self._tasks[task_name] = asyncio.create_task(
+                    eval(f"self.{task_name}()")
+                )
 
-    async def __clear_bandwidth(self) -> None:
+    async def _clear_bandwidth(self) -> None:
         while True:
             await asyncio.sleep(1.0 / self.fps)
             if (
@@ -120,9 +114,6 @@ def set_frame(self, frame: np.ndarray) -> None:
         self._frame = frame
 
 
-CustomStream = StreamBase
-
-
 class Stream(StreamBase):
     def __init__(
         self,
@@ -163,6 +154,7 @@ def __init__(
         mode: str = "fast-on-demand",
         poll_delay_ms: Optional[Union[float, int]] = None,
     ) -> None:
+        super().__init__(name, fps)
         self.source = source
         self.mode = mode
         self._available_modes: List[str,] = ["fast-on-demand", "full-on-demand"]
@@ -175,14 +167,9 @@ def __init__(
         self._cap: cv2.VideoCapture = None
         self._cap_background_task: Optional[asyncio.Task] = None
         self._is_running: bool = False
-        super().__init__(name, fps)
-
-    async def _ensure_background_tasks(self) -> None:
-        await super()._ensure_background_tasks()
-        if self._cap_background_task is None or self._cap_background_task.done():
-            self._cap_background_task = asyncio.create_task(self.__manage_cap_state())
+        self._tasks["_manage_cap_state"] = None
 
-    async def __manage_cap_state(self) -> None:
+    async def _manage_cap_state(self) -> None:
         while True:
             await asyncio.sleep(self.poll_delay_seconds)
             if self.mode == "full-on-demand":
@@ -223,7 +210,6 @@ async def __read_frame(self) -> None:
     async def _process_current_frame(self) -> np.ndarray:
         if not self.has_demand():
             return self._last_processed_frame
-        print("reading frame")
         await self.__read_frame()
         frame = await self._resize_and_encode_frame(
             self._frame,
@@ -248,7 +234,7 @@ def set_quality(self, quality: int) -> None:
 
     def set_frame(self, frame: np.ndarray) -> None:
         raise NotImplementedError(
-            "This method is not available for ManagedStream, use Stream or CustomStream instead."
+            "This method is not available for ManagedStream, use Stream or StreamBase instead."
         )
 
     def change_mode(self, mode: str) -> None:

tests/test_fast_od.py

@@ -0,0 +1,21 @@
+import time
+
+from mjpeg_streamer import ManagedStream, Server
+
+server = Server()
+stream = ManagedStream(
+    "test",
+    source=0,
+    fps=30,
+    size=(640, 480),
+    quality=50,
+    mode="fast-on-demand",
+    poll_delay_ms=None,
+)
+
+server.add_stream(stream)
+server.start()
+stream.start()
+
+while True:
+    time.sleep(1 / 30)

tests/test_full_od.py

@@ -0,0 +1,21 @@
+import time
+
+from mjpeg_streamer import ManagedStream, Server
+
+server = Server()
+stream = ManagedStream(
+    "test",
+    source=0,
+    fps=30,
+    size=(640, 480),
+    quality=50,
+    mode="full-on-demand",
+    poll_delay_ms=None,
+)
+
+server.add_stream(stream)
+server.start()
+stream.start()
+
+while True:
+    time.sleep(1 / 30)

tests/test_stream.py

@@ -0,0 +1,14 @@
+import cv2
+
+from mjpeg_streamer import Server, Stream
+
+capture = cv2.VideoCapture(0)
+
+server = Server()
+stream = Stream("test", fps=30, size=(640, 480), quality=50)
+server.add_stream(stream)
+server.start()
+
+while True:
+    frame = capture.read()[1]
+    stream.set_frame(frame)

tests/test_streambase.py

@@ -0,0 +1,16 @@
+import cv2
+
+from mjpeg_streamer import Server, StreamBase
+
+capture = cv2.VideoCapture(0)
+
+server = Server()
+stream = StreamBase("test", fps=30)
+
+server.add_stream(stream)
+server.start()
+
+while True:
+    frame = capture.read()[1]
+    frame = cv2.imencode(".jpg", frame)[1]
+    stream.set_frame(frame)