Submodules

These are the submodules of the scopes package.

scopes.scheduler_components module

The scopes.scheduler_components module includes various components and helper functions used by the scheduler.

class scopes.scheduler_components.Merit(name, func, merit_type, parameters={}, weight=1.0)

Bases: object

evaluate(observation, **kwargs)

Evaluate the function with the given observation and additional arguments.

Parameters:

• observation (Observation) – The input observation.

• **kwargs – Additional keyword arguments.

Returns:

float

Return type:

The evaluation result.

class scopes.scheduler_components.Night(night_date, observations_within, observer)

Bases: object

calculate_culmination_window(obs_list)

Calculate the culmination window for the night based on the given list of observations.

Parameters:

obs_list (List[Observation]) – A list of Observation objects.

calculate_solar_midnight()

Calculate the solar midnight for the night.

Returns:

solar_midnight – The calculated solar midnight in UTC.

Return type:

astropy.time.Time

class scopes.scheduler_components.Observation(target, duration, instrument=None)

Bases: object

efficiency()

Determines the efficiency of the target based on the efficiency merits.

Returns:

The efficiency value, which is the product of all efficiency merit values.

Return type:

float

evaluate_score()

Evaluates the score of the observation based on fairness, sensibility, and efficiency.

Returns:

The score of the observation.

Return type:

float

fairness()

Calculate the fairness score of the target.

The fairness score is calculated by multiplying the priority of the target+program with the product of the evaluations of all fairness merits associated with the target.

Returns:

The fairness score of the target.

Return type:

float

feasible()

Determines the feasibility of the target based on the veto merits.

Returns:

The sensibility value, which is the product of all veto merit values.

Return type:

float

set_night(night)

Set the night for the observation.

Parameters:

night (Night) – The Night object representing the night during which the observation takes place.

set_start_time(start_time)

Set the start time of the observation.

Parameters:

start_time (float) – The start time of the observation in JD (Julian Date).

skypath()

Calculate the skypath of the target during the night. To be run at the start of the scheduling process.

update_alt_airmass()

Update the altitude and airmass values throughout the observation based on the start time and exposure time of the observation.

update_start_and_score(start_time)

Update the start time of the observation and recalculate the score.

Parameters:

start_time (float) – The new start time to set for the observation.

class scopes.scheduler_components.Overheads(slew_rate_az, slew_rate_alt, cable_wrap_angle=None)

Bases: object

A class for managing the calculation of the overheads between two consecutive observations.

This class handles the calculation of various overheads involved in transitioning from one observation position to another.

add_overhead(overhead_func, can_overlap_with_slew=False)

Add an overhead function to the list of overheads.

The function must have exactly two parameters named ‘observation1’ and ‘observation2’.

Parameters:

• overhead_func (Callable) – The overhead function to be added.

• can_overlap_with_slew (bool, optional) – Whether this overhead can overlap with the slew time. Defaults to False.

calculate_slew_time(obs1, obs2)

Calculate the slew time between two observations, taking into account cable wrap.

Parameters:

• obs1 (Observation) – The first observation.

• obs2 (Observation) – The second observation.

• cable_wrap_angle (float) – The azimuth angle where the cable wrap limit is at, in degrees.

Returns:

slew_time – The slew time in seconds.

Return type:

float

calculate_transition(observation1, observation2)

Calculate the total overhead time between two observations.

Parameters:

• observation1 (Observation) – The first observation.

• observation2 (Observation) – The second observation.

Returns:

The total overhead time (in days) between the two observations.

Return type:

float

class scopes.scheduler_components.Plan

Bases: object

A container class for Observation objects representing a plan for scheduling observations.

add_observation(observation)

Add an observation to the plan.

Parameters:

observation (Observation) – The Observation object to be added to the plan.

Returns:

Returns self, to allow method chaining.

Return type:

Plan

calculate_avg_airmass()

Calculate the average airmass of the observations in the plan.

calculate_overhead()

Calculates the overheads for the entire plan, as well as the total observation time.

This method sets the attributes: observation_time, overhead_time, unused_time, overhead_ratio, and observation_ratio.

evaluate_plan(w_score=0.2, w_overhead=0.8)

Calculates the evaluation of the plan. This is the mean of the individual scores of all observations times the observation ratio of the plan. This is to compensate between maximum score of the observations but total observation time.

Parameters:

• w_score (float, optional) – The weight of the score in the evaluation. Defaults to 0.2.

• w_overhead (float, optional) – The weight of the overhead in the evaluation. Defaults to 0.8.

Returns:

The evaluation of the plan.

Return type:

float

plot(display=True, path=None)

Plot the schedule for the night.

Parameters:

• display (bool, optional) – Option to display the plot. Defaults to True.

• path (str, optional) – The path to the file where the plot will be saved. Defaults to None.

plot_altaz(color_by='program', display=True, path=None)

Plot the azimuth and altitude of the observations.

Parameters:

• color_by (str, optional) – Determines how to color the observation segments. Can be ‘program’ (default) or ‘instrument’.

• display (bool, optional) – Option to display the plot. Defaults to True.

• path (str, optional) – The path to the file where the plot will be saved. Defaults to None.

Return type:

None

plot_interactive(path=None)

Makes an interactive plot of the Plan using Plotly.

Parameters:

path (str, optional) – The path to the file where the plot will be saved. Defaults to None.

plot_polar(display=True, path=None)

Plot the azimuth vs. altitude of the observations in a polar plot.

Parameters:

• display (bool, optional) – Option to display the plot. Defaults to True.

• path (str, optional) – The path to the file where the plot will be saved. Defaults to None.

print_stats()

Prints some general information and statistics of the plan.

to_csv(*args, **kwargs)

Create a CSV file for the pandas DataFrame representation of the Plan. This method can take any keyword argument that the pd.DataFrame.to_csv() method can take.

Parameters:

• *args – Positional arguments that will be passed to pd.DataFrame.to_csv()

• **kwargs – Keyword arguments that will be passed to pd.DataFrame.to_csv()

to_df()

Create a pandas DataFrame of the plan and return it. It saves this table in an attribute called ‘df’.

class scopes.scheduler_components.Program(progID, priority, time_share_allocated=0.0, plot_color=None)

Bases: object

set_current_time_usage(current_time_usage)

Update the time share for the program.

Parameters:

current_time_usage (float) – The current time share used by the program as a fraction (0 to 1) of the total.

class scopes.scheduler_components.Target(name, program, coords, priority=None, comment='')

Bases: object

add_merit(merit)

Adds a merit to the corresponding list based on its merit type.

Parameters:

merit (Merit) – The merit object to be added

add_merits(merits)

Adds a list of Merit objects to the instance.

Parameters:

merits (List[Merit]) – A list of Merit objects to be added

scopes.merits module

The scopes.merits module provides functions and classes for calculating various merit functions used in scheduling observations.

Definition of all the generic merit functions

scopes.merits.airmass(observation, limit, alpha=0.0001)

Merit function on the current airmass of the target. It uses a hyperbolic tangent function to gradually increase the merit as the airmass decreases. The specific shape can be set with the parameters limit and alpha. The airmass considered is the maximum that the observation reaches during its exposure time. The exact formula is:

m = (tanh((limit - current_airmass) / alpha) + 1) / 2

By default the alpha parameter is set to 0.0001 which in practice is equivalent to a step function, or a simple boolean with a hard limit.

Parameters:

• observation (Observation) – The Observation object to be used

• limit (float, optional) – The limit airmass.

• alpha (float, optional) – A measure of the tolerance around the maximum airmass. Defaults to 0.0001, equivalent to a step function.

Return type:

float

scopes.merits.airmass_efficiency(observation)

Airmass efficiency merit function. Defined as the inverse of the maximum airmass reached during the observation.

Parameters:

observation (Observation) – The observation object.

Return type:

float

scopes.merits.altitude(observation, min=20, max=90)

Altitude merit function.

Parameters:

observation (Observation) – The Observation object to be used

Return type:

float

scopes.merits.at_night(observation)

Merit function that returns 1 if the observation is within the chosen night time limits, and 0 otherwise.

Parameters:

observation (Observation) – The Observation object to be used

Return type:

float

scopes.merits.culmination(observation)

Culmination constraint merit function. This merit calculates the current height of the target and the proportion to the maximum height it will reach during the current night.

Return type:

float

scopes.merits.culmination_efficiency(observation)

This merit is designed to make the scheduling of astronomical observations more efficient by expanding the selection of observable stars beyond those reaching their culmination (highest point in the sky) during the night. The normal Culmination merit misses out on stars that culminate before the night begins or after it ends, even though these stars are still observable at acceptable altitudes during the night.

To address this, it uses an extended time range for calculating merits beyond the actual observable hours of the night. This extended range includes the earliest culmination point of any star still observable at the night’s start and the latest culmination point of any star still observable at the night’s end.

This extended timerange is then mapped to the actual night time range where observations will be taken (typically within nautical or astronomical twilights). Its a simple one-to-one mapping between two different time ranges. Then to calcualte the merit, the time at which the star actually culminates is mapped from the larger timerange to the actual night range. That new mapped time is when that star will peak in this merit function.

The method shifts observation priorities throughout the night—prioritizing setting stars early on, stars near their culmination around the middle, and rising stars towards the end. This method is a compromise between observing stars at their highest point but also giving priority to stars that are setting or rising.

Parameters:

observation (Observation) – The Observation object to be used

Return type:

float

scopes.merits.end_time(observation, time)

Veto merit function that returns 0 if the observation ends after the given time, and 1 otherwise.

Parameters:

• observation (Observation) – The Observation object to be used

• time (float) – The end time of the observation

Return type:

float

scopes.merits.gaussian(x, sigma)

A simple Gaussian.

Parameters:

• x (float) – The x value at which to evaluate the Gaussian.

• sigma (float) – Measure of the width of the Gaussian.

scopes.merits.moon_separation(observation, theta_lim=20.0, theta_start=30.0, alpha=3.0)

Moon separation constraint merit function TODO: this entire merit function has to be tested.

m = ((sep - min_sep) / (max_sep - min_sep)) ** alpha

Parameters:

• observation (Observation) – The Observation object to be used

• min (float, optional) – The minimum separation to the moon in degrees. Defaults to 30.0.

Return type:

float

scopes.merits.phase_specific(observation, epoch, period, sigma, phases=[0.0])

This merit is used when the observation should be taken at specific phases of a periodic time interval. In exoplanet detections, for example, this is used when observations have to be taken at a specific phase of the orbit of a planet. Its analytic expression is a slight modification of the merit presented by Granzer (2004):

\[m(x, \phi) = \sum_{i=-1}^{1} \exp\left(-\frac{(x(t) - (\phi - i))^2}{2 \sigma^2}\right)\]

\[x(t) = \frac{\mod(t - t0, p)}{p}, \text{ for } |x| \leq 0.5\]

where t is time, p is the period, phi is the desired phase at which to observe, and sigma is the standard deviation of the Gaussian in phase space. If more than one phase is given, the merit will be the max of the merits for each phase:

\[m = \max_{\phi} m(x, \phi)\]

Parameters:

• observation (Observation) – The Observation object to be used

• epoch (float) – The phase at which the merit will be centered

• period (float) – The period of the observation

• phases (list, optional) – List of phases at which the gaussians should peak. Defaults to [0.0].

Return type:

float

scopes.merits.priority(observation, prog_base=1.0, prog_offset=0.1, tar_base=0.0, tar_offset=0.05)

Priority merit function. This is a fairness merit function to schedule based on priority. In this implementation it assumes that the priority values given is between 0 and 3, where 0 is the highest priority and 3 is the lowest. This priority value is mapped to a base values for the priority level of 2, and then an offset is added above or below that base value depending if the priority is 3, 1, or 0. This is to convert this priority scale to something close to 0 or 1 usually.

The default values reflect a system where the program priorities are centered at 1, and then the target priorities are centered at 0 which work as a modifier to the program priority. This means that the priority of the program takes precedent, and then the priority of the targets help to decide within the program which targets to observe first. There is a small overlap, where a target with priority 0 can have a mapped priority higher than a target with priority 3 of the next highest priority program. At this stage we don’t have to worry that this generates more observations for the higher priority programs because that is balanced by the time share merit.

Parameters:

• observation (Observation) – The Observation object to be used

• prog_base (float) – The base value for the priority level of 2, for programs.

• prog_offset (float) – The offset to be added to the base value of programs for the priority levels of 3, 1 and 0.

• tar_base (float) – The base value for the priority level of 2, for targets.

• tar_offset (float) – The offset to be added to the base value of targets for the priority levels of 3, 1 and 0.

Return type:

float

scopes.merits.start_time(observation, time)

Veto merit function that returns 0 if the observation starts before the given time, and 1 otherwise.

Parameters:

• observation (Observation) – The Observation object to be used

• time (float) – The start time of the observation

Return type:

float

scopes.merits.time_critical(observation, start_time, start_time_tolerance, steepness=0.0014)

Calculate the time criticality merit of an observation. It uses a double hyperbolic tangent function to gradually increase the merit as the observation approaches the desired start time. The center times of the increasing tanh and decresing tanh, are start_time - start_time_tolerance and start_time + start_time_tolerance, respectively.

TODO Rethink this merit and how to ensure that time critical observations are done without failure. This merit is not enough to ensure that the observation is done at the desired time as a previous observation can cover the entire tolerance range and thus block the time critical observation from being done.

Parameters:

• observation (Observation) – The observation object.

• start_time (float) – The desired start time for the observation. In Julian Date (JD).

• start_time_tolerance (float) – The tolerance around the desired start time in days.

• steepness (float, optional) – The steepness of the hyperbolic tangent function. A measure of how much time it takes the function to go from 0 to the max value in days. Defaults to 0.0014 days, which is ~2 minutes.

Return type:

float

scopes.merits.time_share(observation, alpha=3, beta=5.0, delta=0.05)

Time share fairness merit. It uses a modified sigmoid function to calculate the merit. The specific shape can be set with the parameters alpha and beta. The exact formula is:

m = (delta / (1 + np.exp((pct_diff / beta) ** alpha))) + (1 - delta / 2)

It’s shaped in a way so that there is some permissiveness. This means that a program can be over or under the allocated time by a certain percentage before its priority is decreased or increased. The alpha parameter controls how sharp the sigmoid is, and the beta parameter controls how much difference is allowed (in percentage).

Parameters:

• observation (Observation) – The Observation object to be used

• alpha (float, optional) – The attack parameter for how sharp the sigmoid is. Defaults to 3.

• beta (float, optional) – The leeway parameter for how much difference in time use is allowed. Defaults to 5.

• delta (float, optional) – The maximum percentage increase or decrease that will be applied if a program is over or under its allocated time. Defaults to 0.1.

Returns:

merit – The time share merit of the observation

Return type:

float

scopes.scheduler module

The scopes.scheduler module contains the core scheduling algorithms and utilities for managing observation schedules.

class scopes.scheduler.BeamSearchPlanner(*args, **kwargs)

Bases: Scheduler

class PrioritizedItem(score, plan, obs)

Bases: object

obs: Any

plan: Any

score: float

run(max_plan_length=None, K=5)

Return type:

Plan

class scopes.scheduler.DPPlanner(*args, **kwargs)

Bases: Scheduler

dp_recursion(remaining_observations, current_plan, max_plan_length, K=5)

Dynamic programming recursive function to find the best plan from a given state. It uses beam search to consider only the top K observations at each step.

Return type:

Tuple[float, Plan]

reset_dp()

run(max_plan_length=None, K=5)

Return type:

Plan

class scopes.scheduler.Scheduler(night, obs_list, overheads, plan_start_time=None, plan_end_time=None)

Bases: object

Base class for all schedulers.

This class doesn’t actually do any scheduling, but it sets up the necessary initializations and provides some helper functions that are common to all schedulers.

lsh_optimization(plan, method, metric, iterations=None, verbose=False)

Optimize the plan using the LSH (Local Search Heuristic) algorithm. The algorithm works by going through all possible swaps or moves of observations and checking if the chosen metric is improved and takes the best one. If it is, the swap or move is accepted, otherwise it is rejected. The process is repeated for a number of iterations or until the metric doesn’t improve. The algorithm can be used to optimize the plan based on the overhead time or the score of the plan. The method can be either “swap” or “move”. Swap swaps two observations in the plan, while move moves an observation to a new spot in the plan.

Parameters:

• plan (Plan) – The plan to optimize

• method (str) – The method to use for the optimization. Can be either “swap” or “move”.

• metric (str) – The metric to use for the optimization. Can be either “overheads” or “score”.

• iterations (int, optional) – The number of iterations to run the optimization for. If None, it runs until the score doesn’t improve, by default None

• verbose (bool, optional) – If True, print the progress of the optimization, by default False

Returns:

The optimized plan

Return type:

Plan

move_observation(plan, index, new_index)

Move an observation to a new index in the plan.

V2: Implement it more efficiently. Not as a series of swaps but directly placing the observation at the new location and moving all the observations in the middle as a block.

Parameters:

• plan (Plan) – The plan to move the observation in

• index (int) – The index of the observation to move

• new_index (int) – The new index to move the observation to

Returns:

The plan with the moved observation

Return type:

Plan

swap_observations(plan, index1, index2)

Swaps any two observations in the plan adjusts the start times and overheads for this new schedule.

Parameters:

• plan (Plan) – The plan to swap observations in

• index1 (int) – The index of the first observation to swap

• index2 (int) – The index of the second observation to swap

Returns:

The plan with the swapped observations

Return type:

Plan

transition(obs1, obs2)

Use the overheads class to calculate the transition time from obs1 to obs2. Then update the start time of obs2 and recalculate the score of obs2.

Parameters:

• obs1 (Observation) – The first observation

• obs2 (Observation) – The second observation

update_start_from_prev(observations, previous_obs)

Update the start time of all observations in the list based on the previous observation.

Parameters:

• observations (List[Observation]) – The list of observations to update

• previous_obs (Observation) – The previous observation

update_start_times(observations, new_start_time)

Update the start time of all observations in the list based on defined start time.

Parameters:

• observations (List[Observation]) – The list of observations to update

• new_start_time (float) – The new start time to set for all observations

class scopes.scheduler.generateQ(*args, **kwargs)

Bases: Scheduler

forwardP(start_obs, available_obs, lookahead_distance=None)

Basic scheduler that simply continues a Plan from the starting observation by sequentially choosing the highest scoring observation.

run(max_plan_length=None, K=5)

The run function for the generateQ Scheduler. The way it works is by using the forwardP function to generate a plan from the starting time to the end of the night but at each step it does this for the top K observations, and it chooses the observation that has the highest plan score at the end of the night. That would become the first observation of the plan. After each step it performs a Local Search Heuristic optimization that optimizes the order of the observation to minimize overheads. Then for the second it repeats the same process. It takes the top K observations runs forwardP until the end of the night, and assesses the plan score (but of the entire night, including the observation that was added before). It then chooses the observation that has the highest plan score at the end of the night, and that becomes the second observation of the plan. It repeats this process until the plan is full or it reaches the maximum plan length.

At the end a final LSH optimization is run to maximize the score of the overall plan. This usually means reordering the observation to achieve a higher average airmass for the observations.

Parameters:

• max_plan_length (int, optional) – The maximum length of the plan, by default None meaning it will go until the end of the night

• K (int, optional) – The number of top observations to consider at each step, by default 5

Returns:

The final plan

Return type:

Plan