Source code for herosdevices.core.templates.camera

"""Templates for creating camera device representations."""

from abc import abstractmethod

from herosdevices.core.templates.acq_device import AcquisitionDeviceTemplate
from herosdevices.interfaces import atomiq




class CameraTemplate(atomiq.Camera, AcquisitionDeviceTemplate):
    """
    Template (base class) for cameras.

    To make a functional camera device, the user needs to implement all abstract methods.
    In addition, this call does not cover the mechanism the actually retrieve the image buffers
    from the device since it is typically special to each camera vendor/API. A general guideline
    should be to start a separate thread for the acquisition which uses the _acquisition_lock
    to prohibit concurrent exposures. For each received image, the event :meth:`acquisition_data`
    should be called. In addition, :meth:`acquisition_stopped` should be emitted
    after the acquisition.
    """

    get_camera = AcquisitionDeviceTemplate.get_device

    @abstractmethod
    def _get_exposure_time(self) -> float | None:
        """Return the current exposure time in seconds from the active configuration.

        Note:
            Implementations must read this value from the stored configuration only,
            never by querying the camera hardware directly.

        Returns:
            exposure time in seconds or `None` if not set
        """
        raise NotImplementedError

    @abstractmethod
    def _set_exposure_time(self, exposure_time: float) -> dict:
        """Return a config patch dict for the given exposure time.

        Note:
            Implementations must only compute and return the patch; they must not
            write to the camera or mutate `self._config` directly.

        Args:
            exposure_time: exposure time in seconds

        Returns:
            dict to merge into the active configuration
        """
        raise NotImplementedError

    @abstractmethod
    def _get_roi_coordinates(self) -> tuple[int, int, int, int] | None:
        """Return the current ROI from the active configuration.

        Note:
            Implementations must read this value from the stored configuration only,
            never by querying the camera hardware directly.

        Returns:
            (x_offset, y_offset, width, height) in sensor pixels
        """
        raise NotImplementedError

    @abstractmethod
    def _set_roi_coordinates(self, roi: tuple[int, int, int, int]) -> dict:
        """Return a config patch dict for the given ROI.

        Note:
            Implementations must only compute and return the patch; they must not
            write to the camera or mutate `self._config` directly.

        Args:
            roi: (x_offset, y_offset, width, height) in sensor pixels

        Returns:
            dict to merge into the active configuration
        """
        raise NotImplementedError

    @abstractmethod
    def _get_binning(self) -> tuple[int, int] | None:
        """Return the current binning from the active configuration.

        Note:
            Implementations must read this value from the stored configuration only,
            never by querying the camera hardware directly.

        Returns:
            (horizontal, vertical) binning factors
        """
        raise NotImplementedError

    @abstractmethod
    def _set_binning(self, binning: tuple[int, int]) -> dict:
        """Return a config patch dict for the given binning factors.

        Note:
            Implementations must only compute and return the patch; they must not
            write to the camera or mutate `self._config` directly.

        Args:
            binning: (horizontal, vertical) binning factors

        Returns:
            dict to merge into the active configuration
        """
        raise NotImplementedError

    @property
    def exposure_time(self) -> float | None:
        """Exposure time in seconds."""
        return self._get_exposure_time()

    @exposure_time.setter
    def exposure_time(self, exposure_time: float) -> None:
        patch = self._set_exposure_time(exposure_time)
        self.update_configuration({self._config: patch})
        if not self.configure(self._config):
            msg = "Could not set exposure time. Is the acquisition currently running? Then call .stop() and try again."
            raise RuntimeError(msg)

    @property
    def roi_coordinates(self) -> tuple[int, int, int, int] | None:
        """ROI as (x_offset, y_offset, width, height) in sensor pixels."""
        return self._get_roi_coordinates()

    @roi_coordinates.setter
    def roi_coordinates(self, roi: tuple[int, int, int, int]) -> None:
        patch = self._set_roi_coordinates(roi)
        self.update_configuration({self._config: patch})
        if not self.configure(self._config):
            msg = (
                "Could not set ROI coordinates. Is the acquisition currently running? Then call .stop() and try again."
            )
            raise RuntimeError(msg)

    @property
    def binning(self) -> tuple[int, int] | None:
        """Binning as (horizontal, vertical)."""
        return self._get_binning()

    @binning.setter
    def binning(self, binning: tuple[int, int]) -> None:
        patch = self._set_binning(binning)
        self.update_configuration({self._config: patch})
        if not self.configure(self._config):
            msg = "Could not set binning. Is the acquisition currently running? Then call .stop() and try again."
            raise RuntimeError(msg)