Make it possible to use worker thread with --autoreload

[MarcSkovMadsen]

I'm trying to help a user in https://discourse.holoviz.org/t/streaming-local-video/6929. As I've seen questions about using server camera multiple times on discord I think its a good candidate for an Intermediate Streaming Tutorial.

I'm using a worker thread in a seperate module. But I've run into the issue that the --autoreload does not delete the existing worker thread when it reloads the module and starts a new worker thread. This is a big problem when using a camera as a camera can only be instantiated and used once, i.e. lots of exceptions are starting to be raised when two threads are trying to use it at the same time.

Please explain and document how to stop a thread when the module it is in is being autoreloaded.

Please do this by extending the documentation in Setup Manual Threading to include a reference example for setting up a thread outside of the main app.py file. I.e. a thread that is run once and results shared between all sessions.

Reproducible Example

pip install opencv panel pillow

server_video_stream.py

import cv2 as cv

import panel as pn
from PIL import Image
import param
import time
import threading
import logging
import sys

FORMAT = "%(asctime)s | %(levelname)s | %(name)s | %(message)s"

@pn.cache
def get_logger(name, format_=FORMAT, level=logging.INFO):
    logger = logging.getLogger(name)

    logger.handlers.clear()

    handler = logging.StreamHandler()
    handler.setStream(sys.stdout)
    formatter = logging.Formatter(format_)
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.propagate = False

    logger.setLevel(level)
    logger.info("Logger successfully configured")
    return logger

class AllReadyStarted(Exception):
    """Raised if the camera is already started"""

class CannotOpenCamera(Exception):
    """Raised if the camera cannot be opened"""

class CannotReadCamera(Exception):
    """Raised if the camera cannot be read"""

class ServerVideoStream(pn.viewable.Viewer):
    value = param.Parameter(doc="String representation of the current snapshot")
    paused = param.Boolean(default=False, doc="Whether the video stream is paused")
    fps = param.Number(10, doc="Frames per second", inclusive_bounds=(0, None))
    camera_index = param.Integer(0, doc="The index of the active camera")


    def __init__(self, log_level=logging.INFO, **params):
        super().__init__(**params)

        self._cameras={}
        
        self._stop_thread = False
        self._thread = threading.Thread(target=self._take_images)
        self._thread.daemon=True

        self._logger = get_logger(f"ServerVideoStream {id(self)}", level=log_level)
        
    def start(self, camera_indicies):
        if camera_indicies:
            for index in camera_indicies:
                self._logger.debug("Getting Camera %s", index)
                self.get_camera(index)

        if not self._thread.is_alive():
            self._logger.debug("Starting Camera Thread")
            self._thread.start()

    def __panel__(self):
        settings = pn.Column(
            self.param.paused,
            self.param.fps,
            self.param.camera_index,
            width=300,
        )
        image = pn.pane.Image(self.param.value, sizing_mode="stretch_both")
        return pn.Row(settings, image)
        
    @staticmethod
    def _cv2_to_pil(bgr_image):
        rgb_image = cv.cvtColor(bgr_image, cv.COLOR_BGR2RGB)
        image = Image.fromarray(rgb_image)
        return image
    
    def get_camera(self, index):
        if index in self._cameras:
            return self._cameras[index]
        
        self._logger.debug("Camera %s Opening", index)
        cap = cv.VideoCapture(index)

        if not cap.isOpened():
            raise CannotOpenCamera(f"Cannot open the camera {index}")
        
        self._cameras[index]=cap
        self._logger.debug("Camera %s Opened", index)
        return cap
        

    def _take_image(self):
        self._logger.debug("Taking image with camera %s", self.camera_index)
        camera = self.get_camera(self.camera_index)
        ret, frame = camera.read()
        if not ret:
            raise CannotReadCamera("Are you sure the camera exists and is not being read by other processes?")
        else:
            self.value = self._cv2_to_pil(frame)
    
    def _take_images(self):
        while not self._stop_thread:
            start_time = time.time()  # Record the start time of the capture
            if not self.paused:
                try:
                    self._take_image()
                except Exception as ex:
                    self._logger.error("Error. Could not take image", exc_info=ex)
            
            if self.fps>0:
                interval = 1/self.fps
                elapsed_time = time.time() - start_time
                sleep_time = max(0, interval - elapsed_time)
                time.sleep(sleep_time)
    
    def __del__(self):
        self._logger.debug("Stopping Camera Thread")
        self._stop_thread=True
        if self._thread.is_alive():
            self._thread.join()
        self._logger.debug("Releasing Cameras")
        for camera in self._cameras.values():
            camera.release()
        cv.destroyAllWindows()

# some text
server_video_stream = ServerVideoStream(fps=3, log_level=logging.DEBUG)
server_video_stream.start(camera_indicies=[0])

# https://discourse.holoviz.org/t/best-practice-for-displaying-high-resolution-camera-images-captured-on-server/4285/12
# https://discourse.holoviz.org/t/streaming-local-video/6929

script.py

import panel as pn
from server_video_stream import server_video_stream

pn.extension()
server_video_stream.servable()

panel serve script.py --autoreload

• Open the app in your browser
• Change # some text in server_video_stream.py to # some texts.
• Watch the application reload and exceptions being raised in the terminal.

video_stream_issue.mp4

[philippjfr]

As discussed elsewhere, I'd consider creating a worker thread a singleton in a module as bad practice. If you want to share a worker thread put it in the pn.state.cache.

[MarcSkovMadsen]

I reopen as this should be documented which is the initial request. I plan to document this.