v1.2.0

version 1.2.0

Author: thusser

pyobs_qhyccd/__init__.py

@@ -1 +1 @@
-from .qhyccdcamera import QHYCCDCamera
+from .qhyccdcamera import QHYCCDCamera

pyobs_qhyccd/libqhyccd.pxd

@@ -103,4 +103,5 @@ cdef extern from "../lib/usr/local/include/qhyccd.h":
     unsigned int CancelQHYCCDExposingAndReadout(libusb_device_handle *handle)
     unsigned int GetQHYCCDExposureRemaining(libusb_device_handle *handle)
     unsigned int GetQHYCCDCameraStatus(libusb_device_handle *handle, unsigned char *buf);
-    void SetQHYCCDLogLevel(unsigned int logLevel);
+    unsigned int ControlQHYCCDTemp(libusb_device_handle *handle, double targettemp);
+    void SetQHYCCDLogLevel(unsigned int logLevel);

pyobs_qhyccd/qhyccdcamera.py

@@ -17,20 +17,18 @@
 log = logging.getLogger(__name__)
 
 
-class QHYCCDCamera(BaseCamera, ICamera, IWindow, IBinning, IAbortable):
+class QHYCCDCamera(BaseCamera, ICamera, IWindow, IBinning, IAbortable, ICooling):
     """A pyobs module for QHYCCD cameras."""
 
     __module__ = "pyobs_qhyccd"
 
-    def __init__(self, **kwargs: Any):
+    def __init__(self, setpoint: float=-10, **kwargs: Any):
         """Initializes a new QHYCCDCamera.
         """
         BaseCamera.__init__(self, **kwargs)
 
-        # driver
         self._driver: Optional[QHYCCDDriver] = None
-
-        # window and binning
+        self._setpoint = setpoint
         self._window = (0, 0, 0, 0)
         self._binning = (1, 1)
 
@@ -39,7 +37,7 @@ async def open(self) -> None:
         await BaseCamera.open(self)
 
         # disable logs
-        set_log_level(0)
+        set_log_level(0) #TODO:
 
         # get devices
         devices = QHYCCDDriver.list_devices()
@@ -72,8 +70,8 @@ async def open(self) -> None:
         self._window = self._driver.get_effective_area()
 
         # set cooling
-        #if self._temp_setpoint is not None:
-        #    await self.set_cooling(True, self._temp_setpoint)
+        if self._setpoint is not None:
+            await self.set_cooling(True, self._setpoint)
 
     async def close(self) -> None:
         """Close the module."""
@@ -142,33 +140,24 @@ async def list_binnings(self, **kwargs: Any) -> List[Tuple[int, int]]:
         Returns:
             List of available binnings as (x, y) tuples.
         """
-        return [(1, 1), (2, 2), (3, 3), (4, 4)]
-
-    async def _expose(self, exposure_time: float, open_shutter: bool, abort_event: asyncio.Event) -> Image:
-        """Actually do the exposure, should be implemented by derived classes.
-
-        Args:
-            exposure_time: The requested exposure time in seconds.
-            open_shutter: Whether to open the shutter.
-            abort_event: Event that gets triggered when exposure should be aborted.
-
-        Returns:
-            The actual image.
 
-        Raises:
-            GrabImageError: If exposure was not successful.
-        """
-        # check driver
+        binnings = []
+        if self._driver.is_control_available(Control.CAM_BIN1X1MODE):
+            binnings.append((1, 1))
+        if self._driver.is_control_available(Control.CAM_BIN2X2MODE):
+            binnings.append((2, 2))
+        if self._driver.is_control_available(Control.CAM_BIN3X3MODE):
+            binnings.append((3, 3))
+        if self._driver.is_control_available(Control.CAM_BIN4X4MODE):
+            binnings.append((4, 4))
+        return binnings
+
+    async def _prepare_driver_for_exposure(self, exposure_time) -> None:
         if self._driver is None:
             raise ValueError("No camera driver.")
-
-        # set binning
         log.info("Set binning to %dx%d.", self._binning[0], self._binning[1])
         self._driver.set_bin_mode(*self._binning)
 
-        # set window, divide width/height by binning, from libfli:
-        # "Note that the given lower-right coordinate must take into account the horizontal and
-        # vertical bin factor settings, but the upper-left coordinate is absolute."
         width = int(math.floor(self._window[2]) / self._binning[0])
         height = int(math.floor(self._window[3]) / self._binning[1])
         log.info(
@@ -181,101 +170,117 @@ async def _expose(self, exposure_time: float, open_shutter: bool, abort_event: a
             self._window[1],
         )
         self._driver.set_resolution(self._window[0], self._window[1], width, height)
+        self._driver.set_param(Control.CONTROL_EXPOSURE, int(exposure_time * 1000.0 * 1000.0))
 
-        # exposure time
-        self._driver.set_param(Control.CONTROL_EXPOSURE, int(exposure_time * 1000.0))
-
-        # get date obs
-        log.info(
-            "Starting exposure with %s shutter for %.2f seconds...", "open" if open_shutter else "closed", exposure_time
-        )
-        date_obs = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")
-
-        # expose
-        print("before expose", time.time())
-        self._driver.expose_single_frame()
-        print("after expose", time.time())
-
-        # wait for exposure
-        if exposure_time > 0.5:
-            await event_wait(abort_event, exposure_time - 0.5)
-        #if abort_event.is_set():
-        #    raise
-
-        # get image
-        print("before get", time.time())
-        img = self._driver.get_single_frame()
-        #loop = asyncio.get_running_loop()
-        #img = await loop.run_in_executor(None, self._driver.get_single_frame)
-        print("after get", time.time())
-
-        # wait exposure
-        await self._wait_exposure(abort_event, exposure_time, open_shutter)
-
-        # create FITS image and set header
-        image = Image(img)
+    async def _get_image_with_header(self, image_data, date_obs, exposure_time) -> Image:
+        image = Image(image_data)
         image.header["DATE-OBS"] = (date_obs, "Date and time of start of exposure")
         image.header["EXPTIME"] = (exposure_time, "Exposure time [s]")
-        #image.header["DET-TEMP"] = (self._driver.get_temp(FliTemperature.CCD), "CCD temperature [C]")
-        #image.header["DET-COOL"] = (self._driver.get_cooler_power(), "Cooler power [percent]")
-        #image.header["DET-TSET"] = (self._temp_setpoint, "Cooler setpoint [C]")
-
-        # instrument and detector
-        #image.header["INSTRUME"] = (self._driver.name, "Name of instrument")
-
-        # binning
+        image.header["DET-TEMP"] = (await self._get_ccd_temperature(), "CCD temperature [C]")
+        image.header["DET-COOL"] = (await self._get_cooling_power(), "Cooler power [percent]")
+        image.header["DET-TSET"] = (self._setpoint, "Cooler setpoint [C]")
+        # image.header["INSTRUME"] = (self._driver.name, "Name of instrument")
         image.header["XBINNING"] = image.header["DET-BIN1"] = (self._binning[0], "Binning factor used on X axis")
         image.header["YBINNING"] = image.header["DET-BIN2"] = (self._binning[1], "Binning factor used on Y axis")
-
-        # window
         image.header["XORGSUBF"] = (self._window[0], "Subframe origin on X axis")
         image.header["YORGSUBF"] = (self._window[1], "Subframe origin on Y axis")
-
-        # statistics
-        image.header["DATAMIN"] = (float(np.min(img)), "Minimum data value")
-        image.header["DATAMAX"] = (float(np.max(img)), "Maximum data value")
-        image.header["DATAMEAN"] = (float(np.mean(img)), "Mean data value")
+        image.header["DATAMIN"] = (float(np.min(image_data)), "Minimum data value")
+        image.header["DATAMAX"] = (float(np.max(image_data)), "Maximum data value")
+        image.header["DATAMEAN"] = (float(np.mean(image_data)), "Mean data value")
 
         # biassec/trimsec
-        #full = self._driver.get_visible_frame()
-        #self.set_biassec_trimsec(image.header, *full)
-
-        # return FITS image
+        # full = self._driver.get_visible_frame()
+        # self.set_biassec_trimsec(image.header, *full)
         log.info("Readout finished.")
         return image
 
-    async def _wait_exposure(self, abort_event: asyncio.Event, exposure_time: float, open_shutter: bool) -> None:
-        """Wait for exposure to finish.
+    async def _expose(self, exposure_time: float, open_shutter: bool, abort_event: asyncio.Event) -> Image:
+        """Actually do the exposure, should be implemented by derived classes.
 
-        Params:
-            abort_event: Event that aborts the exposure.
-            exposure_time: Exp time in sec.
-        """
+        Args:
+            exposure_time: The requested exposure time in seconds.
+            open_shutter: Whether to open the shutter.
+            abort_event: Event that gets triggered when exposure should be aborted.
 
-        """
-        while True:
-            # aborted?
-            if abort_event.is_set():
-                await self._change_exposure_status(ExposureStatus.IDLE)
-                raise InterruptedError("Aborted exposure.")
-
-            # is exposure finished?
-            if self._driver.is_exposing():
-                break
-            else:
-                # sleep a little
-                await asyncio.sleep(0.01)
+        Returns:
+            The actual image.
+
+        Raises:
+            GrabImageError: If exposure was not successful.
         """
 
+        await self._prepare_driver_for_exposure(exposure_time)
+        log.info("Starting exposure with %s shutter for %.2f seconds...", "open" if open_shutter else "closed", exposure_time)
+        date_obs = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")
+        self._driver.expose_single_frame()
+        await event_wait(abort_event, exposure_time-0.5)
+        loop = asyncio.get_running_loop()
+        image_data = await loop.run_in_executor(None, self._driver.get_single_frame)
+        return await self._get_image_with_header(image_data, date_obs, exposure_time)
+
     async def _abort_exposure(self) -> None:
         """Abort the running exposure. Should be implemented by derived class.
-
         Raises:
             ValueError: If an error occured.
         """
         if self._driver is None:
             raise ValueError("No camera driver.")
         #self._driver.cancel_exposure()
 
+    async def _get_cooling_power(self):
+        return self._driver.get_param(Control.CONTROL_CURPWM) /256 * 100 # TODO:
+
+    async def get_cooling(self, **kwargs: Any) -> Tuple[bool, float, float]:
+        enabled = self._driver.is_control_available(Control.CONTROL_COOLER)
+        setpoint = self._setpoint
+        power = await self._get_cooling_power()
+        return enabled, setpoint, power
+
+    async def set_cooling(self, enabled: bool, setpoint: float, **kwargs: Any) -> None:
+        #if not enabled:
+        #    self._driver.set_param(Control.CONTROL_CURPWM, 0)  #TODO: einfach PWM auf 0?
+        self._setpoint = setpoint
+        await self._cool_stepwise(setpoint)
+
+    async def _wait_for_reaching_temperature(self, target_temperature, wait_step=1):
+        while await self._get_ccd_temperature() > target_temperature:
+            print("Current temperature is", await self._get_ccd_temperature(), "Target temperature is", target_temperature)
+            if await self._cooling_bug_occured():
+                break
+            time.sleep(wait_step)
+
+    async def _cooling_bug_occured(self):
+        return (self._driver.get_param(Control.CONTROL_CURPWM) > 250) & (await self._get_ccd_temperature() < 0)
+
+    async def _get_ccd_temperature(self):
+        return self._driver.get_param(Control.CONTROL_CURTEMP)
+
+    async def _cool_stepwise(self, target_temperature, temperature_stepwidth=1):
+        print("Start stepwise cooling to ", target_temperature)
+        while await self._get_ccd_temperature() - target_temperature > temperature_stepwidth:
+            intermediate_temperature = await self._get_ccd_temperature() - temperature_stepwidth
+            print("Set temperature to", intermediate_temperature)
+            self._driver.set_temperature(intermediate_temperature)
+            await self._wait_for_reaching_temperature(intermediate_temperature)
+            if await self._cooling_bug_occured():
+                await self._handle_cooling_bug(intermediate_temperature)
+                return
+        print("Set temperature to", target_temperature)
+        self._driver.set_temperature(target_temperature)
+        print("End stepwise cooling to", target_temperature)
+
+    async def _handle_cooling_bug(self, original_target_temperature, puffer=5, correction_step=1):
+        print(f"Setpoint of {original_target_temperature:.2f} °C too low for cooler. Temporarily resetting it to {await self._get_ccd_temperature() + puffer:.2f} °C.")
+        while await self._cooling_bug_occured():
+            time.sleep(1)
+            await self._cool_stepwise(await self._get_ccd_temperature() + puffer)
+        print("Wait a minute")
+        time.sleep(60)
+        self._setpoint = original_target_temperature + correction_step
+        print("Retry stepwise cooling with new setpoint of", self._setpoint)
+        await self._cool_stepwise(self._setpoint)
+
+    async def get_temperatures(self, **kwargs: Any) -> Dict[str, float]:
+        return {"CCD": await self._get_ccd_temperature()}
 
 __all__ = ["QHYCCDCamera"]

pyobs_qhyccd/qhyccddriver.pyx

@@ -188,12 +188,19 @@ cdef class QHYCCDDriver:
         cdef unsigned int imagew, imageh, bpp
         if  GetQHYCCDChipInfo(self._device, &chipw, &chiph, &imagew, &imageh, &pixelw, &pixelh, &bpp) != 0:
             raise ValueError('Could not fetch chip info.')
-        return chipw, chiph, imagew, imageh, pixelw, pixelh, bpp
+        return chipw, chiph, imagew, imageh, pixelw, pixelh,
+
+    def get_param(self, param: Control):
+        return GetQHYCCDParam(self._device, param.value[0])
 
     def set_param(self, param: Control, value: float):
         if SetQHYCCDParam(self._device, param.value[0], value) != 0:
             raise ValueError('Could not set parameter %s to %f.' % (param, value))
 
+    def set_temperature(self, temperature: float):
+        if ControlQHYCCDTemp(self._device, temperature) != 0:
+            raise ValueError("Could not set temperature to %f °C" % temperature)
+
     def set_resolution(self, x, y, width, height):
         if SetQHYCCDResolution(self._device, x, y, width, height) != 0:
             raise ValueError('Could not set resolution to %dx%d at %d,%d.' % (width, height, x, y))

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pyobs-qhyccd"
-version = "1.1.1"
+version = "1.2.0"
 description = "pyobs module for QHYCCD cameras"
 authors = ["Tim-Oliver Husser <[email protected]>"]
 license = "MIT"