From ad762022d0f197280b637281b827f81504d4c489 Mon Sep 17 00:00:00 2001 From: germanhydrogen Date: Mon, 29 Jan 2024 16:24:51 +0100 Subject: [PATCH 1/5] Seperated guiding statistics classes into own module --- pyobs/modules/pointing/_baseguiding.py | 142 ------------------ .../pointing/guidingstatistics/__init__.py | 2 + .../guidingstatistics/guidingstatistics.py | 59 ++++++++ .../pointing/guidingstatistics/pixeloffset.py | 54 +++++++ .../pointing/guidingstatistics/uptime.py | 54 +++++++ .../pointing/test_guiding_statistic_uptime.py | 10 +- .../test_guiding_statistics_pixel_offset.py | 8 +- 7 files changed, 178 insertions(+), 151 deletions(-) create mode 100644 pyobs/modules/pointing/guidingstatistics/__init__.py create mode 100644 pyobs/modules/pointing/guidingstatistics/guidingstatistics.py create mode 100644 pyobs/modules/pointing/guidingstatistics/pixeloffset.py create mode 100644 pyobs/modules/pointing/guidingstatistics/uptime.py diff --git a/pyobs/modules/pointing/_baseguiding.py b/pyobs/modules/pointing/_baseguiding.py index f7a1d9dd..c32bf1f4 100644 --- a/pyobs/modules/pointing/_baseguiding.py +++ b/pyobs/modules/pointing/_baseguiding.py @@ -18,148 +18,6 @@ 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.""" diff --git a/pyobs/modules/pointing/guidingstatistics/__init__.py b/pyobs/modules/pointing/guidingstatistics/__init__.py new file mode 100644 index 00000000..018262ab --- /dev/null +++ b/pyobs/modules/pointing/guidingstatistics/__init__.py @@ -0,0 +1,2 @@ +from .uptime import GuidingStatisticsUptime +from .pixeloffset import GuidingStatisticsPixelOffset diff --git a/pyobs/modules/pointing/guidingstatistics/guidingstatistics.py b/pyobs/modules/pointing/guidingstatistics/guidingstatistics.py new file mode 100644 index 00000000..7dc851fc --- /dev/null +++ b/pyobs/modules/pointing/guidingstatistics/guidingstatistics.py @@ -0,0 +1,59 @@ +from abc import abstractmethod, ABCMeta +from collections import defaultdict +from typing import List, Dict, Tuple, Any + + +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: 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) + diff --git a/pyobs/modules/pointing/guidingstatistics/pixeloffset.py b/pyobs/modules/pointing/guidingstatistics/pixeloffset.py new file mode 100644 index 00000000..ce688b77 --- /dev/null +++ b/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.") + + + diff --git a/pyobs/modules/pointing/guidingstatistics/uptime.py b/pyobs/modules/pointing/guidingstatistics/uptime.py new file mode 100644 index 00000000..4ff7b2f7 --- /dev/null +++ b/pyobs/modules/pointing/guidingstatistics/uptime.py @@ -0,0 +1,54 @@ +from datetime import datetime +from typing import List, Dict, Tuple, Any + +from pyobs.modules.pointing.guidingstatistics.guidingstatistics import GuidingStatistics + + +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 + + +__all__ = ['GuidingStatisticsUptime'] \ No newline at end of file diff --git a/tests/modules/pointing/test_guiding_statistic_uptime.py b/tests/modules/pointing/test_guiding_statistic_uptime.py index 50a418cc..95167587 100644 --- a/tests/modules/pointing/test_guiding_statistic_uptime.py +++ b/tests/modules/pointing/test_guiding_statistic_uptime.py @@ -1,12 +1,12 @@ import pytest from datetime import datetime -from pyobs.modules.pointing._baseguiding import _GuidingStatisticsUptime +from pyobs.modules.pointing.guidingstatistics import GuidingStatisticsUptime -def test_end_to_end(): +def test_end_to_end() -> None: client = "camera" - statistic = _GuidingStatisticsUptime() + statistic = GuidingStatisticsUptime() statistic.init_stats(client) @@ -17,10 +17,10 @@ def test_end_to_end(): assert header["GUIDING UPTIME"][0] == 100.0 -def test_calc_uptime_percentage(): +def test_calc_uptime_percentage() -> None: states = [ (True, datetime.fromtimestamp(100)), (False, datetime.fromtimestamp(110)), (None, datetime.fromtimestamp(120)), ] - assert _GuidingStatisticsUptime()._calc_uptime_percentage(states) == 50 + assert GuidingStatisticsUptime()._calc_uptime_percentage(states) == 50 diff --git a/tests/modules/pointing/test_guiding_statistics_pixel_offset.py b/tests/modules/pointing/test_guiding_statistics_pixel_offset.py index b41cfe2a..4427968d 100644 --- a/tests/modules/pointing/test_guiding_statistics_pixel_offset.py +++ b/tests/modules/pointing/test_guiding_statistics_pixel_offset.py @@ -2,7 +2,7 @@ from pyobs.images import Image from pyobs.images.meta import PixelOffsets, SkyOffsets -from pyobs.modules.pointing._baseguiding import _GuidingStatisticsPixelOffset +from pyobs.modules.pointing.guidingstatistics import GuidingStatisticsPixelOffset @pytest.fixture() @@ -14,7 +14,7 @@ def mock_meta_image(): def test_end_to_end(mock_meta_image): client = "camera" - statistic = _GuidingStatisticsPixelOffset() + statistic = GuidingStatisticsPixelOffset() statistic.init_stats(client) @@ -29,13 +29,13 @@ def test_end_to_end(mock_meta_image): def test_build_header_to_few_values(): - gspo = _GuidingStatisticsPixelOffset() + gspo = GuidingStatisticsPixelOffset() assert gspo._build_header([(1.0, 1.0)]) == {} def test_get_session_data(): image = Image() - gspo = _GuidingStatisticsPixelOffset() + gspo = GuidingStatisticsPixelOffset() with pytest.raises(KeyError): gspo._get_session_data(image) From 2663c631fef8c83d6acc4caaa48c8b3ffd0c801b Mon Sep 17 00:00:00 2001 From: germanhydrogen Date: Mon, 29 Jan 2024 16:46:14 +0100 Subject: [PATCH 2/5] Added skyoffset guidingstatistic class --- .../pointing/guidingstatistics/__init__.py | 1 + .../pointing/guidingstatistics/skyoffsets.py | 50 +++++++++++++++++++ .../test_guiding_statistics_pixel_offset.py | 8 +-- .../test_guiding_statistics_sky_offset.py | 47 +++++++++++++++++ 4 files changed, 102 insertions(+), 4 deletions(-) create mode 100644 pyobs/modules/pointing/guidingstatistics/skyoffsets.py create mode 100644 tests/modules/pointing/test_guiding_statistics_sky_offset.py diff --git a/pyobs/modules/pointing/guidingstatistics/__init__.py b/pyobs/modules/pointing/guidingstatistics/__init__.py index 018262ab..91946b5b 100644 --- a/pyobs/modules/pointing/guidingstatistics/__init__.py +++ b/pyobs/modules/pointing/guidingstatistics/__init__.py @@ -1,2 +1,3 @@ from .uptime import GuidingStatisticsUptime from .pixeloffset import GuidingStatisticsPixelOffset +from .skyoffsets import GuidingStatisticsSkyOffset \ No newline at end of file diff --git a/pyobs/modules/pointing/guidingstatistics/skyoffsets.py b/pyobs/modules/pointing/guidingstatistics/skyoffsets.py new file mode 100644 index 00000000..1817119e --- /dev/null +++ b/pyobs/modules/pointing/guidingstatistics/skyoffsets.py @@ -0,0 +1,50 @@ +import logging +from typing import List, Dict, Tuple, Any, Optional + +import numpy as np + +from pyobs.images import Image +from pyobs.images.meta import SkyOffsets +from .guidingstatistics import GuidingStatistics + +log = logging.getLogger(__name__) + + +class GuidingStatisticsSkyOffset(GuidingStatistics): + @staticmethod + def _calc_rms(data: List[float]) -> Optional[float]: + """ + Calculates RMS of data. + + Args: + data: Data to calculate RMS for. + + Returns: + Tuple of RMS. + """ + if len(data) < 3: + return None + + data_len = len(data) + rms = np.sqrt(np.sum(np.power(data, 2)) / data_len) + return float(rms) + + def _build_header(self, data: List[float]) -> Dict[str, Tuple[Any, str]]: + header = {} + rms = self._calc_rms(data) + + if rms is not None: + header["GUIDING OS RMS"] = (float(rms), "RMS for guiding on sky") + + return header + + def _get_session_data(self, data: Image) -> float: + if data.has_meta(SkyOffsets): + sky_offset = data.get_meta(SkyOffsets) + return float(sky_offset.separation().deg) + else: + log.warning("Image is missing the necessary meta information!") + raise KeyError("Unknown meta.") + + + diff --git a/tests/modules/pointing/test_guiding_statistics_pixel_offset.py b/tests/modules/pointing/test_guiding_statistics_pixel_offset.py index 4427968d..c57965f0 100644 --- a/tests/modules/pointing/test_guiding_statistics_pixel_offset.py +++ b/tests/modules/pointing/test_guiding_statistics_pixel_offset.py @@ -6,13 +6,13 @@ @pytest.fixture() -def mock_meta_image(): +def mock_meta_image() -> Image: image = Image() image.set_meta(PixelOffsets(1.0, 1.0)) return image -def test_end_to_end(mock_meta_image): +def test_end_to_end(mock_meta_image) -> None: client = "camera" statistic = GuidingStatisticsPixelOffset() @@ -28,12 +28,12 @@ def test_end_to_end(mock_meta_image): assert header["GUIDING RMS2"] == (1.0, "RMS for guiding on axis 2") -def test_build_header_to_few_values(): +def test_build_header_to_few_values() -> None: gspo = GuidingStatisticsPixelOffset() assert gspo._build_header([(1.0, 1.0)]) == {} -def test_get_session_data(): +def test_get_session_data() -> None: image = Image() gspo = GuidingStatisticsPixelOffset() diff --git a/tests/modules/pointing/test_guiding_statistics_sky_offset.py b/tests/modules/pointing/test_guiding_statistics_sky_offset.py new file mode 100644 index 00000000..048c677d --- /dev/null +++ b/tests/modules/pointing/test_guiding_statistics_sky_offset.py @@ -0,0 +1,47 @@ +import numpy as np +import pytest +from astropy.coordinates import SkyCoord + +from pyobs.images import Image +from pyobs.images.meta import SkyOffsets +from pyobs.modules.pointing.guidingstatistics import GuidingStatisticsSkyOffset + + +@pytest.fixture() +def mock_meta_image() -> Image: + image = Image() + + coord0 = SkyCoord(alt=90.0, az=0.0, unit="degree", frame="altaz") + coord1 = SkyCoord(alt=80.0, az=0.0, unit="degree", frame="altaz") + + image.set_meta(SkyOffsets(coord0, coord1)) + return image + + +def test_end_to_end(mock_meta_image) -> None: + client = "camera" + statistic = GuidingStatisticsSkyOffset() + + statistic.init_stats(client) + + statistic.add_data(mock_meta_image) + statistic.add_data(mock_meta_image) + statistic.add_data(mock_meta_image) + + header = statistic.add_to_header(client, {}) + + np.testing.assert_almost_equal(header["GUIDING OS RMS"][0], 10.0) + assert header["GUIDING OS RMS"][1] == "RMS for guiding on sky" + + +def test_build_header_to_few_values() -> None: + guiding_stat = GuidingStatisticsSkyOffset() + assert guiding_stat._build_header([1.0]) == {} + + +def test_get_session_data() -> None: + image = Image() + guiding_stat = GuidingStatisticsSkyOffset() + + with pytest.raises(KeyError): + guiding_stat._get_session_data(image) From 5da6793796137f8ba215f79707ba064af877374d Mon Sep 17 00:00:00 2001 From: germanhydrogen Date: Mon, 29 Jan 2024 16:54:44 +0100 Subject: [PATCH 3/5] Added optional parameter to _BaseGuiding to use a custom guiding statistic --- pyobs/modules/pointing/_baseguiding.py | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/pyobs/modules/pointing/_baseguiding.py b/pyobs/modules/pointing/_baseguiding.py index c32bf1f4..67660660 100644 --- a/pyobs/modules/pointing/_baseguiding.py +++ b/pyobs/modules/pointing/_baseguiding.py @@ -1,18 +1,16 @@ 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 log = logging.getLogger(__name__) @@ -32,6 +30,7 @@ def __init__( pid: bool = False, reset_at_focus: bool = True, reset_at_filter: bool = True, + guiding_statistic: Optional[GuidingStatistics] = None, **kwargs: Any, ): """Initializes a new science frame auto guiding system. @@ -63,8 +62,11 @@ def __init__( self._ref_header = None # stats - self._statistics = _GuidingStatisticsPixelOffset() - self._uptime = _GuidingStatisticsUptime() + if guiding_statistic is None: + guiding_statistic = GuidingStatisticsPixelOffset() + + self._statistics = guiding_statistic + self._uptime = GuidingStatisticsUptime() async def start(self, **kwargs: Any) -> None: """Starts/resets auto-guiding.""" From 73eba0c4f5bd675c18f3f6c2ad817550b0d37063 Mon Sep 17 00:00:00 2001 From: GermanHydrogen Date: Tue, 30 Jan 2024 10:52:42 +0100 Subject: [PATCH 4/5] Added dummy skyoffsets for testing --- pyobs/images/processors/offsets/__init__.py | 2 ++ .../processors/offsets/dummyskyoffsets.py | 24 ++++++++++++++++++ .../offsets/test_dummyskyoffsets.py | 25 +++++++++++++++++++ 3 files changed, 51 insertions(+) create mode 100644 pyobs/images/processors/offsets/dummyskyoffsets.py create mode 100644 tests/images/processors/offsets/test_dummyskyoffsets.py diff --git a/pyobs/images/processors/offsets/__init__.py b/pyobs/images/processors/offsets/__init__.py index e4fd968c..1ff322cf 100644 --- a/pyobs/images/processors/offsets/__init__.py +++ b/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" ] diff --git a/pyobs/images/processors/offsets/dummyskyoffsets.py b/pyobs/images/processors/offsets/dummyskyoffsets.py new file mode 100644 index 00000000..d0a86bf2 --- /dev/null +++ b/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"] diff --git a/tests/images/processors/offsets/test_dummyskyoffsets.py b/tests/images/processors/offsets/test_dummyskyoffsets.py new file mode 100644 index 00000000..57bb6555 --- /dev/null +++ b/tests/images/processors/offsets/test_dummyskyoffsets.py @@ -0,0 +1,25 @@ +import pytest +from astropy.coordinates import SkyCoord + +from pyobs.images import Image +from pyobs.images.meta import SkyOffsets +from pyobs.images.processors.offsets import DummySkyOffsets + + +@pytest.mark.asyncio +async def test_call(): + coord0 = SkyCoord(alt=90.0, az=0.0, unit="degree", frame="altaz") + coord1 = SkyCoord(alt=80.0, az=0.0, unit="degree", frame="altaz") + + offsets = DummySkyOffsets(coord0, coord1) + image = Image() + + result_image = await offsets(image) + + result_offset = result_image.get_meta(SkyOffsets) + + assert result_offset.coord0.alt.deg == 90.0 + assert result_offset.coord0.az.deg == 0.0 + + assert result_offset.coord1.alt.deg == 80.0 + assert result_offset.coord1.az.deg == 0.0 From 68b412bb038bf7a3a308f330fc93cf1aa893be5a Mon Sep 17 00:00:00 2001 From: GermanHydrogen Date: Tue, 30 Jan 2024 10:52:55 +0100 Subject: [PATCH 5/5] Fixed guiding config --- pyobs/modules/pointing/_baseguiding.py | 5 +++-- pyobs/modules/pointing/guidingstatistics/__init__.py | 2 +- .../modules/pointing/guidingstatistics/guidingstatistics.py | 6 ++++-- pyobs/modules/pointing/guidingstatistics/skyoffsets.py | 2 +- 4 files changed, 9 insertions(+), 6 deletions(-) diff --git a/pyobs/modules/pointing/_baseguiding.py b/pyobs/modules/pointing/_baseguiding.py index 67660660..86d7e0e7 100644 --- a/pyobs/modules/pointing/_baseguiding.py +++ b/pyobs/modules/pointing/_baseguiding.py @@ -12,6 +12,7 @@ from .guidingstatistics import GuidingStatisticsUptime, GuidingStatisticsPixelOffset from .guidingstatistics.guidingstatistics import GuidingStatistics from ...interfaces import ITelescope +from ...object import get_object log = logging.getLogger(__name__) @@ -30,7 +31,7 @@ def __init__( pid: bool = False, reset_at_focus: bool = True, reset_at_filter: bool = True, - guiding_statistic: Optional[GuidingStatistics] = None, + guiding_statistic: Optional[Union[Dict[str, Any], GuidingStatistics]] = None, **kwargs: Any, ): """Initializes a new science frame auto guiding system. @@ -65,7 +66,7 @@ def __init__( if guiding_statistic is None: guiding_statistic = GuidingStatisticsPixelOffset() - self._statistics = guiding_statistic + self._statistics = get_object(guiding_statistic, GuidingStatistics) self._uptime = GuidingStatisticsUptime() async def start(self, **kwargs: Any) -> None: diff --git a/pyobs/modules/pointing/guidingstatistics/__init__.py b/pyobs/modules/pointing/guidingstatistics/__init__.py index 91946b5b..233c391b 100644 --- a/pyobs/modules/pointing/guidingstatistics/__init__.py +++ b/pyobs/modules/pointing/guidingstatistics/__init__.py @@ -1,3 +1,3 @@ from .uptime import GuidingStatisticsUptime from .pixeloffset import GuidingStatisticsPixelOffset -from .skyoffsets import GuidingStatisticsSkyOffset \ No newline at end of file +from .skyoffsets import GuidingStatisticsSkyOffset diff --git a/pyobs/modules/pointing/guidingstatistics/guidingstatistics.py b/pyobs/modules/pointing/guidingstatistics/guidingstatistics.py index 7dc851fc..7320a0e7 100644 --- a/pyobs/modules/pointing/guidingstatistics/guidingstatistics.py +++ b/pyobs/modules/pointing/guidingstatistics/guidingstatistics.py @@ -2,6 +2,8 @@ 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.""" @@ -42,10 +44,10 @@ def add_to_header(self, client: str, header: Dict[str, Tuple[Any, str]]) -> Dict return header | session_header @abstractmethod - def _get_session_data(self, input_data: Any) -> Any: + def _get_session_data(self, input_data: Image) -> Any: raise NotImplementedError - def add_data(self, input_data: Any) -> None: + def add_data(self, input_data: Image) -> None: """ Adds data to all client measurement sessions. Args: diff --git a/pyobs/modules/pointing/guidingstatistics/skyoffsets.py b/pyobs/modules/pointing/guidingstatistics/skyoffsets.py index 1817119e..b4d5fde5 100644 --- a/pyobs/modules/pointing/guidingstatistics/skyoffsets.py +++ b/pyobs/modules/pointing/guidingstatistics/skyoffsets.py @@ -34,7 +34,7 @@ def _build_header(self, data: List[float]) -> Dict[str, Tuple[Any, str]]: rms = self._calc_rms(data) if rms is not None: - header["GUIDING OS RMS"] = (float(rms), "RMS for guiding on sky") + header["GUIDING RMS"] = (float(rms), "RMS for guiding on sky") return header