Module `enrgdaq.daq.jobs.camera`

Classes

class DAQJobCamera (config: DAQJobCameraConfig, **kwargs)

Expand source code

class DAQJobCamera(DAQJob):
    """
    DAQ job for capturing images from a camera using PIL and pygrabber.
    """

    allowed_message_in_types = []
    config_type = DAQJobCameraConfig
    config: DAQJobCameraConfig
    _cam: Optional[cv2.VideoCapture]

    def __init__(self, config: DAQJobCameraConfig, **kwargs):
        super().__init__(config, **kwargs)
        self._cam = None

    def start(self):
        self._cam = cv2.VideoCapture(self.config.camera_device_index)

        while True:
            start_time = datetime.now()
            self.capture_image()
            sleep_for(self.config.store_interval_seconds, start_time)

    def capture_image(self):
        assert self._cam is not None

        self._logger.debug("Capturing image...")
        res, frame = self._cam.read()
        assert res
        if self.config.enable_time_text:
            # insert date & time
            frame = self._insert_date_and_time(frame)
        # get bytes from frame
        res, buffer = cv2.imencode(".jpg", frame)
        assert res
        image_data = buffer.tobytes()

        self._put_message_out(
            DAQJobMessageStoreRaw(
                data=image_data,
                store_config=self.config.store_config,
            )
        )
        self._logger.debug("Image captured and sent")

    def _insert_date_and_time(self, frame):
        """
        Inserts the current date and time as a text overlay on the given frame.
        The text color adjusts based on the brightness of the background.
        """
        # Get frame dimensions
        frame_height, frame_width, _ = frame.shape

        # Constants for font properties
        FONT_SCALE_FACTOR = 1.8e-3
        THICKNESS_FACTOR = 5e-3

        # Calculate font scale and thickness dynamically based on frame size
        font_scale = min(frame_width, frame_height) * FONT_SCALE_FACTOR
        thickness = max(1, int(min(frame_width, frame_height) * THICKNESS_FACTOR))

        # Generate the text to overlay
        current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        # Draw the background rectangle
        (text_width, text_height), baseline = cv2.getTextSize(
            current_time, cv2.FONT_HERSHEY_SIMPLEX, font_scale, thickness
        )

        # Determine the position of the text based on the configuration
        if self.config.time_text_position == TimeTextPosition.TOP_LEFT:
            x_offset = 0
            y_offset = int(30 * font_scale)
        elif self.config.time_text_position == TimeTextPosition.TOP_RIGHT:
            x_offset = frame_width - text_width
            y_offset = int(30 * font_scale)
        elif self.config.time_text_position == TimeTextPosition.BOTTOM_LEFT:
            x_offset = 0
            y_offset = frame_height - 10
        elif self.config.time_text_position == TimeTextPosition.BOTTOM_RIGHT:
            x_offset = frame_width - text_width
            y_offset = frame_height - 10

        # Create a transparent overlay
        overlay = frame.copy()

        # Draw the rectangle on the overlay
        cv2.rectangle(
            overlay,
            (x_offset, y_offset - text_height - baseline),
            (x_offset + text_width, y_offset + baseline),
            (0, 0, 0),
            cv2.FILLED,
        )

        # Blend the overlay with the frame to achieve the desired opacity
        alpha = self.config.time_text_background_opacity
        cv2.addWeighted(overlay, alpha, frame, 1 - alpha, 0, frame)

        # Put the text on the frame
        cv2.putText(
            frame,
            current_time,
            (x_offset, y_offset),
            cv2.FONT_HERSHEY_SIMPLEX,
            font_scale,
            (255, 255, 255),
            thickness,
        )

        return frame

DAQ job for capturing images from a camera using PIL and pygrabber.

Ancestors

DAQJob

Class variables

var allowed_message_in_types : list[type[DAQJobMessage]]

var config : DAQJobCameraConfig

var config_type : Any

Configuration class for DAQJobCamera.

Attributes

camera_device_index: The index of the camera device to use.
store_interval_seconds: The interval in seconds to store images.
enable_time_text: Whether to enable the time text overlay.
time_text_position: The position of the time text on the image.
time_text_background_opacity: The opacity of the time text background.

Methods

def capture_image(self)

Expand source code

def capture_image(self):
    assert self._cam is not None

    self._logger.debug("Capturing image...")
    res, frame = self._cam.read()
    assert res
    if self.config.enable_time_text:
        # insert date & time
        frame = self._insert_date_and_time(frame)
    # get bytes from frame
    res, buffer = cv2.imencode(".jpg", frame)
    assert res
    image_data = buffer.tobytes()

    self._put_message_out(
        DAQJobMessageStoreRaw(
            data=image_data,
            store_config=self.config.store_config,
        )
    )
    self._logger.debug("Image captured and sent")

def start(self)

Expand source code

def start(self):
    self._cam = cv2.VideoCapture(self.config.camera_device_index)

    while True:
        start_time = datetime.now()
        self.capture_image()
        sleep_for(self.config.store_interval_seconds, start_time)

Inherited members

DAQJob:
- consume
- handle_message

class DAQJobCameraConfig (store_config: DAQJobStoreConfig, camera_device_index: int = 0, store_interval_seconds: int = 5, enable_time_text: bool = True, time_text_position: TimeTextPosition = TimeTextPosition.TOP_LEFT, *, verbosity: LogVerbosity = LogVerbosity.INFO, remote_config: DAQRemoteConfig | None = <factory>, daq_job_type: str)

Expand source code

class DAQJobCameraConfig(StorableDAQJobConfig):
    """
    Configuration class for DAQJobCamera.
    Attributes:
        camera_device_index: The index of the camera device to use.
        store_interval_seconds: The interval in seconds to store images.
        enable_time_text: Whether to enable the time text overlay.
        time_text_position: The position of the time text on the image.
        time_text_background_opacity: The opacity of the time text background.
    """

    camera_device_index: int = 0
    store_interval_seconds: int = 5
    enable_time_text: bool = True
    time_text_position: TimeTextPosition = TimeTextPosition.TOP_LEFT
    time_text_background_opacity = 0.7

Configuration class for DAQJobCamera.

Attributes

camera_device_index: The index of the camera device to use.
store_interval_seconds: The interval in seconds to store images.
enable_time_text: Whether to enable the time text overlay.
time_text_position: The position of the time text on the image.
time_text_background_opacity: The opacity of the time text background.

Ancestors

StorableDAQJobConfig
DAQJobConfig
msgspec.Struct
msgspec._core._StructMixin

Class variables

var time_text_background_opacity

Instance variables

var camera_device_index : int

Expand source code

class DAQJobCameraConfig(StorableDAQJobConfig):
    """
    Configuration class for DAQJobCamera.
    Attributes:
        camera_device_index: The index of the camera device to use.
        store_interval_seconds: The interval in seconds to store images.
        enable_time_text: Whether to enable the time text overlay.
        time_text_position: The position of the time text on the image.
        time_text_background_opacity: The opacity of the time text background.
    """

    camera_device_index: int = 0
    store_interval_seconds: int = 5
    enable_time_text: bool = True
    time_text_position: TimeTextPosition = TimeTextPosition.TOP_LEFT
    time_text_background_opacity = 0.7

var enable_time_text : bool

Expand source code

class DAQJobCameraConfig(StorableDAQJobConfig):
    """
    Configuration class for DAQJobCamera.
    Attributes:
        camera_device_index: The index of the camera device to use.
        store_interval_seconds: The interval in seconds to store images.
        enable_time_text: Whether to enable the time text overlay.
        time_text_position: The position of the time text on the image.
        time_text_background_opacity: The opacity of the time text background.
    """

    camera_device_index: int = 0
    store_interval_seconds: int = 5
    enable_time_text: bool = True
    time_text_position: TimeTextPosition = TimeTextPosition.TOP_LEFT
    time_text_background_opacity = 0.7

var store_interval_seconds : int

Expand source code

class DAQJobCameraConfig(StorableDAQJobConfig):
    """
    Configuration class for DAQJobCamera.
    Attributes:
        camera_device_index: The index of the camera device to use.
        store_interval_seconds: The interval in seconds to store images.
        enable_time_text: Whether to enable the time text overlay.
        time_text_position: The position of the time text on the image.
        time_text_background_opacity: The opacity of the time text background.
    """

    camera_device_index: int = 0
    store_interval_seconds: int = 5
    enable_time_text: bool = True
    time_text_position: TimeTextPosition = TimeTextPosition.TOP_LEFT
    time_text_background_opacity = 0.7

var time_text_position : TimeTextPosition

Expand source code

class DAQJobCameraConfig(StorableDAQJobConfig):
    """
    Configuration class for DAQJobCamera.
    Attributes:
        camera_device_index: The index of the camera device to use.
        store_interval_seconds: The interval in seconds to store images.
        enable_time_text: Whether to enable the time text overlay.
        time_text_position: The position of the time text on the image.
        time_text_background_opacity: The opacity of the time text background.
    """

    camera_device_index: int = 0
    store_interval_seconds: int = 5
    enable_time_text: bool = True
    time_text_position: TimeTextPosition = TimeTextPosition.TOP_LEFT
    time_text_background_opacity = 0.7

class TimeTextPosition (*args, **kwds)

Expand source code

class TimeTextPosition(str, Enum):
    """Position of the time text on the image."""

    TOP_LEFT = "top_left"
    TOP_RIGHT = "top_right"
    BOTTOM_LEFT = "bottom_left"
    BOTTOM_RIGHT = "bottom_right"

Position of the time text on the image.

Ancestors

builtins.str
enum.Enum

Class variables

var BOTTOM_LEFT
var BOTTOM_RIGHT
var TOP_LEFT
var TOP_RIGHT