diff --git a/schedview/app/metric_maps/main.py b/schedview/app/metric_maps/main.py
index 08b3cdc9..38bf1e89 100644
--- a/schedview/app/metric_maps/main.py
+++ b/schedview/app/metric_maps/main.py
@@ -1,105 +1,5 @@
-import bokeh.plotting
+from schedview.app.metric_maps.metric_maps import add_metric_app
+import bokeh
-from rubin_sim import maf
-
-from schedview.plot.SphereMap import ArmillarySphere, Planisphere, MollweideMap
-from schedview.collect.stars import load_bright_stars
-from schedview.collect import get_metric_path
-
-
-def make_metric_figure(metric_values_fname=None, nside=32, mag_limit_slider=True):
- """Create a figure showing multiple projections of a set of a MAF metric.
-
- Parameters
- ----------
- metric_values_fname : `str`, optional
- Name of file from which to load metric values, as saved by MAF
- in a saved metric bundle. If it is None, the look for the
- file name in the ``METRIC_FNAME`` environment varable. By default None
- nside : `int`, optional
- Healpix nside to use for display, by default 8
- mag_limit_slider : `bool`, optional
- Show the mag limit slider for stars?, by default True
-
- Returns
- -------
- fig : `bokeh.models.layouts.LayoutDOM`
- A bokeh figure that can be displayed in a notebook (e.g. with
- ``bokeh.io.show``) or used to create a bokeh app.
-
- Notes
- -----
- If ``mag_limit_slider`` is ``True``, it creates a magnitude limit
- slider for the stars. This is implemented as a python callback, and
- so is only operational in full bokeh app, not standalone output.
- """
-
- if metric_values_fname is None:
- metric_values_fname = get_metric_path()
-
- healpy_values = maf.MetricBundle.load(metric_values_fname).metricValues
-
- star_data = load_bright_stars().loc[:, ["name", "ra", "decl", "Vmag"]]
- star_data["glyph_size"] = 15 - (15.0 / 3.5) * star_data["Vmag"]
- star_data.query("glyph_size>0", inplace=True)
-
- arm = ArmillarySphere()
- hp_ds, cmap, _ = arm.add_healpix(healpy_values, nside=nside)
- hz = arm.add_horizon()
- zd70 = arm.add_horizon(zd=70, line_kwargs={"color": "red", "line_width": 2})
- star_ds = arm.add_stars(
- star_data, mag_limit_slider=mag_limit_slider, star_kwargs={"color": "black"}
- )
- arm.decorate()
-
- pla = Planisphere()
- pla.add_healpix(hp_ds, cmap=cmap, nside=nside)
- pla.add_horizon(data_source=hz)
- pla.add_horizon(
- zd=60, data_source=zd70, line_kwargs={"color": "red", "line_width": 2}
- )
- pla.add_stars(
- star_data,
- data_source=star_ds,
- mag_limit_slider=False,
- star_kwargs={"color": "black"},
- )
- pla.decorate()
-
- mol_plot = bokeh.plotting.figure(plot_width=512, plot_height=256, match_aspect=True)
- mol = MollweideMap(plot=mol_plot)
- mol.add_healpix(hp_ds, cmap=cmap, nside=nside)
- mol.add_horizon(data_source=hz)
- mol.add_horizon(
- zd=70, data_source=zd70, line_kwargs={"color": "red", "line_width": 2}
- )
- mol.add_stars(
- star_data,
- data_source=star_ds,
- mag_limit_slider=False,
- star_kwargs={"color": "black"},
- )
- mol.decorate()
-
- figure = bokeh.layouts.row(
- bokeh.layouts.column(mol.plot, *arm.sliders.values()), arm.plot, pla.plot
- )
-
- return figure
-
-
-def add_metric_app(doc):
- """Add a metric figure to a bokeh document
-
- Parameters
- ----------
- doc : `bokeh.document.document.Document`
- The bokeh document to which to add the figure.
- """
- figure = make_metric_figure()
- doc.add_root(figure)
-
-
-if __name__.startswith("bokeh_app_"):
- doc = bokeh.plotting.curdoc()
- add_metric_app(doc)
+doc = bokeh.plotting.curdoc()
+add_metric_app(doc)
diff --git a/schedview/app/metric_maps/metric_maps.py b/schedview/app/metric_maps/metric_maps.py
new file mode 100644
index 00000000..08b3cdc9
--- /dev/null
+++ b/schedview/app/metric_maps/metric_maps.py
@@ -0,0 +1,105 @@
+import bokeh.plotting
+
+from rubin_sim import maf
+
+from schedview.plot.SphereMap import ArmillarySphere, Planisphere, MollweideMap
+from schedview.collect.stars import load_bright_stars
+from schedview.collect import get_metric_path
+
+
+def make_metric_figure(metric_values_fname=None, nside=32, mag_limit_slider=True):
+ """Create a figure showing multiple projections of a set of a MAF metric.
+
+ Parameters
+ ----------
+ metric_values_fname : `str`, optional
+ Name of file from which to load metric values, as saved by MAF
+ in a saved metric bundle. If it is None, the look for the
+ file name in the ``METRIC_FNAME`` environment varable. By default None
+ nside : `int`, optional
+ Healpix nside to use for display, by default 8
+ mag_limit_slider : `bool`, optional
+ Show the mag limit slider for stars?, by default True
+
+ Returns
+ -------
+ fig : `bokeh.models.layouts.LayoutDOM`
+ A bokeh figure that can be displayed in a notebook (e.g. with
+ ``bokeh.io.show``) or used to create a bokeh app.
+
+ Notes
+ -----
+ If ``mag_limit_slider`` is ``True``, it creates a magnitude limit
+ slider for the stars. This is implemented as a python callback, and
+ so is only operational in full bokeh app, not standalone output.
+ """
+
+ if metric_values_fname is None:
+ metric_values_fname = get_metric_path()
+
+ healpy_values = maf.MetricBundle.load(metric_values_fname).metricValues
+
+ star_data = load_bright_stars().loc[:, ["name", "ra", "decl", "Vmag"]]
+ star_data["glyph_size"] = 15 - (15.0 / 3.5) * star_data["Vmag"]
+ star_data.query("glyph_size>0", inplace=True)
+
+ arm = ArmillarySphere()
+ hp_ds, cmap, _ = arm.add_healpix(healpy_values, nside=nside)
+ hz = arm.add_horizon()
+ zd70 = arm.add_horizon(zd=70, line_kwargs={"color": "red", "line_width": 2})
+ star_ds = arm.add_stars(
+ star_data, mag_limit_slider=mag_limit_slider, star_kwargs={"color": "black"}
+ )
+ arm.decorate()
+
+ pla = Planisphere()
+ pla.add_healpix(hp_ds, cmap=cmap, nside=nside)
+ pla.add_horizon(data_source=hz)
+ pla.add_horizon(
+ zd=60, data_source=zd70, line_kwargs={"color": "red", "line_width": 2}
+ )
+ pla.add_stars(
+ star_data,
+ data_source=star_ds,
+ mag_limit_slider=False,
+ star_kwargs={"color": "black"},
+ )
+ pla.decorate()
+
+ mol_plot = bokeh.plotting.figure(plot_width=512, plot_height=256, match_aspect=True)
+ mol = MollweideMap(plot=mol_plot)
+ mol.add_healpix(hp_ds, cmap=cmap, nside=nside)
+ mol.add_horizon(data_source=hz)
+ mol.add_horizon(
+ zd=70, data_source=zd70, line_kwargs={"color": "red", "line_width": 2}
+ )
+ mol.add_stars(
+ star_data,
+ data_source=star_ds,
+ mag_limit_slider=False,
+ star_kwargs={"color": "black"},
+ )
+ mol.decorate()
+
+ figure = bokeh.layouts.row(
+ bokeh.layouts.column(mol.plot, *arm.sliders.values()), arm.plot, pla.plot
+ )
+
+ return figure
+
+
+def add_metric_app(doc):
+ """Add a metric figure to a bokeh document
+
+ Parameters
+ ----------
+ doc : `bokeh.document.document.Document`
+ The bokeh document to which to add the figure.
+ """
+ figure = make_metric_figure()
+ doc.add_root(figure)
+
+
+if __name__.startswith("bokeh_app_"):
+ doc = bokeh.plotting.curdoc()
+ add_metric_app(doc)
diff --git a/schedview/app/sched_maps/main.py b/schedview/app/sched_maps/main.py
index 9946039a..9b58e34f 100644
--- a/schedview/app/sched_maps/main.py
+++ b/schedview/app/sched_maps/main.py
@@ -1,554 +1,5 @@
-import bokeh.plotting
-from astropy.time import Time
+from schedview.app.sched_maps.sched_maps import add_scheduler_map_app
+import bokeh
-import pandas as pd
-import bokeh.models
-import bokeh.core.properties
-
-from schedview.plot.SphereMap import (
- ArmillarySphere,
- HorizonMap,
- Planisphere,
- MollweideMap,
-)
-
-from schedview.collect import sample_pickle
-
-from schedview.plot.scheduler import SchedulerDisplay
-from schedview.collect import read_scheduler
-from schedview.plot.scheduler import LOGGER, DEFAULT_NSIDE
-
-
-class SchedulerDisplayApp(SchedulerDisplay):
- include_mollweide = False
-
- def make_pickle_entry_box(self):
- """Make the entry box for a file name from which to load state."""
- file_input_box = bokeh.models.TextInput(
- value=sample_pickle("auxtel59628.pickle.gz") + " ",
- title="Pickle path:",
- )
-
- def switch_pickle(attrname, old, new):
- def do_switch_pickle():
- LOGGER.info(f"Loading {new}.")
- try:
- # load updates the survey & conditions, which updates
- # bokeh models...
- self.load(new)
- except Exception as e:
- LOGGER.warning(f"Failed to load file {new}: {e}")
- pass
-
- LOGGER.debug(f"Finished loading {new}")
-
- # If we do not have access to the document, this won't
- # do anything and is unnecessary, but that's okay.
- self.enable_controls()
-
- if file_input_box.document is None:
- # If we don't have access to the document, we can't disable
- # the controls, so just do it.
- do_switch_pickle()
- else:
- # disable the controls, and ask the document to do the update
- # on the following event look tick.
- self.disable_controls()
- file_input_box.document.add_next_tick_callback(do_switch_pickle)
-
- file_input_box.on_change("value", switch_pickle)
- self.bokeh_models["file_input_box"] = file_input_box
-
- def _set_scheduler(self, scheduler):
- LOGGER.info("Setting scheduler")
- super()._set_scheduler(scheduler)
- self.update_tier_selector()
- self.update_reward_summary_table_bokeh_model()
-
- def _set_conditions(self, conditions):
- super()._set_conditions(conditions)
- self.update_healpix_bokeh_model()
- self.update_reward_table_bokeh_model()
- self.update_hovertool_bokeh_model()
- self.update_telescope_marker_bokeh_model()
- self.update_moon_marker_bokeh_model()
- self.update_sun_marker_bokeh_model()
- self.update_survey_marker_bokeh_model()
- self.update_chosen_survey_bokeh_model()
- self.update_mjd_slider_bokeh_model()
- self.update_time_input_box_bokeh_model()
-
- def make_time_selector(self):
- """Create the time selector slider bokeh model."""
- time_selector = bokeh.models.Slider(
- title="MJD",
- start=self.mjd - 1,
- end=self.mjd + 1,
- value=self.mjd,
- step=1.0 / 1440,
- )
-
- def switch_time(attrname, old, new):
- if time_selector.document is None:
- # If we don't have access to the document, we can't disable
- # the controls, so don't try.
- self.mjd = new
- else:
- # To disable controls as the time is being updated, we need to
- # separate the callback so it happens in two event loop ticks:
- # the first tick disables the controls, the next one
- # actually updates the MJD and then re-enables the controls.
- def do_switch_time():
- self.mjd = new
- self.enable_controls()
-
- self.disable_controls()
- time_selector.document.add_next_tick_callback(do_switch_time)
-
- time_selector.on_change("value_throttled", switch_time)
- self.bokeh_models["time_selector"] = time_selector
- self.update_time_selector_bokeh_model()
-
- def update_time_selector_bokeh_model(self):
- """Update the time selector limits and value to match the date."""
- if "time_selector" in self.bokeh_models:
- self.bokeh_models["time_selector"].start = self.conditions.sun_n12_setting
- self.bokeh_models["time_selector"].end = self.conditions.sun_n12_rising
- self.bokeh_models["time_selector"].value = self.conditions.mjd
-
- def add_mjd_slider_callback(self):
- """Create the mjd slider bokeh model."""
- mjd_slider = self.bokeh_models["mjd_slider"]
-
- def switch_time(attrname, old, new):
- if mjd_slider.document is None:
- # If we don't have access to the document, we can't disable
- # the controls, so don't try.
- self.mjd = new
- else:
- # To disable controls as the time is being updated, we need to
- # separate the callback so it happens in two event loop ticks:
- # the first tick disables the controls, the next one
- # actually updates the MJD and then re-enables the controls.
- def do_switch_time():
- self.mjd = new
- self.enable_controls()
-
- self.disable_controls()
- mjd_slider.document.add_next_tick_callback(do_switch_time)
-
- mjd_slider.on_change("value_throttled", switch_time)
- self.update_time_selector_bokeh_model()
-
- def update_mjd_slider_bokeh_model(self):
- """Update the time selector limits and value to match the date."""
- if "mjd_slider" in self.bokeh_models:
- self.bokeh_models["mjd_slider"].start = self.conditions.sun_n12_setting
- self.bokeh_models["mjd_slider"].end = self.conditions.sun_n12_rising
- self.bokeh_models["mjd_slider"].value = self.conditions.mjd
-
- def make_time_input_box(self):
- """Create the time entry box bokeh model."""
- time_input_box = bokeh.models.TextInput(title="Date and time (UTC):")
- self.bokeh_models["time_input_box"] = time_input_box
- self.update_time_input_box_bokeh_model()
-
- def switch_time(attrname, old, new):
- new_mjd = pd.to_datetime(new, utc=True).to_julian_date() - 2400000.5
-
- if time_input_box.document is None:
- # If we don't have access to the document, we can't disable
- # the controls, so don't try.
- self.mjd = new_mjd
- else:
- # To disable controls as the time is being updated, we need to
- # separate the callback so it happens in two event loop ticks:
- # the first tick disables the controls, the next one
- # actually updates the MJD and then re-enables the controls.
- def do_switch_time():
- self.mjd = new_mjd
- self.enable_controls()
-
- self.disable_controls()
- time_input_box.document.add_next_tick_callback(do_switch_time)
-
- time_input_box.on_change("value", switch_time)
-
- def update_time_input_box_bokeh_model(self):
- """Update the time selector limits and value to match the date."""
- if "time_input_box" in self.bokeh_models:
- iso_time = Time(self.mjd, format="mjd", scale="utc").iso
- self.bokeh_models["time_input_box"].value = iso_time
-
- def make_tier_selector(self):
- """Create the tier selector bokeh model."""
- tier_selector = bokeh.models.Select(value=None, options=[None])
-
- def switch_tier(attrname, old, new):
- self.select_tier(new)
-
- tier_selector.on_change("value", switch_tier)
- self.bokeh_models["tier_selector"] = tier_selector
- self.update_tier_selector()
-
- def update_tier_selector(self):
- """Update tier selector to represent tiers for the current survey."""
- if "tier_selector" in self.bokeh_models:
- options = self.tier_names
- self.bokeh_models["tier_selector"].options = options
- self.bokeh_models["tier_selector"].value = options[self.survey_index[0]]
-
- def select_tier(self, tier):
- """Set the tier being displayed."""
- super().select_tier(tier)
- self.update_survey_selector()
-
- def make_survey_selector(self):
- """Create the survey selector bokeh model."""
- survey_selector = bokeh.models.Select(value=None, options=[None])
-
- def switch_survey(attrname, old, new):
- self.select_survey(new)
-
- survey_selector.on_change("value", switch_survey)
- self.bokeh_models["survey_selector"] = survey_selector
-
- def update_survey_selector(self):
- """Uptade the survey selector to the current scheduler and tier."""
- if "survey_selector" in self.bokeh_models:
- options = [
- self._unique_survey_name([self.survey_index[0], s])
- for s in range(len(self.scheduler.survey_lists[self.survey_index[0]]))
- ]
- self.bokeh_models["survey_selector"].options = options
- self.bokeh_models["survey_selector"].value = options[self.survey_index[1]]
-
- def select_survey(self, survey):
- """Set the tier being displayed."""
- super().select_survey(survey)
- # Note that updating the value selector triggers the
- # callback, which updates the maps themselves
- self.update_value_selector()
- self.update_reward_table_bokeh_model()
- self.update_hovertool_bokeh_model()
- self.update_survey_marker_bokeh_model()
-
- def make_value_selector(self):
- """Create the bokeh model to select which value to show in maps."""
- value_selector = bokeh.models.Select(value=None, options=[None])
-
- def switch_value(attrname, old, new):
- self.select_value(new)
-
- value_selector.on_change("value", switch_value)
- self.bokeh_models["value_selector"] = value_selector
-
- def update_value_selector(self):
- """Update the value selector bokeh model to show available options."""
- if "value_selector" in self.bokeh_models:
- self.bokeh_models["value_selector"].options = self.map_keys
- if self.map_key in self.map_keys:
- self.bokeh_models["value_selector"].value = self.map_key
- elif self.init_key in self.map_keys:
- self.bokeh_models["value_selector"].value = self.init_key
- else:
- self.bokeh_models["value_selector"].value = self.map_keys[-1]
-
- def select_value(self, map_key):
- """Set the tier being displayed."""
- super().select_value(map_key)
- self.update_healpix_bokeh_model()
-
- def update_time_display_bokeh_model(self):
- """Update the value of the displayed time."""
- if "mjd" in self.sliders:
- self.update_mjd_slider_bokeh_model()
-
- if "time_input_box" in self.bokeh_models:
- self.update_time_input_box_bokeh_model()
-
- def update_displayed_value_metadata_bokeh_model(self):
- self.update_tier_selector()
-
- def _select_survey_from_summary_table(self, attr, old, new):
- LOGGER.debug("Called select_survey_from_summary_table")
- selected_index = new[0]
- tier_name = self.data_sources["reward_summary_table"].data["tier"][
- selected_index
- ]
- survey_name = self.data_sources["reward_summary_table"].data["survey_name"][
- selected_index
- ]
-
- # Update the selectors, and this will run
- # the callbacks to do all the updates
- self.bokeh_models["tier_selector"].value = tier_name
- self.bokeh_models["survey_selector"].value = survey_name
-
- def make_reward_summary_table(self):
- super().make_reward_summary_table()
-
- self.data_sources["reward_summary_table"].selected.on_change(
- "indices",
- self._select_survey_from_summary_table,
- )
-
- def disable_controls(self):
- """Disable all controls.
-
- Intended to be used while plot elements are updating, and the
- control therefore do not do what the user probably intends.
- """
- LOGGER.info("Disabling controls")
- for model in self.bokeh_models.values():
- try:
- model.disabled = True
- except AttributeError:
- pass
-
- def enable_controls(self):
- """Enable all controls."""
- LOGGER.info("Enabling controls")
- for model in self.bokeh_models.values():
- try:
- model.disabled = False
- except AttributeError:
- pass
-
- def make_figure(self):
- """Create a bokeh figures showing sky maps for scheduler behavior.
-
- Returns
- -------
- fig : `bokeh.models.layouts.LayoutDOM`
- A bokeh figure that can be displayed in a notebook (e.g. with
- ``bokeh.io.show``) or used to create a bokeh app.
- """
- self.make_sphere_map(
- "altaz",
- HorizonMap,
- "Alt Az",
- plot_width=512,
- plot_height=512,
- decorate=False,
- horizon_graticules=True,
- )
-
- self.bokeh_models["key"] = bokeh.models.Div(text=self.key_markup)
-
- self.bokeh_models["reward_table_title"] = bokeh.models.Div(
- text="Basis functions for displayed survey
"
- )
- self.make_reward_table()
-
- self.bokeh_models["reward_summary_table_title"] = bokeh.models.Div(
- text="Rewards for all survey schedulers
"
- )
- self.make_reward_summary_table()
- self.make_chosen_survey()
- self.make_value_selector()
- self.make_survey_selector()
- self.make_tier_selector()
- self.make_pickle_entry_box()
-
- controls = [self.bokeh_models["file_input_box"]]
-
- if self.observatory is not None:
- self.make_time_input_box()
- controls.append(self.bokeh_models["time_input_box"])
-
- controls += [
- self.bokeh_models["tier_selector"],
- self.bokeh_models["survey_selector"],
- self.bokeh_models["value_selector"],
- ]
-
- figure = bokeh.layouts.column(
- self.bokeh_models["key"],
- self.bokeh_models["altaz"],
- *controls,
- self.bokeh_models["chosen_survey"],
- self.bokeh_models["reward_table_title"],
- self.bokeh_models["reward_table"],
- self.bokeh_models["reward_summary_table_title"],
- self.bokeh_models["reward_summary_table"],
- )
-
- return figure
-
- def make_figure_with_many_projections(self):
- """Create a bokeh figures showing sky maps for scheduler behavior.
-
- Returns
- -------
- fig : `bokeh.models.layouts.LayoutDOM`
- A bokeh figure that can be displayed in a notebook (e.g. with
- ``bokeh.io.show``) or used to create a bokeh app.
- """
- self.make_sphere_map(
- "armillary_sphere",
- ArmillarySphere,
- "Armillary Sphere",
- plot_width=512,
- plot_height=512,
- decorate=True,
- )
- self.bokeh_models["alt_slider"] = self.sphere_maps["armillary_sphere"].sliders[
- "alt"
- ]
- self.bokeh_models["az_slider"] = self.sphere_maps["armillary_sphere"].sliders[
- "az"
- ]
- self.bokeh_models["mjd_slider"] = self.sphere_maps["armillary_sphere"].sliders[
- "mjd"
- ]
- # self.bokeh_models["mjd_slider"].visible = False
- self.make_sphere_map(
- "planisphere",
- Planisphere,
- "Planisphere",
- plot_width=512,
- plot_height=512,
- decorate=True,
- )
- self.make_sphere_map(
- "altaz",
- HorizonMap,
- "Alt Az",
- plot_width=512,
- plot_height=512,
- decorate=False,
- horizon_graticules=True,
- )
-
- if self.include_mollweide:
- self.make_sphere_map(
- "mollweide",
- MollweideMap,
- "Mollweide",
- plot_width=512,
- plot_height=512,
- decorate=True,
- )
-
- self.bokeh_models["key"] = bokeh.models.Div(text=self.key_markup)
-
- self.bokeh_models["reward_table_title"] = bokeh.models.Div(
- text="Basis functions for displayed survey
"
- )
- self.make_reward_table()
-
- self.bokeh_models["reward_summary_table_title"] = bokeh.models.Div(
- text="Rewards for all survey schedulers
"
- )
- self.make_reward_summary_table()
- self.make_chosen_survey()
- self.make_value_selector()
- self.make_survey_selector()
- self.make_tier_selector()
- self.make_pickle_entry_box()
-
- # slider was created by SphereMap
- self.add_mjd_slider_callback()
-
- arm_controls = [
- self.bokeh_models["alt_slider"],
- self.bokeh_models["az_slider"],
- ]
-
- controls = [self.bokeh_models["file_input_box"]]
-
- if self.observatory is not None:
- self.make_time_input_box()
- controls.append(self.bokeh_models["time_input_box"])
-
- controls += [
- self.bokeh_models["mjd_slider"],
- self.bokeh_models["tier_selector"],
- self.bokeh_models["survey_selector"],
- self.bokeh_models["value_selector"],
- ]
-
- if self.include_mollweide:
- map_column = bokeh.layouts.column(
- self.bokeh_models["altaz"],
- self.bokeh_models["planisphere"],
- self.bokeh_models["armillary_sphere"],
- *arm_controls,
- self.bokeh_models["mollweide"],
- )
- else:
- map_column = bokeh.layouts.column(
- self.bokeh_models["altaz"],
- self.bokeh_models["planisphere"],
- self.bokeh_models["armillary_sphere"],
- *arm_controls,
- )
-
- figure = bokeh.layouts.row(
- bokeh.layouts.column(
- self.bokeh_models["key"],
- *controls,
- self.bokeh_models["chosen_survey"],
- self.bokeh_models["reward_table_title"],
- self.bokeh_models["reward_table"],
- self.bokeh_models["reward_summary_table_title"],
- self.bokeh_models["reward_summary_table"],
- ),
- map_column,
- )
-
- return figure
-
-
-def make_scheduler_map_figure(
- scheduler_pickle_fname="baseline22_start.pickle.gz",
- init_key="AvoidDirectWind",
- nside=DEFAULT_NSIDE,
-):
- """Create a set of bekeh figures showing sky maps for scheduler behavior.
-
- Parameters
- ----------
- scheduler_pickle_fname : `str`, optional
- File from which to load the scheduler state. If set to none, look for
- the file name in the ``SCHED_PICKLE`` environment variable.
- By default None
- init_key : `str`, optional
- Name of the initial map to show, by default 'AvoidDirectWind'
- nside : int, optional
- Healpix nside to use for display, by default 32
-
- Returns
- -------
- fig : `bokeh.models.layouts.LayoutDOM`
- A bokeh figure that can be displayed in a notebook (e.g. with
- ``bokeh.io.show``) or used to create a bokeh app.
- """
- if scheduler_pickle_fname is None:
- scheduler_map = SchedulerDisplayApp(nside=nside)
- else:
- scheduler, conditions = read_scheduler(sample_pickle(scheduler_pickle_fname))
- scheduler.update_conditions(conditions)
- scheduler_map = SchedulerDisplayApp(nside=nside, scheduler=scheduler)
-
- figure = scheduler_map.make_figure()
-
- return figure
-
-
-def add_scheduler_map_app(doc):
- """Add a scheduler map figure to a bokeh document
-
- Parameters
- ----------
- doc : `bokeh.document.document.Document`
- The bokeh document to which to add the figure.
- """
- figure = make_scheduler_map_figure()
- doc.add_root(figure)
-
-
-if __name__.startswith("bokeh_app_"):
- doc = bokeh.plotting.curdoc()
- add_scheduler_map_app(doc)
+doc = bokeh.plotting.curdoc()
+add_scheduler_map_app(doc)
diff --git a/schedview/app/sched_maps/sched_maps.py b/schedview/app/sched_maps/sched_maps.py
new file mode 100644
index 00000000..9946039a
--- /dev/null
+++ b/schedview/app/sched_maps/sched_maps.py
@@ -0,0 +1,554 @@
+import bokeh.plotting
+from astropy.time import Time
+
+import pandas as pd
+import bokeh.models
+import bokeh.core.properties
+
+from schedview.plot.SphereMap import (
+ ArmillarySphere,
+ HorizonMap,
+ Planisphere,
+ MollweideMap,
+)
+
+from schedview.collect import sample_pickle
+
+from schedview.plot.scheduler import SchedulerDisplay
+from schedview.collect import read_scheduler
+from schedview.plot.scheduler import LOGGER, DEFAULT_NSIDE
+
+
+class SchedulerDisplayApp(SchedulerDisplay):
+ include_mollweide = False
+
+ def make_pickle_entry_box(self):
+ """Make the entry box for a file name from which to load state."""
+ file_input_box = bokeh.models.TextInput(
+ value=sample_pickle("auxtel59628.pickle.gz") + " ",
+ title="Pickle path:",
+ )
+
+ def switch_pickle(attrname, old, new):
+ def do_switch_pickle():
+ LOGGER.info(f"Loading {new}.")
+ try:
+ # load updates the survey & conditions, which updates
+ # bokeh models...
+ self.load(new)
+ except Exception as e:
+ LOGGER.warning(f"Failed to load file {new}: {e}")
+ pass
+
+ LOGGER.debug(f"Finished loading {new}")
+
+ # If we do not have access to the document, this won't
+ # do anything and is unnecessary, but that's okay.
+ self.enable_controls()
+
+ if file_input_box.document is None:
+ # If we don't have access to the document, we can't disable
+ # the controls, so just do it.
+ do_switch_pickle()
+ else:
+ # disable the controls, and ask the document to do the update
+ # on the following event look tick.
+ self.disable_controls()
+ file_input_box.document.add_next_tick_callback(do_switch_pickle)
+
+ file_input_box.on_change("value", switch_pickle)
+ self.bokeh_models["file_input_box"] = file_input_box
+
+ def _set_scheduler(self, scheduler):
+ LOGGER.info("Setting scheduler")
+ super()._set_scheduler(scheduler)
+ self.update_tier_selector()
+ self.update_reward_summary_table_bokeh_model()
+
+ def _set_conditions(self, conditions):
+ super()._set_conditions(conditions)
+ self.update_healpix_bokeh_model()
+ self.update_reward_table_bokeh_model()
+ self.update_hovertool_bokeh_model()
+ self.update_telescope_marker_bokeh_model()
+ self.update_moon_marker_bokeh_model()
+ self.update_sun_marker_bokeh_model()
+ self.update_survey_marker_bokeh_model()
+ self.update_chosen_survey_bokeh_model()
+ self.update_mjd_slider_bokeh_model()
+ self.update_time_input_box_bokeh_model()
+
+ def make_time_selector(self):
+ """Create the time selector slider bokeh model."""
+ time_selector = bokeh.models.Slider(
+ title="MJD",
+ start=self.mjd - 1,
+ end=self.mjd + 1,
+ value=self.mjd,
+ step=1.0 / 1440,
+ )
+
+ def switch_time(attrname, old, new):
+ if time_selector.document is None:
+ # If we don't have access to the document, we can't disable
+ # the controls, so don't try.
+ self.mjd = new
+ else:
+ # To disable controls as the time is being updated, we need to
+ # separate the callback so it happens in two event loop ticks:
+ # the first tick disables the controls, the next one
+ # actually updates the MJD and then re-enables the controls.
+ def do_switch_time():
+ self.mjd = new
+ self.enable_controls()
+
+ self.disable_controls()
+ time_selector.document.add_next_tick_callback(do_switch_time)
+
+ time_selector.on_change("value_throttled", switch_time)
+ self.bokeh_models["time_selector"] = time_selector
+ self.update_time_selector_bokeh_model()
+
+ def update_time_selector_bokeh_model(self):
+ """Update the time selector limits and value to match the date."""
+ if "time_selector" in self.bokeh_models:
+ self.bokeh_models["time_selector"].start = self.conditions.sun_n12_setting
+ self.bokeh_models["time_selector"].end = self.conditions.sun_n12_rising
+ self.bokeh_models["time_selector"].value = self.conditions.mjd
+
+ def add_mjd_slider_callback(self):
+ """Create the mjd slider bokeh model."""
+ mjd_slider = self.bokeh_models["mjd_slider"]
+
+ def switch_time(attrname, old, new):
+ if mjd_slider.document is None:
+ # If we don't have access to the document, we can't disable
+ # the controls, so don't try.
+ self.mjd = new
+ else:
+ # To disable controls as the time is being updated, we need to
+ # separate the callback so it happens in two event loop ticks:
+ # the first tick disables the controls, the next one
+ # actually updates the MJD and then re-enables the controls.
+ def do_switch_time():
+ self.mjd = new
+ self.enable_controls()
+
+ self.disable_controls()
+ mjd_slider.document.add_next_tick_callback(do_switch_time)
+
+ mjd_slider.on_change("value_throttled", switch_time)
+ self.update_time_selector_bokeh_model()
+
+ def update_mjd_slider_bokeh_model(self):
+ """Update the time selector limits and value to match the date."""
+ if "mjd_slider" in self.bokeh_models:
+ self.bokeh_models["mjd_slider"].start = self.conditions.sun_n12_setting
+ self.bokeh_models["mjd_slider"].end = self.conditions.sun_n12_rising
+ self.bokeh_models["mjd_slider"].value = self.conditions.mjd
+
+ def make_time_input_box(self):
+ """Create the time entry box bokeh model."""
+ time_input_box = bokeh.models.TextInput(title="Date and time (UTC):")
+ self.bokeh_models["time_input_box"] = time_input_box
+ self.update_time_input_box_bokeh_model()
+
+ def switch_time(attrname, old, new):
+ new_mjd = pd.to_datetime(new, utc=True).to_julian_date() - 2400000.5
+
+ if time_input_box.document is None:
+ # If we don't have access to the document, we can't disable
+ # the controls, so don't try.
+ self.mjd = new_mjd
+ else:
+ # To disable controls as the time is being updated, we need to
+ # separate the callback so it happens in two event loop ticks:
+ # the first tick disables the controls, the next one
+ # actually updates the MJD and then re-enables the controls.
+ def do_switch_time():
+ self.mjd = new_mjd
+ self.enable_controls()
+
+ self.disable_controls()
+ time_input_box.document.add_next_tick_callback(do_switch_time)
+
+ time_input_box.on_change("value", switch_time)
+
+ def update_time_input_box_bokeh_model(self):
+ """Update the time selector limits and value to match the date."""
+ if "time_input_box" in self.bokeh_models:
+ iso_time = Time(self.mjd, format="mjd", scale="utc").iso
+ self.bokeh_models["time_input_box"].value = iso_time
+
+ def make_tier_selector(self):
+ """Create the tier selector bokeh model."""
+ tier_selector = bokeh.models.Select(value=None, options=[None])
+
+ def switch_tier(attrname, old, new):
+ self.select_tier(new)
+
+ tier_selector.on_change("value", switch_tier)
+ self.bokeh_models["tier_selector"] = tier_selector
+ self.update_tier_selector()
+
+ def update_tier_selector(self):
+ """Update tier selector to represent tiers for the current survey."""
+ if "tier_selector" in self.bokeh_models:
+ options = self.tier_names
+ self.bokeh_models["tier_selector"].options = options
+ self.bokeh_models["tier_selector"].value = options[self.survey_index[0]]
+
+ def select_tier(self, tier):
+ """Set the tier being displayed."""
+ super().select_tier(tier)
+ self.update_survey_selector()
+
+ def make_survey_selector(self):
+ """Create the survey selector bokeh model."""
+ survey_selector = bokeh.models.Select(value=None, options=[None])
+
+ def switch_survey(attrname, old, new):
+ self.select_survey(new)
+
+ survey_selector.on_change("value", switch_survey)
+ self.bokeh_models["survey_selector"] = survey_selector
+
+ def update_survey_selector(self):
+ """Uptade the survey selector to the current scheduler and tier."""
+ if "survey_selector" in self.bokeh_models:
+ options = [
+ self._unique_survey_name([self.survey_index[0], s])
+ for s in range(len(self.scheduler.survey_lists[self.survey_index[0]]))
+ ]
+ self.bokeh_models["survey_selector"].options = options
+ self.bokeh_models["survey_selector"].value = options[self.survey_index[1]]
+
+ def select_survey(self, survey):
+ """Set the tier being displayed."""
+ super().select_survey(survey)
+ # Note that updating the value selector triggers the
+ # callback, which updates the maps themselves
+ self.update_value_selector()
+ self.update_reward_table_bokeh_model()
+ self.update_hovertool_bokeh_model()
+ self.update_survey_marker_bokeh_model()
+
+ def make_value_selector(self):
+ """Create the bokeh model to select which value to show in maps."""
+ value_selector = bokeh.models.Select(value=None, options=[None])
+
+ def switch_value(attrname, old, new):
+ self.select_value(new)
+
+ value_selector.on_change("value", switch_value)
+ self.bokeh_models["value_selector"] = value_selector
+
+ def update_value_selector(self):
+ """Update the value selector bokeh model to show available options."""
+ if "value_selector" in self.bokeh_models:
+ self.bokeh_models["value_selector"].options = self.map_keys
+ if self.map_key in self.map_keys:
+ self.bokeh_models["value_selector"].value = self.map_key
+ elif self.init_key in self.map_keys:
+ self.bokeh_models["value_selector"].value = self.init_key
+ else:
+ self.bokeh_models["value_selector"].value = self.map_keys[-1]
+
+ def select_value(self, map_key):
+ """Set the tier being displayed."""
+ super().select_value(map_key)
+ self.update_healpix_bokeh_model()
+
+ def update_time_display_bokeh_model(self):
+ """Update the value of the displayed time."""
+ if "mjd" in self.sliders:
+ self.update_mjd_slider_bokeh_model()
+
+ if "time_input_box" in self.bokeh_models:
+ self.update_time_input_box_bokeh_model()
+
+ def update_displayed_value_metadata_bokeh_model(self):
+ self.update_tier_selector()
+
+ def _select_survey_from_summary_table(self, attr, old, new):
+ LOGGER.debug("Called select_survey_from_summary_table")
+ selected_index = new[0]
+ tier_name = self.data_sources["reward_summary_table"].data["tier"][
+ selected_index
+ ]
+ survey_name = self.data_sources["reward_summary_table"].data["survey_name"][
+ selected_index
+ ]
+
+ # Update the selectors, and this will run
+ # the callbacks to do all the updates
+ self.bokeh_models["tier_selector"].value = tier_name
+ self.bokeh_models["survey_selector"].value = survey_name
+
+ def make_reward_summary_table(self):
+ super().make_reward_summary_table()
+
+ self.data_sources["reward_summary_table"].selected.on_change(
+ "indices",
+ self._select_survey_from_summary_table,
+ )
+
+ def disable_controls(self):
+ """Disable all controls.
+
+ Intended to be used while plot elements are updating, and the
+ control therefore do not do what the user probably intends.
+ """
+ LOGGER.info("Disabling controls")
+ for model in self.bokeh_models.values():
+ try:
+ model.disabled = True
+ except AttributeError:
+ pass
+
+ def enable_controls(self):
+ """Enable all controls."""
+ LOGGER.info("Enabling controls")
+ for model in self.bokeh_models.values():
+ try:
+ model.disabled = False
+ except AttributeError:
+ pass
+
+ def make_figure(self):
+ """Create a bokeh figures showing sky maps for scheduler behavior.
+
+ Returns
+ -------
+ fig : `bokeh.models.layouts.LayoutDOM`
+ A bokeh figure that can be displayed in a notebook (e.g. with
+ ``bokeh.io.show``) or used to create a bokeh app.
+ """
+ self.make_sphere_map(
+ "altaz",
+ HorizonMap,
+ "Alt Az",
+ plot_width=512,
+ plot_height=512,
+ decorate=False,
+ horizon_graticules=True,
+ )
+
+ self.bokeh_models["key"] = bokeh.models.Div(text=self.key_markup)
+
+ self.bokeh_models["reward_table_title"] = bokeh.models.Div(
+ text="Basis functions for displayed survey
"
+ )
+ self.make_reward_table()
+
+ self.bokeh_models["reward_summary_table_title"] = bokeh.models.Div(
+ text="Rewards for all survey schedulers
"
+ )
+ self.make_reward_summary_table()
+ self.make_chosen_survey()
+ self.make_value_selector()
+ self.make_survey_selector()
+ self.make_tier_selector()
+ self.make_pickle_entry_box()
+
+ controls = [self.bokeh_models["file_input_box"]]
+
+ if self.observatory is not None:
+ self.make_time_input_box()
+ controls.append(self.bokeh_models["time_input_box"])
+
+ controls += [
+ self.bokeh_models["tier_selector"],
+ self.bokeh_models["survey_selector"],
+ self.bokeh_models["value_selector"],
+ ]
+
+ figure = bokeh.layouts.column(
+ self.bokeh_models["key"],
+ self.bokeh_models["altaz"],
+ *controls,
+ self.bokeh_models["chosen_survey"],
+ self.bokeh_models["reward_table_title"],
+ self.bokeh_models["reward_table"],
+ self.bokeh_models["reward_summary_table_title"],
+ self.bokeh_models["reward_summary_table"],
+ )
+
+ return figure
+
+ def make_figure_with_many_projections(self):
+ """Create a bokeh figures showing sky maps for scheduler behavior.
+
+ Returns
+ -------
+ fig : `bokeh.models.layouts.LayoutDOM`
+ A bokeh figure that can be displayed in a notebook (e.g. with
+ ``bokeh.io.show``) or used to create a bokeh app.
+ """
+ self.make_sphere_map(
+ "armillary_sphere",
+ ArmillarySphere,
+ "Armillary Sphere",
+ plot_width=512,
+ plot_height=512,
+ decorate=True,
+ )
+ self.bokeh_models["alt_slider"] = self.sphere_maps["armillary_sphere"].sliders[
+ "alt"
+ ]
+ self.bokeh_models["az_slider"] = self.sphere_maps["armillary_sphere"].sliders[
+ "az"
+ ]
+ self.bokeh_models["mjd_slider"] = self.sphere_maps["armillary_sphere"].sliders[
+ "mjd"
+ ]
+ # self.bokeh_models["mjd_slider"].visible = False
+ self.make_sphere_map(
+ "planisphere",
+ Planisphere,
+ "Planisphere",
+ plot_width=512,
+ plot_height=512,
+ decorate=True,
+ )
+ self.make_sphere_map(
+ "altaz",
+ HorizonMap,
+ "Alt Az",
+ plot_width=512,
+ plot_height=512,
+ decorate=False,
+ horizon_graticules=True,
+ )
+
+ if self.include_mollweide:
+ self.make_sphere_map(
+ "mollweide",
+ MollweideMap,
+ "Mollweide",
+ plot_width=512,
+ plot_height=512,
+ decorate=True,
+ )
+
+ self.bokeh_models["key"] = bokeh.models.Div(text=self.key_markup)
+
+ self.bokeh_models["reward_table_title"] = bokeh.models.Div(
+ text="Basis functions for displayed survey
"
+ )
+ self.make_reward_table()
+
+ self.bokeh_models["reward_summary_table_title"] = bokeh.models.Div(
+ text="Rewards for all survey schedulers
"
+ )
+ self.make_reward_summary_table()
+ self.make_chosen_survey()
+ self.make_value_selector()
+ self.make_survey_selector()
+ self.make_tier_selector()
+ self.make_pickle_entry_box()
+
+ # slider was created by SphereMap
+ self.add_mjd_slider_callback()
+
+ arm_controls = [
+ self.bokeh_models["alt_slider"],
+ self.bokeh_models["az_slider"],
+ ]
+
+ controls = [self.bokeh_models["file_input_box"]]
+
+ if self.observatory is not None:
+ self.make_time_input_box()
+ controls.append(self.bokeh_models["time_input_box"])
+
+ controls += [
+ self.bokeh_models["mjd_slider"],
+ self.bokeh_models["tier_selector"],
+ self.bokeh_models["survey_selector"],
+ self.bokeh_models["value_selector"],
+ ]
+
+ if self.include_mollweide:
+ map_column = bokeh.layouts.column(
+ self.bokeh_models["altaz"],
+ self.bokeh_models["planisphere"],
+ self.bokeh_models["armillary_sphere"],
+ *arm_controls,
+ self.bokeh_models["mollweide"],
+ )
+ else:
+ map_column = bokeh.layouts.column(
+ self.bokeh_models["altaz"],
+ self.bokeh_models["planisphere"],
+ self.bokeh_models["armillary_sphere"],
+ *arm_controls,
+ )
+
+ figure = bokeh.layouts.row(
+ bokeh.layouts.column(
+ self.bokeh_models["key"],
+ *controls,
+ self.bokeh_models["chosen_survey"],
+ self.bokeh_models["reward_table_title"],
+ self.bokeh_models["reward_table"],
+ self.bokeh_models["reward_summary_table_title"],
+ self.bokeh_models["reward_summary_table"],
+ ),
+ map_column,
+ )
+
+ return figure
+
+
+def make_scheduler_map_figure(
+ scheduler_pickle_fname="baseline22_start.pickle.gz",
+ init_key="AvoidDirectWind",
+ nside=DEFAULT_NSIDE,
+):
+ """Create a set of bekeh figures showing sky maps for scheduler behavior.
+
+ Parameters
+ ----------
+ scheduler_pickle_fname : `str`, optional
+ File from which to load the scheduler state. If set to none, look for
+ the file name in the ``SCHED_PICKLE`` environment variable.
+ By default None
+ init_key : `str`, optional
+ Name of the initial map to show, by default 'AvoidDirectWind'
+ nside : int, optional
+ Healpix nside to use for display, by default 32
+
+ Returns
+ -------
+ fig : `bokeh.models.layouts.LayoutDOM`
+ A bokeh figure that can be displayed in a notebook (e.g. with
+ ``bokeh.io.show``) or used to create a bokeh app.
+ """
+ if scheduler_pickle_fname is None:
+ scheduler_map = SchedulerDisplayApp(nside=nside)
+ else:
+ scheduler, conditions = read_scheduler(sample_pickle(scheduler_pickle_fname))
+ scheduler.update_conditions(conditions)
+ scheduler_map = SchedulerDisplayApp(nside=nside, scheduler=scheduler)
+
+ figure = scheduler_map.make_figure()
+
+ return figure
+
+
+def add_scheduler_map_app(doc):
+ """Add a scheduler map figure to a bokeh document
+
+ Parameters
+ ----------
+ doc : `bokeh.document.document.Document`
+ The bokeh document to which to add the figure.
+ """
+ figure = make_scheduler_map_figure()
+ doc.add_root(figure)
+
+
+if __name__.startswith("bokeh_app_"):
+ doc = bokeh.plotting.curdoc()
+ add_scheduler_map_app(doc)
diff --git a/schedview/app/start.py b/schedview/app/start.py
new file mode 100644
index 00000000..528bbbc0
--- /dev/null
+++ b/schedview/app/start.py
@@ -0,0 +1,26 @@
+import bokeh.command.bootstrap
+from pathlib import Path
+
+
+def start_app(app_name):
+ """Start a bokeh app.
+
+ Parameters
+ ----------
+ app_name : `str`
+ The name of the bokeh app (and submodule of schedview.app)
+
+ """
+ base_dir = Path(__file__).resolve().parent
+ app_dir = Path(base_dir, app_name).as_posix()
+ bokeh.command.bootstrap.main(["bokeh", "serve", app_dir])
+
+
+def sched_maps():
+ """Start the sched_maps app."""
+ start_app("sched_maps")
+
+
+def metric_maps():
+ """Start the metric_maps app."""
+ start_app("metric_maps")