Source code for pyobs.mixins.fitsheader
from __future__ import annotations
import asyncio
import logging
import math
import os
from asyncio import Task
from collections.abc import Iterable
from typing import Any, cast
from astropy.io import fits
from pyobs.images import Image
from pyobs.interfaces import FitsHeaderEntry, IFilters, IFitsHeaderAfter, IFitsHeaderBefore, IFocuser
from pyobs.modules import Module
from pyobs.utils import exceptions as exc
from pyobs.utils.fits import format_filename
from pyobs.utils.time import Time
from pyobs.utils.versions import version_fits_headers
log = logging.getLogger(__name__)
async def request_fits_headers(
module: Module, namespaces: list[str] | None = None, before: bool = True, exclude: Iterable[str] = ()
) -> dict[str, Task[Any]]:
"""Request FITS headers from all other modules that provide them.
Args:
module: Module to send the requests from.
namespaces: Namespaces to request, None for all.
before: Request IFitsHeaderBefore headers if True, IFitsHeaderAfter otherwise.
exclude: Names of clients not to ask.
Returns:
Futures from all modules.
"""
# init
futures: dict[str, Task[Any]] = {}
# we can only do this with a comm module
if module._comm:
# get clients that provide fits headers
clients = await module._comm.clients_with_interface(IFitsHeaderBefore if before else IFitsHeaderAfter)
# create and run a threads in which the fits headers are fetched
for client in clients:
if client in exclude:
continue
log.debug("Requesting FITS headers from %s...", client)
if before:
async with module.proxy(client, IFitsHeaderBefore) as proxy:
futures[client] = asyncio.create_task(proxy.get_fits_header_before(namespaces))
else:
async with module.proxy(client, IFitsHeaderAfter) as proxy:
futures[client] = asyncio.create_task(proxy.get_fits_header_after(namespaces))
# finished
return futures
async def add_requested_fits_headers(
image: Image | fits.PrimaryHDU, futures: dict[str, Task[Any]], timeout: float
) -> None:
"""Add requested FITS headers to header of given image.
Args:
image: Image with header to add to.
futures: Futures to get headers from, see request_fits_headers().
timeout: Maximum seconds to wait for all of them together.
"""
# Bound the whole collection to a single deadline: a peer that never answers its
# IQ (e.g. a laptop put to sleep without closing its client) would otherwise stall
# the frame for the full XMPP IQ timeout (~120s) per dead peer. Give up on anything
# still pending and proceed without its headers.
pending: set[Task[Any]] = set()
if futures:
_, pending = await asyncio.wait(futures.values(), timeout=timeout)
for task in pending:
task.cancel()
if pending:
await asyncio.gather(*pending, return_exceptions=True)
# get fits headers from other clients
for client, future in futures.items():
# join thread
log.info("Fetching FITS headers from %s...", client)
if future in pending:
log.warning("Could not fetch FITS headers from %s: timed out.", client)
continue
try:
headers = future.result()
except exc.RemoteError as e:
log.warning("Could not fetch FITS headers from %s: %s.", client, str(e))
continue
except Exception as e:
# XmppComm.execute() only converts IqError/IqTimeout into RemoteError; anything
# else raised while making or deserializing the RPC call (e.g. a peer returning a
# malformed IFitsHeaderBefore/After response) escapes as a raw exception. That must
# not take down the whole exposure -- log and skip this peer's headers (#767).
log.warning("Could not fetch FITS headers from %s: unexpected %s: %s.", client, type(e).__name__, str(e))
continue
# add them to fits file
if headers:
log.debug("Adding additional FITS headers from %s...", client)
for key, entry in headers.items():
image.header[key] = (entry.value, entry.comment)
class FitsHeaderMixin:
"""Helper methods for all modules that implement IImageGrabber."""
__module__ = "pyobs.mixins"
def __init__(
self,
fits_namespaces: list[str] | None = None,
fits_headers: dict[str, Any] | None = None,
filenames: str = "/cache/pyobs-{DAY-OBS|date:}-{FRAMENUM|string:04d}.fits",
frame_number: bool = True,
night_obs: bool = True,
fits_header_timeout: float = 15.0,
**kwargs: Any,
):
"""Initialise the mixin.
Args:
fits_namespaces: List of namespaces for FITS headers that this camera should request.
fits_headers: Additional FITS headers.
filename: Filename pattern for FITS images.
frame_number: Whether to add frame number to FITS file header.
night_obs: If True, DAY-OBS will contain the night of observation, not the calendar day.
fits_header_timeout: Maximum seconds to wait for a peer's FITS headers before skipping
them. A peer that never answers (e.g. a laptop put to sleep without closing its
client) would otherwise stall the frame for the full XMPP IQ timeout.
"""
# store
self._fitsheadermixin_fits_namespaces = fits_namespaces
self._fitsheadermixin_filename_pattern = filenames
self._fitsheadermixin_fits_headers = fits_headers if fits_headers is not None else {}
self._fitsheadermixin_header_timeout = fits_header_timeout
if "OBSERVER" not in self._fitsheadermixin_fits_headers:
self._fitsheadermixin_fits_headers["OBSERVER"] = ["pyobs", "Name of observer"]
self._fitsheadermixin_night_obs = night_obs
# night exposure number
self._fitsheadermixin_enable_frame_number = frame_number
self._fitsheadermixin_frame_number = 0
super().__init__(**kwargs)
@property
def _fitsheadermixin_cache(self) -> str:
# computed lazily (not cached at __init__ time) since it depends on module.name, which in
# turn depends on comm having been set up by Object.__init__ -- in a cooperative super()
# chain, this mixin's __init__ can run before Module has finished its own setup
module = cast(Module, self)
return f"/pyobs/modules/{module.name}/cache.yaml"
[docs]
async def request_fits_headers(self, before: bool = True) -> dict[str, Task[Any]]:
"""Request FITS headers from other modules.
Returns:
Futures from all modules.
"""
module = cast(Module, cast(object, self))
return await request_fits_headers(module, self._fitsheadermixin_fits_namespaces, before=before)
[docs]
async def add_requested_fits_headers(self, image: Image | fits.PrimaryHDU, futures: dict[str, Task[Any]]) -> None:
"""Add requested FITS headers to header of given image.
Args:
image: Image with header to add to.
futures: Futures to get headers from.
"""
await add_requested_fits_headers(image, futures, self._fitsheadermixin_header_timeout)
[docs]
def add_local_fits_headers(self, image: Image | fits.PrimaryHDU) -> None:
"""Add the cheap, local FITS headers to the given image (no I/O, no comm).
This is the always-on half of add_fits_headers(): the configured static
fits_headers dict plus the computed fields from _fitsheadermixin_add_fits_headers().
The frame-sequence-number step is deliberately kept out (it does a VFS read+write
per call), so video-rate callers can reuse this without hitting the VFS on every frame.
Args:
image: Image with header to add to.
"""
# add HDU name
image.header["EXTNAME"] = "SCI"
# add static fits headers
for key, value in self._fitsheadermixin_fits_headers.items():
image.header[key] = tuple(value)
# add more fits headers
self._fitsheadermixin_add_fits_headers(image)
[docs]
async def add_fits_headers(self, image: Image | fits.PrimaryHDU) -> None:
"""Add requested FITS headers to header of given image.
Args:
image: Image with header to add to.
"""
# cheap, local half
self.add_local_fits_headers(image)
# persistent nightly frame-sequence number (VFS I/O -- not per-frame safe)
if self._fitsheadermixin_enable_frame_number:
await self._fitsheadermixin_add_framenum(image)
[docs]
def _fitsheadermixin_add_fits_headers(self, image: Image | fits.PrimaryHDU) -> None:
"""Add FITS header keywords to the given FITS header.
Args:
image: Image with header to add to.
"""
module = cast(Module, self)
# get header
hdr = image.header
# we definitely need a DATE-OBS and IMAGETYP!!
if "DATE-OBS" not in hdr:
log.warning("No DATE-OBS found in FITS header, adding NO further information!")
return
# get date obs
date_obs = Time(hdr["DATE-OBS"])
# set MJD-OBS
hdr["MJD-OBS"] = date_obs.mjd
# UT1-UTC
hdr["UT1_UTC"] = (float(date_obs.delta_ut1_utc), "UT1-UTC")
# basic stuff
hdr["EQUINOX"] = (2000.0, "Equinox of celestial coordinate system")
# do we have an observer (and thus a location)?
if module._observer is not None:
loc = module._observer.location
# add location of telescope
hdr["LONGITUD"] = (float(loc.lon.degree), "Longitude of the telescope [deg E]")
hdr["LATITUDE"] = (float(loc.lat.degree), "Latitude of the telescope [deg N]")
hdr["HEIGHT"] = (float(loc.height.value), "Altitude of the telescope [m]")
# add local sidereal time
lst = module._observer.local_sidereal_time(date_obs)
lst_hours = float(lst.hour)
h = int(lst_hours)
m = int((lst_hours - h) * 60)
s = (lst_hours - h - m / 60) * 3600
hdr["LST"] = (f"{h:02d}:{m:02d}:{s:05.2f}", "Local sidereal time")
# date of night this observation is in
if self._fitsheadermixin_night_obs and module._observer is not None:
hdr["DAY-OBS"] = (date_obs.night_obs(module._observer).strftime("%Y-%m-%d"), "Night of observation")
else:
hdr["DAY-OBS"] = (date_obs.strftime("%Y-%m-%d"), "Day of observation")
# loaded pyobs package versions
for key, entry in version_fits_headers(module.name).items():
hdr[key] = (entry.value, entry.comment)
[docs]
def format_filename(self, image: Image | fits.PrimaryHDU) -> str | None:
"""Format filename according to given pattern and store in header of image.
Args:
image: Image with header to add to.
"""
# no pattern?
if self._fitsheadermixin_filename_pattern is None:
return None
# create a temporary filename
filename = format_filename(image.header, self._fitsheadermixin_filename_pattern)
if filename is None:
return None
# set it
image.header["ORIGNAME"] = (os.path.basename(filename), "The original file name")
image.header["FNAME"] = (os.path.basename(filename), "FITS file file name")
return filename
class ImageFitsHeaderMixin(FitsHeaderMixin):
"""Helper methods for all modules that need FITS headers for an image."""
__module__ = "pyobs.mixins"
def __init__(self, centre: tuple[float, float] | None = None, rotation: float | None = None, **kwargs: Any):
"""Initialise the mixin.
Args:
fits_namespaces: List of namespaces for FITS headers that this camera should request.
fits_headers: Additional FITS headers.
centre: (x, y) tuple of camera centre.
rotation: Rotation east of north.
filename: Filename pattern for FITS images.
"""
super().__init__(**kwargs)
# store
self._fitsheadermixin_centre = centre
self._fitsheadermixin_rotation = rotation
# only warn once per missing-header case, since these repeat on every single frame
self._fitsheadermixin_warned_cdelt = False
self._fitsheadermixin_warned_crpix = False
self._fitsheadermixin_warned_cd_matrix = False
self._fitsheadermixin_warned_centre = False
@property
def rotation(self) -> float | None:
return self._fitsheadermixin_rotation
@property
def centre(self) -> tuple[float, float] | None:
return self._fitsheadermixin_centre
[docs]
def _fitsheadermixin_add_fits_headers(self, image: Image | fits.PrimaryHDU) -> None:
"""Add FITS header keywords to the given FITS header.
Args:
image: Image with header to add to.
"""
# parent method
FitsHeaderMixin._fitsheadermixin_add_fits_headers(self, image)
# get header
hdr = image.header
# convenience function to return value of keyword
def v(k: str) -> Any:
return hdr[k][0] if isinstance(k, list) or isinstance(k, tuple) else hdr[k]
# set CRVAL1/2 from RA/Dec
if "TEL-RA" in hdr and "TEL-DEC" in hdr:
hdr["CRVAL1"] = hdr["TEL-RA"]
hdr["CRVAL2"] = hdr["TEL-DEC"]
# pixel size in world coordinates
if "DET-PIXL" in hdr and "TEL-FOCL" in hdr and "DET-BIN1" in hdr and "DET-BIN2" in hdr:
# calculate plate scale for 1x1 binning
tmp = 360.0 / (2.0 * math.pi) * v("DET-PIXL") / v("TEL-FOCL")
# any focal reduction factor?
if "FOCL-RED" in hdr:
tmp /= hdr["FOCL-RED"]
# set WCS
hdr["CDELT1"] = (-tmp * v("DET-BIN1"), "Coordinate increment on x-axis [deg/px]")
hdr["CDELT2"] = (+tmp * v("DET-BIN2"), "Coordinate increment on y-axis [deg/px]")
hdr["CUNIT1"] = ("deg", "Units of CRVAL1, CDELT1")
hdr["CUNIT2"] = ("deg", "Units of CRVAL2, CDELT2")
hdr["WCSAXES"] = (2, "Number of WCS axes")
elif not self._fitsheadermixin_warned_cdelt:
log.warning("Could not calculate CDELT1/CDELT2 (DET-PIXL/TEL-FOCL/DET-BIN1/DET-BIN2 missing).")
self._fitsheadermixin_warned_cdelt = True
# centre pixel
if self._fitsheadermixin_centre is not None:
hdr["DET-CPX1"] = (self._fitsheadermixin_centre[0], "x-pixel on mechanical axis in unbinned image")
hdr["DET-CPX2"] = (self._fitsheadermixin_centre[1], "y-pixel on mechanical axis in unbinned image")
elif not self._fitsheadermixin_warned_centre:
log.warning("Could not calculate DET-CPX1/DET-CPX2 (centre not given in config).")
self._fitsheadermixin_warned_centre = True
# reference pixel in binned image
if "DET-CPX1" in hdr and "DET-BIN1" in hdr and "DET-CPX2" in hdr and "DET-BIN2" in hdr:
# offset?
off_x, off_y = 0.0, 0.0
if "XORGSUBF" in hdr and "YORGSUBF" in hdr:
off_x = v("XORGSUBF") if "XORGSUBF" in hdr else 0.0
off_y = v("YORGSUBF") if "YORGSUBF" in hdr else 0.0
hdr["CRPIX1"] = ((v("DET-CPX1") - off_x) / v("DET-BIN1"), "Reference x-pixel position in binned image")
hdr["CRPIX2"] = ((v("DET-CPX2") - off_y) / v("DET-BIN2"), "Reference y-pixel position in binned image")
elif not self._fitsheadermixin_warned_crpix:
log.warning(
"Could not calculate CRPIX1/CRPIX2 " "(XORGSUBF/YORGSUBF/DET-CPX1/TEL-CPX2/DET-BIN1/DET-BIN2) missing."
)
self._fitsheadermixin_warned_crpix = True
# only add all this stuff for OBJECT images
if "IMAGETYP" not in hdr or hdr["IMAGETYP"] not in ["dark", "bias"]:
# projection, only override if not already set
if "CTYPE1" not in hdr:
hdr["CTYPE1"] = ("RA---TAN", "RA in tangent plane projection")
if "CTYPE2" not in hdr:
hdr["CTYPE2"] = ("DEC--TAN", "Dec in tangent plane projection")
# PC matrix: rotation only, shift comes from CDELT1/2
if self._fitsheadermixin_rotation is not None:
# position angle is set rotation + DEROTOFF if defined
posang = self._fitsheadermixin_rotation
if "DEROTOFF" in hdr:
posang += hdr["DEROTOFF"]
# write position angle
hdr["POSANG"] = (posang, "Position angle [deg e of n]")
theta_rad = math.radians(posang)
cos_theta = math.cos(theta_rad)
sin_theta = math.sin(theta_rad)
hdr["PC1_1"] = (+cos_theta, "Partial of first axis coordinate w.r.t. x")
hdr["PC1_2"] = (-sin_theta, "Partial of first axis coordinate w.r.t. y")
hdr["PC2_1"] = (+sin_theta, "Partial of second axis coordinate w.r.t. x")
hdr["PC2_2"] = (+cos_theta, "Partial of second axis coordinate w.r.t. y")
elif not self._fitsheadermixin_warned_cd_matrix:
log.warning("Could not calculate CD matrix (rotation or CDELT1/CDELT2 missing.")
self._fitsheadermixin_warned_cd_matrix = True
class SpectrumFitsHeaderMixin(FitsHeaderMixin):
"""Helper methods for all modules that need FITS headers for an image."""
__module__ = "pyobs.mixins"
pass
async def _chain_get_fits_header_before(
parent: Any, namespaces: list[str] | None, **kwargs: Any
) -> dict[str, FitsHeaderEntry]:
"""Cooperatively call get_fits_header_before() on the next class in the MRO, if any.
Used by FilterHeaderMixin/FocuserHeaderMixin so they compose with each other and with a real
implementation elsewhere in the MRO. getattr() alone isn't enough to detect "nothing real
here" -- IFitsHeaderBefore.get_fits_header_before is an abstractmethod, so it's still a
perfectly real, callable attribute (that's how ABCMeta lets a mixin earlier in the MRO satisfy
the abstract requirement in the first place); calling it directly just runs its `...` body and
returns None, not a dict. Skip it via __isabstractmethod__ rather than trusting the return
value, so a well-behaved implementation that legitimately returns {} is still handled the same
way as one that returns None by mistake.
"""
base = getattr(parent, "get_fits_header_before", None)
if base is None or getattr(base, "__isabstractmethod__", False):
return {}
result = await base(namespaces, **kwargs)
return dict(result) if result else {}
class FilterHeaderMixin:
"""For IFilters modules: contributes FILTER from the module's own last-published state, so
drivers don't have to hand-roll get_fits_header_before just for this. Cooperative -- chains to
a get_fits_header_before elsewhere in the MRO if one exists (e.g. a FocuserHeaderMixin on a
combo focuser/filter device), otherwise starts from an empty header."""
__module__ = "pyobs.mixins"
async def get_fits_header_before(
self, namespaces: list[str] | None = None, **kwargs: Any
) -> dict[str, FitsHeaderEntry]:
hdr: dict[str, FitsHeaderEntry] = await _chain_get_fits_header_before(super(), namespaces, **kwargs)
module = cast(Module, self)
state = module.comm.get_own_state(IFilters) if module._comm is not None else None
if state is not None:
hdr["FILTER"] = FitsHeaderEntry(state.filter, "Current filter")
return hdr
class FocuserHeaderMixin:
"""For IFocuser modules: contributes FOCUS/FOCOFF from the module's own last-published state,
so drivers don't have to hand-roll get_fits_header_before just for this. Cooperative -- chains
to a get_fits_header_before elsewhere in the MRO if one exists (e.g. a FilterHeaderMixin on a
combo focuser/filter device), otherwise starts from an empty header."""
__module__ = "pyobs.mixins"
async def get_fits_header_before(
self, namespaces: list[str] | None = None, **kwargs: Any
) -> dict[str, FitsHeaderEntry]:
hdr: dict[str, FitsHeaderEntry] = await _chain_get_fits_header_before(super(), namespaces, **kwargs)
module = cast(Module, self)
state = module.comm.get_own_state(IFocuser) if module._comm is not None else None
if state is not None:
hdr["FOCUS"] = FitsHeaderEntry(state.focus, "Focus value [mm]")
hdr["FOCOFF"] = FitsHeaderEntry(state.focus_offset, "Focus offset [mm]")
return hdr
__all__ = [
"FitsHeaderMixin",
"ImageFitsHeaderMixin",
"SpectrumFitsHeaderMixin",
"FilterHeaderMixin",
"FocuserHeaderMixin",
"request_fits_headers",
"add_requested_fits_headers",
]