v1.1.0b1

thusser · web-flow · commit dc4b67b86c32 · 2022-11-17T11:43:30.000+01:00
version 1.1.0b1
diff --git a/pyobs_fli/__init__.py b/pyobs_fli/__init__.py
@@ -4,6 +4,7 @@
 __title__ = "FLI camera modules"
 
 from .flicamera import FliCamera
+from .flifilterwheel import FliFilterWheel
 
 
 __all__ = ["FliCamera"]
diff --git a/pyobs_fli/flibase.py b/pyobs_fli/flibase.py
@@ -0,0 +1,107 @@
+import asyncio
+import logging
+from typing import Any, Optional, Union
+
+from pyobs_fli.flidriver import DeviceType
+
+log = logging.getLogger(__name__)
+
+
+class FliBaseMixin:
+    """A base class for pyobs module for FLI devices."""
+
+    __module__ = "pyobs_fli"
+
+    def __init__(
+        self,
+        dev_type: DeviceType,
+        dev_name: Optional[str] = None,
+        dev_path: Optional[str] = None,
+        keep_alive_ping: int = 10,
+        **kwargs: Any,
+    ):
+        """Initializes a new FliCamera.
+
+        If neither dev_name nor dev_path are given, the first found device is used.
+
+        Args:
+            dev_type: Device type.
+            dev_name: Optional name for device.
+            dev_path: Optional path to device.
+            keep_alive_ping: Interval for keep alive ping.
+        """
+        from .flidriver import FliDriver  # type: ignore
+
+        # variables
+        self._dev_type = dev_type
+        self._dev_name = dev_name
+        self._dev_path = dev_path
+        self._keep_alive_ping = keep_alive_ping
+        self._driver: Optional[FliDriver] = None
+        self._device: Optional[Any] = None
+
+        # keep alive
+        self.add_background_task(self._keep_alive)
+
+    async def open(self) -> None:
+        """Open module."""
+        from .flidriver import FliDriver
+
+        # list devices
+        devices = FliDriver.list_devices(self._dev_type)
+        if len(devices) == 0:
+            raise ValueError("No FLI device found.")
+
+        # do we have a device name or path given?
+        if self._dev_name is not None or self._dev_path is not None:
+            # yes, find matching device
+            for dev in devices:
+                if dev.name.decode("utf-8") == self._dev_name or dev.filename.decode("utf-8") == self._dev_path:
+                    self._device = dev
+                    break
+            else:
+                raise ValueError("No matching device found, check dev_name/dev_path.")
+
+        else:
+            # no, just pick first one
+            self._device = devices[0]
+
+        # open driver
+        log.info(
+            'Opening connection to "%s" at %s...',
+            self._device.name.decode("utf-8"),
+            self._device.filename.decode("utf-8"),
+        )
+        self._driver = FliDriver(self._device)
+        try:
+            self._driver.open()
+        except ValueError as e:
+            raise ValueError("Could not open FLI camera: %s", e)
+
+    async def close(self) -> None:
+        # not open?
+        if self._driver is not None:
+            # close connection
+            self._driver.close()
+            self._driver = None
+
+    async def _keep_alive(self) -> None:
+        """Keep connection to camera alive."""
+        from .flidriver import FliDriver
+
+        while True:
+            # is there a valid driver?
+            if self._driver is not None:
+                # then we should be able to call it
+                try:
+                    self._driver.get_serial_string()
+                except ValueError:
+                    # no? then reopen driver
+                    log.warning("Lost connection to camera, reopening it.")
+                    self._driver.close()
+                    self._driver = FliDriver(self._device)
+
+            await asyncio.sleep(self._keep_alive_ping)
+
+
+__all__ = ["FliBaseMixin"]
diff --git a/pyobs_fli/flicamera.py b/pyobs_fli/flicamera.py
@@ -10,60 +10,41 @@
 from pyobs.images import Image
 from pyobs.utils.enums import ExposureStatus
 
+from .flibase import FliBaseMixin
+
 
 log = logging.getLogger(__name__)
 
 
-class FliCamera(BaseCamera, ICamera, IWindow, IBinning, ICooling, IAbortable):
+class FliCamera(FliBaseMixin, BaseCamera, ICamera, IWindow, IBinning, ICooling, IAbortable):
     """A pyobs module for FLI cameras."""
 
     __module__ = "pyobs_fli"
 
-    def __init__(self, setpoint: float = -20.0, keep_alive_ping: int = 10, **kwargs: Any):
+    def __init__(self, setpoint: float = -20.0, **kwargs: Any):
         """Initializes a new FliCamera.
 
         Args:
             setpoint: Cooling temperature setpoint.
-            keep_alive_ping: Interval in seconds to ping camera.
         """
         BaseCamera.__init__(self, **kwargs)
-        from .flidriver import FliDriver  # type: ignore
+        FliBaseMixin.__init__(**kwargs)
 
         # variables
-        self._driver: Optional[FliDriver] = None
-        self._device: Optional[Any] = None
         self._temp_setpoint: Optional[float] = setpoint
-        self._keep_alive_ping = keep_alive_ping
 
         # window and binning
         self._window = (0, 0, 0, 0)
         self._binning = (1, 1)
 
-        # keep alive
-        self.add_background_task(self._keep_alive)
-
     async def open(self) -> None:
         """Open module."""
         await BaseCamera.open(self)
-        from .flidriver import FliDriver
-
-        # list devices
-        devices = FliDriver.list_devices()
-        if len(devices) == 0:
-            raise ValueError("No camera found.")
+        await FliBaseMixin.open(self)
 
-        # open first one
-        self._device = devices[0]
-        log.info(
-            'Opening connection to "%s" at %s...',
-            self._device.name.decode("utf-8"),
-            self._device.filename.decode("utf-8"),
-        )
-        self._driver = FliDriver(self._device)
-        try:
-            self._driver.open()
-        except ValueError as e:
-            raise ValueError("Could not open FLI camera: %s", e)
+        # check
+        if self._driver is None:
+            raise ValueError("No driver found.")
 
         # serial number
         serial = self._driver.get_serial_string()
@@ -79,30 +60,7 @@ async def open(self) -> None:
     async def close(self) -> None:
         """Close the module."""
         await BaseCamera.close(self)
-
-        # not open?
-        if self._driver is not None:
-            # close connection
-            self._driver.close()
-            self._driver = None
-
-    async def _keep_alive(self) -> None:
-        """Keep connection to camera alive."""
-        from .flidriver import FliDriver
-
-        while True:
-            # is there a valid driver?
-            if self._driver is not None:
-                # then we should be able to call it
-                try:
-                    self._driver.get_full_frame()
-                except ValueError:
-                    # no? then reopen driver
-                    log.warning("Lost connection to camera, reopening it.")
-                    self._driver.close()
-                    self._driver = FliDriver(self._device)
-
-            await asyncio.sleep(self._keep_alive_ping)
+        await FliBaseMixin.close(self)
 
     async def get_full_frame(self, **kwargs: Any) -> Tuple[int, int, int, int]:
         """Returns full size of CCD.
diff --git a/pyobs_fli/flidriver.pyx b/pyobs_fli/flidriver.pyx
@@ -22,11 +22,16 @@ class FliTemperature(Enum):
     BASE = FLI_TEMPERATURE_BASE
 
 
+class DeviceType(Enum):
+    CAMERA = FLIDEVICE_CAMERA
+    FILTERWHEEL = FLIDEVICE_FILTERWHEEL
+
+
 cdef class FliDriver:
     """Wrapper for the FLI driver."""
 
     @staticmethod
-    def list_devices() -> List[DeviceInfo]:
+    def list_devices(device_type: DeviceType = DeviceType.CAMERA) -> List[DeviceInfo]:
         """List all FLI USB cameras connected to this computer.
 
         Returns:
@@ -39,7 +44,7 @@ cdef class FliDriver:
         cdef char name[1024]
 
         # create list of USB camera
-        if FLICreateList(FLIDOMAIN_USB | FLIDEVICE_CAMERA) != 0:
+        if FLICreateList(FLIDOMAIN_USB | device_type.value) != 0:
             raise ValueError('Could not create list of FLI cameras.')
 
         # init list of devices
@@ -368,6 +373,25 @@ cdef class FliDriver:
         if res != 0:
             raise ValueError('Could not set temperature.')
 
+    def get_model(self) -> str:
+        """Returns the model of the device.
+
+        Returns:
+            str: Model of device.
+        """
+
+        # variables
+        cdef char *model
+        cdef size_t len
+
+        # get it
+        res = FLIGetModel(self._device, model, len)
+        if res != 0:
+            raise ValueError('Could not fetch model.')
+
+        # return it
+        return str(model)
+
     def get_serial_string(self) -> str:
         """Returns serial string for camera."""
 
@@ -380,4 +404,34 @@ cdef class FliDriver:
             raise ValueError('Could not fetch serial string.')
 
         # return it
-        return str(serial)
+        return bytes(serial).decode('utf-8')
+
+    def get_filter_pos(self) -> int:
+        """Returns current filter position.
+
+        Returns:
+            Filter position.
+        """
+
+        # variables
+        cdef long pos
+
+        # get it
+        res = FLIGetFilterPos(self._device, &pos)
+        if res != 0:
+            raise ValueError('Could not fetch filter position.')
+
+        # return it
+        return pos
+
+    def set_filter_pos(self, pos: int) -> None:
+        """Set filter position.
+
+        Args:
+            pos: New filter position.
+        """
+
+        # set filter pos
+        res = FLISetFilterPos(self._device, pos)
+        if res != 0:
+            raise ValueError('Could not set filter position.')
diff --git a/pyobs_fli/flifilterwheel.py b/pyobs_fli/flifilterwheel.py
diff --git a/pyproject.toml b/pyproject.toml
