Add mode-aware expose templates and dynamic camera telemetry (a25f7a4d) · Commits · Davide Ricci / noctua

noctua/config/viewer.ini

+1 −1

Original line number	Diff line number	Diff line
		@@ -5,7 +5,7 @@
		# Paths are relative to the project data/ directory.

		[station1/scicam]
		path = fits/scicam.fits
		path = fits/scicam1.fits

		[station1/teccam]
		path = fits/teccam1.fits

noctua/templates/snapshot_imaging.py

0 → 100644

+297 −0

Original line number	Diff line number	Diff line
		#!/usr/bin/env python3
		# -- coding: utf-8 --

		# System modules
		from time import sleep

		# Third-party modules
		from astropy.time import Time

		# Other templates
		from ..config.constants import (camera_state, filter_name, filter_number,
		filter_state, frame_number, frame_type,
		image_state, temp_fits, viewer_fits_path)
		from .. import devices as _dev
		from ..devices import dom, tel
		from ..utils.coordinates import apply_offset
		from ..utils.logger import log
		from .basetemplate import BaseTemplate
		from .fillheader import Template as FillHeader
		from .lampsoff import Template as LampsOff


		class Template(BaseTemplate):
		def __init__(self):
		super().__init__()
		self.name = "snapshot_imaging"
		self.description = "Observes a field with the imaging camera"
		# self.box = 300
		# self.recenter = False
		# #self.domeslewing = None
		# if not hasattr(self, "domeslewing"):
		# log.warning("domeslewing attribute undefined, setting to False")
		# self.domeslewing = False

		def content(self, params):

		########################
		##### Params check #####
		########################

		try:
		camera_name = params.get("camera") or "cam"
		cam = getattr(_dev, camera_name)
		fits_file = viewer_fits_path(getattr(cam, '_viewer_key', None))

		objname = params.get("object") or "test"
		repeat = params.get("repeat") or 1
		frametype = params.get("frametype") or "Light"
		binning = params.get("binning") or cam.binning
		filt = params.get("filter") or (
		filter_name.get(cam.filter) if hasattr(cam, 'filter') else None)
		xystart = params.get("xystart") or cam.xystart
		xyend = params.get("xyend") or cam.xyend
		exptime = params["exptime"]
		self.domeslewing = params.get("domeslewing") or False
		self.recenter = params.get("recenter") or False
		self.box = params.get("box") or 300

		except KeyError as e:
		msg = f"Parameter {e} not found"
		log.error(msg)
		self.error.append(msg)
		return

		#########################################
		##### Switch off lamps if not Flat #####
		#########################################

		if frametype != "Flat":
		log.info("Not a 'Flat' frametype: switching off lamps")
		lao = LampsOff()
		lao.run({})

		if frametype != "Light":
		log.info("Not a 'Light' frametype: switching off recentering")
		self.recenter = False

		########################
		##### Camera setup #####
		########################

		# Bringing back the camera to Idle state.
		if cam.state != 0:
		log.warning(
		f"Resetting camera from state {camera_state[cam.state]}.")
		cam.abort()
		sleep(0.1)

		log.info(f"Camera is {camera_state[cam.state]}")

		# Changing binning if not the same as the current one
		if binning != cam.binning:
		log.info(f"Changing binning to {binning}")
		cam.binning = [binning, binning]
		sleep(0.1)

		log.info(f"Camera is {camera_state[cam.state]}")

		# Changing filter if not the same as the current one (only if camera has a filter wheel)
		if filt and hasattr(cam, 'filter'):
		# On bad ctrl-c, filter can be 15 (?!)
		log.warning(f"Filter is number {cam.filter}")
		log.warning(
		f"Filter is {filter_name[cam.filter]}, we have selected {filt}")

		if filter_number[filt] != cam.filter:
		msg = f"Changing filter to {filt}, "
		msg += f"Which is number {filter_number[filt]}"
		log.info(msg)

		cam.filter = filter_number[filt]
		sleep(0.2)

		status = 1
		while status != 0:
		self.check_pause_or_abort()

		try:
		status = int(cam.is_moving)
		except ValueError as e:
		msg = f"Error trying to contact the camera: {e}"
		log.error(msg)
		self.error.append(msg)

		if status == 2:
		msg = f"Filter {filter_state[status]}"
		log.error(msg)
		self.error.append(msg)
		return
		else:
		log.warning(f"Filter {filter_state[status]}")
		sleep(0.75)

		# Changing windowing
		if xystart and xyend:
		log.info(
		f"Changing windowing to xystart:{xystart}, xyend:{xyend}")
		cam.set_window(xystart[0],
		xystart[1],
		xyend[0]-xystart[0],
		xyend[1]-xystart[1])


		else:
		msg = f"No windowing: full frame in bininng {binning}"
		log.warning(msg)
		cam.full_frame()

		#########################
		##### Exposure loop #####
		#########################

		log.info(f"Starting {repeat} {frametype} frames of {exptime}s")

		# Looping on the number of exposures
		for exposure in range(repeat):
		if self.check_pause_or_abort():
		return

		# RA, DEC, ALT, AZ, UTC are asked to cabinet
		# simultaneously in order to avoid latency.

		log.info(f"Asking cabinet time and coordinates")
		basic_cabinet = tel.coordinates
		gps = Time(tel.coordinates["utc"], format="unix")

		log.info(f"Starting Exposure {exposure + 1} of {repeat}")

		# ########################
		# ##### Dome Slewing #####
		# ########################

		log.info(f"Dome slewing is {self.domeslewing}")
		if frametype == "Light":
		if self.domeslewing:
		log.info(f"Check Dome position")
		while (abs(tel.coordinates["altaz"][1] - dom.azimuth) > 3):
		if not self.domeslewing:
		log.debug("Stop slewing anyway")
		return

		dom.azimuth = tel.coordinates["altaz"][1]
		sleep(0.5)
		while dom.is_moving:
		if self.check_pause_or_abort():
		log.debug(
		"check_pause_or_abort returned true. I'm here")
		return
		log.warning(f"Waiting for Dome to Stop")
		sleep(0.5)

		##########################
		##### Start exposure #####
		##########################

		try:
		cam.start(exptime,
		frame_number[frametype],
		datetime=gps.isot)
		except Exception as e:
		msg = f"An error occurred : {e}"
		log.error(msg)
		self.error.append(msg)
		return

		# Some debug time counter
		now = Time.now()

		# Checking if the exposure is still running
		status = 1
		while status != 0:
		if self.check_pause_or_abort():
		log.debug("check_pause_or_abort returned true. I'm here")
		return

		try:
		status = int(cam.state)
		except TypeError as e:
		msg = f"Error trying to read camera status: {e}"
		log.error(msg)
		self.error.append(msg)

		if status == 5: # i.e. error in camera
		msg = f"Camera {camera_state[status]}"
		log.error(msg)
		self.error.append(msg)
		return
		else:
		elapsed = (Time.now() - now).sec
		msg = f"Camera {
		camera_state[status]}, {
		elapsed:.1f}/{exptime}s "
		msg += f"of exposure {exposure}/{repeat - 1}"
		log.warning(msg)
		sleep(0.75)

		log.debug("check_pause_or_abort before the download")
		if self.check_pause_or_abort():
		log.debug("check_pause_or_abort returned true. I'm here")
		return

		################################
		##### Download from camera #####
		################################

		log.info(f"Downloading frame as {temp_fits}")

		# Checking if the image is ready to be downloaded
		while cam.ready != 1:
		log.debug(f"Downloading... (cam state {image_state[cam.ready]})")
		sleep(0.5)
		if self.check_pause_or_abort():
		log.warning(f"Buffer still {image_state[cam.ready]}")
		return
		log.debug(f"Downloading... (cam state {image_state[cam.ready]})")

		cam.download()
		log.info(f"Elapsed {(Time.now() - now).sec:.2f}")

		##################################
		##### Update header and save #####
		##################################

		fih_params = {
		"object": objname,
		"camera": camera_name,
		"fits_file": fits_file,
		}

		fih_params.update(basic_cabinet) # radec, altaz, utc, lst

		fih = FillHeader()

		fih.run(fih_params)
		log.debug(f"observation {fih.filename}")

		self.filename = fih.filename
		self.filenames.append(fih.filename)

		log.debug(f"filenames from obs tpl {self.filenames}")

		######################################
		##### Calculate and apply offset #####
		######################################

		log.info(f"Recenter is {self.recenter}")
		if self.recenter:
		log.info(f"... with box={self.box}px")

		if frametype == "Light":
		if self.recenter:
		try:
		apply_offset(
		fits_file=fits_file, box=self.box, display=False)
		except Exception as e:
		log.warning(f"Cannot apply offset: {e}")
		return

noctua/templates/snapshot_spectro.py

0 → 100644

+249 −0

Original line number	Diff line number	Diff line
		#!/usr/bin/env python3
		# -- coding: utf-8 --

		# System modules
		from time import sleep

		# Third-party modules
		from astropy.time import Time

		# Other templates
		from ..config.constants import (camera_state, frame_number, frame_type,
		image_state, temp_fits, viewer_fits_path)
		from .. import devices as _dev
		from ..devices import dom, tel
		from ..utils.coordinates import apply_offset
		from ..utils.logger import log
		from .basetemplate import BaseTemplate
		from .fillheader import Template as FillHeader
		from .lampsoff import Template as LampsOff


		class Template(BaseTemplate):
		def __init__(self):
		super().__init__()
		self.name = "snapshot_spectro"
		self.description = "Observes a field with the spectroscopic camera"

		def content(self, params):

		########################
		##### Params check #####
		########################

		try:
		camera_name = params.get("camera") or "cam2"
		cam = getattr(_dev, camera_name)
		fits_file = viewer_fits_path(getattr(cam, '_viewer_key', None))

		objname = params.get("object") or "test"
		repeat = params.get("repeat") or 1
		frametype = params.get("frametype") or "Light"
		binning = params.get("binning") or cam.binning
		xystart = params.get("xystart") or cam.xystart
		xyend = params.get("xyend") or cam.xyend
		exptime = params["exptime"]
		self.domeslewing = params.get("domeslewing") or False
		self.recenter = params.get("recenter") or False
		self.box = params.get("box") or 300

		except KeyError as e:
		msg = f"Parameter {e} not found"
		log.error(msg)
		self.error.append(msg)
		return

		#########################################
		##### Switch off lamps if not Flat #####
		#########################################

		if frametype != "Flat":
		log.info("Not a 'Flat' frametype: switching off lamps")
		lao = LampsOff()
		lao.run({})

		if frametype != "Light":
		log.info("Not a 'Light' frametype: switching off recentering")
		self.recenter = False

		########################
		##### Camera setup #####
		########################

		# Bringing back the camera to Idle state.
		if cam.state != 0:
		log.warning(
		f"Resetting camera from state {camera_state[cam.state]}.")
		cam.abort()
		sleep(0.1)

		log.info(f"Camera is {camera_state[cam.state]}")

		# Changing binning if not the same as the current one
		if binning != cam.binning:
		log.info(f"Changing binning to {binning}")
		cam.binning = [binning, binning]
		sleep(0.1)

		log.info(f"Camera is {camera_state[cam.state]}")

		# Changing windowing
		if xystart and xyend:
		log.info(
		f"Changing windowing to xystart:{xystart}, xyend:{xyend}")
		cam.set_window(xystart[0],
		xystart[1],
		xyend[0]-xystart[0],
		xyend[1]-xystart[1])

		else:
		msg = f"No windowing: full frame in bininng {binning}"
		log.warning(msg)
		cam.full_frame()

		#########################
		##### Exposure loop #####
		#########################

		log.info(f"Starting {repeat} {frametype} frames of {exptime}s")

		# Looping on the number of exposures
		for exposure in range(repeat):
		if self.check_pause_or_abort():
		return

		log.info(f"Asking cabinet time and coordinates")
		basic_cabinet = tel.coordinates
		gps = Time(tel.coordinates["utc"], format="unix")

		log.info(f"Starting Exposure {exposure + 1} of {repeat}")

		# ########################
		# ##### Dome Slewing #####
		# ########################

		log.info(f"Dome slewing is {self.domeslewing}")
		if frametype == "Light":
		if self.domeslewing:
		log.info(f"Check Dome position")
		while (abs(tel.coordinates["altaz"][1] - dom.azimuth) > 3):
		if not self.domeslewing:
		log.debug("Stop slewing anyway")
		return

		dom.azimuth = tel.coordinates["altaz"][1]
		sleep(0.5)
		while dom.is_moving:
		if self.check_pause_or_abort():
		log.debug(
		"check_pause_or_abort returned true. I'm here")
		return
		log.warning(f"Waiting for Dome to Stop")
		sleep(0.5)

		##########################
		##### Start exposure #####
		##########################

		try:
		cam.start(exptime,
		frame_number[frametype],
		datetime=gps.isot)
		except Exception as e:
		msg = f"An error occurred : {e}"
		log.error(msg)
		self.error.append(msg)
		return

		# Some debug time counter
		now = Time.now()

		# Checking if the exposure is still running
		status = 1
		while status != 0:
		if self.check_pause_or_abort():
		log.debug("check_pause_or_abort returned true. I'm here")
		return

		try:
		status = int(cam.state)
		except TypeError as e:
		msg = f"Error trying to read camera status: {e}"
		log.error(msg)
		self.error.append(msg)

		if status == 5: # i.e. error in camera
		msg = f"Camera {camera_state[status]}"
		log.error(msg)
		self.error.append(msg)
		return
		else:
		elapsed = (Time.now() - now).sec
		msg = f"Camera {
		camera_state[status]}, {
		elapsed:.1f}/{exptime}s "
		msg += f"of exposure {exposure}/{repeat - 1}"
		log.warning(msg)
		sleep(0.75)

		log.debug("check_pause_or_abort before the download")
		if self.check_pause_or_abort():
		log.debug("check_pause_or_abort returned true. I'm here")
		return

		################################
		##### Download from camera #####
		################################

		log.info(f"Downloading frame as {temp_fits}")

		# Checking if the image is ready to be downloaded
		while cam.ready != 1:
		log.debug(f"Downloading... (cam state {image_state[cam.ready]})")
		sleep(0.5)
		if self.check_pause_or_abort():
		log.warning(f"Buffer still {image_state[cam.ready]}")
		return
		log.debug(f"Downloading... (cam state {image_state[cam.ready]})")

		cam.download()
		log.info(f"Elapsed {(Time.now() - now).sec:.2f}")

		##################################
		##### Update header and save #####
		##################################

		fih_params = {
		"object": objname,
		"camera": camera_name,
		"fits_file": fits_file,
		}

		fih_params.update(basic_cabinet) # radec, altaz, utc, lst

		fih = FillHeader()

		fih.run(fih_params)
		log.debug(f"observation {fih.filename}")

		self.filename = fih.filename
		self.filenames.append(fih.filename)

		log.debug(f"filenames from obs tpl {self.filenames}")

		######################################
		##### Calculate and apply offset #####
		######################################

		log.info(f"Recenter is {self.recenter}")
		if self.recenter:
		log.info(f"... with box={self.box}px")

		if frametype == "Light":
		if self.recenter:
		try:
		apply_offset(
		fits_file=fits_file, box=self.box, display=False)
		except Exception as e:
		log.warning(f"Cannot apply offset: {e}")
		return

noctua/web/pages/control.html

+5 −5

Original line number	Diff line number	Diff line
		@@ -43,8 +43,8 @@

		<div class="card bg-black border-secondary p-2 small font-monospace text-success">
		<div>Mode: <span id="current-mode-label" class="text-info fw-bold">Imaging</span></div>
		<div>Camera State: <var data-status="camera-snapshot-state">N/A</var></div>
		<div>Cooler: <var data-status="camera-cooler">N/A</var> (<var data-status="camera-settings-temperature">N/A</var>°C)</div>
		<div>Camera State: <var data-status="camera-snapshot-state" data-status-suffix="-snapshot-state">N/A</var></div>
		<div>Cooler: <var data-status="camera-cooler" data-status-suffix="-cooler">N/A</var> (<var data-status="camera-settings-temperature" data-status-suffix="-settings-temperature">N/A</var>°C)</div>
		</div>
		</section>

		@@ -67,7 +67,7 @@
		{{ ctrl.fits_viewer_section('/api') }}
		</div>
		</div>
		<div class="tab-pane fade" id="mon-webcam">
		<div class="tab-pane fade show active" id="mon-webcam">
		{{ webcam.webcam_panel() }}
		</div>
		<div class="tab-pane fade" id="mon-synoptic">

noctua/web/pages/init.html

+0 −1

Original line number	Diff line number	Diff line
		@@ -220,7 +220,6 @@

		</section>

		{# ── RIGHT: Webcam + Synoptic ── #}
		<section class="col-xl-4 col-lg-12">
		<div class="card bg-dark border-secondary shadow-sm">
		<div class="card-header p-0 border-secondary">