Add a DitheredObservation type - WIP

pocs/scheduler/observation/__init__.py

@@ -0,0 +1 @@
+from pocs.scheduler.observation.base import Observation  # pragma: no flakes

pocs/scheduler/observation/dithered.py

@@ -0,0 +1,81 @@
+from astropy import units as u
+
+from contextlib import suppress
+from pocs.scheduler.field import Field
+from pocs.scheduler.observation import Observation
+
+from pocs.utils import listify
+
+
+class DitheredObservation(Observation):
+
+    """ Observation that dithers to different points.
+
+    Dithered observations will consist of both multiple exposure time as well as multiple
+    `Field` locations, which are used as a simple dithering mechanism.
+
+    Note:
+        For now the new observation must be created like a normal `Observation`,
+        with one `exp_time` and one `field`. Then use direct property assignment
+        for the list of `exp_time` and `field`. New `field`/`exp_time` combos can
+        more conveniently be set with `add_field`
+    """
+
+    def __init__(self, *args, **kwargs):
+        super(DitheredObservation, self).__init__(*args, **kwargs)
+
+        # Set initial list to original values
+        self._exp_time = listify(self.exp_time)
+        self._field = listify(self.field)
+
+        self.extra_config = kwargs
+
+    @property
+    def exp_time(self):
+        exp_time = self._exp_time[self.exposure_index]
+
+        if not isinstance(exp_time, u.Quantity):
+            exp_time *= u.second
+
+        return exp_time
+
+    @exp_time.setter
+    def exp_time(self, values):
+        assert all(t > 0.0 for t in listify(values)), \
+            self.logger.error("Exposure times (exp_time) must be greater than 0")
+
+        self._exp_time = listify(values)
+
+    @property
+    def field(self):
+        return self._field[self.exposure_index]
+
+    @field.setter
+    def field(self, values):
+        assert all(isinstance(f, Field) for f in listify(values)), \
+            self.logger.error("All fields must be a valid Field instance")
+
+        self._field = listify(values)
+
+    @property
+    def exposure_index(self):
+        _exp_index = 0
+        with suppress(AttributeError):
+            _exp_index = self.current_exp_num % len(self._exp_time)
+
+        return _exp_index
+
+    def add_field(self, new_field, new_exp_time):
+        """ Add a new field to observe along with exposure time
+
+        Args:
+            new_field (pocs.scheduler.field.Field): A `Field` object
+            new_exp_time (float): Number of seconds to expose
+
+        """
+        self.logger.debug("Adding new field {} {}".format(new_field, new_exp_time))
+        self._field.append(new_field)
+        self._exp_time.append(new_exp_time)
+
+    def __str__(self):
+        return "DitheredObservation: {}: {}".format(self._field, self._exp_time)

pocs/tests/test_dithered_observation.py

@@ -0,0 +1,129 @@
+import pytest
+
+from astropy import units as u
+from pocs.scheduler.field import Field
+from pocs.scheduler.observation.dithered import DitheredObservation as Observation
+from pocs.utils import dither
+
+
+@pytest.fixture
+def field():
+    return Field('Test Observation', '20h00m43.7135s +22d42m39.0645s')
+
+
+def test_create_observation_no_field():
+    with pytest.raises(TypeError):
+        Observation()
+
+
+def test_create_observation_bad_field():
+    with pytest.raises(AssertionError):
+        Observation('20h00m43.7135s +22d42m39.0645s')
+
+
+def test_create_observation_exp_time_no_units(field):
+    with pytest.raises(TypeError):
+        Observation(field, exp_time=1.0)
+
+
+def test_create_observation_exp_time_bad(field):
+    with pytest.raises(AssertionError):
+        Observation(field, exp_time=0.0 * u.second)
+
+
+def test_create_observation_exp_time_minutes(field):
+    obs = Observation(field, exp_time=5.0 * u.minute)
+    assert obs.exp_time == 300 * u.second
+
+
+def test_bad_priority(field):
+    with pytest.raises(AssertionError):
+        Observation(field, priority=-1)
+
+
+def test_good_priority(field):
+    obs = Observation(field, priority=5.0)
+    assert obs.priority == 5.0
+
+
+def test_priority_str(field):
+    obs = Observation(field, priority="5")
+    assert obs.priority == 5.0
+
+
+def test_bad_min_set_combo(field):
+    with pytest.raises(AssertionError):
+        Observation(field, exp_set_size=7)
+    with pytest.raises(AssertionError):
+        Observation(field, min_nexp=57)
+
+
+def test_small_sets(field):
+    obs = Observation(field, exp_time=1 * u.second, min_nexp=1, exp_set_size=1)
+    assert obs.minimum_duration == 1 * u.second
+    assert obs.set_duration == 1 * u.second
+
+
+def test_good_min_set_combo(field):
+    obs = Observation(field, min_nexp=21, exp_set_size=3)
+    assert isinstance(obs, Observation)
+
+
+def test_default_min_duration(field):
+    obs = Observation(field)
+    assert obs.minimum_duration == 7200 * u.second
+
+
+def test_default_set_duration(field):
+    obs = Observation(field)
+    assert obs.set_duration == 1200 * u.second
+
+
+def test_print(field):
+    obs = Observation(field, exp_time=17.5 * u.second, min_nexp=27, exp_set_size=9)
+    assert str(obs) == "Test Observation: 17.5 s exposures in blocks of 9, minimum 27, priority 100"
+
+
+def test_seq_time(field):
+    obs = Observation(field, exp_time=17.5 * u.second, min_nexp=27, exp_set_size=9)
+    assert obs.seq_time is None
+
+
+def test_no_exposures(field):
+    obs = Observation(field, exp_time=17.5 * u.second, min_nexp=27, exp_set_size=9)
+    assert obs.first_exposure is None
+    assert obs.last_exposure is None
+    assert obs.pointing_image is None
+
+
+def test_last_exposure_and_reset(field):
+    obs = Observation(field, exp_time=17.5 * u.second, min_nexp=27, exp_set_size=9)
+    status = obs.status()
+    assert status['current_exp'] == obs.current_exp_num
+
+    # Mimic taking exposures
+    obs.merit = 112.5
+
+    for i in range(5):
+        obs.exposure_list['image_{}'.format(i)] = 'full_image_path_{}'.format(i)
+
+    last = obs.last_exposure
+    assert isinstance(last, tuple)
+    assert obs.merit > 0.0
+    assert obs.current_exp_num == 5
+
+    assert last[0] == 'image_4'
+    assert last[1] == 'full_image_path_4'
+
+    assert isinstance(obs.first_exposure, tuple)
+    assert obs.first_exposure[0] == 'image_0'
+    assert obs.first_exposure[1] == 'full_image_path_0'
+
+    obs.reset()
+    status2 = obs.status()
+
+    assert status2['current_exp'] == 0
+    assert status2['merit'] == 0.0
+    assert obs.first_exposure is None
+    assert obs.last_exposure is None
+    assert obs.seq_time is None