Loading noctua/devices/atik2.py +56 −3 Original line number Diff line number Diff line Loading @@ -6,11 +6,14 @@ # System modules import threading import time from ctypes import byref, c_int, string_at # Third-party modules import numpy as np from astropy.io import fits from AtikSDK import AtikSDKCamera, ArtemisAbortExposure, CoolingInfo from AtikSDK import (ARTEMIS_OK, AtikSDKCamera, ArtemisAbortExposure, ArtemisGetImageData, ArtemisImageBuffer, CameraError, CoolingInfo) # Custom modules from .basedevice import BaseDevice Loading Loading @@ -144,6 +147,23 @@ class Camera(BaseDevice): self.error.append(msg) return None def _download_image(self): """Fetch the last frame via ``_get_image()``, with the same connect-check / lock / error-capture convention as get()/put() above (not routed through them: ``_get_image`` is our own method, not one dispatched by name to ``self._cam``).""" if not self._check_connection(): return None try: with self._lock: return self._get_image() except Exception as e: msg = f"Atik get(get_image) failed: {e}" log.error(msg) self.error.append(msg) return None @property def connection(self): """bool : True if the camera is currently connected.""" Loading @@ -162,6 +182,39 @@ class Camera(BaseDevice): # convenience class's own connection handle. ArtemisAbortExposure(self._cam._handle) def _get_image(self): """Download the last acquired frame directly via the Artemis SDK. Bypasses AtikSDKCamera.get_image()/take_image(): both build the array with np.fromstring() in binary mode, which numpy>=2.0 (this project pins numpy==2.3.5) removed outright — raising ValueError on every call, regardless of frame size. Same module-level-function pattern as abort() above, calling ArtemisGetImageData/ArtemisImageBuffer on the convenience class's own connection handle and building the array with np.frombuffer() ourselves. Returns ------- ndarray 2-D uint16 array of the image, shape (height, width). """ x, y, w, h, bx, by = c_int(), c_int(), c_int(), c_int(), c_int(), c_int() error = ArtemisGetImageData(self._cam._handle, byref(x), byref(y), byref(w), byref(h), byref(bx), byref(by)) if error != ARTEMIS_OK: raise CameraError(error) image_buffer = ArtemisImageBuffer(self._cam._handle) if not image_buffer: raise CameraError(False) raw_string = string_at(image_buffer, w.value * h.value * 2) arr = np.frombuffer(raw_string, dtype=np.uint16, count=w.value * h.value) return arr.reshape(h.value, w.value) @property def looping(self): """bool : True if the camera is continuously acquiring frames Loading Loading @@ -296,7 +349,7 @@ class Camera(BaseDevice): time.sleep(0.1) log.debug(f"Getting image") data = self.get('get_image') data = self._download_image() if data is None: return None log.debug(f"Got image") Loading Loading @@ -361,7 +414,7 @@ class Camera(BaseDevice): if not self._check_connection() or not self.get('image_ready'): return None data = self.get('get_image') data = self._download_image() if data is None: return None Loading Loading
noctua/devices/atik2.py +56 −3 Original line number Diff line number Diff line Loading @@ -6,11 +6,14 @@ # System modules import threading import time from ctypes import byref, c_int, string_at # Third-party modules import numpy as np from astropy.io import fits from AtikSDK import AtikSDKCamera, ArtemisAbortExposure, CoolingInfo from AtikSDK import (ARTEMIS_OK, AtikSDKCamera, ArtemisAbortExposure, ArtemisGetImageData, ArtemisImageBuffer, CameraError, CoolingInfo) # Custom modules from .basedevice import BaseDevice Loading Loading @@ -144,6 +147,23 @@ class Camera(BaseDevice): self.error.append(msg) return None def _download_image(self): """Fetch the last frame via ``_get_image()``, with the same connect-check / lock / error-capture convention as get()/put() above (not routed through them: ``_get_image`` is our own method, not one dispatched by name to ``self._cam``).""" if not self._check_connection(): return None try: with self._lock: return self._get_image() except Exception as e: msg = f"Atik get(get_image) failed: {e}" log.error(msg) self.error.append(msg) return None @property def connection(self): """bool : True if the camera is currently connected.""" Loading @@ -162,6 +182,39 @@ class Camera(BaseDevice): # convenience class's own connection handle. ArtemisAbortExposure(self._cam._handle) def _get_image(self): """Download the last acquired frame directly via the Artemis SDK. Bypasses AtikSDKCamera.get_image()/take_image(): both build the array with np.fromstring() in binary mode, which numpy>=2.0 (this project pins numpy==2.3.5) removed outright — raising ValueError on every call, regardless of frame size. Same module-level-function pattern as abort() above, calling ArtemisGetImageData/ArtemisImageBuffer on the convenience class's own connection handle and building the array with np.frombuffer() ourselves. Returns ------- ndarray 2-D uint16 array of the image, shape (height, width). """ x, y, w, h, bx, by = c_int(), c_int(), c_int(), c_int(), c_int(), c_int() error = ArtemisGetImageData(self._cam._handle, byref(x), byref(y), byref(w), byref(h), byref(bx), byref(by)) if error != ARTEMIS_OK: raise CameraError(error) image_buffer = ArtemisImageBuffer(self._cam._handle) if not image_buffer: raise CameraError(False) raw_string = string_at(image_buffer, w.value * h.value * 2) arr = np.frombuffer(raw_string, dtype=np.uint16, count=w.value * h.value) return arr.reshape(h.value, w.value) @property def looping(self): """bool : True if the camera is continuously acquiring frames Loading Loading @@ -296,7 +349,7 @@ class Camera(BaseDevice): time.sleep(0.1) log.debug(f"Getting image") data = self.get('get_image') data = self._download_image() if data is None: return None log.debug(f"Got image") Loading Loading @@ -361,7 +414,7 @@ class Camera(BaseDevice): if not self._check_connection() or not self.get('image_ready'): return None data = self.get('get_image') data = self._download_image() if data is None: return None Loading