Source code for pyobs.modules.image.seeing

import logging
import numpy as np
from typing import List, Union, Any, Optional, Dict
from astropy.wcs import WCS
from astropy.wcs.utils import proj_plane_pixel_scales

from pyobs.modules import Module
from pyobs.events import NewImageEvent, Event
from pyobs.utils.publisher import Publisher
from pyobs.utils.time import Time

log = logging.getLogger(__name__)


class Seeing(Module):
    """Measures seeing on reduced images with a catalog."""

    __module__ = "pyobs.modules.image"

    def __init__(
        self,
        sources: Optional[Union[str, List[str]]] = None,
        publisher: Optional[Union[Publisher, Dict[str, Any]]] = None,
        max_ellipticity: float = 0.2,
        correct_for_airmass: bool = True,
        **kwargs: Any,
    ):
        """Creates a new seeing estimator.

        Args:
            sources: List of sources (e.g. cameras) to process images from or None for all.
            publisher: Publisher to publish results to.
            max_ellipticity: Maximum ellipticity for sources to consider.
            correct_for_zenith: Whether to correct seeing for airmass.
        """
        Module.__init__(self, **kwargs)

        # stuff
        self._sources = [sources] if isinstance(sources, str) else sources
        self._publisher = self.add_child_object(publisher, Publisher)
        self._max_ellipticity = max_ellipticity
        self._correct_for_airmass = correct_for_airmass

[docs] async def open(self) -> None: """Open module.""" await Module.open(self) # subscribe to channel with new images log.info("Subscribing to new image events...") await self.comm.register_event(NewImageEvent, self.process_new_image_event)
[docs] async def process_new_image_event(self, event: Event, sender: str) -> bool: """Puts a new images in the DB with the given ID. Args: event: New image event sender: Who sent the event? Returns: Success """ if not isinstance(event, NewImageEvent): return False # filter by source if self._sources is not None and sender not in self._sources: return False # put into queue log.info("Received new image event from %s.", sender) # download image try: log.info("Downloading file %s...", event.filename) image = await self.vfs.read_image(event.filename) except FileNotFoundError: log.error("Could not download image.") return False # get catalog cat = image.catalog if cat is None: # no catalog found in file return False # filter by ellipticity cat = cat[cat["ellipticity"] < self._max_ellipticity] # get WCS and pixel size wcs = WCS(image.header) pix_size = abs(proj_plane_pixel_scales(wcs)[0] * 3600.0) # calculate seeing seeing = np.mean(cat["fwhm"]) * pix_size # correct for airmass? if self._correct_for_airmass: # Seeing S as function of seeing S0 at zenith and airmass a: # S = S0 * a^0.6 # see https://www.astro.auth.gr/~seeing-gr/seeing_gr_files/theory/node17.html # need airmass if "AIRMASS" in image.header: seeing /= image.header["AIRMASS"] ** 0.6 else: # could not correct return False # log it if self._publisher is not None: await self._publisher(time=Time.now().isot, seeing=seeing) return True
__all__ = ["Seeing"]