displayarray.frame

Read Updates

read_updates(*vids, callbacks: Union[Dict[Any, Union[Callable[[numpy.ndarray], Optional[numpy.ndarray]], List[Callable[[numpy.ndarray], Optional[numpy.ndarray]]]]], List[Callable[[numpy.ndarray], Optional[numpy.ndarray]]], Callable[[numpy.ndarray], Optional[numpy.ndarray]], None] = None, fps_limit=inf, size=(-1, -1), end_callback: Callable[[], bool] = <function <lambda>>, blocking=True)[source]

Read back all updates from the requested videos.

Example usage:

>>> from examples.videos import test_video
>>> f = 0
>>> for f, r in enumerate(read_updates(test_video, end_callback=lambda :f==2)):
...   print(f"Frame:{f}. Array:{r}")

Frame Passing

Handles publishing arrays, videos, and cameras.

CamCtrl handles sending and receiving commands to specific camera (or array/video) publishers VideoHandlerThread updates the frames for the global displayer, since OpenCV can only update on the main thread get_cam_ids gets the ids for all cameras that OpenCV can detect pub_cam_thread continually publishes updates to arrays, videos, and cameras np_cam simulates numpy arrays as OpenCV cameras

class FrameUpdater(video_source: Union[int, str, numpy.ndarray] = 0, callbacks: Union[List[Callable[[numpy.ndarray], Optional[numpy.ndarray]]], Callable[[numpy.ndarray], Optional[numpy.ndarray]], None] = None, request_size: Tuple[int, int] = (-1, -1), high_speed: bool = True, fps_limit: float = 240)[source]

Thread for updating frames from a video source.

display(callbacks: List[Callable[[numpy.ndarray], Any]] = None)[source]

Start default display operation.

For multiple video sources, please use something outside of this class.

Parameters

callbacks – List of callbacks to be run on frames before displaying to the screen.

loop()[source]

Continually get frames from the video publisher, run callbacks on them, and listen to commands.

read_updates(*vids, callbacks: Union[Dict[Any, Union[Callable[[numpy.ndarray], Optional[numpy.ndarray]], List[Callable[[numpy.ndarray], Optional[numpy.ndarray]]]]], List[Callable[[numpy.ndarray], Optional[numpy.ndarray]]], Callable[[numpy.ndarray], Optional[numpy.ndarray]], None] = None, fps_limit=inf, size=(-1, -1), end_callback: Callable[[], bool] = <function <lambda>>, blocking=True)[source]

Read back all updates from the requested videos.

Example usage:

>>> from examples.videos import test_video
>>> f = 0
>>> for f, r in enumerate(read_updates(test_video, end_callback=lambda :f==2)):
...   print(f"Frame:{f}. Array:{r}")
get_cam_ids() → List[int][source]

Get all cameras that OpenCV can currently detect.

class NpCam(img)[source]

Add OpenCV camera controls to a numpy array.

static get(*args, **kwargs)[source]

Get OpenCV args. Currently only a fake CAP_PROP_FRAME_COUNT to fix detecting video ends.

isOpened()[source]

Hack to tell OpenCV we’re opened until we call release.

read()[source]

Read back the numpy array in standard “did it work”, “the array”, OpenCV format.

release()[source]

Let OpenCV know we’re finished.

set(*args, **kwargs)[source]

Set CAP_PROP_FRAME_WIDTH or CAP_PROP_FRAME_HEIGHT to scale a numpy array to that size.

pub_cam_thread(cam_id: Union[int, str], request_ize: Tuple[int, int] = (-1, -1), high_speed: bool = True, fps_limit: float = 240) → threading.Thread[source]

Run pub_cam_loop in a new thread. Starts on creation.