from __future__ import annotations
import asyncio
import logging
from abc import ABCMeta, abstractmethod
from datetime import UTC, datetime
from typing import Any, NamedTuple
from astropy.io import fits
from pyobs.events import ExposureStatusChangedEvent, NewSpectrumEvent
from pyobs.interfaces import ExposureState, IExposure, IResettable, ISpectrograph, default_reset
from pyobs.mixins.datasequence import DataSequenceMixin
from pyobs.mixins.fitsheader import SpectrumFitsHeaderMixin
from pyobs.modules import Module, timeout
from pyobs.utils import exceptions as exc
from pyobs.utils.enums import ExposureStatus
log = logging.getLogger(__name__)
class ExposureInfo(NamedTuple):
"""Info about a running exposure."""
start: datetime
class BaseSpectrograph(
Module, SpectrumFitsHeaderMixin, ISpectrograph, IExposure, IResettable, DataSequenceMixin, metaclass=ABCMeta
):
"""Base class for all spectrograph modules."""
__module__ = "pyobs.modules.camera"
def __init__(
self,
fits_headers: dict[str, Any] | None = None,
filenames: str = "/cache/pyobs-{DAY-OBS|date:}-{FRAMENUM|string:04d}.fits.gz",
fits_namespaces: list[str] | None = None,
**kwargs: Any,
):
"""Creates a new BaseCamera.
Args:
fits_headers: Additional FITS headers.
flip: Whether or not to flip the image along its first axis.
filenames: Template for file naming.
fits_namespaces: List of namespaces for FITS headers that this camera should request
"""
super().__init__(fits_namespaces=fits_namespaces, fits_headers=fits_headers, filenames=filenames, **kwargs)
# init camera
self._exposure: ExposureInfo | None = None
self._spectrograph_status = ExposureStatus.IDLE
# multi-threading
self._expose_lock = asyncio.Lock()
self.expose_abort = asyncio.Event()
# check
if self._comm is None:
log.warning("No comm module given, will not be able to signal new images!")
[docs]
async def open(self) -> None:
"""Open module."""
await Module.open(self)
# publish initial state
await self.comm.set_state(
IExposure, ExposureState(status=self._spectrograph_status, progress=0.0, exposure_time_left=0.0)
)
await self._datasequence_open()
@abstractmethod
async def _expose(self, abort_event: asyncio.Event) -> fits.HDUList:
"""Actually do the exposure, should be implemented by derived classes.
Args:
abort_event: Event that gets triggered when exposure should be aborted.
Returns:
The actual image and, if present, a filename.
Raises:
AbortedError: If the exposure was cancelled via abort_event.
ValueError: If exposure was not successful.
"""
...
async def __expose(self, broadcast: bool) -> tuple[fits.HDUList | None, str | None]:
"""Wrapper for a single exposure.
Args:
broadcast: Whether or not the new image should be broadcasted.
Returns:
Tuple of the image itself and its filename.
Raises:
ValueError: If exposure was not successful.
"""
# request fits headers
header_futures_before = await self.request_fits_headers(before=True)
# do the exposure
self._exposure = ExposureInfo(start=datetime.now(UTC))
try:
hdulist = await self._expose(abort_event=self.expose_abort)
if hdulist is None or hdulist[0].data is None:
self._exposure = None
return None, None
except:
# exposure was not successful (aborted?), so reset everything
self._exposure = None
raise
# request fits headers again
header_futures_after = await self.request_fits_headers(before=False)
# add HDU name
hdulist[0].header["EXTNAME"] = "SCI"
# add fits headers and format filename
await self.add_requested_fits_headers(hdulist[0], header_futures_before)
await self.add_requested_fits_headers(hdulist[0], header_futures_after)
await self.add_fits_headers(hdulist[0])
# format filename
filename = self.format_filename(hdulist[0])
# don't want to save?
if filename is None:
return hdulist, None
# store spectrum
await self.store_spectrum(hdulist, filename, broadcast)
# return spectrum and unique
self._exposure = None
log.info("Finished spectrum %s.", filename)
return hdulist, filename
[docs]
async def store_spectrum(self, hdulist: fits.HDUList, filename: str, broadcast: bool) -> None:
"""Store spectrum at given destination.
Can be overwritten by derived classes to custom store a file. In those cases, this version should be called!
Args:
hdulist: HDU list with spectrum.
filename: Name to store file with.
broadcast: Whether to broadcast new spectrum.
"""
# upload file
try:
log.info("Uploading spectrum to file server...")
await self.vfs.write_fits(filename, hdulist)
except FileNotFoundError:
raise ValueError("Could not upload spectrum.")
# broadcast image path
if broadcast and self._comm:
log.info("Broadcasting spectrum ID...")
await self.comm.send_event(NewSpectrumEvent(filename))
[docs]
@timeout(10)
async def grab_data(self, broadcast: bool = True, **kwargs: Any) -> str:
"""Grabs a spectrum and returns reference.
Args:
broadcast: Broadcast existence of image.
Returns:
Name of image that was taken.
Raises:
DeviceBusyError: If the spectrograph is already busy (exposing).
GrabImageError: If there was a problem grabbing the spectrum.
"""
# are we exposing?
if self._spectrograph_status != ExposureStatus.IDLE:
raise exc.DeviceBusyError("Cannot start new exposure because spectrograph is not idle.")
await self._change_exposure_status(ExposureStatus.EXPOSING)
# expose
hdu, filename = await self.__expose(broadcast)
if hdu is None:
raise exc.GrabImageError("Could not take spectrum.")
else:
if filename is None:
raise ValueError("Spectrum has not been saved, so cannot be retrieved by filename.")
# return filename
await self._change_exposure_status(ExposureStatus.IDLE)
return filename
async def _change_exposure_status(self, status: ExposureStatus) -> None:
"""Change exposure status and send event,
Args:
status: New exposure status.
"""
# send event, if it changed
if self._spectrograph_status != status:
await self.comm.send_event(ExposureStatusChangedEvent(last=self._spectrograph_status, current=status))
# set it
self._spectrograph_status = status
await self.comm.set_state(IExposure, ExposureState(status=status, progress=0.0, exposure_time_left=0.0))
def _sequence_busy(self) -> bool:
"""Whether the spectrograph is busy exposing outside of a running sequence."""
return self._spectrograph_status != ExposureStatus.IDLE
[docs]
@default_reset
async def reset(self, **kwargs: Any) -> None:
"""Reset the spectrograph. The base class has no settings of its own, this is a hook
for drivers.
Raises:
DeviceBusyError: If the spectrograph is currently exposing or running a sequence.
"""
if self._spectrograph_status != ExposureStatus.IDLE or self._sequence_count_left > 0:
raise exc.DeviceBusyError("Cannot reset spectrograph while it is busy.")
[docs]
@default_reset
async def full_reset(self, **kwargs: Any) -> None:
"""Reset the spectrograph completely to its configured defaults, hardware included.
Raises:
DeviceBusyError: If the spectrograph is currently exposing or running a sequence.
"""
await self.reset(**kwargs)
[docs]
async def abort(self, **kwargs: Any) -> None:
"""Aborts the current exposure and sequence. Derived class must implement the actual
hardware abort for this!
Raises:
ValueError: If exposure could not be aborted.
"""
self._abort_data_sequence()
__all__ = ["BaseSpectrograph"]