SequentialScheduler

class lsst.ts.scheduler.driver.SequentialScheduler(models: dict[str, Any], raw_telemetry: dict[str, Any], observing_blocks: dict[str, lsst.ts.observing.block.ObservingBlock], parameters: Any = None, log: Logger | None = None)

Bases: Driver

A simple scheduler driver that implements a sequential scheduler algorithm.

The driver reads from an input file of targets provided by the user and send one target at a time.

Parameters:

modelsdict`[`str, typing.Any]

Models.

raw_telemetrydict`[`str, typing.Any]

Raw telemetry

observing_blocksdict`[`str, observing.ObservingBlock]

Observing blocks.

parameterstyping.Any, optional

Driver parameters, by default None

loglogging.Logger | None, optional

Logger, by default None

Methods Summary

assert_survey_observing_script(survey_name)	Assert that the input survey name has a dedicated observing script.
cold_start(observations)	Rebuilds the internal state of the scheduler from a list of observations.
configure_scheduler([config])	This method is responsible for running the scheduler configuration and returning the survey topology, which specifies the number, name and type of projects running by the scheduler.
convert_efd_observations_to_targets(...)	Convert EFD dataframe into list of driver targets.
get_state_as_file_object(targets_queue)	Get the current state of the scheduling algorithm as a file object.
get_stop_tracking_target()
get_survey_observing_block(survey_name)	Return the appropriate survey observing block.
get_survey_topology(config)	Get the survey topology.
load(config)	Load a new set of targets.
parse_observation_database(filename)	Parse an observation database into a list of observations.
register_observation(target)	Register observations.
register_observed_target(target)	Validates observed target and returns an observation.
reset_from_state(filename)	Load the state from a file.
save_state()	Save the current state of the scheduling algorithm to a file.
schema()	Get schema for the sequential scheduler algorithm configuration files.
select_next_target()	Picks a target and returns it as a target object.
select_next_targets()	Select a list of targets.
update_conditions()	Update driver internal conditions.

Methods Documentation

assert_survey_observing_script(survey_name: str) → None

Assert that the input survey name has a dedicated observing script.

Parameters:

survey_namestr

Name of the survey.

Raises:

AssertError:

If survey_name is not in the list of valid survey observing scripts.

cold_start(observations: list[lsst.ts.scheduler.driver.observation.Observation]) → None

Rebuilds the internal state of the scheduler from a list of observations.

Parameters:

observationslist`[`Observation]

List of observations.

configure_scheduler(config=None)

This method is responsible for running the scheduler configuration and returning the survey topology, which specifies the number, name and type of projects running by the scheduler.

By default it will just return a test survey topology.

Parameters:

configtypes.SimpleNamespace

Configuration, as described by schema/Scheduler.yaml

Returns:

survey_topology: lsst.ts.scheduler.kernel.SurveyTopology

convert_efd_observations_to_targets(efd_observations: DataFrame) → List[DriverTarget]

Convert EFD dataframe into list of driver targets.

Parameters:

efd_observationspandas.DataFrame

Data frame returned from a query to the EFD for observations.

get_state_as_file_object(targets_queue: list[lsst.ts.scheduler.driver.driver_target.DriverTarget])

Get the current state of the scheduling algorithm as a file object.

Returns:

file_objectio.BytesIO

File object with the current.

get_stop_tracking_target() → DriverTarget

get_survey_observing_block(survey_name: str) → ObservingBlock

Return the appropriate survey observing block.

Parameters:

survey_namestr

Name of the survey.

Returns:

observing.ObservingBlock

Survey observing block.

get_survey_topology(config: SimpleNamespace) → SurveyTopology

Get the survey topology.

Parameters:

configtypes.SimpleNamespace

Configuration, as described by schema/Scheduler.yaml

Returns:

survey_topologySurveyTopology

Survey topology.

load(config)

Load a new set of targets.

Format of the input file must be the same as that for the configure method. The targets will be appended to the existing queue.

Raises:

RuntimeError:

If driver is not configured.

parse_observation_database(filename: str) → None

Parse an observation database into a list of observations.

Parameters:

filenamestr

Returns:

observationslist of DriverTarget

register_observation(target: DriverTarget) → None

Register observations.

This method should store the observation in a way that can be retrieved afterwards by the driver.

The default implementation is not implemented.

Parameters:

targetDriverTarget

Observation to register.

register_observed_target(target: DriverTarget) → Observation

Validates observed target and returns an observation.

Parameters:

targetDriverTarget

Observed target to register.

Returns:

Observation

Registered observation.

reset_from_state(filename)

Load the state from a file.

Parameters:

filenamestr

Name of the file with the state.

save_state()

Save the current state of the scheduling algorithm to a file.

Returns:

filenamestr

Name of the file with the state.

schema() → dict[str, Any]

Get schema for the sequential scheduler algorithm configuration files.

Returns:

dict`[str, `typing.Any]

Schema.

select_next_target() → DriverTarget

Picks a target and returns it as a target object.

Returns:

DriverTarget

Target to observe.

select_next_targets() → list[lsst.ts.scheduler.driver.driver_target.DriverTarget]

Select a list of targets.

By default it will just return a dummy test target.

Returns:

list [DriverTarget]

update_conditions() → None

Update driver internal conditions.

When subclassing this method, make sure to call it at the start of the method, as it performs operations like running the observatory through the current targets on the queue.