diff --git a/pocs/scheduler/observation/__init__.py b/pocs/scheduler/observation/__init__.py new file mode 100644 index 000000000..a7c8e0aaf --- /dev/null +++ b/pocs/scheduler/observation/__init__.py @@ -0,0 +1 @@ +from pocs.scheduler.observation.base import Observation # pragma: no flakes diff --git a/pocs/scheduler/observation.py b/pocs/scheduler/observation/base.py similarity index 100% rename from pocs/scheduler/observation.py rename to pocs/scheduler/observation/base.py diff --git a/pocs/scheduler/observation/dithered.py b/pocs/scheduler/observation/dithered.py new file mode 100644 index 000000000..e69804802 --- /dev/null +++ b/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) diff --git a/pocs/tests/test_dithered_observation.py b/pocs/tests/test_dithered_observation.py new file mode 100644 index 000000000..4e64bfe54 --- /dev/null +++ b/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