AI webcam utility for online meetings. Make your introduction fun!
"""Definition of constants and types"""

from typing import NewType

# Definition of maximum image size for virtual camera.
# Higher resolutions are usually not supported on online meeting tools
# 720p:(1280, 720)  540p:(960, 540)
MAX_WIDTH = 1280
MAX_HEIGHT = 720

# Definition of plugin types (avoid spelling errors with explicit type)
PluginType = NewType("PluginType", str)
DEPTHAI = PluginType("depthai")
WEBCAM = PluginType("webcam")

src/meetingcam/constants.py
nengelmann/MeetingCam
#!/usr/bin/env python3

"""This script facilitates the management and utilization of virtual and real camera devices."""

import re
import signal
import sys
from pathlib import Path
from types import FrameType
from typing import Any

import depthai
from constants import DEPTHAI, WEBCAM
from rich.console import Console
from rich.markdown import Markdown
from rich.style import Style
from utils import DepthaiCapture, VideoCapture
from v4l2ctl import V4l2Capabilities, V4l2Device


class V4l2Capture(V4l2Device):
    """Handle real camera devices.

    Inherits from v4l2ctl.V4l2Device and extends functionalities with
    methods to safely enter and exit.
    """

    def __enter__(self) -> V4l2Device:
        """Enter method for context management, returning self."""
        return self

    def __exit__(self, *args: tuple[Any]) -> None:
        """Exit method for context management, releasing the devices."""
        self.flush()
        self.close()


class DeviceHandler:
    """Base class for management and mapping of real and virtual devices."""

    def __init__(self) -> None:
        """Initialize the DeviceHandler by setting up necessary mappings and signal handler."""
        signal.signal(signal.SIGINT, self._interrupt)

        self.mapping = self.device_mapping()
        self.pprint = DevicePrinter()

    def _interrupt(self, signal: int, frame: FrameType | None) -> None:
        """Handle SIGINT signal by stopping the device and exiting the program."""
        self.pprint.device_stopped()
        sys.exit(0)

    def init_device(self, device_path: str | None) -> tuple[str, str]:
        """Initialize the device by analyzing the device path and mapping it to a real or virtual device.

        Args:
            device_path -- the path of the device to initialize, None if not specified

        Returns:
            A tuple containing paths to the real and virtual devices.
        """
        device_map = {
            device["path_real"]: device["path_virtual"]
            for device in self.mapping.values()
        }
        if device_path:
            # device is specified as argument in command line
            if device_path in device_map.keys():
                # check if specified device is available in real devices and has virtual counterpart
                self.real_path = device_path
                virtual_path = device_map[device_path]
                return virtual_path
            else:
                # if not print that the specified device needs to have a virtual counterpart and how to do it
                self.pprint.device_not_available(device_path)
                self.pprint.available_devices(
                    self.mapping, self.available_devices_real
                )
                self.pprint.add_virtual_devices(
                    self.available_devices_real, self.type
                )
                self.pprint.run_main_command(self.mapping, self.type)
                sys.exit()
        else:
            # device is not specified
            if device_map:
                # no device is specified, but a real device with virtual counterpart is available
                # print in the available devices and point out that the first available device with virtual counterpart will be used
                device = list(device_map.items())[0]
                self.real_path = device[0]
                virtual_path = device[1]
                self.pprint.available_devices(
                    self.mapping, self.available_devices_real
                )
                self.pprint.run_main_command(self.mapping, self.type)
                sys.exit()
            else:
                # no device is specified and no real device with virtual counterpart is available
                self.pprint.device_not_available(device_path)
                self.pprint.available_devices(
                    self.mapping, self.available_devices_real
                )
                self.pprint.add_virtual_devices(
                    self.available_devices_real, self.type
                )
                self.pprint.run_main_command(self.mapping, self.type)
                sys.exit()

    def device_mapping(self) -> dict[int, dict[str, str]]:
        """Create a mapping from real camera devices to virtual camera devices.

        Returns:
            A dictionary that contains information about real and virtual devices.
        """

        device_map = {}

        (paths_real, labels_real) = self.available_devices_real
        (paths_virtual, labels_virtual) = self.available_devices_virtual

        # extract the IDs (first number in virtual camera label) of available virtual cameras for mapping it to real cameras
        # the mapping index is same as real cameras path id. e.g. /dev/video_n_ with n as id
        if self.type == WEBCAM:
            mapping_ids = [
                int(re.findall(r"\d+", label)[0]) for label in labels_virtual
            ]
            ids_real = [
                int(path.replace("/dev/video", "")) for path in paths_real
            ]
            try:
                mapping_indices = [ids_real.index(id) for id in mapping_ids]
            except:
                print(
                    "The virtual cameras are not matching with attached real"
                    " cameras. Has a camera be unplugged?\n"
                )
                print(
                    "The specified virtual label is"
                    f" {str(labels_virtual)[1:-1]}."
                )
                print(
                    "May you wanted to run MeetingCam with --depthai argument?"
                )
                sys.exit(0)

        elif self.type == DEPTHAI:
            mapping_indices = []
            for i, mxid in enumerate(paths_real):
                for j, label in enumerate(labels_virtual):
                    if mxid in label:
                        mapping_indices.append(j)
        else:
            raise NotImplementedError(
                "Device type needs to be either 'webcam' or 'depthai'."
            )

        for i, j in enumerate(mapping_indices):
            device_map[i] = {
                "path_real": paths_real[j],
                "label_real": labels_real[j],
                "path_virtual": paths_virtual[i],
                "label_virtual": labels_virtual[i],
            }

        return device_map

    def get_available(self, real: bool) -> tuple[list[str], list[str]]:
        """Return a list of available real or virtual camera devices based on the input flag.

        Args:
            real --- a flag to determine whether to return real or virtual devices.

        Returns:
            A tuple containing lists of device paths and labels.
        """

        device_paths = []
        labels = []

        with V4l2Capture() as v4l2:
            for device in v4l2.iter_devices(skip_links=True):
                if (
                    real
                    and not self._is_virtual_device(device)
                    and V4l2Capabilities.VIDEO_CAPTURE in device.capabilities
                ):
                    device_path, label = self._get_device_info(device)
                elif (
                    not real
                    and self._is_virtual_device(device)
                    and V4l2Capabilities.VIDEO_OUTPUT in device.capabilities
                ):
                    device_path, label = self._get_device_info(device)
                else:
                    continue

                labels.append(label)
                device_paths.append(device_path)

        return device_paths, labels

    def device_running(self) -> None:
        """Print the running devices."""
        self.pprint.device_running()

    def _is_virtual_device(self, device: V4l2Device) -> bool:
        """Check if the device is a virtual device.

        Args:
            device (V4l2Device) --- the device to check.

        Returns:
            True if the device is virtual, False otherwise.
        """
        virtual = True if "platform:v4l2loopback" in str(device.bus) else False
        return virtual

    def _get_device_info(self, device: V4l2Device) -> tuple[str, str]:
        """Extract and return device information including path and label.

        Args:
            device (V4l2Device) --- the device to extract information from.

        Returns:
            A tuple containing the device path (str) and label (str).

        Raises:
            AssertionError: If the device path does not follow the expected format or does not exist.
        """
        device_path = str(device.device)
        card_label = str(device.name)

        assert device_path.__contains__(
            "/dev/video"
        ), "Device name '{device_name}' should be of format '/dev/video_n_'."
        assert Path(
            device_path
        ).exists(), "Device path '{device_name}' does not exist."
        return device_path, card_label


class WebcamDevice(DeviceHandler):
    """Handles the management and mapping of real and virtual webcam devices."""

    def __init__(self) -> None:
        """Initialize the WebcamDevice class."""

        self.type = WEBCAM
        self.real_path = None
        self.available_devices_real = None
        self.available_devices_virtual = None

        self.update_available()
        super().__init__()

    def get_device(self):
        """Get a VideoCapture class instance can be used for webcam image acquisition.

        Raises:
            LookupError in case the device path is incorrect.

        Returns:
            A VideoCapture class instance.
        """
        if self.real_path:
            return VideoCapture(self.real_path)
        else:
            raise LookupError(
                "self.real_path is {self.real_path}. Device initialization"
                " 'self.init_device(device_path)' is needed before a device"
                " instance is created."
            )

    def update_available(self) -> None:
        """Update the list of available real and virtual devices."""
        self.available_devices_real = self.get_available(real=True)
        self.available_devices_virtual = self.get_available(real=False)


class DepthaiDevice(DeviceHandler):
    """Handles the management and mapping of depthai and virtual devices."""

    def __init__(self, pipeline) -> None:
        """Initialize the DepthaiDevice class."""

        self.type = DEPTHAI
        self.real_path = None
        self.available_devices_real = None
        self.available_devices_virtual = None

        self.update_available()
        super().__init__()

        self.usb_speed = depthai.UsbSpeed.SUPER_PLUS
        self.device_info = None
        self.pipeline = pipeline

    def get_device(self):
        """Get a DepthaiCapture class instance can be used for depthai image acquisition.

        Raises:
            LookupError in case the device path is incorrect.

        Returns:
            A DepthaiCapture class instance.
        """
        if self.real_path:
            self.device_info = depthai.DeviceInfo(self.real_path)
            return DepthaiCapture(
                pipeline=self.pipeline,
                deviceInfo=self.device_info,
                maxUsbSpeed=self.usb_speed,
            )
        else:
            raise LookupError(
                "self.real_path is {self.real_path}. Device initialization"
                " 'self.init_device(device_path)' is needed before a device"
                " instance is created."
            )

    def update_available(self) -> None:
        """Update the list of available real and virtual devices."""
        self.available_devices_real = self.get_available_depthai()
        self.available_devices_virtual = self.get_available(real=False)

    def get_available_depthai(self) -> dict:
        """Get available depthai camera devices.

        Returns:
            Device paths and labels.
        """
        device_paths = []
        labels = []

        device_info = depthai.Device.getAllAvailableDevices()

        for device in device_info:
            labels.append(f"OAK Device on port {device.name}")
            device_paths.append(device.mxid)

        return device_paths, labels


class DevicePrinter:
    """Handles printing functions for device handling."""

    def __init__(self) -> None:
        """
        Initialize the DevicePrinter instance.

        Instantiates a Console object and initializes styles for different types of messages (danger, warning, ok).
        """
        self.console = Console()
        self.danger_style = Style(color="red", blink=True, bold=True)
        self.warning_style = Style(color="yellow", blink=True, bold=True)
        self.ok_style = Style(color="green", blink=True, bold=True)

    def available_devices(
        self,
        device_map: dict[int, dict[str, Any]],
        available_devices_real: tuple[
            list[str], list[str], list[int], list[str]
        ],
    ) -> None:
        """
        Print available real and virtual devices to the console.

        Args:
            device_map --- a mapping of device ids to device properties including paths and labels.
            available_devices_real --- a tuple containing lists of paths, labels, ids, and another attribute of real devices.

        The method extracts real and mapped device information and formats them for console printing.
        """

        (paths_real, labels_real) = available_devices_real

        devices_real = [
            "|" + label + "|" + str(path) + "|"
            for label, path in zip(labels_real, paths_real)
        ]
        devices_real = (
            str(devices_real).replace("'", "").replace(", ", "\n")[1:-1]
        )

        devices_mapped = [
            "|"
            + v["label_real"]
            + "|"
            + v["path_real"]
            + " -> |"
            + v["label_virtual"]
            + "|"
            + v["path_virtual"]
            + "|"
            for v in device_map.values()
        ]
        devices_mapped = (
            str(devices_mapped).replace("'", "").replace(", ", "\n")[1:-1]
        )

        text = f"""
        # Available devices
        
        | Device      | Path       |
        |-------------|------------|
        {devices_real}
        
        # Available devices with virtual counterpart
        
        | Device      | Path       | Virtual Device | Virtual Path |
        |-------------|------------|----------------|--------------|
        {devices_mapped}

        """
        self._print(text)

    def add_virtual_devices(
        self,
        available_devices_real: tuple[list[str], list[str]],
        type: str | int,
    ) -> None:
        """
        Print instructions to add virtual devices to the console.

        Args:
            available_devices_real --- a tuple containing lists of paths, labels, ids, and another attribute of real devices.

        The method extracts device labels and ids to create instructions for adding virtual devices and prints them to the console.
        """
        (device_paths, labels) = available_devices_real

        cli_cmd_single = []
        labels_str = []

        if type == WEBCAM:
            ids = [
                int(path.replace("/dev/video", "")) for path in device_paths
            ]
            vd_nrs = ids
            for idx, label in zip(ids, labels):
                cli_cmd_single.append(
                    "`sudo modprobe v4l2loopback devices=1"
                    f" video_nr={idx} card_label='MeetingCam{idx} {label}'`"
                )
                labels_str.append(f"MeetingCam{idx} {label}")

            labels_str = str(labels_str).replace(", ", ",")[1:-1]
            vd_nrs = str(vd_nrs).replace(" ", "")[1:-1]
            cli_cmd_multi = (
                "`sudo modprobe v4l2loopback"
                f" devices={len(ids)} video_nr={str(vd_nrs)} card_label='{labels_str[1:-1]}'`"
            )
        elif type == DEPTHAI:
            ids = device_paths
            for idx, label in zip(ids, labels):
                cli_cmd_single.append(
                    "`sudo modprobe v4l2loopback devices=1"
                    f" card_label='MeetingCam{idx} {label}'`"
                )
                labels_str.append(f"MeetingCam{idx} {label}")

            labels_str = str(labels_str).replace(", ", ",")[1:-1]
            cli_cmd_multi = (
                "`sudo modprobe v4l2loopback"
                f" devices={len(ids)} card_label='{labels_str[1:-1]}'`"
            )

        else:
            raise NotImplementedError(
                "Device type needs to be either 'webcam' or 'depthai'."
            )

        cli_cmd_single = (
            str(cli_cmd_single).replace(", ", "\n\n").replace('"', "")[1:-1]
        )

        self.console.print(
            "Follow the instruction below! :arrow_down:", style=self.ok_style
        )

        text = f"""
        # Add virtual devices

        The following describes on how to add virtual devices. A virtual device is needed to publish a modified camera stream.
        You can add a single device or multiple devices.
        
        ## Add a singel device
        Add a single device by running **one** of the following commands.
        
        {cli_cmd_single}
        
        ## Add multiple devices
        Add all your devices by running
        
        {cli_cmd_multi}
        
        You can **reset** the **virtual devices** running **`sudo modprobe -r v4l2loopback`**.
        If this isn't working you need to restart your system.
        """

        self._print(text)

    def device_not_available(self, device: V4l2Device) -> None:
        """Print a warning message indicating the specified device is not available.

        Args:
            device (V4l2Device) --- the device that is not available.
        """
        text = f"""
        **Device not available** 
        The specified device '{device}' is not available or does not have a virtual counterpart.
        
        """
        self.console.print("Warning! :warning:", style=self.warning_style)
        self._print(text)

    def used_device(self, real_path: str) -> None:
        self.console.print("Device ready! :ok_hand:", style=self.ok_style)
        self.console.print(
            f"Using device on {real_path}.", style=Style(bold=True)
        )
        self.console.print(
            "To use another device, specify it via command line and the"
            " devices real path, e.g. '/dev/video0'."
        )

    def run_main_command(
        self, device_map: dict[int, dict[str, str]], type: str | int
    ) -> None:
        """Print a message guiding the user to run the main command depending on the device map status.

        Args:
            device_map --- a dictionary containing the device map information.
        """
        if not device_map:
            self.console.print(
                "\nAdd a device with one of the above commands now.\n",
                style=Style(bold=True),
            )
            return

        elif len(device_map) == 1:
            self.console.print(
                "\nAlright, a device has been added. :white_check_mark:"
            )

        else:
            ticks = (
                str([":white_check_mark:" for i in range(len(device_map))])
                .replace("'", "")
                .replace(", ", " ")[1:-1]
            )
            self.console.print(
                f"\nAlright, multiple devices have been added. {ticks}"
            )

        self.console.print("Now it's time to run MeetingCam!\n")

        for key, device in device_map.items():
            self.console.print(
                f"For {device['label_real']} run:", style=Style(bold=True)
            )
            if type == WEBCAM:
                self.console.print(
                    "python src/meetingcam/main.py --device-path"
                    f" {device['path_real']} --name YourName",
                    style="bold cyan on black",
                )
            elif type == DEPTHAI:
                self.console.print(
                    "python src/meetingcam/main.py --device-path"
                    f" {device['path_real']} --depthai",
                    style="bold cyan on black",
                )

    def device_running(self) -> None:
        """Print a message indicating that the device is currently running and how to access the stream."""
        self.console.print(
            "Device running! :arrow_forward:", style=self.ok_style
        )
        self.console.print(
            "You can now access the modified camera stream in Meets, Teams or"
            " Zoom. :rocket:",
            style=Style(bold=True),
        )
        self.console.print(
            "Press `Ctrl+C` to stop the running stream and access your device"
            " normally."
        )

    def device_stopped(self) -> None:
        """Print a message indicating that the device has stopped and can now be accessed normally."""
        self.console.print(
            "Device stopped. :raised_hand:", style=self.danger_style
        )
        self.console.print(
            "You can access your device now as usual. MeetingCam has stopped.",
            style=Style(bold=True),
        )

    def _print(self, text: str) -> None:
        """Format text to markdown and print formatted text to the console.

        Args:
            text --- text to be formatted and displayed on the console.
        """
        text = self._reformat(text)
        md = Markdown(text)
        self.console.print(md)

    def _reformat(self, text: str) -> str:
        """Removes the indent spaces from indented multiline text, necessary for compatibility with the rich library's markdown format.

        Args:
            text --- the text to be reformatted.

        Returns:
            The reformatted text with removed indent spaces.
        """
        return text.replace("      ", "")


if __name__ == "__main__":
    pass

src/meetingcam/device.py
nengelmann/MeetingCam
#!/usr/bin/env python3

"""This script is a camera processing pipeline that allows for real-time manipulation of video streams,
so that a modified video stream can be used in web meeting tools like Teams, Meets or Zoom.

The script utilizes is processing of video frames from a real camera device, and output the modified frames to a virtual camera device. 
A command line interface is provided for user interaction, with keyboard listeners enabling real-time adjustments 
like color space conversion, image flipping, and face detection functionalities.
"""

import cv2
import pyvirtualcam
import typer
from constants import DEPTHAI, WEBCAM
from device import DepthaiDevice, WebcamDevice
from plugins.depthai_yolov5_coco.plugin import Yolov5
from plugins.face_detection.plugin import FaceDetection
from plugins.plugin_utils import PluginBase
from utils import ArgumentHandler, InvalidArgumentError, KeyHandler

app = typer.Typer()


@app.command(
    context_settings={"allow_extra_args": True, "ignore_unknown_options": True}
)
def main(
    ctx: typer.Context,
    device_path: str | None = None,
    depthai: bool = False,
) -> None:
    """Initialize the several handlers, real and virtual camera devices for video frame capture and processing and run the whole pipeline.
    It integrates keyboard listeners for real-time user inputs to manipulate video streams.

    Args:
        ctx (typer.Context): Context information from the command line.
        device_path (str | None): The device path for the real camera. Defaults to None.
    """

    # preprocess extra arguments, e.g. for custom plugin
    extra_args = ArgumentHandler(ctx.args)
    extra_args.print(ctx.args)

    if depthai:
        # define your plugin
        plugin = Yolov5()

        # initialize camera device handler with OAK as input device
        device_handler = DepthaiDevice(plugin.pipeline)
        virtual_path = device_handler.init_device(device_path)

    else:
        # handle extra argument --name, if defined
        name = extra_args.get("--name", ctx.args)
        # define your plugin
        plugin = FaceDetection(name=name)

        # initialize camera device handler with webcam as input device
        device_handler = WebcamDevice()
        virtual_path = device_handler.init_device(device_path)

    # perform some checks on given configuration before camera is started
    if plugin.type is None:
        raise NotImplementedError(
            "Your plugin needs to have a plugin type asigend. plugin.type"
            " WEBCAM | DEPTHAI"
        )
    if depthai and plugin.type == WEBCAM:
        raise InvalidArgumentError(
            "The cli flag 'depthai' is set but the plugin you run is meant for"
            " webcam usage."
        )

    if not depthai and plugin.type == DEPTHAI:
        raise InvalidArgumentError(
            "The cli flag 'depthai' is not set but the plugin you run is meant"
            " for depthai usage."
        )

    # initialize real camera, to get frames
    with device_handler.get_device() as r_cam:
        # custom setup for depthai (on device handling)
        if depthai:
            r_cam.setup(plugin.device_setup, plugin.acquisition)

        # initialize a virtual camera where the modified frames will be sent to
        with pyvirtualcam.Camera(
            width=r_cam.width,
            height=r_cam.height,
            fps=24,
            device=virtual_path,
        ) as v_cam:
            # initialize a keyboard listener to get and use keystroke during runtime as trigger or switch
            with KeyHandler() as listener:
                listener.start()

                # print in command line that the pipeline is running
                device_handler.device_running()

                # get frames from real camera, process it and sent it out via virtual camera
                while True:
                    # get a frame and optionally some on camera detections
                    frame, detection = r_cam.get_frame()

                    # convert bgr to rgb if <Ctrl+Alt+r> keys are pressed
                    if listener.bgr2rgb_sw:
                        frame = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR)

                    # pass trigger <Ctrl+Alt+f>, <Ctrl+Alt+l> and <Ctrl+Alt+n> for usage in plugin
                    trigger = (
                        listener.f_trig,
                        listener.l_trig,
                        listener.n_trig,
                    )
                    frame = plugin.process(frame, detection, trigger)

                    # flip image if <Ctrl+Alt+m> keys are pressed
                    if listener.mirror_sw:
                        frame = cv2.flip(frame, 1)

                    frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)

                    # sent out the modified frame to the virtual camera
                    v_cam.send(frame)
                    v_cam.sleep_until_next_frame()


if __name__ == "__main__":
    app()

src/meetingcam/main.py
nengelmann/MeetingCam
import json
from pathlib import Path
from typing import Any

import blobconverter
import depthai
from numpy.typing import NDArray

from .utils import *


class PipelineHandler:
    """
    This class defines the depthai pipeline and the on device processing.

    Attributes:
        model --- a compiled neural network model used for face detection.
        confidence_thr --- the confidence threshold for filtering detections.
        overlap_thr --- the overlap threshold for non-maximum suppression.
    Note:
        Code partly from:
        https://github.com/luxonis/depthai-experiments/blob/master/gen2-yolo/device-decoding/main_api.py
    """

    def __init__(self, model_dir: str) -> None:
        """
        Initialize the FaceDetector instance.

        Args:
            model_path --- the path to the neural network model file.

        Raises:
            Various exceptions can be raised by openvino runtime methods if the model file is not found or has errors.
        """

        self.model = "yolov5n_coco_416x416"
        self.config = "yolov5_config.json"
        self.model_dir = model_dir

        # parse config
        configPath = Path(__file__).resolve().parent / Path(self.config)
        if not configPath.exists():
            raise ValueError("Path {} does not exist!".format(configPath))

        self.nnPath = str(
            blobconverter.from_zoo(
                name=self.model,
                shaves=6,
                zoo_type="depthai",
                use_cache=True,
                output_dir=self.model_dir,
            )
        )

        with configPath.open() as f:
            config = json.load(f)
        nnConfig = config.get("nn_config", {})

        # parse input shape
        if "input_size" in nnConfig:
            self.width, self.height = tuple(
                map(int, nnConfig.get("input_size").split("x"))
            )

        # parse labels
        nnMappings = config.get("mappings", {})
        self.labels = nnMappings.get("labels", {})

        # extract metadata
        metadata = nnConfig.get("NN_specific_metadata", {})
        self.classes = metadata.get("classes", {})
        self.coordinates = metadata.get("coordinates", {})
        self.anchors = metadata.get("anchors", {})
        self.anchorMasks = metadata.get("anchor_masks", {})
        self.iouThreshold = metadata.get("iou_threshold", {})
        self.confidenceThreshold = metadata.get("confidence_threshold", {})

    def create(self):
        # Create pipeline
        pipeline = depthai.Pipeline()

        # Define sources and outputs
        camRgb = pipeline.create(depthai.node.ColorCamera)
        detectionNetwork = pipeline.create(depthai.node.YoloDetectionNetwork)
        xoutRgb = pipeline.create(depthai.node.XLinkOut)
        nnOut = pipeline.create(depthai.node.XLinkOut)

        xoutRgb.setStreamName("rgb")
        nnOut.setStreamName("nn")

        # Properties
        camRgb.setPreviewSize(self.width, self.height)
        camRgb.setResolution(
            depthai.ColorCameraProperties.SensorResolution.THE_1080_P
        )
        camRgb.setInterleaved(False)
        camRgb.setColorOrder(depthai.ColorCameraProperties.ColorOrder.BGR)
        camRgb.setFps(40)
        camRgb.setPreviewKeepAspectRatio(False)

        # Network specific settings
        detectionNetwork.setConfidenceThreshold(self.confidenceThreshold)
        detectionNetwork.setNumClasses(self.classes)
        detectionNetwork.setCoordinateSize(self.coordinates)
        detectionNetwork.setAnchors(self.anchors)
        detectionNetwork.setAnchorMasks(self.anchorMasks)
        detectionNetwork.setIouThreshold(self.iouThreshold)
        detectionNetwork.setBlobPath(self.nnPath)
        detectionNetwork.setNumInferenceThreads(2)
        detectionNetwork.input.setBlocking(False)

        # Linking
        camRgb.preview.link(detectionNetwork.input)
        camRgb.video.link(xoutRgb.input)
        detectionNetwork.out.link(nnOut.input)

        return pipeline

src/meetingcam/plugins/depthai_yolov5_coco/pipeline.py
nengelmann/MeetingCam
import sys
from math import ceil
from pathlib import Path
from typing import Any

import cv2
import depthai
from numpy.typing import NDArray

from ..plugin_utils import PluginDepthai
from .pipeline import PipelineHandler
from .utils import displayFrame


class Yolov5(PluginDepthai):
    """A plugin for running a pretrained yolov5 model (COCO dataset) on a depthai device (OAK)."""

    def __init__(self) -> None:
        """Initialize the base class and create depthai pipeline."""
        super().__init__()
        self.pipeline_handler = PipelineHandler(self.model_dir)
        self.pipeline = self.pipeline_handler.create()

    def device_setup(self, device):
        """Setup of device before image acquisition loop, e.g. for get queue definition etc.

        Args:
            device --- depthai device
        """
        device.qRgb = device.getOutputQueue(
            name="rgb", maxSize=1, blocking=False
        )
        device.qDet = device.getOutputQueue(
            name="nn", maxSize=1, blocking=False
        )

        # Overwrite image height and width if needed, otherwise MAX_HEIGHT and MAX_WIDTH are used
        # device.height = MAX_HEIGHT
        # device.width = MAX_WIDTH

    def acquisition(
        self, device
    ) -> tuple[NDArray[Any], list[depthai.ImgDetection]]:
        """Acquire an image and optionally detections from camera queue and return them.

        Args:
            device --- depthai device
        """
        inRgb = device.qRgb.get()
        inDet = device.qDet.get()

        if inRgb is not None:
            image = inRgb.getCvFrame()

            if inDet is not None:
                detections = inDet.detections

        return image, detections

    def process(
        self,
        image: NDArray[Any],
        detection: Any,
        trigger: tuple[bool, bool, bool],
    ) -> NDArray[Any]:
        """Process the input image and return the image with detected face annotations.

        Args:
            image --- the input image to be processed.
            detections --- detections from depthai device
            trigger --- a tuple containing boolean values indicating whether to draw bounding boxes and name annotations.

        Returns:
            The processed image with face and name annotations.
        """
        if trigger[1]:
            image = displayFrame(
                image, detection, self.pipeline_handler.labels
            )

        # do more host processing here..

        return image

src/meetingcam/plugins/depthai_yolov5_coco/plugin.py
nengelmann/MeetingCam
"""
Utility functions for depthai yolov5

Note:
    Code adapted from: https://github.com/luxonis/depthai-experiments/blob/master/gen2-yolo/device-decoding/main_api.py
"""

import argparse
import json
import sys
import time
from pathlib import Path

import blobconverter
import cv2
import depthai as dai
import numpy as np


# nn data, being the bounding box locations, are in <0..1> range - they need to be normalized with frame width/height
def _frameNorm(frame, bbox):
    normVals = np.full(len(bbox), frame.shape[0])
    normVals[::2] = frame.shape[1]
    return (np.clip(np.array(bbox), 0, 1) * normVals).astype(int)


def displayFrame(frame, detections, labels):
    color = (255, 0, 0)
    for detection in detections:
        bbox = _frameNorm(
            frame,
            (detection.xmin, detection.ymin, detection.xmax, detection.ymax),
        )
        cv2.putText(
            frame,
            labels[detection.label],
            (bbox[0] + 10, bbox[1] + 50),
            cv2.FONT_HERSHEY_SIMPLEX,
            2,
            255,
            3,
            0,
        )
        cv2.putText(
            frame,
            f"{int(detection.confidence * 100)}%",
            (bbox[0] + 10, bbox[1] + 100),
            cv2.FONT_HERSHEY_SIMPLEX,
            1,
            255,
            2,
            0,
        )
        cv2.rectangle(frame, (bbox[0], bbox[1]), (bbox[2], bbox[3]), color, 3)
    return frame

src/meetingcam/plugins/depthai_yolov5_coco/utils.py
nengelmann/MeetingCam
from typing import Any

import cv2
import numpy as np
from numpy.typing import NDArray
from openvino.runtime import Core

from .utils import non_max_suppression


class FaceDetector:
    """
    This class utilizes a pre-trained model for detecting faces in images.

    Attributes:
        model --- a compiled neural network model used for face detection.
        confidence_thr --- the confidence threshold for filtering detections.
        overlap_thr --- the overlap threshold for non-maximum suppression.
    Note:
        Check this documentation for more detail:
        https://github.com/openvinotoolkit/open_model_zoo/blob/master/models/public/ultra-lightweight-face-detection-rfb-320/README.md

        Code partly from:
        https://github.com/Gabriellgpc/ultra-lightweight-face-detection
    """

    model = None

    def __init__(
        self,
        model_path: str,
        confidence_thr: float = 0.5,
        overlap_thr: float = 0.7,
    ) -> None:
        """
        Initialize the FaceDetector instance.

        Args:
            model_path --- the path to the neural network model file.
            confidence_thr --- the confidence threshold for filtering detections (default is 0.5).
            overlap_thr --- the overlap threshold for non-maximum suppression (default is 0.7).

        Raises:
            Various exceptions can be raised by openvino runtime methods if the model file is not found or has errors.
        """
        core = Core()
        model = core.read_model(str(model_path))
        self.model = core.compile_model(model)
        self.confidence_thr = confidence_thr
        self.overlap_thr = overlap_thr

    def preprocess(self, image: NDArray[Any]) -> NDArray[Any]:
        """
        Resize and prepare BGR image for neural network.

        Args:
            image --- the input image array.

        Returns:
            The processed image array ready to be fed into the neural network.
        """
        input_image = cv2.resize(image, dsize=[320, 240])
        input_image = np.expand_dims(input_image.transpose(2, 0, 1), axis=0)
        return input_image

    def postprocess(
        self,
        pred_scores: NDArray[Any],
        pred_boxes: NDArray[Any],
        image_shape: tuple[int, int],
    ):
        """
        Processes the neural network's predictions to obtain usable results.

        This method filters detections below a confidence threshold, applies non-maximum suppression
        on bounding boxes, and returns bounding boxes in absolute pixel values.

        Args:
            pred_scores --- the predicted scores from the neural network.
            pred_boxes --- the predicted bounding boxes from the neural network.
            image_shape --- the shape of the input image.

        Returns:
            A tuple containing two lists: one for bounding boxes and another for scores.
        """

        # filter
        filtered_indexes = np.argwhere(
            pred_scores[0, :, 1] > self.confidence_thr
        ).tolist()
        filtered_boxes = pred_boxes[0, filtered_indexes, :]
        filtered_scores = pred_scores[0, filtered_indexes, 1]

        if len(filtered_scores) == 0:
            return [], []

        # convert all boxes to image coordinates
        h, w = image_shape

        def _convert_bbox_format(*args: list[int]):
            bbox = args[0]
            x_min, y_min, x_max, y_max = bbox
            x_min = int(w * x_min)
            y_min = int(h * y_min)
            x_max = int(w * x_max)
            y_max = int(h * y_max)
            return x_min, y_min, x_max, y_max

        bboxes_image_coord = np.apply_along_axis(
            _convert_bbox_format, axis=2, arr=filtered_boxes
        )

        # apply non-maximum supressions
        bboxes_image_coord, indexes = non_max_suppression(
            bboxes_image_coord.reshape([-1, 4]), threshold=self.overlap_thr
        )
        filtered_scores = filtered_scores[indexes]
        return bboxes_image_coord, filtered_scores

    def inference(
        self, image: NDArray[Any]
    ) -> tuple[NDArray[Any], NDArray[Any]]:
        """
        Run the entire inference pipeline on an input image.

        This method preprocesses the input image, performs inference using the neural network,
        and postprocesses the predictions to obtain final results.

        Args:
            image --- the input image array.

        Returns:
            A tuple containing two arrays: one for faces and another for scores.
        """
        input_image = self.preprocess(image)

        pred = self.model([input_image])
        pred_scores = pred[self.model.output(0)]
        pred_boxes = pred[self.model.output(1)]

        image_shape = image.shape[:2]
        faces, scores = self.postprocess(pred_scores, pred_boxes, image_shape)
        return faces, scores

src/meetingcam/plugins/face_detection/model.py
nengelmann/MeetingCam
import sys
from math import ceil
from pathlib import Path
from typing import Any

import cv2
from constants import WEBCAM
from numpy.typing import NDArray

from ..plugin_utils import PluginBase
from .model import FaceDetector
from .utils import box_area, draw_bbox


class FaceDetection(PluginBase):
    """A plugin for detecting faces in an image.

    This plugin extends the PluginBase class and is used to detect faces in an image using a pretrained face detection model.
    """

    def __init__(self, name: str | None = None) -> None:
        """Initialize the FaceDetection plugin with a specified name.

        This method initializes the FaceDetection plugin and sets up the necessary parameters including model paths, and name.

        Args:
            name --- the name of the user. If not provided, a default name is assigned.

        Raises:
            SystemExit: If the model files are not found in the specified directory.
        """
        super().__init__()

        self.fn_scale = 3e-3  # Adjust for larger font size in all images
        self.th_scale = 6e-3  # relative thickness in all images
        self.off_scale = (
            3e-2  # Adjust for larger Y-offset of text and bounding box
        )

        model_path = (
            Path(self.model_dir)
            / "public/ultra-lightweight-face-detection-rfb-320/FP16/ultra-lightweight-face-detection-rfb-320.xml"
        )

        if not model_path.exists():
            print(
                "\nModel files are not downloaded or converted.\n Download and"
                " convert the model with the following commands."
            )
            print(
                "\n`omz_downloader --name"
                " ultra-lightweight-face-detection-rfb-320 --output_dir"
                " src/meetingcam/models`"
            )
            sys.exit()

        if name is not None:
            self.name = name
        else:
            self.name = "Code Ninja"
            print("\nYou have not specified your name ('--name' YourName).")
            print(f"I simply name you {self.name}\n")

        self.detector = FaceDetector(
            model_path=model_path, confidence_thr=0.9, overlap_thr=0.7
        )
        self.type = WEBCAM

    def process(
        self,
        image: NDArray[Any],
        detection: Any,
        trigger: tuple[bool, bool, bool],
    ) -> NDArray[Any]:
        """Process the input image and return the image with detected face annotations.

        This method takes an image and a trigger tuple as inputs, performs face detection, and returns the image with annotations such as bounding boxes and names.

        Args:
            image --- the input image to be processed.
            trigger --- a tuple containing boolean values indicating whether to draw bounding boxes and name annotations.

        Returns:
            The processed image with face and name annotations.
        """
        bboxes, scores = self.detector.inference(image)

        if len(bboxes) > 0:
            # Assumption: The persons face in front of the web cam is the one which is closest to the camera and hence the biggest.
            # So we get the detection result with the biggest bbox area.
            box_areas = [box_area(box) for box in bboxes]
            idx = box_areas.index(max(box_areas))
            h, w = image.shape[0:2]

            # If f_trigger <Ctrl+Alt+f> is True, print in the face bbox
            if trigger[0]:
                image = draw_bbox(
                    image,
                    bboxes[idx][:2],
                    bboxes[idx][2:],
                    color=(0, 255, 0),
                    thickness=ceil(h * self.th_scale),
                    radius=ceil(h * self.th_scale * 4),
                    corner_len=ceil(h * self.th_scale * 8),
                )
            # If n_trigger <Ctrl+Alt+n> is True, print in the name above the bbox
            if trigger[2] and self.name:
                x1, y1 = bboxes[idx][:2]
                image = cv2.putText(
                    image,
                    self.name,
                    (x1, y1 - int(h * self.off_scale)),
                    fontFace=cv2.FONT_HERSHEY_SIMPLEX,
                    fontScale=h * self.fn_scale,
                    color=(0, 255, 0),
                    thickness=ceil(h * self.th_scale),
                )
        return image

src/meetingcam/plugins/face_detection/plugin.py
nengelmann/MeetingCam
from typing import Any

import cv2
import numpy as np
from numpy.typing import NDArray


def non_max_suppression(
    boxes: NDArray[Any], threshold: float
) -> tuple[list[Any]]:
    """Apply fast non-maximum suppression on bounding boxes.

    This function takes a set of bounding boxes and a threshold as input, and returns
    a tuple containing a list of suppressed bounding boxes and their respective indexes.

    Args:
        boxes --- a numpy array of bounding boxes.
        threshold --- the overlap threshold for suppression.

    Returns:
        A tuple containing a list of suppressed bounding boxes and their indexes.

    Note:
        Code adapted from: https://github.com/Gabriellgpc/ultra-lightweight-face-detection and
    """

    # if there are no boxes, return an empty list
    if len(boxes) == 0:
        return []

    # initialize the list of picked indexes
    pick = []

    # grab the coordinates of the bounding boxes
    x1 = boxes[:, 0]
    y1 = boxes[:, 1]
    x2 = boxes[:, 2]
    y2 = boxes[:, 3]

    # compute the area of the bounding boxes and sort the boxes by
    # their bottom-right y-coordinate
    area = (x2 - x1 + 1) * (y2 - y1 + 1)
    idxs = np.argsort(y2)

    # keep looping while some indexes still remain in the indexes list
    while len(idxs) > 0:
        # grab the last index in the indexes list and add the index
        # value to the list of picked indexes
        last = len(idxs) - 1
        i = idxs[last]
        pick.append(i)

        # find the largest (x, y) coordinates for the start of the
        # bounding box and the smallest (x, y) coordinates for the
        # end of the bounding box
        xx1 = np.maximum(x1[i], x1[idxs[:last]])
        yy1 = np.maximum(y1[i], y1[idxs[:last]])
        xx2 = np.minimum(x2[i], x2[idxs[:last]])
        yy2 = np.minimum(y2[i], y2[idxs[:last]])

        # compute the width and height of the bounding box
        w = np.maximum(0, xx2 - xx1 + 1)
        h = np.maximum(0, yy2 - yy1 + 1)

        # compute the ratio of overlap between the bounding box and
        # other bounding boxes
        overlap = (w * h) / area[idxs[:last]]

        # delete all indexes from the index list that have overlap
        # greater than the provided overlap threshold
        idxs = np.delete(
            idxs,
            np.concatenate(([last], np.where(overlap > threshold)[0])),
        )

    # return only the bounding boxes that were picked
    return boxes[pick], pick


def draw_bbox(
    img: NDArray[Any],
    pt1: tuple[int, int],
    pt2: tuple[int, int],
    color: tuple[int, int, int],
    thickness: int,
    radius: int,
    corner_len: int,
    connected: bool = True,
    c_thickness: int = 1,
) -> NDArray[Any]:
    """Draw a bounding box with rounded corners onto an image.

    This function draws a bounding box with rounded corners onto a given image,
    according to the specified parameters such as the coordinates of the corners,
    the color, and the thickness of the lines.

    Args:
        img --- the input image as a numpy array.
        pt1 --- the coordinates of the first point (top left corner).
        pt2 --- the coordinates of the second point (bottom right corner).
        color --- the color of the bounding box as a tuple of RGB values.
        thickness --- the thickness of the lines forming the bounding box.
        radius --- the radius of the rounded corners.
        corner_len --- the length of the corners.
        connected --- Flag to indicate if corners are connected. Defaults to True.
        c_thickness --- the thickness of the connecting lines. Defaults to 1.

    Returns:
        The image with the bounding box drawn on it.

    Note:
        Code adapted from: https://stackoverflow.com/questions/46036477/drawing-fancy-rectangle-around-face
    """
    x1, y1 = pt1
    x2, y2 = pt2
    r = radius
    d = corner_len

    # Top left
    cv2.line(img, (x1 + r, y1), (x1 + r + d, y1), color, thickness)
    cv2.line(img, (x1, y1 + r), (x1, y1 + r + d), color, thickness)
    cv2.ellipse(img, (x1 + r, y1 + r), (r, r), 180, 0, 90, color, thickness)

    # Top right
    cv2.line(img, (x2 - r, y1), (x2 - r - d, y1), color, thickness)
    cv2.line(img, (x2, y1 + r), (x2, y1 + r + d), color, thickness)
    cv2.ellipse(img, (x2 - r, y1 + r), (r, r), 270, 0, 90, color, thickness)

    # Bottom left
    cv2.line(img, (x1 + r, y2), (x1 + r + d, y2), color, thickness)
    cv2.line(img, (x1, y2 - r), (x1, y2 - r - d), color, thickness)
    cv2.ellipse(img, (x1 + r, y2 - r), (r, r), 90, 0, 90, color, thickness)

    # Bottom right
    cv2.line(img, (x2 - r, y2), (x2 - r - d, y2), color, thickness)
    cv2.line(img, (x2, y2 - r), (x2, y2 - r - d), color, thickness)
    cv2.ellipse(img, (x2 - r, y2 - r), (r, r), 0, 0, 90, color, thickness)

    if connected:
        # Top left to top right
        cv2.line(img, (x1 + r + d, y1), (x2 - r, y1), color, c_thickness)
        # Bottom left to Bottom right
        cv2.line(img, (x1 + r + d, y2), (x2 - r, y2), color, c_thickness)
        # Bottom left to top left
        cv2.line(img, (x1, y2 - r - d), (x1, y1 + r + d), color, c_thickness)
        # Bottom right to top right
        cv2.line(img, (x2, y2 - r - d), (x2, y1 + r + d), color, c_thickness)

    return img


def box_area(box: list[int]) -> int:
    """Calculate the area of a bounding box.

    Args:
        box --- a list of integers representing the bounding box in the format [x_min, y_min, x_max, y_max].

    Returns:
       The area of the bounding box.
    """
    area = int(box[2] - box[0]) * (box[3] - box[1])
    return area

src/meetingcam/plugins/face_detection/utils.py
nengelmann/MeetingCam
from abc import ABC, abstractmethod
from typing import Any

import cv2
import depthai
from constants import DEPTHAI, WEBCAM
from numpy.typing import NDArray


class PluginBase(ABC):
    """PluginBase class acts as a base class for plugins with image processing capabilities.

    This class is intended to be subclassed, with subclasses implementing the abstract
    `process` method to perform specific image processing tasks.

    Attributes:
        model_dir --- directory where models are stored. Defaults to "./src/meetingcam/models".
    """

    @abstractmethod
    def __init__(self, *args: tuple[Any], **kwargs: tuple[Any]) -> None:
        """Initialize the PluginBase instance with the default model directory.

        Args:
            *args --- Variable length argument list.
            **kwargs --- Arbitrary keyword arguments.

            With this custom arguments can be passed to plugins, e.g. --name YourName
        """
        self.model_dir = "./src/meetingcam/models"
        self.type = WEBCAM

    @abstractmethod
    def process(
        self,
        image: NDArray[Any],
        detection: Any,
        trigger: tuple[bool, bool, bool],
    ) -> NDArray[Any]:
        """Process an image through the plugin's image processing method.

        This method is abstract and should be overridden in subclasses to implement
        specific image processing functionality.

        Args:
            image --- the image to be processed.

        Returns:
            The processed image.
        """
        pass


class PluginDepthai(PluginBase):
    """PluginBase class for Depthai plugins with on device compute."""

    @abstractmethod
    def __init__(self, *args: tuple[Any], **kwargs: tuple[Any]) -> None:
        """Initialize the PluginBase instance with the default model directory.

        Args:
            *args --- Variable length argument list.
            **kwargs --- Arbitrary keyword arguments.

            With this custom arguments can be passed to plugins, e.g. --name YourName
        """
        super().__init__()
        self.type = DEPTHAI

    @abstractmethod
    def device_setup(self, device) -> None:
        """Setup of device before image acquisition loop, e.g. for get queue definition etc.

        Args:
            device --- depthai device
        """
        pass

    @abstractmethod
    def acquisition(
        self, device
    ) -> tuple[NDArray[Any], list[depthai.ImgDetection]]:
        """Acquire an image and optionally detections from camera queue and return them.

        Args:
            device --- depthai device
        """
        pass

src/meetingcam/plugins/plugin_utils.py
nengelmann/MeetingCam
from types import MethodType
from typing import Any, Callable

import cv2
import depthai
from constants import MAX_HEIGHT, MAX_WIDTH
from numpy.typing import NDArray
from pynput import keyboard
from typing_extensions import Self


class VideoCapture(cv2.VideoCapture):
    """Handle video capture functionalities with OpenCV.

    Inherits from cv2.VideoCapture and extends functionalities with
    methods to safely enter, exit, get frames and frame rates from
    a video capture.
    """

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Initialize the VideoCapture object with width and height properties."""
        super().__init__(*args, **kwargs)
        self.width = int(self.get(cv2.CAP_PROP_FRAME_WIDTH))
        self.height = int(self.get(cv2.CAP_PROP_FRAME_HEIGHT))
        self.img_handler = ImageHandler()

    def __enter__(self) -> Self:
        """Enter method for context management, returning self."""
        return self

    def __exit__(self, *args: Any) -> None:
        """Exit method for context management, releasing the video capture object."""
        self.release()

    def get_frame(self) -> NDArray[Any]:
        """Capture a frame from the video stream.

        Raises:
            RuntimeError: If image acquisition fails.

        Returns:
            A frame which has been captured by the camera.
        """
        (grabbed, frame) = self.read()

        if not grabbed:
            raise RuntimeError(
                "Image acquisition failed. Make sure the specified camera is"
                " not running in another application."
            )

        # high image resolution is usually not supported by online meeting tools
        frame = self.img_handler.correct_img_size(frame)

        # return the frame and None, there is no detection available from a webcam
        return frame, None

    def get_fps(self) -> int:
        """Retrieve the frames per second (FPS) of the video capture.

        Prints the total FPS and returns the FPS as an integer.
        """
        fps = self.get(cv2.CAP_PROP_FPS)
        print(f"total FPS: {int(fps)}")
        return int(fps)


class DepthaiCapture(depthai.Device):
    """Handle video capture functionalities with depthai.

    Inherits from depthai.Device and extends functionalities with
    methods to safely enter, exit, get frames and frame rates from
    a depthai device.
    """

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Initialize the depthai.Device object and assign width and height properties."""
        super().__init__(*args, **kwargs)
        self.width = MAX_WIDTH
        self.height = MAX_HEIGHT
        self.img_handler = ImageHandler()

    def __enter__(self) -> Self:
        """Enter method for context management, returning self."""
        return self

    def __exit__(self, *args: Any) -> None:
        """Exit method for context management, releasing the video capture object."""
        self.close()

    def get_frame(self) -> NDArray[Any]:
        """
        Capture a frame from the video stream.

        Returns:
            A frame and potentially detections which has been captured by the camera.
        """
        # get image and detections from depthai plugin (image acquisition function)
        img, det = self.acquisition()

        # high image resolution is usually not supported by online meeting tools
        img = self.img_handler.correct_img_size(img)

        return img, det

    def setup(self, setup_func: Callable, acquisition_func: Callable):
        """Setup initializations defined in the plugins setup function and create an acquisition function also based on plugin specification.

        Args:
            setup_func --- initialization function of plugin
            acquisition_func --- handling function for image acquisition of plugin
        """
        setup_func(self)
        self.acquisition = MethodType(acquisition_func, self)

    def get_fps(self) -> int:
        """Retrieve the frames per second (FPS) of the video capture.

        Prints the total FPS and returns the FPS as an integer.
        """

        # fps =
        # print(f"total FPS: {int(fps)}")
        # return int(fps)
        raise NotImplementedError


class KeyHandler(keyboard.GlobalHotKeys):
    """A class to handle global hotkeys and their functionalities.

    Inherits from keyboard.GlobalHotKeys and defines hotkey triggers and switches.
    """

    def __init__(self) -> None:
        """Initialize the KeyHandler with hotkeys and their respective states."""
        self.hotkeys = {
            "<ctrl>+<alt>+f": self.f_trigger,
            "<ctrl>+<alt>+l": self.l_trigger,
            "<ctrl>+<alt>+n": self.n_trigger,
            # '<ctrl>+<alt>+x':self.your_hotkey_x,
            "<ctrl>+<alt>+r": self.bgr2rgb_switch,
            "<ctrl>+<alt>+m": self.mirror_switch,
        }
        super().__init__(self.hotkeys)

        self.f_trig: bool = False
        self.l_trig: bool = False
        self.n_trig: bool = False
        # self.x_trig: bool = False
        self.bgr2rgb_sw: bool = False
        self.mirror_sw: bool = True

    def __enter__(self) -> Self:
        """Enter method for context management, returning self."""
        return self

    def __exit__(self, *args: tuple[Any]) -> None:
        """Exit method for context management, stopping the KeyHandler."""
        self.stop()

    def f_trigger(self) -> None:
        """Toggle the state of f_trigger and print its status."""
        self.f_trig = not self.f_trig
        print("Keyboard trigger 1: ", str(self.f_trig))

    def l_trigger(self) -> None:
        """Toggle the state of l_trigger and print its status."""
        self.l_trig = not self.l_trig
        print("Keyboard trigger 2: ", str(self.l_trig))

    def n_trigger(self) -> None:
        """Toggle the state of n_trigger and print its status."""
        self.n_trig = not self.n_trig
        print("Keyboard trigger 3: ", str(self.n_trig))

    # def your_trigger_x(self) -> None:
    #    """Toggle the state of custom_trigger and print its status."""
    #    print('your hotkey is pressed')
    #    self.x_trig = not self.x_trig

    def bgr2rgb_switch(self) -> None:
        """Toggle the state of bgr2rgb_switch and print its status."""
        self.bgr2rgb_sw = not self.bgr2rgb_sw
        print("Keyboard switch BGR2RGB: ", str(self.bgr2rgb_sw))

    def mirror_switch(self) -> None:
        """Toggle the state of mirror_switch and print its status."""
        self.mirror_sw = not self.mirror_sw
        print("Keyboard switch Mirror Camera: ", str(self.mirror_sw))


class InvalidArgumentError(ValueError):
    """Custom exception for invalid arguments, inherits from ValueError."""

    def __init__(self, message: str = "Invalid argument") -> None:
        """Initialize the exception with a custom or default message."""
        super().__init__(message)


class ArgumentHandler:
    """A class to handle and validate command-line arguments."""

    def __init__(self, ctx_args: list[str]) -> None:
        """Initialize the ArgumentHandler and check the provided arguments."""
        self.check(ctx_args)

    def check(self, ctx_args: list[str]) -> None:
        """Check the validity of the command-line arguments.

        Raises:
            InvalidArgumentError: If the arguments are not correctly formatted.
        """
        if len(ctx_args) < 2 and len(ctx_args) != 0:
            raise InvalidArgumentError(
                "You need to provide extra arguments in the form '--name'"
                " 'argument'. "
            )

        if bool(len(ctx_args) % 2):
            raise InvalidArgumentError(
                "You need to provide extra arguments in the form '--name'"
                " 'argument'. Multiple arguments per parameter are not"
                " allowed. The following is invalid: '--your_param argument1"
                " argument2', while '--your_param argument1argument2' or"
                " '--your_param1 argument1 --your_param2 argument2' are valid."
            )

    def print(self, ctx_args: list[str]) -> None:
        """Print the list of provided command-line arguments or a message if none were provided."""
        if len(ctx_args) > 0:
            print("\nProvided extra arguments are:")
            for param, arg in zip(ctx_args[0::2], ctx_args[1::2]):
                print(param, ": ", arg)
        else:
            print("\nNo extra arguments provided.\n")

    def get(self, pram_name: str, ctx_args: list[str]) -> str | None:
        """Retrieve the value of a specific parameter from the command-line arguments.

        Args:
            pram_name --- the name of the parameter to retrieve.
            ctx_args --- a list of command-line arguments.

        Returns:
            The value of the parameter if found, otherwise None.
        """
        if pram_name in ctx_args:
            idx = ctx_args.index(pram_name) + 1
            arg = ctx_args[idx]
            return arg
        else:
            return None


class ImageHandler:
    """A class to handle image processing functions."""

    def __init__(self) -> None:
        """Initialize placeholder."""
        pass

    def correct_img_size(self, image: NDArray[Any]) -> NDArray[Any]:
        """Check image for maximum image size and resize if exceeded.
        The resizing will just be applied on the exceeding dimension (width, height or both) without taking the aspect ratio into account.

        Args:
            image --- the image to be processed.

        Returns:
            The processed image.
        """
        if image.shape[0] > MAX_HEIGHT or image.shape[1] > MAX_WIDTH:
            h = min(image.shape[0], MAX_HEIGHT)
            w = min(image.shape[1], MAX_WIDTH)
            image = cv2.resize(image, (w, h))
        return image

src/meetingcam/utils.py
nengelmann/MeetingCam
#!/usr/bin/env python3

# Code from luxonis/depthai-python github repository:
# https://github.com/luxonis/depthai-python/blob/main/examples/ColorCamera/rgb_uvc.py

import argparse
import time

parser = argparse.ArgumentParser()
parser.add_argument(
    "-fb", "--flash-bootloader", default=False, action="store_true"
)
parser.add_argument("-f", "--flash-app", default=False, action="store_true")
parser.add_argument(
    "-l", "--load-and-exit", default=False, action="store_true"
)
args = parser.parse_args()

if args.load_and_exit:
    import os

    # Disabling device watchdog, so it doesn't need the host to ping periodically.
    # Note: this is done before importing `depthai`
    os.environ["DEPTHAI_WATCHDOG"] = "0"

import depthai as dai


def getPipeline():
    enable_4k = True  # Will downscale 4K -> 1080p

    pipeline = dai.Pipeline()

    # Define a source - color camera
    cam_rgb = pipeline.createColorCamera()
    cam_rgb.setBoardSocket(dai.CameraBoardSocket.CAM_A)
    cam_rgb.setInterleaved(False)
    # cam_rgb.initialControl.setManualFocus(130)

    if enable_4k:
        cam_rgb.setResolution(
            dai.ColorCameraProperties.SensorResolution.THE_4_K
        )
        cam_rgb.setIspScale(1, 2)
    else:
        cam_rgb.setResolution(
            dai.ColorCameraProperties.SensorResolution.THE_1080_P
        )

    # Create an UVC (USB Video Class) output node
    uvc = pipeline.createUVC()
    cam_rgb.video.link(uvc.input)

    # Note: if the pipeline is sent later to device (using startPipeline()),
    # it is important to pass the device config separately when creating the device
    config = dai.Device.Config()
    # config.board.uvc = dai.BoardConfig.UVC()  # enable default 1920x1080 NV12
    config.board.uvc = dai.BoardConfig.UVC(1920, 1080)
    config.board.uvc.frameType = dai.ImgFrame.Type.NV12
    # config.board.uvc.cameraName = "My Custom Cam"
    pipeline.setBoardConfig(config.board)

    return pipeline


# Will flash the bootloader if no pipeline is provided as argument
def flash(pipeline=None):
    (f, bl) = dai.DeviceBootloader.getFirstAvailableDevice()
    bootloader = dai.DeviceBootloader(bl, True)

    # Create a progress callback lambda
    def progress(p):
        return print(f"Flashing progress: {p * 100:.1f}%")

    startTime = time.monotonic()
    if pipeline is None:
        print("Flashing bootloader...")
        bootloader.flashBootloader(progress)
    else:
        print("Flashing application pipeline...")
        bootloader.flash(progress, pipeline)

    elapsedTime = round(time.monotonic() - startTime, 2)
    print("Done in", elapsedTime, "seconds")


if args.flash_bootloader or args.flash_app:
    if args.flash_bootloader:
        flash()
    if args.flash_app:
        flash(getPipeline())
    print("Flashing successful. Please power-cycle the device")
    quit()

if args.load_and_exit:
    device = dai.Device(getPipeline())
    print("\nDevice started. Attempting to force-terminate this process...")
    print("Open an UVC viewer to check the camera stream.")
    print(
        "To reconnect with depthai, a device power-cycle may be required in"
        " some cases"
    )
    # We do not want the device to be closed, so terminate the process uncleanly.
    # (TODO add depthai API to be able to cleanly exit without closing device)
    import signal

    os.kill(os.getpid(), signal.SIGTERM)

# Standard UVC load with depthai
with dai.Device(getPipeline()) as device:
    print("\nDevice started, please keep this process running")
    print("and open an UVC viewer to check the camera stream.")
    print("\nTo close: Ctrl+C")

    # Doing nothing here, just keeping the host feeding the watchdog
    while True:
        try:
            time.sleep(0.1)
        except KeyboardInterrupt:
            break

tools/oak_as_uvc.py
nengelmann/MeetingCam