Source code for rubin_scheduler.scheduler.surveys.too_surveys

__all__ = ("TooMaster", "TooSurvey")

import copy

import numpy as np

from rubin_scheduler.scheduler.surveys import BaseSurvey, BlobSurvey


[docs] class TooMaster(BaseSurvey): """ A target of opportunity class. Every time a new ToO comes in, it will spawn a new sub-survey. Parameters ---------- example__to_o_survey : rubin_scheduler.scheduler.surveys.ToO_survey A survey object that will be coppied and have a new target map set for each incoming ToO. """ def __init__(self, example__to_o_survey): self.example__to_o_survey = example__to_o_survey self.surveys = [] self.highest_reward = -np.inf self.scheduled_obs = None def add_observation(self, observation, indx=None): if len(self.surveys) > 0: for survey in self.surveys: survey.add_observation(observation, indx=indx) def _spawn_new_survey(self, too): """Create a new survey object for a ToO we haven't seen before. Parameters ---------- too : rubin_scheduler.scheduler.utils.TargetoO object """ new_survey = copy.deepcopy(self.example__to_o_survey) new_survey.set_id(too.id) new_survey.set_target_map(too.footprint) return new_survey def _check_survey_list(self, conditions): """There is a current ToO in the conditions.""" running_ids = [survey.too_id for survey in self.surveys] current_ids = [too.id for too in conditions.targets_of_opportunity] # delete any ToO surveys that are no longer relevant self.surveys = [survey for survey in self.surveys if survey.too_id in current_ids] # Spawn new surveys that are needed new_surveys = [] for too in conditions.targets_of_opportunity: if too.id not in running_ids: new_surveys.append(self._spawn_new_survey(too)) self.surveys.extend(new_surveys)
[docs] def calc_reward_function(self, conditions): # Catch if a new ToO has happened if conditions.targets_of_opportunity is not None: self._check_survey_list(conditions) if len(self.surveys) > 0: rewards = [np.nanmax(survey.calc_reward_function(conditions)) for survey in self.surveys] self.reward = np.nanmax(rewards) self.highest_reward = np.min(np.where(rewards == self.reward)) else: self.reward = -np.inf self.highest_reward = None return self.reward
def generate_observations(self, conditions): if self.reward > -np.inf: result = self.surveys[self.highest_reward].generate_observations(conditions) return result
[docs] class TooSurvey(BlobSurvey): """Survey class to catch incoming target of opportunity anouncements and try to observe them. The idea is that we can dynamically update the target footprint basis function, and add new features as more ToOs come in. Parameters ---------- too_id : int (None) A unique integer ID for the ToO getting observed """ def __init__( self, basis_functions, basis_weights, filtername1="r", filtername2=None, slew_approx=7.5, filter_change_approx=140.0, read_approx=2.4, exptime=30.0, nexp=2, ideal_pair_time=22.0, min_pair_time=15.0, search_radius=30.0, alt_max=85.0, az_range=180.0, flush_time=30.0, smoothing_kernel=None, nside=None, dither=True, seed=42, ignore_obs=None, scheduler_note="ToO", detailers=None, camera="LSST", too_id=None, survey_name=None, ): super(TooSurvey, self).__init__( basis_functions=basis_functions, basis_weights=basis_weights, filtername1=filtername1, filtername2=filtername2, slew_approx=slew_approx, filter_change_approx=filter_change_approx, read_approx=read_approx, exptime=exptime, nexp=nexp, ideal_pair_time=ideal_pair_time, min_pair_time=min_pair_time, search_radius=search_radius, alt_max=alt_max, az_range=az_range, flush_time=flush_time, smoothing_kernel=smoothing_kernel, nside=nside, dither=dither, seed=seed, ignore_obs=ignore_obs, scheduler_note=scheduler_note, detailers=detailers, camera=camera, survey_name=survey_name, ) # Include the ToO id in the note self.scheduler_note_base = self.scheduler_note if self.survey_name is None: self.survey_name = self.scheduler_note self.set_id(too_id)
[docs] def set_id(self, newid): """Set the id""" self.too_id = newid self.scheduler_note = self.scheduler_note_base + ", " + str(newid)
[docs] def set_target_map(self, newmap): """ Expect one of the basis functions to be Footprint_nvis_basis_function """ for basis_func in self.basis_functions: if hasattr(basis_func, "footprint"): basis_func.footprint = newmap
[docs] def generate_observations_rough(self, conditions): # Always spin the tesselation before generating a new block. if self.dither: self._spin_fields(conditions) result = super(TooSurvey, self).generate_observations_rough(conditions) return result