merge with weather

docs/source/configurations.rst

@@ -73,11 +73,16 @@ Many of these variables are dense, so they are given default values, but explain
     :type name: float
     :param cloud_extinction: Rate clouds appear
     :type name: float
+    :param weather_sim: Include a rudimentary weather simulation based on historical data
+    :type name: boolean 
+    :param weather_config: Setting for the `Weather` engine class
+    :type name: dictionary
 
 .. code-block:: yaml
 
     seeing: 0.9
     cloud_extinction: 0.0
+    weather_sim: False
 
 
 .. attribute:: Camera

docs/source/survey.rst

@@ -5,4 +5,8 @@ Survey
 
 
 .. autoclass:: DeepSurveySim.Survey.ObservationVariables
+    :members:
+
+
+.. autoclass:: telescope_positioning_simulation.Survey.Weather
     :members:

telescope_positioning_simulation/Survey/__init__.py

@@ -2,6 +2,7 @@
 from telescope_positioning_simulation.Survey.observation_variables import (
     ObservationVariables,
 )
+from telescope_positioning_simulation.Survey.weather import Weather
 from telescope_positioning_simulation.Survey.cummulative_survey import (
     UniformSurvey,
     LowVisiblitySurvey,

telescope_positioning_simulation/Survey/observation_variables.py

@@ -83,7 +83,9 @@ def __init__(self, observator_configuration: dict):
 
         self.seeing = observator_configuration["seeing"]
         self.clouds = observator_configuration["cloud_extinction"]
-
+        if observator_configuration['weather_sim']: 
+            self._init_weather(observator_configuration['weather_config'])
+
         self.optics_fwhm = observator_configuration["fwhm"]
 
         self.default_locations = self._default_locations(
@@ -103,27 +105,34 @@ def __init__(self, observator_configuration: dict):
         self.readout_seconds = observator_configuration["readout_seconds"]
 
     def _init_skybright(self, skybright_config):
+        from configparser import ConfigParser
+
         try:
             from skybright import skybright
-            from configparser import ConfigParser
-
-            default_config_path = f"{os.path.dirname(__file__).rstrip('/')}/../settings/skybright_config.conf"
-
-            config_path = (
-                default_config_path
-                if skybright_config["config"] == "default"
-                else skybright_config["config"]
-            )
-            skybright_config_file = ConfigParser()
-            skybright_config_file.read(config_path)
-
-            self.skybright = skybright.MoonSkyModel(skybright_config_file)
 
         except ModuleNotFoundError:
-            print(
+            raise ModuleNotFoundError(
                 "ERROR: skybright module not found, please install it from https://github.com/ehneilsen/skybright.git"
             )
 
+        default_config_path = (
+            f"{os.path.dirname(__file__).rstrip('/')}/../settings/skybright_config.conf"
+        )
+
+        config_path = (
+            default_config_path
+            if skybright_config["config"] == "default"
+            else skybright_config["config"]
+        )
+        skybright_config_file = ConfigParser()
+        skybright_config_file.read(config_path)
+
+        self.skybright = skybright.MoonSkyModel(skybright_config_file)
+
+    def _init_weather(self, weather_config): 
+        from telescope_positioning_simulation.Survey import Weather
+        self.weather = Weather(base_seeing=self.seeing, base_clouds=self.clouds, **weather_config)
+
     def update(
         self,
         time: Union[float, list[float]],
@@ -157,6 +166,11 @@ def update(
         self.band = band if band is not None else self.band
         self.location = location
 
+        if hasattr(self, "weather"): 
+            conditions = self.weather.condition(self.time)
+            self.seeing = self.weather.seeing(conditions)
+            self.clouds = self.weather.clouds(conditions)
+
     def _angular_distance(self, location):
         ra1, ra2 = np.array(
             (self.location.ra.value * np.pi / 180) + 10**-9

telescope_positioning_simulation/Survey/weather.py

@@ -0,0 +1,124 @@
+import pandas as pd
+from astropy.time import Time
+
+
+class Weather:
+    def __init__(
+        self,
+        weather_source_file: str,
+        seeing_tolerance: float = 0.5,
+        cloud_tolerance: float = 0.5,
+        base_seeing: float = 0.9,
+        base_clouds: float = 0,
+        csv_configuration: dict = {},
+        **kwargs
+    ) -> None:
+        """Run a deteriministic weather simulation based on historical data. 
+        Assumes all weather conditions are a function of the month. 
+        Update either "clouds" or "seeing" based on given tolerances. 
+
+        Args:
+            weather_source_file (str): Path to csv file containing historical weather data. 
+            seeing_tolerance (float, optional): Allowed percent of cloudiness before seeing is impacted. Defaults to 0.5.
+            cloud_tolerance (float, optional): Allowed percent of cloudiness before clouds are set to full extiniction. Defaults to 0.5.
+            base_seeing (float, optional): Best allowed seeing conditions. Defaults to 0.9.
+            base_clouds (float, optional): Best allowed cloud conditions. Defaults to 0.
+            csv_configuration (dict, optional): Instructions on how to read the csv, add in any parameters to the configuration dictionary with this argument. Defaults to {"seeing": "HourlySkyConditions","date": "DATE", "allowed_conditions": ["FEW", "CLR"]}
+        }
+.
+        """
+
+        csv_configuration = {**csv_configuration, **self._default_configuration()}
+        self.seeing_name = csv_configuration["seeing"]
+        self.date_name = csv_configuration["date"]
+
+        self.seeing_tolerance = seeing_tolerance
+        self.clouds_tolerance = cloud_tolerance
+
+        self.reference_seeing = base_seeing
+        self.reference_clouds = base_clouds
+
+        self.weather_source = pd.read_csv(weather_source_file)[
+            [self.seeing_name, self.date_name]
+        ]
+        self._format_source(csv_configuration)
+
+    def _default_configuration(self):
+        return {
+            "seeing": "HourlySkyConditions",
+            "date": "DATE",
+            "allowed_conditions": ["FEW", "CLR"],
+        }
+
+    def _format_source(self, configuration):
+
+        self.weather_source.dropna(inplace=True)
+
+        allowed = "|".join(configuration["allowed_conditions"])
+
+        self.weather_source[self.seeing_name] = self.weather_source[
+            self.seeing_name].str.contains(allowed).astype(int)
+
+        self.weather_source[self.date_name] = pd.to_datetime(
+            self.weather_source[self.date_name], infer_datetime_format=True
+        )
+
+    def _find_date(self, mjd):
+        date = Time(mjd, format="mjd")
+        month = date.datetime.month
+        day = date.datetime.day
+        return month, day
+
+    def condition(self, mjd):
+        """
+        Return the sky conditions for all data with the same month as the supplied mjd. 
+
+        Args:
+            mjd (Union(int, float)): Date in MJD to get sky conditions for. 
+
+        Returns:
+           pd.Series: Conditions of the sky with the same month as the supplied mjd. 
+        """
+        month, day = self._find_date(mjd)
+        matching = self.weather_source[
+            self.weather_source[self.date_name].dt.month == month
+        ]
+        matching = matching[matching[self.date_name].dt.day.isin([day, day+1, day+2])]
+        return matching
+
+    def seeing(self, condition):
+        """ 
+        Approximate seeing conditions. If the seeing condition is above the tolerance level, seeing is scaled by the percent removed from perfect seeing
+
+        Args:
+            condition (pd.Series): Series to arggergate into a seeing score
+
+        Returns:
+            float: Updated seeing conditions
+        """
+        seeing_conditions = 1 - condition[self.seeing_name].mean()
+
+        if seeing_conditions >= self.seeing_tolerance:
+            seeing = self.reference_seeing * (1 - round(seeing_conditions, 1))
+        else:
+            seeing = self.reference_seeing
+
+        return seeing
+
+    def clouds(self, condition):
+        """
+        Approximate cloud extiction. If the cloud conditions are above the tolerance levels, clouds are set to 1. Else returns them to base levels. 
+
+        Args:
+            condition (pd.Series): Series to arggergate into a cloud score
+
+        Returns:
+            float: clouds for the passed condition
+        """
+        cloud_condition = 1 -  condition[self.seeing_name].mean()
+        if cloud_condition >= self.clouds_tolerance:
+            clouds = 1
+        else:
+            clouds = self.reference_clouds
+
+        return clouds