from __future__ import annotations
import asyncio
import datetime
import logging
from abc import ABCMeta, abstractmethod
from typing import Any, NamedTuple
import numpy as np
from astropy.io import fits
from astropy.table import Table
from astropy.time import TimeDelta
from numpy.typing import NDArray
from pyobs.events import BadWeatherEvent, Event, ExposureStatusChangedEvent, NewImageEvent
from pyobs.images import Image
from pyobs.interfaces import (
Binning,
DataStackState,
ExposureState,
ExposureTimeState,
IBinning,
ICamera,
IDataStack,
IExposure,
IExposureTime,
IImageType,
ImageTypeState,
IResettable,
IWindow,
default_reset,
)
from pyobs.mixins.datapipeline import DataPipelineMixin
from pyobs.mixins.datasequence import DataSequenceMixin
from pyobs.mixins.fitsheader import ImageFitsHeaderMixin
from pyobs.modules import Module, timeout
from pyobs.utils import exceptions as exc
from pyobs.utils.enums import ExposureStatus, ImageType
from pyobs.utils.time import Time
log = logging.getLogger(__name__)
class ExposureInfo(NamedTuple):
"""Info about a running exposure."""
start: datetime.datetime
exposure_time: float
async def calc_expose_timeout(camera: BaseCamera, *args: Any, **kwargs: Any) -> float:
"""Calculates timeout for expose()."""
return camera._exposure_time + 30
async def calc_stack_timeout(camera: BaseCamera, count: int = 1, *args: Any, **kwargs: Any) -> float:
"""Calculates timeout for grab_stack()."""
return count * (camera._exposure_time + camera._stack_frame_overhead) + camera._stack_timeout_margin
class BaseCamera(
Module,
ImageFitsHeaderMixin,
ICamera,
IExposure,
IExposureTime,
IImageType,
IResettable,
DataSequenceMixin,
DataPipelineMixin,
IDataStack,
metaclass=ABCMeta,
):
"""Base class for all camera modules."""
__module__ = "pyobs.modules.camera"
def __init__(
self,
fits_headers: dict[str, Any] | None = None,
centre: tuple[float, float] | None = None,
rotation: float = 0.0,
flip: bool | None = None,
flip_x: bool = False,
flip_y: bool = False,
filenames: str = "/cache/pyobs-{DAY-OBS|date:}-{FRAMENUM|string:04d}-{IMAGETYP|type}00.fits.gz",
fits_namespaces: list[str] | None = None,
meridian_flip_on: str | None = None,
fits_header_timeout: float = 15.0,
max_stack_bytes: int = 2 * 1024**3,
stack_frame_overhead: float = 10.0,
stack_timeout_margin: float = 60.0,
**kwargs: Any,
):
"""Creates a new BaseCamera.
Args:
fits_headers: Additional FITS headers.
centre: (x, y) tuple of camera centre.
rotation: Rotation east of north.
flip: Whether or not to flip the image along its first axis (deprecated, use flip_x).
flip_x: Whether or not to flip the image along its first axis.
flip_y: Whether or not to flip the image along its second axis.
filenames: Template for file naming.
fits_namespaces: List of namespaces for FITS headers that this camera should request
fits_header_timeout: Maximum seconds to wait for a peer's FITS headers before skipping them.
max_stack_bytes: Memory cap for a grab_stack() cube (count * frame size), in bytes.
stack_frame_overhead: Estimated per-frame readout/transfer time in seconds, added to
the exposure time when computing grab_stack()'s timeout.
stack_timeout_margin: Extra seconds added to grab_stack()'s timeout, covering header
requests, cube assembly and the data pipeline.
"""
super().__init__(
fits_namespaces=fits_namespaces,
fits_headers=fits_headers,
centre=centre,
rotation=rotation,
filenames=filenames,
fits_header_timeout=fits_header_timeout,
**kwargs,
)
# check
if self._comm is None:
log.warning("No comm module given, will not be able to signal new images!")
# store
self._flip_x = flip if flip is not None else flip_x
self._flip_y = flip_y
self._exposure_time: float = 0.0
self._image_type = ImageType.OBJECT
self._meridian_flip_on = meridian_flip_on
# init camera
self._exposure: ExposureInfo | None = None
self._camera_status = ExposureStatus.IDLE
# stacks
self._max_stack_bytes = max_stack_bytes
self._stack_frame_overhead = stack_frame_overhead
self._stack_timeout_margin = stack_timeout_margin
self._last_frame_nbytes: int | None = None
self._stack_running = False
self._stack_abort = False
# multi-threading
self.expose_abort = asyncio.Event()
# register exception
self._register_exception(exc.GrabImageError, 3, timespan=600, callback=self._default_remote_error_callback)
# re-publish exposure progress while exposing, so clients see a live progress bar and
# remaining time instead of the frozen values from the EXPOSING status transition
self.add_background_task(self._publish_exposure_progress)
[docs]
async def open(self) -> None:
"""Open module."""
await Module.open(self)
# subscribe to events
if self._comm:
await self.comm.register_event(NewImageEvent)
await self.comm.register_event(ExposureStatusChangedEvent)
await self.comm.register_event(BadWeatherEvent, self._abort_weather)
# publish initial states
await self.comm.set_state(IExposureTime, ExposureTimeState(exposure_time=self._exposure_time))
await self.comm.set_state(IImageType, ImageTypeState(image_type=self._image_type))
await self.comm.set_state(
IExposure,
ExposureState(
status=self._camera_status,
progress=0.0,
exposure_time_left=0.0,
),
)
await self._datasequence_open()
await self._datapipeline_open()
await self.comm.set_state(IDataStack, DataStackState(count_total=0, count_left=0))
[docs]
async def set_exposure_time(self, exposure_time: float, **kwargs: Any) -> None:
"""Set the exposure time in seconds.
Args:
exposure_time: Exposure time in seconds.
Raises:
ValueError: If exposure time could not be set.
"""
log.info("Setting exposure time to %.5fs...", exposure_time)
self._exposure_time = exposure_time
await self.comm.set_state(IExposureTime, ExposureTimeState(exposure_time=exposure_time))
[docs]
async def set_image_type(self, image_type: ImageType, **kwargs: Any) -> None:
"""Set the image type.
Args:
image_type: New image type.
"""
log.info("Setting image type to %s...", image_type)
self._image_type = image_type
await self.comm.set_state(IImageType, ImageTypeState(image_type=image_type))
[docs]
@default_reset
async def reset(self, **kwargs: Any) -> None:
"""Reset exposure time, image type, binning and window to their defaults.
Raises:
DeviceBusyError: If the camera is currently exposing, or running a sequence or stack.
"""
if self._camera_status != ExposureStatus.IDLE or self._sequence_count_left > 0 or self._stack_running:
raise exc.DeviceBusyError("Cannot reset camera while it is busy.")
await self.set_exposure_time(0.0)
await self.set_image_type(ImageType.OBJECT)
if isinstance(self, IBinning):
binning_caps = self.comm.get_own_capabilities(IBinning)
if binning_caps is None or not binning_caps.binnings or Binning(1, 1) in binning_caps.binnings:
await self.set_binning(1, 1)
else:
log.warning("Cannot reset binning to 1x1, not in the camera's advertised binnings.")
if isinstance(self, IWindow):
window_caps = self.comm.get_own_capabilities(IWindow)
if window_caps is not None and window_caps.full_frame_width > 0 and window_caps.full_frame_height > 0:
await self.set_window(
window_caps.full_frame_x,
window_caps.full_frame_y,
window_caps.full_frame_width,
window_caps.full_frame_height,
)
else:
log.warning("Cannot reset window to full frame, no full-frame capabilities advertised.")
await self.set_pipeline(self._default_pipeline)
[docs]
@default_reset
async def full_reset(self, **kwargs: Any) -> None:
"""Reset the camera completely to its configured defaults, hardware included.
Raises:
DeviceBusyError: If the camera is currently exposing, or running a sequence or stack.
"""
await self.reset(**kwargs)
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._camera_status != status:
await self.comm.send_event(ExposureStatusChangedEvent(last=self._camera_status, current=status))
# set it
self._camera_status = status
await self.comm.set_state(
IExposure,
ExposureState(
status=status,
progress=await self._get_exposure_progress(),
exposure_time_left=await self._get_exposure_time_left(),
),
)
async def _get_exposure_time_left(self) -> float:
"""Returns the remaining exposure time on the current exposure in seconds."""
if self._exposure is None:
return 0.0
duration = datetime.timedelta(seconds=self._exposure.exposure_time)
diff = self._exposure.start + duration - datetime.datetime.now(datetime.UTC)
return max(diff.total_seconds(), 0.0)
async def _get_exposure_progress(self) -> float:
"""Returns the progress of the current exposure in percent.
Returns:
Progress of the current exposure in percent.
"""
# if we're not exposing, there is no progress
if self._exposure is None:
return 0.0
# calculate difference between start of exposure and now
diff = datetime.datetime.now(datetime.UTC) - self._exposure[0]
# zero exposure time?
if self._exposure.exposure_time == 0.0 or self._camera_status == ExposureStatus.READOUT:
return 100.0
else:
# return max of 100
percentage = diff.total_seconds() / self._exposure[1] * 100.0
return min(percentage, 100.0)
async def _publish_exposure_progress(self) -> None:
"""Background task: re-publish IExposure state while exposing.
_change_exposure_status() only publishes IExposure on status transitions, and at the
EXPOSING transition self._exposure isn't set yet, so progress and exposure_time_left are
both 0. Without this periodic re-publish, the GUI shows a frozen 0% bar and 0.0s remaining
for the whole exposure.
"""
while True:
try:
if self._camera_status == ExposureStatus.EXPOSING:
await self.comm.set_state(
IExposure,
ExposureState(
status=self._camera_status,
progress=await self._get_exposure_progress(),
exposure_time_left=await self._get_exposure_time_left(),
),
)
except Exception:
pass
await asyncio.sleep(1.0)
@abstractmethod
async def _expose(self, exposure_time: float, open_shutter: bool, abort_event: asyncio.Event) -> Image:
"""Actually do the exposure, should be implemented by derived classes.
Args:
exposure_time: The requested exposure time in seconds.
open_shutter: Whether to open the shutter.
abort_event: Event that gets triggered when exposure should be aborted.
Returns:
The actual image.
Raises:
AbortedError: If the exposure was cancelled via abort_event.
GrabImageError: If exposure was not successful.
"""
...
async def _init_exposure(self) -> None:
"""Method that is always called at the very beginning of __expose and can be used to set stuff up."""
...
async def __acquire_frame(self, exposure_time: float, open_shutter: bool) -> Image:
"""Acquire a single frame: the exposure itself, error wrapping, and the axis flip.
Shared by __expose() (single grabs) and grab_stack() (one call per frame). Sets
self._last_frame_nbytes, used for grab_stack()'s advisory memory-cap check.
Args:
exposure_time: The requested exposure time in seconds.
open_shutter: Whether to open the shutter.
Returns:
The acquired frame.
Raises:
GrabImageError: If there was a problem grabbing the image.
"""
# reset abort event
self.expose_abort.clear()
# do the exposure
self._exposure = ExposureInfo(start=datetime.datetime.now(datetime.UTC), exposure_time=exposure_time)
try:
image = await self._expose(exposure_time, open_shutter, abort_event=self.expose_abort)
if image is None or image.safe_data is None:
raise exc.GrabImageError("Could not take image.")
except exc.PyobsError:
# exposure was not successful (aborted?), so reset everything and re-raise
self._exposure = None
raise
except Exception as e:
# exposure was not successful (aborted?), so reset everything and wrap exception
self._exposure = None
raise exc.GrabImageError(str(e))
# flip it?
if self._flip_x or self._flip_y:
# do we have three dimensions in array? need this for deciding which axis to flip
is_3d = len(image.data.shape) == 3
# flip image and make contiguous again
tmp = image.data
if self._flip_x:
tmp = np.flip(tmp, axis=1 if is_3d else 0)
if self._flip_y:
tmp = np.flip(tmp, axis=2 if is_3d else 1)
image.data = np.ascontiguousarray(tmp)
image.header["FLIPX"] = (self._flip_x, "Image flipped along first axis at capture time")
image.header["FLIPY"] = (self._flip_y, "Image flipped along second axis at capture time")
self._last_frame_nbytes = image.data.nbytes
return image
async def __finish_product(
self,
image: Image,
image_type: ImageType,
header_futures_before: dict[str, asyncio.Task[Any]],
header_futures_after: dict[str, asyncio.Task[Any]],
broadcast: bool,
pipeline: str | None,
) -> tuple[Image, str]:
"""Add headers, run the data pipeline, and store the finished product (single frame or
stack cube).
Args:
image: The acquired frame, or the assembled stack cube.
image_type: Type of image.
header_futures_before: FITS headers requested before the exposure(s).
header_futures_after: FITS headers requested after the exposure(s).
broadcast: Whether the new image should be broadcasted.
pipeline: Name of the data pipeline to run before storing, or None for raw data.
Returns:
Tuple of the image itself and its filename.
Raises:
GrabImageError: If there was a problem finishing the product.
"""
# add HDU name
image.header["EXTNAME"] = "SCI"
# add image type
image.header["IMAGETYP"] = image_type
# add fits headers and format filename
await self.add_custom_fits_headers(image)
await self.add_requested_fits_headers(image, header_futures_before)
await self.add_requested_fits_headers(image, header_futures_after)
await self.add_fits_headers(image)
await self.apply_meridian_flip(image)
# run data pipeline -- replaces raw data with the pipeline's result
try:
image = await self._run_data_pipeline(image, pipeline)
except exc.GrabImageError:
self._exposure = None
raise
filename = self.format_filename(image)
# don't want to save?
if filename is None:
self._exposure = None
raise exc.GrabImageError("No filename given.")
# upload file
try:
log.info("Uploading image to file server...")
await self.vfs.write_image(filename, image)
except FileNotFoundError:
raise ValueError("Could not upload image.")
# broadcast image path
if broadcast and self._comm:
log.info("Broadcasting image ID...")
await self.comm.send_event(NewImageEvent(filename, image_type))
# return image and unique
self._exposure = None
log.info("Finished image %s.", filename)
return image, filename
async def __expose(
self, exposure_time: float, image_type: ImageType, broadcast: bool, pipeline: str | None
) -> tuple[Image, str]:
"""Wrapper for a single exposure.
Args:
exposure_time: The requested exposure time in seconds.
image_type: Type of image.
broadcast: Whether the new image should be broadcasted.
pipeline: Name of the data pipeline to run before storing, or None for raw data.
Returns:
Tuple of the image itself and its filename.
Raises:
GrabImageError: If there was a problem grabbing the image.
"""
# init stuff
await self._init_exposure()
# request fits headers
header_futures_before = await self.request_fits_headers(before=True)
# open the shutter?
open_shutter = image_type not in [ImageType.BIAS, ImageType.DARK]
# acquire the frame
image = await self.__acquire_frame(exposure_time, open_shutter)
# request fits headers again
header_futures_after = await self.request_fits_headers(before=False)
# add headers, run pipeline, store
return await self.__finish_product(
image, image_type, header_futures_before, header_futures_after, broadcast, pipeline
)
[docs]
@timeout(calc_expose_timeout)
async def grab_data(self, broadcast: bool = True, **kwargs: Any) -> str:
"""Grabs an image ans returns reference.
Args:
broadcast: Broadcast existence of image.
Returns:
Name of image that was taken.
Raises:
DeviceBusyError: If the camera is already busy (exposing or running a sequence).
GrabImageError: If there was a problem grabbing the image.
"""
# are we exposing?
if self._camera_status != ExposureStatus.IDLE:
raise exc.DeviceBusyError("Cannot start new exposure because camera is not idle.")
await self._change_exposure_status(ExposureStatus.EXPOSING)
# capture the pipeline at grab start -- a change during the grab applies to the next one
pipeline = self._data_pipeline
# expose
try:
image, filename = await self.__expose(self._exposure_time, self._image_type, broadcast, pipeline)
finally:
await self._change_exposure_status(ExposureStatus.IDLE)
# check
if image is None:
raise exc.GrabImageError("Could not take image.")
else:
if filename is None:
raise ValueError("Image has not been saved, so cannot be retrieved by filename.")
# return filename
return filename
[docs]
@timeout(calc_stack_timeout)
async def grab_stack(self, count: int, broadcast: bool = True, **kwargs: Any) -> str:
"""Grab count consecutive frames into one product and return its name.
Without a pipeline selected (IDataPipeline), the product is a 3D cube of all frames.
With one, the pipeline runs on the cube (e.g. to combine it) and only its result is
stored.
Args:
count: Number of frames.
broadcast: Broadcast existence of the product.
Returns:
Name of the stored product.
Raises:
InvalidArgumentError: If count < 1, or the stack would exceed the module's memory cap.
DeviceBusyError: If the camera is exposing, or running a sequence or another stack.
AbortedError: If the stack was aborted.
GrabImageError: If a frame could not be grabbed, frames don't match, or the
pipeline failed.
"""
if count < 1:
raise exc.InvalidArgumentError("count must be >= 1.")
if self._camera_status != ExposureStatus.IDLE or self._sequence_count_left > 0 or self._stack_running:
raise exc.DeviceBusyError("Cannot start new stack because camera is not idle.")
# advisory check, using the size of the last grabbed frame -- settings may have
# changed since then, so this can only reject, never guarantee it's safe
if self._last_frame_nbytes is not None and count * self._last_frame_nbytes > self._max_stack_bytes:
raise exc.InvalidArgumentError(
f"Stack of {count} frames would exceed the memory cap of {self._max_stack_bytes} bytes."
)
# capture settings once, at the start of the stack
exposure_time = self._exposure_time
image_type = self._image_type
pipeline = self._data_pipeline
open_shutter = image_type not in [ImageType.BIAS, ImageType.DARK]
self._stack_running = True
self._stack_abort = False
await self._change_exposure_status(ExposureStatus.EXPOSING)
await self.comm.set_state(IDataStack, DataStackState(count_total=count, count_left=count))
try:
await self._init_exposure()
header_futures_before = await self.request_fits_headers(before=True)
cube: NDArray[Any] | None = None
first_image: Image | None = None
frame_rows: list[tuple[int, str, float]] = []
for i in range(count):
if self._stack_abort:
raise exc.AbortedError("Stack was aborted.")
image = await self.__acquire_frame(exposure_time, open_shutter)
if len(image.data.shape) != 2:
raise exc.GrabImageError("Stacking color frames is not supported.")
if cube is None:
# authoritative check, now with the real frame size
if count * image.data.nbytes > self._max_stack_bytes:
raise exc.InvalidArgumentError(
f"Stack of {count} frames would exceed the memory cap of " f"{self._max_stack_bytes} bytes."
)
cube = np.empty((count, *image.data.shape), dtype=image.data.dtype)
first_image = image
elif image.data.shape != cube.shape[1:] or image.data.dtype != cube.dtype:
raise exc.GrabImageError("Frames in stack do not match in shape or dtype.")
cube[i] = image.data
frame_rows.append(
(i, str(image.header.get("DATE-OBS", "")), float(image.header.get("EXPTIME", exposure_time)))
)
await self.comm.set_state(IDataStack, DataStackState(count_total=count, count_left=count - i - 1))
header_futures_after = await self.request_fits_headers(before=False)
assert cube is not None and first_image is not None
frames_table = Table(rows=frame_rows, names=("FRAME", "DATE-OBS", "EXPTIME"))
stack_image = Image(cube, header=first_image.header, frames=frames_table)
stack_image.header["CTYPE3"] = "FRAME"
stack_image.header["NFRAMES"] = count
last_date_obs, last_exptime = frame_rows[-1][1], frame_rows[-1][2]
if last_date_obs:
stack_image.header["DATE-END"] = (Time(last_date_obs) + TimeDelta(last_exptime, format="sec")).isot
_, filename = await self.__finish_product(
stack_image, image_type, header_futures_before, header_futures_after, broadcast, pipeline
)
return filename
finally:
self._exposure = None
self._stack_running = False
await self._change_exposure_status(ExposureStatus.IDLE)
await self.comm.set_state(IDataStack, DataStackState(count_total=0, count_left=0))
def _sequence_busy(self) -> bool:
"""Whether the camera is busy exposing outside of a running sequence."""
return self._camera_status != ExposureStatus.IDLE or self._stack_running
async def _abort_exposure(self) -> None:
"""Abort the running exposure. Should be implemented by derived class.
Raises:
ValueError: If an error occured.
"""
pass
[docs]
async def abort(self, **kwargs: Any) -> None:
"""Aborts the current exposure and sequence. Derived class must implement IAbortable for this!
Returns:
Success or not.
"""
# set abort event
log.info("Aborting current image and sequence...")
self.expose_abort.set()
self._abort_data_sequence()
self._stack_abort = True
# do camera-specific abort
await self._abort_exposure()
# wait until state is not EXPOSING anymore
while self._camera_status == ExposureStatus.EXPOSING:
await asyncio.sleep(0.1)
async def _abort_weather(self, event: Event, sender: str, **kwargs: Any) -> bool:
"""Abort on bad weather, unless the shutter is closed (dark/bias) -- those don't
expose anything to the weather, and image_type is fixed for a whole sequence, so
this also lets an in-progress dark/bias sequence run to completion undisturbed."""
if self._image_type not in (ImageType.BIAS, ImageType.DARK):
await self.abort()
return True
[docs]
@staticmethod
def set_biassec_trimsec(hdr: fits.Header, left: int, top: int, width: int, height: int) -> None:
"""Calculates and sets the BIASSEC and TRIMSEC areas.
Args:
hdr: FITS header (in/out)
left: left edge of data area
top: top edge of data area
width: width of data area
height: height of data area
"""
# get image area in unbinned coordinates
img_left = hdr["XORGSUBF"]
img_top = hdr["YORGSUBF"]
img_width = hdr["NAXIS1"] * hdr["XBINNING"]
img_height = hdr["NAXIS2"] * hdr["YBINNING"]
# get intersection
is_left = max(left, img_left)
is_right = min(left + width, img_left + img_width)
is_top = max(top, img_top)
is_bottom = min(top + height, img_top + img_height)
# prescan/overscan exists only where the window extends beyond the visible frame
x_ovs = img_left < left or img_left + img_width > left + width
y_ovs = img_top < top or img_top + img_height > top + height
# window extends beyond the visible frame on both axes: the simple calculation
# below only supports prescan/overscan on one axis
if x_ovs and y_ovs:
log.warning(
"Cannot compute BIASSEC/TRIMSEC: window %dx%d at %d,%d has prescan/overscan on both axes "
"(only one axis is supported).",
img_width,
img_height,
img_left,
img_top,
)
return
# window lies fully inside the visible frame: there is no prescan/overscan at all
if (
not x_ovs
and not y_ovs
and (
img_left > left
or img_left + img_width < left + width
or img_top > top
or img_top + img_height < top + height
)
):
log.info(
"No BIASSEC/TRIMSEC: window %dx%d at %d,%d lies fully inside the visible frame "
"(no prescan/overscan).",
img_width,
img_height,
img_left,
img_top,
)
return
# comments
c1 = "Bias overscan area [x1:x2,y1:y2] (binned)"
c2 = "Image area [x1:x2,y1:y2] (binned)"
# rectangle empty?
if is_right <= is_left or is_bottom <= is_top:
# easy case, all is BIASSEC, no TRIMSEC at all
hdr["BIASSEC"] = ("[1:{},1:{}]".format(hdr["NAXIS1"], hdr["NAXIS2"]), c1) # noqa: UP031
return
# we got a TRIMSEC, calculate its binned and windowd coordinates
is_left_binned = np.floor((is_left - hdr["XORGSUBF"]) / hdr["XBINNING"]) + 1
is_right_binned = np.ceil((is_right - hdr["XORGSUBF"]) / hdr["XBINNING"])
is_top_binned = np.floor((is_top - hdr["YORGSUBF"]) / hdr["YBINNING"]) + 1
is_bottom_binned = np.ceil((is_bottom - hdr["YORGSUBF"]) / hdr["YBINNING"])
# set it
sec = f"[{is_left_binned:.0f}:{is_right_binned:.0f},{is_top_binned:.0f}:{is_bottom_binned:.0f}]"
hdr["TRIMSEC"] = (sec, c2)
hdr["DATASEC"] = (sec, c2)
# now get BIASSEC -- whatever we do, we only take the last (!) one
# which axis?
if img_left + img_width > left + width:
left_binned = np.floor((is_right - hdr["XORGSUBF"]) / hdr["XBINNING"]) + 1
hdr["BIASSEC"] = ("[{:.0f}:{},1:{}]".format(left_binned, hdr["NAXIS1"], hdr["NAXIS2"]), c1)
elif img_left < left:
right_binned = np.ceil((is_left - hdr["XORGSUBF"]) / hdr["XBINNING"])
hdr["BIASSEC"] = ("[1:{:.0f},1:{}]".format(right_binned, hdr["NAXIS2"]), c1)
elif img_top + img_height > top + height:
top_binned = np.floor((is_bottom - hdr["YORGSUBF"]) / hdr["YBINNING"]) + 1
hdr["BIASSEC"] = ("[1:{},{:.0f}:{}]".format(hdr["NAXIS1"], top_binned, hdr["NAXIS2"]), c1)
elif img_top < top:
bottom_binned = np.ceil((is_top - hdr["YORGSUBF"]) / hdr["YBINNING"])
hdr["BIASSEC"] = ("[1:{},1:{:.0f}]".format(hdr["NAXIS1"], bottom_binned), c1)
[docs]
async def apply_meridian_flip(self, image: Image) -> None:
"""If the telescope has a meridian flip (MERIDIAN keyword in header), flip the image on both axes.
Args:
image: Image to be flipped.
"""
flipped = False
if "MERIDIAN" in image.header:
if image.header["MERIDIAN"].upper() == self._meridian_flip_on:
# flip both axes
image.data = image.data[..., ::-1, ::-1]
flipped = True
image.header["FLIPDONE"] = (flipped, "Image flipped for meridian side at capture time")
__all__ = ["BaseCamera"]