make it possible to replace seeing, clouds, downtime in ModelObservatory

rubin_sim/scheduler/model_observatory/model_observatory.py

@@ -63,6 +63,9 @@ def __init__(
         kinem_model=None,
         cloud_db=None,
         cloud_offset_year=0,
+        cloud_data=None,
+        seeing_data=None,
+        downtimes=None,
     ):
         """
         Parameters
@@ -95,6 +98,15 @@ def __init__(
             The file to use for clouds. Default of None uses the database in rubin_sim_data.
         cloud_offset_year : 0
             The year offset to be passed to CloudData.
+        cloud_data : None
+            If one wants to replace the default cloud data. Should be an object with a
+            __call__ method that takes MJD and returns cloudy level.
+        seeing_data : None
+            If one wants to replace the default seeing_data object. Should be an object with a
+            __call__ method that takes MJD and returns zenith fwhm_500 in arcsec.
+        downtimes : None
+            If one wants to replace the default downtimes. Should be a np.array with columns
+            of "start" and "end" with MJD values and should include both scheduled and unscheduled downtime
         """
 
         if nside is None:
@@ -121,47 +133,52 @@ def __init__(
 
         mjd_start_time = Time(self.mjd_start, format="mjd")
         # Downtime
-        self.down_nights = []
-        self.sched_downtime_data = ScheduledDowntimeData(mjd_start_time)
-        self.unsched_downtime_data = UnscheduledDowntimeData(mjd_start_time)
-
-        sched_downtimes = self.sched_downtime_data()
-        unsched_downtimes = self.unsched_downtime_data()
-
-        down_starts = []
-        down_ends = []
-        for dt in sched_downtimes:
-            down_starts.append(dt["start"].mjd)
-            down_ends.append(dt["end"].mjd)
-        if not ideal_conditions:
-            for dt in unsched_downtimes:
+        if downtimes is None:
+            self.down_nights = []
+            self.sched_downtime_data = ScheduledDowntimeData(mjd_start_time)
+            self.unsched_downtime_data = UnscheduledDowntimeData(mjd_start_time)
+
+            sched_downtimes = self.sched_downtime_data()
+            unsched_downtimes = self.unsched_downtime_data()
+
+            down_starts = []
+            down_ends = []
+            for dt in sched_downtimes:
                 down_starts.append(dt["start"].mjd)
                 down_ends.append(dt["end"].mjd)
-
-        self.downtimes = np.array(
-            list(zip(down_starts, down_ends)),
-            dtype=list(zip(["start", "end"], [float, float])),
-        )
-        self.downtimes.sort(order="start")
-
-        # Make sure there aren't any overlapping downtimes
-        diff = self.downtimes["start"][1:] - self.downtimes["end"][0:-1]
-        while np.min(diff) < 0:
-            # Should be able to do this wihtout a loop, but this works
-            for i, dt in enumerate(self.downtimes[0:-1]):
-                if self.downtimes["start"][i + 1] < dt["end"]:
-                    new_end = np.max([dt["end"], self.downtimes["end"][i + 1]])
-                    self.downtimes[i]["end"] = new_end
-                    self.downtimes[i + 1]["end"] = new_end
-
-            good = np.where(
-                self.downtimes["end"] - np.roll(self.downtimes["end"], 1) != 0
+            if not ideal_conditions:
+                for dt in unsched_downtimes:
+                    down_starts.append(dt["start"].mjd)
+                    down_ends.append(dt["end"].mjd)
+
+            self.downtimes = np.array(
+                list(zip(down_starts, down_ends)),
+                dtype=list(zip(["start", "end"], [float, float])),
             )
-            self.downtimes = self.downtimes[good]
+            self.downtimes.sort(order="start")
+
+            # Make sure there aren't any overlapping downtimes
             diff = self.downtimes["start"][1:] - self.downtimes["end"][0:-1]
+            while np.min(diff) < 0:
+                # Should be able to do this wihtout a loop, but this works
+                for i, dt in enumerate(self.downtimes[0:-1]):
+                    if self.downtimes["start"][i + 1] < dt["end"]:
+                        new_end = np.max([dt["end"], self.downtimes["end"][i + 1]])
+                        self.downtimes[i]["end"] = new_end
+                        self.downtimes[i + 1]["end"] = new_end
+
+                good = np.where(
+                    self.downtimes["end"] - np.roll(self.downtimes["end"], 1) != 0
+                )
+                self.downtimes = self.downtimes[good]
+                diff = self.downtimes["start"][1:] - self.downtimes["end"][0:-1]
+        else:
+            self.downtimes = downtimes
 
         if ideal_conditions:
             self.seeing_data = NominalSeeing()
+        elif seeing_data is not None:
+            self.seeing_data = seeing_data
         else:
             self.seeing_data = SeeingData(mjd_start_time, seeing_db=seeing_db)
         self.seeing_model = SeeingModel()
@@ -171,6 +188,8 @@ def __init__(
 
         if ideal_conditions:
             self.cloud_data = NoClouds()
+        elif cloud_data is not None:
+            self.cloud_data = cloud_data
         else:
             self.cloud_data = CloudData(
                 mjd_start_time, cloud_db=cloud_db, offset_year=cloud_offset_year

tests/scheduler/test_modelobs.py

@@ -0,0 +1,58 @@
+import numpy as np
+import unittest
+
+from rubin_sim.scheduler.model_observatory import ModelObservatory
+
+
+class KindaClouds(object):
+    """Dummy class that always sets the clouds level to a unique float"""
+
+    def __call__(self, mjd):
+        return 0.23659
+
+
+class ArbSeeing(object):
+    """Dummy class to always return a specific seeing value"""
+
+    def __call__(self, mjd):
+        fwhm_500 = 1.756978
+        return fwhm_500
+
+
+class TestModelObservatory(unittest.TestCase):
+    def test_replace(self):
+        """test that we can replace default downtimes, seeing, and clouds"""
+
+        mjd_start = 60796.0
+        mo_default = ModelObservatory(mjd_start=mjd_start)
+        cond_default = mo_default.return_conditions()
+
+        # Define new downtimes
+        downtimes = np.zeros(2, dtype=list(zip(["start", "end"], [float, float])))
+        downtimes["start"] = np.array([1, 10]) + mjd_start
+        downtimes["end"] = np.array([2, 11]) + mjd_start
+
+        seeing_data = ArbSeeing()
+        cloud_data = KindaClouds()
+
+        mo_new = ModelObservatory(
+            mjd_start=mjd_start,
+            seeing_data=seeing_data,
+            cloud_data=cloud_data,
+            downtimes=downtimes,
+        )
+        cond_new = mo_new.return_conditions()
+
+        # Make sure the internal downtimes are different
+        assert np.all(mo_default.downtimes != mo_new.downtimes)
+
+        # Make sure seeing is not the same
+        diff = cond_default.fwhm_eff["r"] - cond_new.fwhm_eff["r"]
+        assert np.nanmin(np.abs(diff)) > 0
+
+        # Make sure cloudyness level is not the same
+        assert cond_default.bulk_cloud != cond_new.bulk_cloud
+
+
+if __name__ == "__main__":
+    unittest.main()