Prima notte di debug di luglio (d30e1ab5) · Commits · Davide Ricci / noctua

noctua/api/telescope.py

+3 −4

Original line number	Diff line number	Diff line
		@@ -120,6 +120,7 @@ class CoordinatesResolve(BaseResource):
		@expects(param_type="string", unit="hms ±dms / id", placeholder="Polaris")
		async def post(self):
		'''Return the resolved canonical coordinate string, or an error if unresolvable.'''

		target = await self.get_payload()
		radec = await self.run_blocking(to_radec, target)

		@@ -134,13 +135,11 @@ class CoordinatesResolve(BaseResource):
		class CoordinatesMovementRadec(BaseResource):
		'''Point the telescope in Ra, Dec.'''

		@expects(param_type="string", count=1, unit="hms ±dms / id", placeholder="Polaris")
		@expects(param_type="string", unit="hms ±dms / id", placeholder="Polaris")
		async def post(self):
		'''Set new Ra and Dec coordinates.'''

		target = await self.get_payload()

		# Coordinate resolution can be slow (network catalogs)
		radec = await self.run_blocking(to_radec, target)

		def action():

noctua/config/constants.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -75,7 +75,7 @@ xmax = [0, 4145] # max xrange in binning 1.
		ymax = [0, 4126] # max yrange in binning 1.

		pixscale = 0.283 # arcsec/px in binning 1. From a resolved FITS
		rotangle = -89.67 # -90 # typical rotation angle. From a resolved FITS
		rotangle = -60 # -90 # typical rotation angle. From a resolved FITS

		temp_fits = str(_DATA_DIR / "fits" / "temp.fits")
		temp_fits0 = str(_DATA_DIR / "fits" / "temp0.fits")

noctua/devices/astelco.py

+3 −3

Original line number	Diff line number	Diff line
		@@ -270,17 +270,17 @@ class Telescope(OpenTSI):

		self.tracking = False

		def load_pointing_model(self, port=0, orientation=0):
		def load_pointing_model(self, filename="cerbero-2026-07-12", port=0, orientation=0):
		"""
		Load the pointing model measurement file from ini and recalculate coefficients.
		Port: Nasmyth 1
		Orientation: Normal pointing
		"""

		filename = getattr(self, '_pointing_model', '/opt/tsi/pm/Modello di puntamento 2025 Novembre Davide')
		#filename = getattr(self, '_pointing_model', '/opt/tsi/pm/cerbero-2026-07-12')
		self.put("TELESCOPE.MEASUREMENT.MODEL.FILE.NAME", f'"{filename}"')
		self.put("TELESCOPE.MEASUREMENT.MODEL.FILE.LOAD", 1)
		# self.put(f"TELESCOPE.CONFIG.PORT[{port}].MODEL[{orientation}].CALCULATE", 1)
		self.put(f"TELESCOPE.CONFIG.PORT[{port}].MODEL[{orientation}].CALCULATE", 1)

		@property
		def is_moving(self):

noctua/devices/mako.py

+210 −143

Original line number	Diff line number	Diff line
		@@ -2,51 +2,47 @@
		# -- coding: utf-8 --

		"""
		Driver for Allied Vision Mako cameras using VmbPy.
		Utilizes external image and streaming utilities.
		Driver for Allied Vision Mako GigE cameras via VmbPy.
		Camera ID is the IP address (e.g. 10.185.119.111).
		"""

		# System modules
		import threading
		import time
		from datetime import datetime as dt
		from pathlib import Path
		import numpy as np

		# Third-party modules
		from astropy.io import fits
		from vmbpy import VmbSystem

		# Custom modules
		from .basedevice import BaseDevice
		from ..utils.image import make_png, Streamer
		from ..utils.image import make_png
		from ..utils.logger import log


		class Mako(BaseDevice):
		"""
		Base wrapper class for Allied Vision Mako cameras
		with persistent connection.
		"""
		"""Low-level wrapper for Allied Vision Mako GigE cameras via VmbPy.

		def __init__(self, url):
		"""
		Constructor
		Manages the persistent VmbSystem and Camera context.
		The camera ID is the IP address on the GigE network.
		"""

		def __init__(self, url):
		super().__init__(url)
		self.id = url
		self.vmb = VmbSystem.get_instance()
		self._cam = None

		def _check_connection(self):
		"""
		Internal method to manage persistent Vimba and Camera context.
		"""

		"""Open and cache the VmbSystem + Camera context."""
		if self._cam is None:
		self.vmb.__enter__()
		try:
		self._cam = self.vmb.get_camera_by_id(self.id)
		self._cam.__enter__()
		# Setup GigE packet size
		try:
		stream = self._cam.get_streams()[0]
		stream.GVSPAdjustPacketSize.run()
		@@ -56,58 +52,173 @@ class Mako(BaseDevice):
		pass
		except Exception as e:
		self._cam = None
		raise ConnectionError(f"Connection failed: {e}")
		msg = f"Mako connection failed: {e}"
		if msg not in self.error:
		self.error.append(msg)
		log.error(msg)
		raise ConnectionError(msg)
		return self._cam

		def get(self, feature_name):
		"""Return the value of a Vimba feature."""

		"""Read a VmbPy camera feature by name."""
		cam = self._check_connection()
		return getattr(cam, feature_name).get()
		return cam.get_feature_by_name(feature_name).get()

		def put(self, feature_name, value):
		"""Set the value of a Vimba feature."""

		"""Write a VmbPy camera feature by name. Returns True on success."""
		try:
		cam = self._check_connection()
		getattr(cam, feature_name).set(value)
		cam.get_feature_by_name(feature_name).set(value)
		return True
		except Exception as e:
		log.warning(f"Mako {feature_name}: {e}")
		return False

		def __del__(self):
		"""Clean shutdown of camera and Vimba system."""

		if self._cam:
		try:
		self._cam.__exit__(None, None, None)
		except Exception:
		pass
		if self.vmb:
		try:
		self.vmb.__exit__(None, None, None)
		except Exception:
		pass


		class Guider(Mako):
		"""High-level interface for Mako cameras with frame capture and loop support.
		"""High-level interface for Allied Vision Mako GigE cameras.

		Single-capture mode: call ``start()`` to grab one frame via ``get_frame()``.
		Capture-mode camera: call ``start()`` to acquire one frame synchronously.
		The result is immediately available in ``matrix``.
		Use ``looping`` to run continuous acquisitions in a background thread.
		The latest frame is always available via ``matrix``.

		States
		------
		0 : Idle
		2 : Exposing / acquiring
		5 : Error
		"""

		dtype = np.uint8 # raw sensor data is 8-bit grayscale
		dtype = np.uint8
		acquisition = 'capture'

		def __init__(self, url):
		"""
		Constructor
		"""

		super().__init__(url)
		self._last_frame = None
		self._last_exptime = None
		self._last_datetime = None
		self._lock = threading.Lock()
		self._streamer = None
		self._state = 0

		self._looping = False
		self._loop_thread = None
		self.loop_exposure = 1.0

		# --- Connection ---

		@property
		def connection(self):
		"""bool : True if the camera is reachable."""
		try:
		self.get('DeviceID')
		return True
		except Exception:
		return False

		# --- State ---

		@property
		def state(self):
		"""int : Current camera state (0=Idle, 2=Exposing, 5=Error)."""
		return self._state

		# --- Low-level feature helpers ---

		def _set_exposure(self, exptime_s):
		self.put('ExposureAuto', 'Off')
		us = exptime_s * 1e6
		# Mako GigE uses ExposureTimeAbs; newer cameras use ExposureTime
		if not self.put('ExposureTimeAbs', us):
		self.put('ExposureTime', us)

		def _set_gain(self, gain):
		self.put('GainAuto', 'Off')
		self.put('Gain', float(gain))

		def _set_binning(self, b):
		try:
		cam = self._check_connection()
		lo, hi = cam.get_feature_by_name('BinningHorizontal').get_range()
		if not (lo <= b <= hi):
		log.error(f"Mako: binning {b} non nel range [{lo}, {hi}]")
		return
		except Exception:
		pass
		self.put('BinningHorizontal', b)
		self.put('BinningVertical', b)
		# after a binning change reset W/H to sensor max, or the old ROI stays
		try:
		self.put('Width', self.get('WidthMax'))
		self.put('Height', self.get('HeightMax'))
		except Exception:
		pass

		def _acquire(self, cam, exptime_s=None):
		"""Call get_frame() with a timeout derived from the exposure time."""
		if exptime_s is None:
		exptime_s = self.loop_exposure
		timeout_ms = int(exptime_s * 1000) + 5000
		frame = cam.get_frame(timeout_ms=timeout_ms)
		return np.squeeze(frame.as_numpy_ndarray().copy())

		# --- Acquisition ---

		def start(self, exptime=None):
		"""Acquire a single frame synchronously.

		Refuses with an error if a loop is active; set ``looping = False`` first.
		The result is stored in ``matrix``.

		Parameters
		----------
		exptime : float, optional
		Exposure time in seconds. If None, the current camera setting is used.
		"""
		if self._looping:
		log.error("Mako: camera is looping. Set looping=False first.")
		self.error.append("Camera is looping")
		return
		self._start(exptime)

		def _start(self, exptime=None):
		"""Trigger acquisition unconditionally (no loop guard)."""
		if exptime is not None:
		self._set_exposure(exptime)
		self._last_exptime = exptime
		try:
		cam = self._check_connection()
		except ConnectionError:
		self._state = 5
		return
		self._state = 2
		self._last_datetime = dt.utcnow().isoformat(timespec='milliseconds')
		try:
		raw = self._acquire(cam, exptime)
		with self._lock:
		self._last_frame = raw
		self._state = 0
		except Exception as e:
		log.error(f"Mako acquisition error: {e}")
		self.error.append(str(e))
		self._state = 5

		# --- Looping ---

		@property
		def looping(self):
		"""bool : True if the camera is continuously acquiring frames in a background loop."""
		"""bool : True if the camera is continuously acquiring in a background loop."""
		return self._looping

		@looping.setter
		@@ -121,52 +232,53 @@ class Guider(Mako):
		self._loop_thread.start()

		def _run_loop(self):
		try:
		self._set_exposure(self.loop_exposure)
		cam = self._check_connection()
		cam.ExposureTime.set(self.loop_exposure * 1e6)
		except Exception:
		self._state = 5
		self._looping = False
		return
		while self._looping:
		try:
		raw = cam.get_frame().as_numpy_ndarray()
		if raw is not None:
		self._state = 2
		self._last_datetime = dt.utcnow().isoformat(timespec='milliseconds')
		raw = self._acquire(cam, self.loop_exposure)
		with self._lock:
		self._last_frame = raw.copy()
		self._last_frame = raw
		self._state = 0
		except Exception as e:
		log.error(f"Mako loop error: {e}")
		self._state = 5
		time.sleep(1.0)
		self._state = 0

		def start(self, exptime=None):
		"""Capture a single frame via a blocking get_frame() call.
		# --- Binning ---

		Refuses with an error if a loop is active; set ``looping = False`` first.
		The result is stored in the internal buffer and accessible via ``matrix``.
		@property
		def binning(self):
		"""list of int : Current [X, Y] binning factors."""
		try:
		return [int(self.get('BinningHorizontal')), int(self.get('BinningVertical'))]
		except Exception:
		return [None, None]

		Parameters
		----------
		exptime : float, optional
		Exposure time in seconds. If given, sets the camera ExposureTime
		feature before capture. If None, the current camera setting is used.
		"""
		@binning.setter
		def binning(self, b):
		"""Set binning. Accepts int (symmetric) or [bx, by]."""
		if self._looping:
		log.error("Camera is looping. Set looping=False first.")
		log.error("Mako: cannot change binning while looping.")
		return
		cam = self._check_connection()
		if exptime is not None:
		cam.ExposureTime.set(exptime * 1e6)
		raw = cam.get_frame().as_numpy_ndarray()
		if raw is not None:
		with self._lock:
		self._last_frame = raw.copy()
		bx = int(b[0]) if isinstance(b, (list, tuple)) else int(b)
		self._set_binning(bx)

		# --- Image output ---

		@property
		def matrix(self):
		"""
		Get the latest image as a 2-D uint8 numpy array ``(H, W)``.
		Updated by start() or by the loop.
		"""
		"""ndarray or None : Last acquired frame as a 2-D uint8 array (H, W)."""
		with self._lock:
		raw = self._last_frame
		if raw is None:
		return None
		return np.squeeze(raw)
		return self._last_frame

		def image(self, vmin=None, vmax=None, color=True):
		"""Return the current frame as PNG-encoded bytes.
		@@ -174,16 +286,15 @@ class Guider(Mako):
		Parameters
		----------
		vmin : float, optional
		Lower clip value for contrast scaling. Defaults to data minimum.
		Lower clip value. Defaults to data minimum.
		vmax : float, optional
		Upper clip value for contrast scaling. Defaults to data maximum.
		Upper clip value. Defaults to data maximum.
		color : bool, optional
		If True apply viridis colormap, if False use grayscale. Default True.
		Apply viridis colormap if True, grayscale if False. Default True.

		Returns
		-------
		bytes or None
		PNG image bytes, or None if no frame is available.
		"""
		data = self.matrix
		if data is None:
		@@ -198,11 +309,8 @@ class Guider(Mako):
		filepath : str, optional
		Destination path. Defaults to the PNG path from cameras.ini.
		vmin : float, optional
		Lower clip value for contrast scaling.
		vmax : float, optional
		Upper clip value for contrast scaling.
		color : bool, optional
		If True apply viridis colormap, if False use grayscale. Default True.

		Returns
		-------
		@@ -211,85 +319,44 @@ class Guider(Mako):
		"""
		if filepath is None:
		from ..config.constants import viewer_fits_path
		filepath = viewer_fits_path(getattr(self, "_viewer_key", None)).replace(".fits", ".png")
		filepath = viewer_fits_path(getattr(self, '_viewer_key', None)).replace('.fits', '.png')
		png = self.image(vmin=vmin, vmax=vmax, color=color)
		if png is not None:
		from pathlib import Path
		Path(filepath).parent.mkdir(parents=True, exist_ok=True)
		with open(filepath, "wb") as f:
		with open(filepath, 'wb') as f:
		f.write(png)
		return filepath


		def download(self, filename=None):
		def download(self, filepath=None):
		"""Save the current frame as a FITS file.

		Parameters
		----------
		filename : str, optional
		filepath : str, optional
		Destination path. Defaults to the path from cameras.ini.

		Returns
		-------
		str
		Absolute path to the saved FITS file.
		str or None
		Absolute path to the saved FITS file, or None if no image is available.
		"""
		if filename is None:
		if filepath is None:
		from ..config.constants import viewer_fits_path
		filename = viewer_fits_path(getattr(self, '_viewer_key', None))
		data = self.matrix # already (H, W)
		if data is not None:
		filepath = viewer_fits_path(getattr(self, '_viewer_key', None))
		data = self.matrix
		if data is None:
		return None
		Path(filepath).parent.mkdir(parents=True, exist_ok=True)
		hdu = fits.PrimaryHDU(data)
		hdu.writeto(filename, overwrite=True)
		return filename


		@property
		def autoexpose(self):
		"""tuple : ExposureAuto feature current value and allowed options."""
		cam = self._check_connection()
		return cam.ExposureAuto.as_tuple()


		@autoexpose.setter
		def autoexpose(self, value):
		"""Set ExposureAuto ('Continuous', 'Off', or 'Once')."""

		self.put("ExposureAuto", value)


		def stream(self, active=True, host='0.0.0.0', port=5534, fps=2):
		"""Start or stop the HTTP MJPEG frame server.

		Serves ``matrix`` frames while ``looping`` is active.

		Parameters
		----------
		active : bool, optional
		True to start, False to stop. Default True.
		host : str, optional
		Bind address. Default '0.0.0.0'.
		port : int, optional
		TCP port. Default 5534.
		fps : int, optional
		Target frame rate. Default 2.
		"""
		if active:
		if not self._streamer:
		self._streamer = Streamer(
		host=host,
		port=port,
		image_provider=lambda: self.matrix,
		status_provider=lambda: self.looping,
		fps=fps
		)
		self._streamer.start()
		log.info(f"Started server http://{host}:{port} fps={fps}")

		else:
		if self._streamer:
		self._streamer.stop()
		log.info(f"Stopped server http://{host}:{port}")
		self._streamer = None


		hdr = hdu.header
		hdr['INSTRUME'] = (f'Mako GigE {self.id}', 'Camera identifier')
		if self._last_exptime is not None:
		hdr['EXPTIME'] = (float(self._last_exptime), '[s] Exposure duration')
		if self._last_datetime is not None:
		hdr['DATE-OBS'] = (self._last_datetime, '(UTC) Date the exposure was started')
		bx, by = self.binning
		if bx is not None:
		hdr['XBINNING'] = (bx, 'X binning factor')
		hdr['YBINNING'] = (by, 'Y binning factor')
		hdu.writeto(filepath, overwrite=True)
		return filepath

noctua/web/pages/macros/control_panel.html

+1 −1

Original line number	Diff line number	Diff line
		@@ -33,7 +33,7 @@
		id="btn-mount-slew"
		data-method="POST"
		data-url="/telescope/coordinates/movement/radec"
		data-inputs="in-radec">Slew</button>
		data-inputs="val1-radec">Slew</button>
		<button class=" col-md-2 btn btn-outline-danger btn-universal"
		data-method="DELETE"
		data-url="/telescope/coordinates/movement">Stop</button>