Add orchestrator

Dockerfile

@@ -111,7 +111,7 @@ python -c "from ortools.linear_solver import pywraplp as p; p.Solver.CreateSolve
 python -c "from ortools.linear_solver import pywraplp as p; p.Solver.CreateSolver('GLPK')"
 
 # assumption: if it's trying to read the licence, it's probably linked properly
-strace -e openat python -c "from ortools.linear_solver import pywraplp as p; p.Solver.CreateSolver('GUROBI')" 2>&1 | grep -q gurobi.lic
+# strace -e openat python -c "from ortools.linear_solver import pywraplp as p; p.Solver.CreateSolver('GUROBI')" 2>&1 | grep -q gurobi.lic
 EOT
 
 USER app

adaptive_scheduler/configdb_connections.py

@@ -22,7 +22,7 @@ class ConfigDBInterface(SendMetricMixin):
     """
 
     def __init__(self, configdb_url, telescope_classes, telescopes_file='data/telescopes.json',
-                 active_instruments_file='data/active_instruments.json'):
+                 active_instruments_file='data/active_instruments.json', overrides=None):
         self.configdb_url = configdb_url
         if not self.configdb_url.endswith('/'):
             self.configdb_url += '/'
@@ -31,11 +31,28 @@ def __init__(self, configdb_url, telescope_classes, telescopes_file='data/telesc
         self.active_instruments_file = active_instruments_file
         self.active_instruments = None
         self.telescope_info = None
+        self.overrides = overrides
         self.update_configdb_structures()
 
     def update_configdb_structures(self):
         self.update_telescope_info()
+        self.apply_overrides_to_telescopes()
         self.update_active_instruments()
+        self.apply_overrides_to_instruments()
+
+    def apply_overrides_to_telescopes(self):
+        if self.overrides:
+            for telescope in self.overrides.get('telescopes', {}).keys():
+                if telescope in self.telescope_info and 'status' in self.overrides['telescopes'][telescope]:
+                    self.telescope_info[telescope]['status'] = self.overrides['telescopes'][telescope]['status']
+
+    def apply_overrides_to_instruments(self):
+        if self.overrides and self.overrides.get('instruments', {}):
+            for instrument in self.active_instruments:
+                if instrument['code'] in self.overrides['instruments']:
+                    instrument['state'] = self.overrides['instruments'][instrument['code']].get('state', instrument['state'])
+                if instrument['instrument_type']['code'] in self.overrides['instruments']:
+                    instrument['state'] = self.overrides['instruments'][instrument['instrument_type']['code']].get('state', instrument['state'])
 
     def update_active_instruments(self):
         try:

adaptive_scheduler/scheduler.py

@@ -394,7 +394,7 @@ def run_scheduler(self, scheduler_input, estimated_scheduler_end, semester_detai
 
         print_compound_reservations(compound_reservations)
         # Prepare scheduler result
-        scheduler_result = SchedulerResult()
+        scheduler_result = SchedulerResult(input_reservations=compound_reservations)
         scheduler_result.schedule = {}
         scheduler_result.resource_schedules_to_cancel = list(available_resources)
 
@@ -569,14 +569,15 @@ class SchedulerResult(object):
     '''Aggregates together output of a scheduler run
     '''
 
-    def __init__(self, schedule=None, resource_schedules_to_cancel=None):
+    def __init__(self, schedule=None, resource_schedules_to_cancel=None, input_reservations=None):
         '''
         schedule - Expected to be a dict mapping resource to scheduled reservations
         resource_schedules_to_cancel - List of resources to cancel schedules on - this is the list of all available
         resources that have any request scheduled on them. Resources with no requests scheduled on them will be
         removed from the list.
         '''
         self.schedule = schedule if schedule else {}
+        self.input_reservations = input_reservations if input_reservations else []
         self.resource_schedules_to_cancel = resource_schedules_to_cancel if resource_schedules_to_cancel else []
 
     def count_reservations(self):
@@ -626,11 +627,13 @@ def __init__(self, sched_params, scheduler, network_interface, network_model, in
         self.sched_params = sched_params
         self.warm_starts_setting = sched_params.warm_starts
         self.scheduler = scheduler
+        self.normal_scheduler_input = None
+        self.rr_scheduler_input = None
+        self.normal_scheduler_result = None
+        self.rr_scheduler_result = None
         self.network_interface = network_interface
         self.network_model = network_model
         self.input_factory = input_factory
-        self.normal_scheduled_requests_by_rg = {}
-        self.rr_scheduled_requests_by_rg = {}
         self.log = logging.getLogger(__name__)
         # List of strings to be printed in final scheduling summary
         self.summary_events = []
@@ -828,6 +831,10 @@ def _can_apply_scheduler_result(self, scheduler_result, apply_deadline):
             self.log.warn("Empty scheduler result. Schedule will not be saved.")
             return False
 
+        if self.sched_params.simulate_now:
+            # Don't care about deadlines if you are simulating a time in the past
+            return True
+
         estimated_apply_timedelta = self.avg_save_time_per_reservation_timedelta * scheduler_result.count_reservations()
         estimated_apply_completion = datetime.utcnow() + estimated_apply_timedelta
         self.log.info(
@@ -973,7 +980,7 @@ def create_rr_schedule(self, scheduler_input):
             rr_scheduler_result = self.call_scheduler(scheduler_input, deadline)
 
             try:
-                self.rr_scheduled_requests_by_rg = rr_scheduler_result.get_scheduled_requests_by_request_group_id()
+                self.rr_scheduler_result = rr_scheduler_result
                 self.apply_rr_result(rr_scheduler_result, scheduler_input, deadline)
                 rr_scheduling_end = datetime.utcnow()
                 rr_scheduling_timedelta = rr_scheduling_end - rr_scheduling_start
@@ -1019,7 +1026,7 @@ def create_normal_schedule(self, scheduler_input):
             resources_to_clear = list(self.network_model.keys())
             try:
                 before_apply = datetime.utcnow()
-                self.normal_scheduled_requests_by_rg = scheduler_result.get_scheduled_requests_by_request_group_id()
+                self.normal_scheduler_result = scheduler_result
                 n_submitted = self.apply_normal_result(scheduler_result,
                                                        scheduler_input,
                                                        resources_to_clear, deadline)
@@ -1066,18 +1073,18 @@ def scheduling_cycle(self, schedule_type, network_state_timestamp, rr_schedule_r
         set_schedule_type(schedule_type)
         result = None
         if schedule_type == NORMAL_OBSERVATION_TYPE:
-            scheduler_input = self.input_factory.create_normal_scheduling_input(
+            self.normal_scheduler_input = self.input_factory.create_normal_scheduling_input(
                 self.estimated_normal_run_timedelta.total_seconds(),
-                scheduled_requests_by_rg=self.normal_scheduled_requests_by_rg,
+                scheduled_requests_by_rg=self.normal_scheduler_result.get_scheduled_requests_by_request_group_id() if self.normal_scheduler_result else {},
                 rr_schedule=rr_schedule_result.schedule,
                 network_state_timestamp=network_state_timestamp)
-            result = self.create_normal_schedule(scheduler_input)
+            result = self.create_normal_schedule(self.normal_scheduler_input)
         elif schedule_type == RR_OBSERVATION_TYPE:
-            scheduler_input = self.input_factory.create_rr_scheduling_input(
+            self.rr_scheduler_input = self.input_factory.create_rr_scheduling_input(
                 self.estimated_rr_run_timedelta.total_seconds(),
-                scheduled_requests_by_rg=self.rr_scheduled_requests_by_rg,
+                scheduled_requests_by_rg=self.rr_scheduler_result.get_scheduled_requests_by_request_group_id() if self.rr_scheduler_result else {},
                 network_state_timestamp=network_state_timestamp)
-            result = self.create_rr_schedule(scheduler_input)
+            result = self.create_rr_schedule(self.rr_scheduler_input)
         return result
 
 

adaptive_scheduler/scheduler_input.py

@@ -49,7 +49,8 @@ def __init__(self,
                  ignore_ipp=to_bool(os.getenv('IGNORE_IPP_VALUES', 'False')),
                  avg_reservation_save_time_seconds=float(os.getenv('INITIAL_PER_RESERVATION_SAVE_TIME', 0.05)),
                  normal_runtime_seconds=float(os.getenv('INITIAL_NORMAL_RUNTIME', 360.0)),
-                 rr_runtime_seconds=float(os.getenv('INITIAL_RAPID_RESPONSE_RUNTIME', 120.0))):
+                 rr_runtime_seconds=float(os.getenv('INITIAL_RAPID_RESPONSE_RUNTIME', 120.0)),
+                 simulation_opensearch_index=os.getenv('SIMULATION_OPENSEARCH_INDEX', '')):
         self.dry_run = dry_run
         self.no_weather = no_weather
         self.no_singles = no_singles
@@ -92,6 +93,7 @@ def __init__(self,
             self.opensearch_excluded_observatories = opensearch_excluded_observatories.split(',')
         else:
             self.opensearch_excluded_observatories = []
+        self.simulation_opensearch_index = simulation_opensearch_index
 
 
 class SchedulingInputFactory(object):

adaptive_scheduler/simulation/README.md

@@ -0,0 +1,63 @@
+# Adaptive Scheduler Simulator Orchestrator
+
+The orchestrator allows for running the adaptive scheduler in a simulated environment in order to facilitate testing.
+It allows the user to dump input request data to the [Observation Portal](https://github.com/observatorycontrolsystem/observation-portal)
+which is then passed to the scheduler. The orchestrator runs the scheduler and passes off the scheduler result to a
+metric calculation file, which calculates metrics to send to an OpenSearch database. Work is still being done to enable the
+orchestrator to step through a time range and run the scheduler repeatedly on different points of the input data.
+
+## Overview of Metrics
+The available metrics center around priority distributions, utilization, and miscellaneous data including airmass data
+and slew distance. Certain metrics sent to OpenSearch are pre-binned by priority level. To get the best understanding of
+the data structures, inspect the raw JSON in OpenSearch directly.
+
+## Prerequisites
+* Python 3.9
+* A running [Configuration Database](https://github.com/observatorycontrolsystem/configdb) with instruments
+* A running [Observation Portal](https://github.com/observatorycontrolsystem/observation-portal) with requests
+* A running OpenSearch with index created to store scheduler simulation results
+
+## Environment Variables
+Consult the adaptive scheduler README for general environment variables related to the scheduler. Additional environment
+variables specific to the orchestrator are as follows:
+| Variable                         | Description                                                                                 | Default                 |
+|----------------------------------|---------------------------------------------------------------------------------------------|-------------------------|
+| `SIMULATION_RUN_ID`              | The run ID of the scheduler. This will be saved as `simulation_id` in OpenSearch            | `1`                     |
+| `SIMULATION_START_TIME`          | The simulation start time, which allows the orchestrator to step through a time range (WIP) | `2022-06-23`            |
+| `SIMULATION_END_TIME`            | The end time of the time range. This should match the start time if only a single run is desired.                                                        | `2022-06-23`            |
+| `SIMULATION_TIME_STEP_MINUTES`   | The time step in minutes for the time range (WIP)                                           | `60`                    |
+| `SIMULATION_AIRMASS_COEFFICIENT` | The airmass optimization weighting value                                                    | `0.1`                   |
+| `SIMULATION_OPENSEARCH_INDEX`    | The OpenSearch index where metrics will be saved to to                                                     | `scheduler-simulations` |
+| `OPENSEARCH_URL`                 | OpenSearch endpoint (needed for the plotting interface)                                     | _`Empty_string`_        |
+|                                  |                                                                                             |                         |
+
+## How to Run
+When running in a Docker container, the entry point can be modified to point to the orchestrator instead of the scheduler,
+e.g. `sh -c "sleep 20s; simulation-orchestrator"`. The twenty second wait time is to ensure all the relevant services (configdb, redis, etc.) are
+spun up and available. Otherwise, run the orchestrator locally on a machine with `poetry run simulation-orchestrator`
+
+## Simulation Process
+The general workflow for running a scheduler simulation is as follows:
+1. Make changes to the adaptive scheduler. If running with Docker, build the image using the suggested build command in the adaptive scheduler README.
+2. If necessary, adjust the `metrics.py` file to conform with the tests you are running, such as adjusting binning for priority values.
+3. Modify environment variables accordingly, making sure to set and verify the run ID.
+4. Run the orchestrator.
+
+## Plotting
+A plotting interface is included with the simulator to facilitate data visualization. The interface features OpenSearch searching by
+either OpenSearch ID or `simulation_id`, the ability to save plots in various formats, and zsh-style TAB autocompletion. 
+Note that the environment variable `OPENSEARCH_URL` must be set on whatever machine you are running the plots from.
+To use the plotting interface, run `python -m adaptive_scheduler.simulation.plots`
+(`-h` to show the available command line arguments).
+
+## Creating Your Own Plots
+The plotting framework provides a `Plot` class defined in `plotutils.py` to help initialize plots and get data from OpenSearch. 
+`Plot` is initialized with a user-defined plotting function to generate the plot, the plot title, and either a single string or a list
+of strings. It searches the `_id` or `simulation_id` field in OpenSearch for the strings and plots the data. To write your own plotting
+functions, follow the example functions in `plotfuncs.py`. Plotting functions should take in either a list of datasets
+or a single dataset (to match the initialization in `plots.py`). The plot title should be passed into the plotting function as well. 
+This title is used to generate the descriptions for the command-line interface of the plotting framework.
+
+The plot creation process is as simple as:
+1. Creating a function (e.g. `plot_my_plot`) in `plotfuncs.py`
+2. Adding the plot to the list of plots in `plots.py`, e.g. `Plot(plotfuncs.plot_my_plot, 'My Plot Title', 'some-data-id')`