[QUESTION] Multiple PoE cameras not working?

[franferraz98]

Hi,

I'm new to harvesters and I'm trying to make a code that enables me to see the video streaming from two PoE cameras through it. I've sucessfully been able to make the code for streaming with one camera, but at the moment I try to use 2 cameras only one streaming appears and it gets blocked. The final objective of my project is to standarize the usage of multiple libraries for my company, so I'm implementing this in a kind of weird way.

This is how I define my Camera Interface for the different libraries that I want to standardize:

import time
import logging
import numpy as np
from typing import Any, Tuple
from pathlib import Path
from threading import Thread, Semaphore
from abc import ABC, abstractmethod

class Camera(ABC):
    def __init__(self, config: dict[str, Any] or Path, cam_id: int = 0):
        """
        Generic abstract camera superclass for the different types of
        cameras.
        :param config: Dictionary containing the configuration variables
            of the camera.
        :param cam_id: Camera id
        """
        self.width = None
        self.height = None
        self.framerate = 10
        self.camera = None
        self.is_streaming = False
        self.stream_thread = None
        self.last_image = None
        self.cam_id = cam_id
        self.semaphore = None

    @abstractmethod
    def configure_camera(self, config: dict[str, Any] or Path) -> None:
        """
        Configures the camera with the desired parameters via input dict
        :param config: Configuration dictionary or path to the
            configuration file
        """

    @abstractmethod
    def start_camera(self) -> None:
        """Initializes the camera."""

    def set_semaphore(self, semaphore: Semaphore) -> None:
        """
        Sets the semaphore to the input semaphore
        :param semaphore: External shared semaphore
        """
        # self.semaphore = semaphore
        pass

    @abstractmethod
    def capture_frame(self) -> np.ndarray:
        """Captures a frame."""

    def capture_frame_loop(self) -> None:
        """Loops capture_frame and updates last_image"""
        while self.is_streaming:
            time.sleep(1/self.framerate)
            self.last_image = self.capture_frame()
        logging.info("Exiting capture loop")

    def is_aquiring(self) -> bool:
        """Retruns true if the streaming is on, false otherwise"""
        if self.stream_thread is not None:
            return self.stream_thread.is_alive()
        else:
            return False

    def get_last_frame(self) -> np.ndarray:
        """Returns the last frame taken by the camera"""
        return self.last_image

    def take_photo(self) -> np.ndarray:
        """
        Captures a frame with the camera,
        counts the time it takes and logs it.
        :return: An OpenCV image in RGB format
        """
        start_take_photo = time.perf_counter()
        frame = self.capture_frame()
        t_take_photo = (time.perf_counter() - start_take_photo) * 1000
        logging.info(f"Image taken in {t_take_photo:.0f} milliseconds")
        return frame

    def start_video_stream(self) -> bool:
        """
        Starts an inner thread that takes pictures
        """
        try:
            self.is_streaming = True
            self.stream_thread = Thread(target=self.capture_frame_loop)
            self.stream_thread.daemon = True
            self.stream_thread.start()
        except Exception as e:
            logging.error(f"Video stream couldn't be started... \n{e}")
            return False
        return True

    @abstractmethod
    def close(self) -> None:
        """Closes the camera."""

This is how I define the Genicam Class:

import threading
import time

import cv2
import logging
import numpy as np

from typing import Any, Tuple
from pathlib import Path
from src.new.interface_cam import Camera
from src.new.sem import sem
from harvesters.core import Harvester
from genicam.gentl import TimeoutException, ClosedException
from harvesters.util.pfnc import mono_location_formats, bgr_formats

class GeniCam(Camera):

    def __init__(self, config: dict[str, Any] or Path):
        super().__init__(config)
        self.h = None
        if "harvester" in config.keys():
            self.h = config["harvester"]
        self.cti = None
        self.image_adquire = None

    def configure_camera(self, config: dict[str, Any] or Path) -> None:
        if self.h is None:
            self.h = Harvester()
            self.cti = config["cti"]
            self.h.add_file(self.cti)
            self.h.update()

        if "serial_number" not in config:
            config["serial_number"] = None

        if "user_name" not in config:
            config["user_name"] = None

        self.image_adquire = self.h.create(
            search_key={"serial_number": config["serial_number"],
                        "user_defined_name": config["user_name"]}
        )
        self.image_adquire.remote_device.node_map.PixelFormat.value = "BayerBG8"

    def start_camera(self) -> None:
        try:
            # self.image_adquire.start(run_as_thread=True)
            self.image_adquire.start()
        except Exception as e:
            logging.exception(e)

    def capture_frame(self) -> np.ndarray:
        try:
            # sem.acquire()
            GeniCam.semaphore.acquire()
            with self.image_adquire.fetch(timeout=0) as buffer:
                img = self.to_2d_array(buffer).copy()
                img = cv2.cvtColor(img, cv2.COLOR_BayerBGGR2BGR)
            if self.white_balance_ratio is not None:
                img = self._white_balance(img)
            # sem.release()
            GeniCam.semaphore.release()
            time.sleep(0.01)
            return img
        except (TimeoutException or ClosedException or Exception) as e:
            logging.exception(e)
            sem.release()
            self.close()

    def close(self) -> None:
        try:
            self.image_adquire.stop()
            self.image_adquire.destroy()
        except Exception as e:
            logging.exception(e)

    @staticmethod
    def to_2d_array(buffer: Any) -> np.ndarray:
        """
        Transforms the fetched buffer to OpenCV image array
        :param buffer: Fetched buffer from the camera
        :return: Resulting OpenCV image
        """
        payload = buffer.payload
        component = payload.components[0]
        width = component.width
        height = component.height
        data_format = component.data_format

        # Reshape the image so that it can be drawn on the VisPy canvas:
        if data_format in mono_location_formats:
            content = component.data.reshape(height, width)
        else:
            # The image requires you to reshape it to draw it on the
            # canvas:
            content = component.data.reshape(
                height, width,
                int(component.num_components_per_pixel)
                # Set of R, G, B, and Alpha
            )
            if data_format in bgr_formats:
                # Swap every R and B:
                content = content[:, :, ::-1]
        return content

This is how I define my multicamera manager:

import numpy as np
from pathlib import Path
from typing import List, Any, Dict, Optional
from threading import Semaphore
from src.new.interface_cam import Camera

class InterfaceMulticamera:
    def __init__(self, semaphore: Optional[Semaphore] = None):
        """
        Interface class for managing multicamera deployments
        """
        self.camera_list = []
        self.semaphore = semaphore

    def add_one_camera(self, camera: Camera) -> int:
        """
        Adds a camera to the list
        :param camera: Camera to add
        :return: Index of the camera
        """
        self.camera_list.append(camera)
        return len(self.camera_list) - 1

    def set_one_semaphore(self, cam_id: int) -> None:
        """
        Sets the semaphore of one camera to the shared semaphore
        """
        self.camera_list[cam_id].set_semaphore(self.semaphore)

    def configure_one_camera(self, cam_id: int,
                             config: Dict[str, Any] or Path) -> None:
        """
        Configures one camera with the desired parameters via input dict
        :param cam_id: Camera id
        :param config: Configuration dictionary or path to the
            configuration file
        """
        self.camera_list[cam_id].configure_camera(config)
        self.camera_list[cam_id].set_semaphore(self.semaphore)

    def start_one_camera(self, cam_id: int) -> None:
        """
        Starts one camera by id
        :param cam_id: Camera id
        """
        self.camera_list[cam_id].start_camera()

    def start_all_cameras(self) -> None:
        """
        Starts all cameras on the list
        """
        for camera in self.camera_list:
            camera.start_camera()

    def close_one_camera(self, cam_id: int) -> None:
        """
        Closes one camera by id
        :param cam_id: Camera id
        """
        self.camera_list[cam_id].close()

    def close_all_cameras(self) -> None:
        """
        Close all cameras on the list
        """
        for camera in self.camera_list:
            camera.close()

    def take_one_photo(self, cam_id: int) -> np.ndarray:
        """
        Takes a photo from the camera from the id
        :param cam_id: Camera id
        :return: Picure taken
        """
        return self.camera_list[cam_id].take_photo()

    def take_all_photos(self) -> List[np.ndarray]:
        """
        Takes one picture from each camera on the list.
        :return: List of pictures taken
        """
        img_list = []
        for camera in self.camera_list:
            img_list.append(camera.take_photo())
        return img_list

    def start_one_video_stream(self, cam_id: int) -> bool:
        """
        Starts the video streaming of one camera by id
        :param cam_id: Camera id
        :return: True if the thread started properly, false otherwise
        """
        return self.camera_list[cam_id].start_video_stream()

    def start_all_video_streams(self) -> List[bool]:
        """
        Starts the video streaming of all cameras on the list.
        :return: List of true if the thread started properly, false
            otherwise
        """
        return_list = []
        for camera in self.camera_list:
            return_list.append(camera.start_video_stream())
        return return_list

And this is how I instance and try the classes:

import time
import logging
import os.path
import string
import random

import cv2
import threading
from threading import Semaphore
from harvesters.core import Harvester
from src.new.webcam_new import WebCam
from src.new.ids_new import IDSCamera
from src.new.interface_cam import Camera
from src.new.genicam_new import GeniCam
from src.new.interface_multicamera import InterfaceMulticamera

def get_random_string(length: int) -> str:
    """Choose from all lowercase letter"""
    letters = string.ascii_lowercase
    result_str = ''.join(random.choice(letters) for i in range(length))
    return result_str

def cam_test(camera: Camera) -> None:
    print(f"Resolution: {camera.get_resolution()}")
    print(f"Framerate: {camera.get_framerate()}")

    imname = get_random_string(5)
    while True:
        try:
            image = camera.get_last_frame()
            if image is not None:
                cv2.imshow(imname, image)
                if cv2.waitKey(1) == ord('q'):
                    break

        except KeyboardInterrupt:
            break
        except Exception as e:
            logging.error(e)
            camera.close()
            break

    camera.close()

def multigenicam_test() -> None:
    harvester = Harvester()
    harvester.add_file("C:/Program Files/MATRIX VISION/mvIMPACT Acquire/bin/"
                       "x64/mvGenTLProducer.cti")  # Se incluye el archivo CTI
    harvester.update()
    config1 = {"cti": "C:/Program Files/MATRIX VISION/mvIMPACT Acquire/bin/"
                      "x64/mvGenTLProducer.cti",
               "serial_number": "3205536",
               "user_name": "DD2",
               "harvester": harvester}
    geni1 = GeniCam(config=config1)
    config2 = {"cti": "C:/Program Files/MATRIX VISION/mvIMPACT Acquire/bin/"
                      "x64/mvGenTLProducer.cti",
               "serial_number": "3205539",
               "user_name": "DI1",
               "harvester": harvester}
    geni2 = GeniCam(config=config2)

    # semaphore = Semaphore()
    semaphore = None
    int_mul = InterfaceMulticamera(semaphore=semaphore)
    int_mul.add_one_camera(geni1)
    int_mul.add_one_camera(geni2)

    int_mul.configure_one_camera(0, config1)
    int_mul.configure_one_camera(1, config2)
    int_mul.start_all_cameras()
    int_mul.start_all_video_streams()

    x = threading.Thread(target=cam_test, args=(geni1,))
    x.start()
    y = threading.Thread(target=cam_test, args=(geni2,))
    y.start()
    x.join()
    y.join()

if __name__ == '__main__':
    multigenicam_test()

Am I doing something wrong? I even tried adding semaphores to let independence between the pictures, but it gets blocked anyway.

Thanks in advance! EDIT: Simplified the code a bit to make it readable.

[laserK3000]

Did you get it to work?

[franferraz98]

I did! Just needed to add a Semaphore to the camera manager class instead:

import logging
import time

import numpy as np
from pathlib import Path
from typing import List, Any, Dict
from threading import Semaphore, Thread
from src.interface_cam import Camera

class MulticameraManager:
    def __init__(self):
        """
        Interface class for managing multicamera deployments
        """
        self.camera_list = []
        self.semaphore = Semaphore()
        self.stream_threads = {}
        self.last_pictures = {}

    def add_one_camera(self, camera: Camera) -> int:
        """
        Adds a camera to the list
        :param camera: Camera to add
        :return: Index of the camera
        """
        self.camera_list.append(camera)
        return len(self.camera_list) - 1

    def add_many_cameras(self, cameras: List[Camera]) -> int:
        """
        Adds a set of cameras to the list.
        :param cameras: List of cameras to add
        :return: Index of the last camera
        """
        self.camera_list.extend(cameras)
        return len(self.camera_list) - 1

    def configure_one_camera(self, cam_id: int,
                             config: Dict[str, Any] or Path) -> None:
        """
        Configures one camera with the desired parameters via input dict
        :param cam_id: Camera id
        :param config: Configuration dictionary or path to the
            configuration file
        """
        self.camera_list[cam_id].configure_camera(config)

    def configure_many_camera(self, cam_ids: List[int],
                              config: Dict[str, Any] or Path) -> None:
        """
        Configures one camera with the desired parameters via input dict
        :param cam_ids: List of camera ids
        :param config: Configuration dictionary or path to the
            configuration file
        """
        for cam_id in cam_ids:
            self.camera_list[cam_id].configure_camera(config)

    def configure_all_cameras(self,
                              config: Dict[str, Any] or Path) -> None:
        """
        Configures all cameras on the list with the desired parameters
        via input dict
        :param config: Configuration dictionary or path to the
            configuration file
        """
        for camera in self.camera_list:
            camera.configure_camera(config)

    def start_one_camera(self, cam_id: int) -> None:
        """
        Starts one camera by id
        :param cam_id: Camera id
        """
        self.camera_list[cam_id].start_camera()

    def start_many_cameras(self, cam_ids: List[int]) -> None:
        """
        Starts a list of cameras by id
        :param cam_ids: List of camera ids
        """
        for cam_id in cam_ids:
            self.camera_list[cam_id].start_camera()

    def start_all_cameras(self) -> None:
        """
        Starts all cameras on the list
        """
        for camera in self.camera_list:
            camera.start_camera()

    def close_one_camera(self, cam_id: int) -> None:
        """
        Closes one camera by id
        :param cam_id: Camera id
        """
        self.camera_list[cam_id].close()
        if cam_id in self.stream_threads:
            self.stream_threads[cam_id].join()

    def close_many_cameras(self, cam_ids: List[int]) -> None:
        """
        Closes a list cameras by id
        :param cam_ids: List of camera ids
        """
        for cam_id in cam_ids:
            self.camera_list[cam_id].close()

    def close_all_cameras(self) -> None:
        """
        Close all cameras on the list
        """
        for camera in self.camera_list:
            camera.close()

    def take_one_photo(self, cam_id: int) -> np.ndarray:
        """
        Takes a photo from the camera from the id
        :param cam_id: Camera id
        :return: Picure taken
        """
        return self.camera_list[cam_id].take_photo()

    def take_many_photos(self, cam_ids: List[int]) -> List[np.ndarray]:
        """
        Takes a photo from each camera from the id list.
        :param cam_ids: List of camera ids
        :return: List of picures taken
        """
        img_list = []
        for cam_id in cam_ids:
            img_list.append(self.camera_list[cam_id].take_photo())
        return img_list

    def take_all_photos(self) -> List[np.ndarray]:
        """
        Takes one picture from each camera on the list.
        :return: List of pictures taken
        """
        img_list = []
        for camera in self.camera_list:
            img_list.append(camera.take_photo())
        return img_list

    def get_one_attr(self, cam_id: int, attr_name: str) -> Any:
        """
        Gets the attribute of one camera by name
        :param cam_id: Camera id
        :param attr_name: Name of the attribute to get
        :return: The value of the attribute
        """
        return self.camera_list[cam_id].get_attr(attr_name)

    def get_many_attrs(self, cam_ids: List[int], attr_name: str) -> List[Any]:
        """
        Gets the attribute of many cameras by name
        :param cam_ids: List of camera ids
        :param attr_name: Name of the attribute to get
        :return: The value of the attributes for each camera
        """
        attr_list = []
        for cam_id in cam_ids:
            attr_list.append(self.camera_list[cam_id].get_attr(attr_name))
        return attr_list

    def get_all_attrs(self, attr_name: str) -> List[Any]:
        """
        Gets the attribute of all cameras on the list by name.
        :param attr_name: Name of the attribute to get
        :return: List of values
        """
        attr_list = []
        for camera in self.camera_list:
            attr_list.append(camera.get_attr(attr_name))
        return attr_list

    def set_one_attr(self, cam_id: int, attr_name: str, attr_value: Any) -> \
            None:
        """
        Sets the attribute of one camera by name
        :param cam_id: Camera id
        :param attr_name: Name of the attribute to set
        :param attr_value: Value to set
        """
        return self.camera_list[cam_id].set_attr(attr_name, attr_value)

    def set_many_attrs(self, cam_ids: List[int], attr_name: str,
                       attr_value: Any) -> List[Any]:
        """
        Sets the attribute of many cameras by name
        :param cam_ids: List of camera ids
        :param attr_name: Name of the attribute to set
        :param attr_value: Value to set
        """
        attr_list = []
        for cam_id in cam_ids:
            attr_list.append(self.camera_list[cam_id].set_attr(
                attr_name, attr_value))
        return attr_list

    def set_all_attrs(self, attr_name: str, attr_value: Any) -> List[Any]:
        """
        Sets the attribute of all cameras on the list by name.
        :param attr_name: Name of the attribute to set
        :param attr_value: Value to set
        """
        attr_list = []
        for camera in self.camera_list:
            attr_list.append(camera.set_attr(attr_name, attr_value))
        return attr_list

    def start_one_video_stream(self, cam_id: int):
        """
        Starts the video streaming of one camera by id
        :param cam_id: Camera id
        :return: True if the thread started properly, false otherwise
        """
        camera = self.camera_list[cam_id]
        camera.is_streaming = True
        cam_thread = Thread(target=self.capture_frame_loop,
                            args=(camera.cam_id, camera, self.semaphore,))
        self.stream_threads[camera.cam_id] = cam_thread
        self.stream_threads[camera.cam_id].daemon = True
        self.stream_threads[camera.cam_id].start()

    def start_many_video_streams(self, cam_ids: List[int]):
        """
        Starts the video streaming of many cameras by id
        :param cam_ids: List of camera ids
        :return: List of rue if the thread started properly, false
            otherwise
        """
        for cam_id in cam_ids:
            camera = self.camera_list[cam_id]
            camera.is_streaming = True
            cam_thread = Thread(target=self.capture_frame_loop,
                                args=(camera.cam_id, camera, self.semaphore,))
            self.stream_threads[camera.cam_id] = cam_thread
            self.stream_threads[camera.cam_id].daemon = True
            self.stream_threads[camera.cam_id].start()

    def start_all_video_streams(self):
        """
        Starts the video streaming of all cameras on the list.
        :return: List of true if the thread started properly, false
            otherwise
        """
        for camera in self.camera_list:
            camera.is_streaming = True
            cam_thread = Thread(target=self.capture_frame_loop,
                                args=(camera.cam_id, camera, self.semaphore,))
            self.stream_threads[camera.cam_id] = cam_thread
            self.stream_threads[camera.cam_id].daemon = True
            self.stream_threads[camera.cam_id].start()

    def capture_frame_loop(self, cam_id: int, camera: Camera,
                           semaphore: Semaphore):
        while camera.is_streaming:
            semaphore.acquire()
            last_image = camera.capture_frame()
            semaphore.release()
            self.last_pictures[cam_id] = last_image
            time.sleep(0.00001)

        logging.info(f"Exiting capture loop for camera {cam_id}")

    def reboot_one(self, cam_id: int) -> None:
        """
        Reboots one camera by id
        :param cam_id: Camera id
        """
        self.camera_list[cam_id].reboot()

    def reboot_many(self, cam_ids: List[int]) -> None:
        """
        Reboots many cameras by id
        :param cam_ids: List of camera ids
        """
        for cam_id in cam_ids:
            self.camera_list[cam_id].reboot()

    def reboot_all(self) -> None:
        """
        Reboots all cameras on the list
        """
        for camera in self.camera_list:
            camera.reboot()

And then

import time
import logging
import numpy as np
from typing import Any, Tuple
from pathlib import Path
from threading import Thread, Semaphore
from abc import ABC, abstractmethod

class Camera(ABC):
    def __init__(self, config: dict[str, Any] or Path, cam_id: int = 0):
        """
        Generic abstract camera superclass for the different types of
        cameras.
        :param config: Dictionary containing the configuration variables
            of the camera.
        :param cam_id: Camera id
        """
        self.width = None
        self.height = None
        self.camera = None
        self.is_streaming = False
        self.stream_thread = None
        self.last_image = None
        self.cam_id = cam_id
        self.config = config

    @abstractmethod
    def configure_camera(self, config: dict[str, Any] or Path) -> None:
        """
        Configures the camera with the desired parameters via input dict
        :param config: Configuration dictionary or path to the
            configuration file
        """

    @abstractmethod
    def start_camera(self) -> None:
        """Initializes the camera."""

    @abstractmethod
    def get_framerate(self) -> float:
        """
        Gets the framerate
        :return: Number of frames per second
        """

    @abstractmethod
    def get_exposure(self) -> float:
        """
        Gets the exposure time.
        :return: Exposure time, in milliseconds
        """

    @abstractmethod
    def get_resolution(self) -> Tuple[int, int]:
        """
        Gets the camera resolution
        :return: Number of pixels in (width, height) format
        """

    @abstractmethod
    def get_white_balance(self) -> float:
        """
        Set white balance.
        :return: White balancing ratio
        """

    @abstractmethod
    def set_framerate(self, framerate: float) -> None:
        """
        Sets the framerate
        :param framerate: Number of frames per second
        :return: True if success, False otherwise
        """

    @abstractmethod
    def set_exposure(self, exposure: float) -> None:
        """
        Gets the exposure time.
        :param exposure: Exposure time, in milliseconds.
        """

    @abstractmethod
    def set_resolution(self, resolution: Tuple[int, int]) -> None:
        """
        Sets the camera resolution
        :param resolution: Number of pixels in (width, height) format
        """

    @abstractmethod
    def set_white_balance(self, wbr: float) -> None:
        """
        Sets the white balance ratio.
        :param wbr: White balance ratio
        """

    @abstractmethod
    def capture_frame(self) -> np.ndarray:
        """Captures a frame."""

    def capture_frame_loop(self) -> None:
        """Loops capture_frame and updates last_image"""
        while self.is_streaming:
            self.last_image = self.capture_frame()
        logging.info("Exiting capture loop")

    def is_aquiring(self) -> bool:
        """Retruns true if the streaming is on, false otherwise"""
        if self.stream_thread is not None:
            return self.stream_thread.is_alive()
        else:
            return False

    def get_last_frame(self) -> np.ndarray:
        """Returns the last frame taken by the camera"""
        return self.last_image

    def take_photo(self) -> np.ndarray:
        """
        Captures a frame with the camera,
        counts the time it takes and logs it.
        :return: An OpenCV image in RGB format
        """
        start_take_photo = time.perf_counter()
        frame = self.capture_frame()
        t_take_photo = (time.perf_counter() - start_take_photo) * 1000
        logging.info(f"Image taken in {t_take_photo:.0f} milliseconds")
        return frame

    def start_video_stream(self) -> bool:
        """
        Starts an inner thread that takes pictures
        """
        try:
            self.is_streaming = True
            self.stream_thread = Thread(target=self.capture_frame_loop,
                                        args=())
            self.stream_thread.daemon = True
            self.stream_thread.start()
        except Exception as e:
            logging.error(f"Video stream couldn't be started... \n{e}")
            return False
        return True

    @abstractmethod
    def close(self) -> None:
        """Closes the camera."""

    def reboot(self) -> None:
        """Reboots the camera."""
        self.close()
        time.sleep(10)
        self.start_camera()

This is working fine aside from the fact that the cameras are a bit slowed down. I'll close the issue now.