Viewer/guider read live device state instead of re-reading FITS

GET /panoramic?vmin=&vmax=&max_px=

    Small PNG thumbnail (longest side ≤ *max_px*, default 256).

GET /tile?cx=&cy=&rx=&ry=

    Raw ``float32`` arraybuffer tile centred on *(cx, cy)*.

    Raw ``uint16`` arraybuffer tile centred on *(cx, cy)*.

GET /fits

    Raw FITS file download.

# Third-party modules

import numpy as np

from astropy.io import fits

from quart import Blueprint, request, Response, jsonify, send_file

# Custom modules

from noctua import devices

from noctua.config.constants import DATA_FOLDER

from noctua.utils.image import auto_range, fits_to_png

# ---------------------------------------------------------------------------

# FITS data cache

# Live device access

# ---------------------------------------------------------------------------

# Maps cam_id → (file_mtime, float32_ndarray).

_fits_cache: dict[str, tuple[float, np.ndarray]] = {}

def _fits_path(cam_id: str) -> Path | None:

    """Return the filesystem path for a camera's latest FITS file."""

    """Return the filesystem path for a camera's latest FITS file.

    Only used by the raw /fits download route below — every other route

    reads the live frame straight from the device (see get_live_data).

    """

    raw_path = get_cameras_cfg().get(cam_id, 'fits_path', fallback=None)

    if raw_path is None:

    return PROJECT_ROOT / DATA_FOLDER / raw_path

def _load_data(cam_id: str) -> np.ndarray | None:

    """Read the primary HDU from disk and return it as a float32 array."""

    path = _fits_path(cam_id)

    if path is None or not path.exists():

        return None

    try:

        with fits.open(path) as hdul:

            data = hdul[0].data

            if data is None:

                return None

            return data.astype(np.float32)

    except OSError:

        # Reader lost a race with a writer that isn't using an atomic

        # rename yet, or hit a genuinely corrupt file — either way, "no

        # data this round" is the right degradation, not an unhandled

        # exception (this fed straight into an uncaught OSError that used

        # to kill the whole /web/socket connection via _handle_tile_request).

        return None

def get_cached_data(cam_id: str) -> np.ndarray | None:

    """Return the FITS image data for a camera, backed by an mtime cache."""

def get_live_data(cam_id: str) -> np.ndarray | None:

    """Return a camera's current frame, read straight from the device.

    path = _fits_path(cam_id)

    if path is None or not path.exists():

        return None

    Each device's own `matrix` property already holds the last acquired

    frame in memory (or de-duplicates the fetch for network cameras like

    STX — see stx.py's _fetch()), so there's no file re-read/cache layer

    needed here anymore.

    """

    try:

        current_mtime = os.path.getmtime(path)

    except OSError:

    device = devices.viewer_devices.get(cam_id)

    if device is None:

        return None

    cached_mtime, cached_array = _fits_cache.get(cam_id, (None, None))

    if cached_mtime is not None and cached_mtime == current_mtime:

        return cached_array

    data = _load_data(cam_id)

    if data is not None:

        _fits_cache[cam_id] = (current_mtime, data)

    return data

    return device.matrix

# ---------------------------------------------------------------------------

async def image_info(cam_id):

    """Return shape, dtype, and auto-range statistics for the image."""

    data = get_cached_data(cam_id)

    data = get_live_data(cam_id)

    if data is None:

        return jsonify({'error': 'FITS not found'}), 404

async def image_png(cam_id):

    """Render the image as a PNG with Viridis colormap."""

    data = get_cached_data(cam_id)

    data = get_live_data(cam_id)

    if data is None:

        return Response('FITS not found', status=404)

async def image_panoramic(cam_id):

    """Render a thumbnail PNG."""

    data = get_cached_data(cam_id)

    data = get_live_data(cam_id)

    if data is None:

        return Response('FITS not found', status=404)

@viewer_blueprint.route('/viewer/<cam_id>/tile')

async def image_tile(cam_id):

    """Return a raw float32 arraybuffer tile centred on (cx, cy)."""

    """Return a raw uint16 arraybuffer tile centred on (cx, cy).

    uint16, not float32: native ADC bit depth for every camera behind this

    endpoint (12-16 bit sensors, values 0-65535) — half the bytes over the

    wire, no precision loss.

    """

    data = get_cached_data(cam_id)

    data = get_live_data(cam_id)

    if data is None:

        return Response('FITS not found', status=404)

    tw, th = tile.shape[1], tile.shape[0]

    header = np.array([tw, th], dtype=np.int32).tobytes()

    body   = tile.astype(np.float32).tobytes()

    body   = tile.astype(np.uint16).tobytes()

    return Response(

        header + body,

for dev in devs.sections():

    print(f"importing {dev}")

    dynamic_import(this_module, dev)

# Reverse lookup: cam_id (viewer_key, e.g. "scicam1") -> live device instance.

# Lets the web/viewer/guider layers read a camera's last frame straight from

# memory (device.matrix) instead of re-reading the FITS file it last wrote.

viewer_devices = {

    getattr(this_module, dev)._viewer_key: getattr(this_module, dev)

    for dev in devs.sections()

    if getattr(getattr(this_module, dev), '_viewer_key', None)

}

        self._last_exptime   = None

        self._last_imagetype = None

        self._last_datetime  = None

        self._last_binning   = None

        self._setpoint       = None

        self._looping      = False

    def _run_loop(self):

        while self._looping:

            self._start(self.loop_exposure, 1)

            # Explicit datetime: _start()'s default (None) would leave

            # _last_datetime unset during looping, which viewer/guider now

            # rely on in-memory as the "new frame" signal (see matrix).

            self._start(self.loop_exposure, 1,

                        datetime=datetime.utcnow().isoformat(timespec='milliseconds'))

            while self._looping and self.ready != 1:

                time.sleep(0.3)

            if self._looping:

        # Only the last completed frame of the loop is written to disk —

        # intermediate frames live only in `matrix` (see viewer/guider,

        # both now read the live camera state instead of re-reading the

        # file). Skip the write if stop interrupted an in-flight exposure.

        if self.ready == 1:

            self.download()

        log.info("Atik: Loop stopped")

            self._last_exptime   = duration

            self._last_imagetype = imagetype

            self._last_datetime  = datetime

            self._last_binning   = self.binning

        is_dark = True if imagetype in [0, 2, "Dark", "Bias"] else False

        self.put('ArtemisSetDarkMode', is_dark)

        self.put('ArtemisStartExposure', ctypes.c_float(duration))

import threading

import time

from ctypes import byref, c_int, string_at

from datetime import datetime as dt

# Third-party modules

import numpy as np

        self._last_exptime   = None

        self._last_imagetype = None

        self._last_datetime  = None

        self._last_binning   = None

        self._setpoint       = None

        self._looping      = False

    def _run_loop(self):

        while self._looping:

            self._start(self.loop_exposure, 1)

            # Explicit datetime: _start()'s default (None) would leave

            # _last_datetime unset during looping, which viewer/guider now

            # rely on in-memory as the "new frame" signal (see matrix).

            self._start(self.loop_exposure, 1,

                        datetime=dt.utcnow().isoformat(timespec='milliseconds'))

            while self._looping and self.ready != 1:

                time.sleep(0.3)

            if self._looping:

        # Only the last completed frame of the loop is written to disk —

        # intermediate frames live only in `matrix` (see viewer/guider,

        # both now read the live camera state instead of re-reading the

        # file). Skip the write if stop interrupted an in-flight exposure.

        if self.ready == 1:

            self.download()

        log.info("Atik: Loop stopped")

            self._last_exptime   = duration

            self._last_imagetype = imagetype

            self._last_datetime  = datetime

            self._last_binning   = self.binning

        is_dark = True if imagetype in [0, 2, "Dark", "Bias"] else False

        # AtikSDKCamera.set_dark_mode() calls ArtemisSetDarkMode

        # unconditionally, with no support check (unlike its own

        self._last_frame    = None

        self._last_exptime  = None

        self._last_datetime = None

        self._last_binning  = None

        self._lock          = threading.Lock()

        self._state         = 0

            return

        self._state         = 2

        self._last_datetime = dt.utcnow().isoformat(timespec='milliseconds')

        self._last_binning  = self.binning

        try:

            raw = self._acquire(cam, exptime)

            with self._lock:

            try:

                self._state         = 2

                self._last_datetime = dt.utcnow().isoformat(timespec='milliseconds')

                self._last_binning  = self.binning

                raw = self._acquire(cam, self.loop_exposure)

                with self._lock:

                    self._last_frame = raw

                self._state = 0

                self.download()

            except Exception as e:

                log.error(f"Mako loop error: {e}")

                self._state = 5

                time.sleep(1.0)

        # Only the last completed frame of the loop is written to disk —

        # intermediate frames live only in `matrix` (see viewer/guider,

        # both now read the live camera state instead of re-reading the file).

        self.download()

        log.info("Mako: Loop stopped")

        self._state = 0

    @property

    def matrix(self):

        """ndarray or None : Last acquired frame as a 2-D uint8 array (H, W)."""

        """ndarray or None : Last acquired frame as a 2-D uint8 array (H, W).

        Flipped along axis 0 to match the orientation written to FITS by

        download() — every consumer of ``matrix`` (viewer, guider,

        ao_calibration) gets the same, already-correct orientation.

        """

        with self._lock:

            return self._last_frame

            if self._last_frame is None:

                return None

            return np.flip(self._last_frame, axis=0)

    def image(self, vmin=None, vmax=None, color=True):

        """Return the current frame as PNG-encoded bytes.

        log.debug(f"Getting original data")

        data = np.flip(self.matrix, axis=0)

        data = self.matrix

        if data is None:

            return None