Merge pull request #339

Skycoord Guiding Statistics

pyobs/images/processors/offsets/__init__.py

@@ -10,6 +10,7 @@
 from .projected import ProjectedOffsets
 from .fitsheader import FitsHeaderOffsets
 from .dummyoffsets import DummyOffsets
+from .dummyskyoffsets import DummySkyOffsets
 
 __all__ = [
     "Offsets",
@@ -19,4 +20,5 @@
     "FitsHeaderOffsets",
     "BrightestStarOffsets",
     "DummyOffsets",
+    "DummySkyOffsets"
 ]

pyobs/images/processors/offsets/dummyskyoffsets.py

@@ -0,0 +1,24 @@
+from copy import copy
+from typing import Any, Union, Dict
+
+from astropy.coordinates import SkyCoord
+
+from pyobs.images import Image
+from pyobs.images.meta import SkyOffsets
+from pyobs.images.processors.offsets import Offsets
+from pyobs.object import get_object
+
+
+class DummySkyOffsets(Offsets):
+    def __init__(self, coord0: Union[SkyCoord, Dict[str, Any]], coord1: Union[SkyCoord, Dict[str, Any]], **kwargs: Any) -> None:
+        super().__init__(**kwargs)
+        sky_coord0 = get_object(coord0, SkyCoord)
+        sky_coord1 = get_object(coord1, SkyCoord)
+        self._offset = SkyOffsets(sky_coord0, sky_coord1)
+
+    async def __call__(self, image: Image) -> Image:
+        image.set_meta(copy(self._offset))
+        return image
+
+
+__all__ = ["DummySkyOffsets"]

pyobs/modules/pointing/_baseguiding.py

@@ -1,165 +1,22 @@
 import logging
-from abc import ABCMeta, abstractmethod, ABC
-from collections import defaultdict
-from datetime import datetime
+from abc import ABCMeta
 from typing import Union, List, Dict, Tuple, Any, Optional
 
 import astropy.units as u
-import numpy as np
 from astropy.coordinates import SkyCoord
 
 from pyobs.images import Image
 from pyobs.interfaces import IAutoGuiding, IFitsHeaderBefore, IFitsHeaderAfter
 from pyobs.utils.time import Time
 from ._base import BasePointing
-from ...images.meta import PixelOffsets
+from .guidingstatistics import GuidingStatisticsUptime, GuidingStatisticsPixelOffset
+from .guidingstatistics.guidingstatistics import GuidingStatistics
 from ...interfaces import ITelescope
+from ...object import get_object
 
 log = logging.getLogger(__name__)
 
 
-class _GuidingStatistics(ABC):
-    """Calculates statistics for guiding."""
-
-    def __init__(self):
-        self._sessions: Dict[str, List[Any]] = defaultdict(list)
-
-    def init_stats(self, client: str, default: Any = None) -> None:
-        """
-        Inits a stat measurement session for a client.
-
-        Args:
-            client: name/id of the client
-            default: first entry in session
-        """
-
-        self._sessions[client] = []
-
-        if default is not None:
-            self._sessions[client].append(self._get_session_data(default))
-
-    @abstractmethod
-    def _build_header(self, data: Any) -> Dict[str, Tuple[Any, str]]:
-        raise NotImplementedError
-
-    def add_to_header(self, client: str, header: Dict[str, Tuple[Any, str]]) -> Dict[str, Tuple[Any, str]]:
-        """
-        Add statistics to given header.
-
-        Args:
-            client: id/name of the client
-            header: Header dict to add statistics to.
-        """
-
-        data = self._sessions.pop(client)
-        session_header = self._build_header(data)
-
-        return header | session_header
-
-    @abstractmethod
-    def _get_session_data(self, input_data: Any) -> Any:
-        raise NotImplementedError
-
-    def add_data(self, input_data: Any) -> None:
-        """
-        Adds data to all client measurement sessions.
-        Args:
-            input_data: Image witch metadata
-        """
-
-        data = self._get_session_data(input_data)
-
-        for k in self._sessions.keys():
-            self._sessions[k].append(data)
-
-
-class _GuidingStatisticsPixelOffset(_GuidingStatistics):
-    @staticmethod
-    def _calc_rms(data: List[Tuple[float, float]]) -> Optional[Tuple[float, float]]:
-        """
-        Calculates RMS of data.
-
-        Args:
-            data: Data to calculate RMS for.
-
-        Returns:
-            Tuple of RMS.
-        """
-        if len(data) < 3:
-            return None
-
-        flattened_data = np.array(list(map(list, zip(*data))))
-        data_len = len(flattened_data[0])
-        rms = np.sqrt(np.sum(np.power(flattened_data, 2), axis=1) / data_len)
-        return tuple(rms)
-
-    def _build_header(self, data: List[Tuple[float, float]]) -> Dict[str, Tuple[Any, str]]:
-        header = {}
-        rms = self._calc_rms(data)
-
-        if rms is not None:
-            header["GUIDING RMS1"] = (float(rms[0]), "RMS for guiding on axis 1")
-            header["GUIDING RMS2"] = (float(rms[1]), "RMS for guiding on axis 2")
-
-        return header
-
-    def _get_session_data(self, data: Image) -> Tuple[float, float]:
-        if data.has_meta(PixelOffsets):
-            meta = data.get_meta(PixelOffsets)
-            primitive = tuple(meta.__dict__.values())
-            return primitive
-        else:
-            log.warning("Image is missing the necessary meta information!")
-            raise KeyError("Unknown meta.")
-
-
-class _GuidingStatisticsUptime(_GuidingStatistics):
-    @staticmethod
-    def _calc_uptime(states: List[Tuple[bool, datetime]]) -> int:
-        uptimes: List[int] = []
-        for i, entry in enumerate(states):
-            state, timestamp = entry
-            # is not closed?
-            if not state or i + 1 == len(states):
-                continue
-
-            uptime = (states[i + 1][1] - timestamp).seconds
-            uptimes.append(uptime)
-
-        return sum(uptimes)
-
-    @staticmethod
-    def _calc_total_time(states: List[Tuple[bool, datetime]]) -> int:
-        initial_time = states[0][1]
-        end_time = states[-1][1]
-        return (end_time - initial_time).seconds
-
-    @staticmethod
-    def _calc_uptime_percentage(states: List[Tuple[bool, datetime]]) -> float:
-        uptime = _GuidingStatisticsUptime._calc_uptime(states)
-        total_time = _GuidingStatisticsUptime._calc_total_time(states)
-
-        """
-        If no time has passed, return 100 if the loop is closed, 0 else.
-        We have to take the second last value in the state array, since the last value is the stop value.
-        """
-        if total_time == 0:
-            return int(states[-2][0]) * 100.0
-
-        return uptime / total_time * 100.0
-
-    def _build_header(self, data: List[Tuple[bool, datetime]]) -> Dict[str, Tuple[Any, str]]:
-        now = datetime.now()
-        data.append((None, now))
-
-        uptime_percentage = self._calc_uptime_percentage(data)
-        return {"GUIDING UPTIME": (uptime_percentage, "Time the guiding loop was closed [%]")}
-
-    def _get_session_data(self, input_data: bool) -> Tuple[bool, datetime]:
-        now = datetime.now()
-        return input_data, now
-
-
 class BaseGuiding(BasePointing, IAutoGuiding, IFitsHeaderBefore, IFitsHeaderAfter, metaclass=ABCMeta):
     """Base class for guiding modules."""
 
@@ -174,6 +31,7 @@ def __init__(
         pid: bool = False,
         reset_at_focus: bool = True,
         reset_at_filter: bool = True,
+        guiding_statistic: Optional[Union[Dict[str, Any], GuidingStatistics]] = None,
         **kwargs: Any,
     ):
         """Initializes a new science frame auto guiding system.
@@ -205,8 +63,11 @@ def __init__(
         self._ref_header = None
 
         # stats
-        self._statistics = _GuidingStatisticsPixelOffset()
-        self._uptime = _GuidingStatisticsUptime()
+        if guiding_statistic is None:
+            guiding_statistic = GuidingStatisticsPixelOffset()
+
+        self._statistics = get_object(guiding_statistic, GuidingStatistics)
+        self._uptime = GuidingStatisticsUptime()
 
     async def start(self, **kwargs: Any) -> None:
         """Starts/resets auto-guiding."""

pyobs/modules/pointing/guidingstatistics/__init__.py

@@ -0,0 +1,3 @@
+from .uptime import GuidingStatisticsUptime
+from .pixeloffset import GuidingStatisticsPixelOffset
+from .skyoffsets import GuidingStatisticsSkyOffset

pyobs/modules/pointing/guidingstatistics/guidingstatistics.py

@@ -0,0 +1,61 @@
+from abc import abstractmethod, ABCMeta
+from collections import defaultdict
+from typing import List, Dict, Tuple, Any
+
+from pyobs.images import Image
+
+
+class GuidingStatistics(object, metaclass=ABCMeta):
+    """Calculates statistics for guiding."""
+
+    def __init__(self) -> None:
+        self._sessions: Dict[str, List[Any]] = defaultdict(list)
+
+    def init_stats(self, client: str, default: Any = None) -> None:
+        """
+        Inits a stat measurement session for a client.
+
+        Args:
+            client: name/id of the client
+            default: first entry in session
+        """
+
+        self._sessions[client] = []
+
+        if default is not None:
+            self._sessions[client].append(self._get_session_data(default))
+
+    @abstractmethod
+    def _build_header(self, data: Any) -> Dict[str, Tuple[Any, str]]:
+        raise NotImplementedError
+
+    def add_to_header(self, client: str, header: Dict[str, Tuple[Any, str]]) -> Dict[str, Tuple[Any, str]]:
+        """
+        Add statistics to given header.
+
+        Args:
+            client: id/name of the client
+            header: Header dict to add statistics to.
+        """
+
+        data = self._sessions.pop(client)
+        session_header = self._build_header(data)
+
+        return header | session_header
+
+    @abstractmethod
+    def _get_session_data(self, input_data: Image) -> Any:
+        raise NotImplementedError
+
+    def add_data(self, input_data: Image) -> None:
+        """
+        Adds data to all client measurement sessions.
+        Args:
+            input_data: Image witch metadata
+        """
+
+        data = self._get_session_data(input_data)
+
+        for k in self._sessions.keys():
+            self._sessions[k].append(data)
+

pyobs/modules/pointing/guidingstatistics/pixeloffset.py

@@ -0,0 +1,54 @@
+import logging
+from typing import List, Dict, Tuple, Any, Optional
+
+import numpy as np
+
+from pyobs.images import Image
+from .guidingstatistics import GuidingStatistics
+from pyobs.images.meta import PixelOffsets
+
+
+log = logging.getLogger(__name__)
+
+
+class GuidingStatisticsPixelOffset(GuidingStatistics):
+    @staticmethod
+    def _calc_rms(data: List[Tuple[float, float]]) -> Optional[Tuple[float, float]]:
+        """
+        Calculates RMS of data.
+
+        Args:
+            data: Data to calculate RMS for.
+
+        Returns:
+            Tuple of RMS.
+        """
+        if len(data) < 3:
+            return None
+
+        flattened_data = np.array(list(map(list, zip(*data))))
+        data_len = len(flattened_data[0])
+        rms = np.sqrt(np.sum(np.power(flattened_data, 2), axis=1) / data_len)
+        return tuple(rms)
+
+    def _build_header(self, data: List[Tuple[float, float]]) -> Dict[str, Tuple[Any, str]]:
+        header = {}
+        rms = self._calc_rms(data)
+
+        if rms is not None:
+            header["GUIDING RMS1"] = (float(rms[0]), "RMS for guiding on axis 1")
+            header["GUIDING RMS2"] = (float(rms[1]), "RMS for guiding on axis 2")
+
+        return header
+
+    def _get_session_data(self, data: Image) -> Tuple[float, float]:
+        if data.has_meta(PixelOffsets):
+            meta = data.get_meta(PixelOffsets)
+            primitive = tuple(meta.__dict__.values())
+            return primitive
+        else:
+            log.warning("Image is missing the necessary meta information!")
+            raise KeyError("Unknown meta.")
+
+
+