Fix atik.py: IndentationError bloccante, race condition su temperatura/SDK;... (e6622964) · Commits · Davide Ricci / noctua

noctua/devices/atik.py

+258 −227

Original line number	Diff line number	Diff line
		@@ -74,6 +74,13 @@ class Camera(BaseDevice):
		self._loop_thread = None
		self.loop_exposure = 1.0

		# Serializes all Artemis SDK calls: the SDK is not thread-safe for
		# a single handle, and this camera (unlike mako/stl) reads most
		# properties live from the SDK instead of a cached last-frame, so
		# request threads and the background loop thread can otherwise call
		# into it concurrently (observed as a corrupted CCD-TEMP reading).
		self._lock = threading.RLock()

		try:
		self._lib = ctypes.CDLL("/usr/lib/libatikcameras.so")
		self._lib.ArtemisConnect.restype = ctypes.c_void_p
		@@ -88,6 +95,7 @@ class Camera(BaseDevice):
		def _check_connection(self):
		"""Internal method to manage the persistent camera handle."""

		with self._lock:
		if self._handle is None and self._lib:
		count = self._lib.ArtemisDeviceCount()
		log.debug(f"SDK reported {count} devices.")
		@@ -137,6 +145,7 @@ class Camera(BaseDevice):
		@property
		def connection(self):
		if not self._lib: return False
		with self._lock:
		h = self._check_connection()
		return bool(self._lib.ArtemisIsConnected(h)) if h else False

		@@ -144,6 +153,7 @@ class Camera(BaseDevice):

		def abort(self):
		"""Abort the current exposure."""
		with self._lock:
		h = self._check_connection()
		if h: self._lib.ArtemisAbortExposure(h)

		@@ -201,6 +211,7 @@ class Camera(BaseDevice):
		datetime : str, optional
		UTC ISO-8601 string stored for later FITS header use.
		"""
		with self._lock:
		h = self._check_connection()
		if not h: return
		if self.state != 0:
		@@ -258,9 +269,20 @@ class Camera(BaseDevice):
		if not h: return

		log.debug(f"Getting original data")
		while not self._lib.ArtemisImageReady(h):
		# Poll with the lock released between checks so a long exposure
		# wait doesn't stall other threads' SDK calls (e.g. temperature).
		while True:
		with self._lock:
		ready = self._lib.ArtemisImageReady(h)
		if ready:
		break
		time.sleep(0.1)

		from pathlib import Path
		from ..config.constants import frame_type as _frame_type
		Path(filepath).parent.mkdir(parents=True, exist_ok=True)

		with self._lock:
		x, y, w, h_img, bx, by = [ctypes.c_int() for _ in range(6)]
		self._lib.ArtemisGetImageData(h, ctypes.byref(x), ctypes.byref(y),
		ctypes.byref(w), ctypes.byref(h_img),
		@@ -271,14 +293,9 @@ class Camera(BaseDevice):
		if not buf_ptr:
		return

		from pathlib import Path
		from ..config.constants import frame_type as _frame_type
		Path(filepath).parent.mkdir(parents=True, exist_ok=True)


		size = w.value * h_img.value
		buffer = (ctypes.c_uint16 * size).from_address(buf_ptr)
		data = np.frombuffer(buffer, dtype=np.uint16).reshape(h_img.value, w.value)
		data = np.array(np.frombuffer(buffer, dtype=np.uint16).reshape(h_img.value, w.value))

		hdu = fits.PrimaryHDU(data)
		hdr = hdu.header
		@@ -331,6 +348,7 @@ class Camera(BaseDevice):

		Returns None if no image is ready in the SDK buffer.
		"""
		with self._lock:
		h = self._check_connection()
		if not h or not self._lib.ArtemisImageReady(h):
		return None
		@@ -413,6 +431,7 @@ class Camera(BaseDevice):
		height : int
		Sub-frame height in unbinned pixels.
		"""
		with self._lock:
		h = self._check_connection()
		if h:
		self._lib.ArtemisSubframe(h, start_x, start_y, width, height)
		@@ -426,6 +445,7 @@ class Camera(BaseDevice):
		list of int
		Full sensor [width, height] in unbinned pixels.
		"""
		with self._lock:
		h = self._check_connection()
		nx, ny = self._props.nPixelsX, self._props.nPixelsY
		if h:
		@@ -442,6 +462,7 @@ class Camera(BaseDevice):
		list of int
		Sub-frame [width, height] in unbinned pixels.
		"""
		with self._lock:
		h = self._check_connection()
		nx, ny = self._props.nPixelsX, self._props.nPixelsY
		coords = camera_frame(getattr(self, '_viewer_key', None), 'half_frame')
		@@ -465,6 +486,7 @@ class Camera(BaseDevice):
		list of int
		Sub-frame [width, height] in unbinned pixels.
		"""
		with self._lock:
		h = self._check_connection()
		nx = self._props.nPixelsX
		coords = camera_frame(getattr(self, '_viewer_key', None), 'small_frame')
		@@ -499,6 +521,7 @@ class Camera(BaseDevice):
		"""int : Current camera state, in the shared camera_state convention
		(0=Idle, 1=Waiting, 2=Exposing, 3=Readout, 4=Downloading, 5=Error,
		6=Flushing)."""
		with self._lock:
		h = self._check_connection()
		raw = self._lib.ArtemisCameraState(h) if h else -1
		return self._STATE_MAP.get(raw, 5)
		@@ -506,6 +529,7 @@ class Camera(BaseDevice):
		@property
		def temperature(self):
		"""float or None : CCD temperature in degrees Celsius."""
		with self._lock:
		h = self._check_connection()
		if not h: return None
		temp = ctypes.c_int()
		@@ -519,6 +543,7 @@ class Camera(BaseDevice):
		@property
		def cooler(self):
		"""bool : Cooler active state (True=On, False=Off)."""
		with self._lock:
		h = self._check_connection()
		if not h: return False
		flags = ctypes.c_int()
		@@ -529,6 +554,7 @@ class Camera(BaseDevice):

		@cooler.setter
		def cooler(self, b):
		with self._lock:
		h = self._check_connection()
		if not h: return
		if b:
		@@ -540,6 +566,7 @@ class Camera(BaseDevice):
		@property
		def binning(self):
		"""list of int : Current [X, Y] binning factors."""
		with self._lock:
		h = self._check_connection()
		bx, by = ctypes.c_int(), ctypes.c_int()
		self._lib.ArtemisGetBin(h, ctypes.byref(bx), ctypes.byref(by))
		@@ -547,6 +574,7 @@ class Camera(BaseDevice):

		@binning.setter
		def binning(self, b):
		with self._lock:
		h = self._check_connection()
		if h: self._lib.ArtemisBin(h, b[0], b[1])

		@@ -580,6 +608,7 @@ class Camera(BaseDevice):
		@property
		def ready(self):
		"""int : 1 if an image is ready in the SDK buffer, 0 otherwise."""
		with self._lock:
		h = self._check_connection()
		if not h: return 0
		return 1 if self._lib.ArtemisImageReady(h) else 0
		@@ -592,6 +621,7 @@ class Camera(BaseDevice):
		@property
		def fan(self):
		"""int : Cooler power level as a percentage (0-100)."""
		with self._lock:
		h = self._check_connection()
		if not h: return None
		flags, level, minl, maxl, setp = [ctypes.c_int() for _ in range(5)]
		@@ -642,6 +672,7 @@ class Camera(BaseDevice):
		@property
		def all(self):
		"""dict : Comprehensive camera state matching the STX driver structure."""
		with self._lock:
		h = self._check_connection()
		if not h:
		return {}

noctua/devices/stl.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -705,7 +705,7 @@ class Camera(STL):
		Path(filepath).parent.mkdir(parents=True, exist_ok=True)
		from ..config.constants import frame_type as _frame_type

		hdu = fits.PrimaryHDU(data)
		hdu = fits.PrimaryHDU(data.astype(np.uint16))
		hdr = hdu.header
		info = self._get_ccd_info()
		hdr['INSTRUME'] = (info.name.decode() if info else "SBIG STL-11000M", "Camera model")