Source code for pyobs.modules.focus.focusseries

import logging
from typing import Union, Tuple, Dict, Any, Optional
import threading
import numpy as np
from numpy.typing import NDArray

from pyobs.interfaces import IAutoFocus
from import FocusFoundEvent
from pyobs.interfaces import IExposureTime, IImageType, IFocuser, IFilters, IData
from pyobs.object import get_object
from pyobs.mixins import CameraSettingsMixin
from pyobs.modules import timeout, Module
from pyobs.utils.enums import ImageType
from pyobs.utils.focusseries import FocusSeries
from pyobs.utils import exceptions as exc

log = logging.getLogger(__name__)

class AutoFocusSeries(Module, CameraSettingsMixin, IAutoFocus):
    """Module for auto-focusing a telescope."""

    __module__ = "pyobs.modules.focus"

    def __init__(
        focuser: Union[str, IFocuser],
        camera: Union[str, IData],
        series: Union[Dict[str, Any], FocusSeries],
        offset: bool = False,
        filters: Optional[Union[str, IFilters]] = None,
        filter_name: Optional[str] = None,
        binning: Optional[int] = None,
        **kwargs: Any,
        """Initialize a new auto focus system.

            focuser: Name of IFocuser.
            camera: Name of ICamera.
            filters: Name of IFilters, if any.
            filter_name: Name of filter to set.
            offset: If True, offsets are used instead of absolute focus values.
        Module.__init__(self, **kwargs)

        # store focuser and camera
        self._focuser = focuser
        self._camera = camera
        self._filters = filters
        self._offset = offset
        self._abort = threading.Event()

        # create focus series
        self._series: FocusSeries = get_object(series, FocusSeries)

        # init camera settings mixin
        CameraSettingsMixin.__init__(self, filters=filters, filter_name=filter_name, binning=binning, **kwargs)

        # register exceptions
        if isinstance(camera, str):
                exc.RemoteError, 3, timespan=600, module=camera, callback=self._default_remote_error_callback
        if isinstance(focuser, str):
                exc.RemoteError, 3, timespan=600, module=focuser, callback=self._default_remote_error_callback

[docs] async def open(self) -> None: """Open module""" await # register event await self.comm.register_event(FocusFoundEvent) # check focuser and camera try: await self.proxy(self._focuser, IFocuser) await self.proxy(self._camera, IData) except ValueError: log.warning("Either camera or focuser do not exist or are not of correct type at the moment.")
[docs] @timeout(600) async def auto_focus(self, count: int, step: float, exposure_time: float, **kwargs: Any) -> Tuple[float, float]: """Perform an auto-focus series. This method performs an auto-focus series with "count" images on each side of the initial guess and the given step size. With count=3, step=1 and guess=10, this takes images at the following focus values: 7, 8, 9, 10, 11, 12, 13 Args: count: Number of images to take on each side of the initial guess. Should be an odd number. step: Step size. exposure_time: Exposure time for images. Returns: Tuple of obtained best focus value and its uncertainty. Or Nones, if focus series failed. Raises: FileNotFoundException: If image could not be downloaded. """"Performing auto-focus...") # get focuser"Getting proxy for focuser...") focuser = await self.proxy(self._focuser, IFocuser) # get camera"Getting proxy for camera...") camera = await self.proxy(self._camera, IData) # do camera settings await self._do_camera_settings(camera) # get filter wheel and current filter filter_name = "unknown" try: filter_wheel = await self.proxy(self._filters, IFilters) filter_name = await filter_wheel.get_filter() except ValueError: log.warning("Filter module is not of type IFilters. Could not get filter.") # get focus as first guess try: if self._offset: guess = 0.0"Using focus offset of 0mm as initial guess.") else: guess = await focuser.get_focus()"Using current focus of %.2fmm as initial guess.", guess) except exc.RemoteError: raise ValueError("Could not fetch current focus value.") # define array of focus values to iterate focus_values: NDArray[float] = np.linspace(guess - count * step, guess + count * step, 2 * count + 1) # reset self._series.reset() self._abort = threading.Event() # loop focus values"Starting focus series...") for foc in focus_values: # set focus if self._offset:"Changing focus offset to %.2fmm...", foc) else:"Changing focus to %.2fmm...", foc) if self._abort.is_set(): raise InterruptedError() try: if self._offset: await focuser.set_focus_offset(float(foc)) else: await focuser.set_focus(float(foc)) except exc.RemoteError: raise ValueError("Could not set new focus value.") # do exposure"Taking picture...") if self._abort.is_set(): raise InterruptedError() try: if isinstance(camera, IExposureTime): await camera.set_exposure_time(exposure_time) if isinstance(camera, IImageType): await camera.set_image_type(ImageType.FOCUS) filename = await camera.grab_data() except exc.RemoteError: log.error("Could not take image.") continue # download image"Downloading image...") image = await self.vfs.read_image(filename) # analyse"Analysing picture...") try: self._series.analyse_image(image, foc) except: # do nothing.. log.error("Could not analyse image.") continue # fit focus if self._abort.is_set(): raise InterruptedError() focus = self._series.fit_focus() # did focus series fail? if focus is None or focus[0] is None or np.isnan(focus[0]): log.warning("Focus series failed.") # reset to initial values if self._offset:"Resetting focus offset to 0.", guess) await focuser.set_focus_offset(0) else:"Resetting focus to initial guess of %.3f mm.", guess) await focuser.set_focus(guess) # raise error raise ValueError("Could not find best focus.") # log and set focus if self._offset:"Setting new focus offset of (%.3f+-%.3f) mm.", focus[0], focus[1]) absolute = focus[0] + await focuser.get_focus() await focuser.set_focus_offset(focus[0]) else:"Setting new focus value of (%.3f+-%.3f) mm.", focus[0], focus[1]) absolute = focus[0] + await focuser.get_focus_offset() await focuser.set_focus(focus[0]) # send event await self.comm.send_event(FocusFoundEvent(absolute, focus[1], filter_name)) # return result return focus[0], focus[1]
[docs] async def auto_focus_status(self, **kwargs: Any) -> Dict[str, Any]: """Returns current status of auto focus. Returned dictionary contains a list of focus/fwhm pairs in X and Y direction. Returns: Dictionary with current status. """ return {}
[docs] @timeout(20) async def abort(self, **kwargs: Any) -> None: """Abort current actions.""" self._abort.set()
__all__ = ["AutoFocusSeries"]