Command Manager

Module that contains CommandManager where commands to DSC are managed.

class CommandManager(logger, immutable_user_id, scu, submit_task, update_component_state, command_progress_callback=None)[source]

Wrap all the commands to DSC.

__init__(logger, immutable_user_id, scu, submit_task, update_component_state, command_progress_callback=None)[source]

Wrap all the commands to DSC.

Parameters:

  • logger (logging.Logger) – logger

  • immutable_user_id (str) – The immutable user ID for the DSC

  • scu (SteeringControlUnit) – SteeringControlUnit instance, None if it is not established yet.

  • submit_task (Callable) – Submit a task to the task executor

execute_opcua_command(command_path, command_args, task_callback=None)[source]

Execute an OPC-UA command on the DSC.

Parameters:

  • command_path (str) – The opcua path to the command node.

  • command_args (list[Any]) – The list of args for the requested command.

  • task_callback (Optional[Callable], optional) – Update LRC status, defaults to None

Returns:

Task status

Return type:

Tuple[TaskStatus, str]

indexer_position_to_band_type(band_type)[source]

Map the OPC-UA BandType enum to the IndexerPosition enum.

Return type:

Optional[Enum]

interlock_acknowledge(task_callback=None)[source]

Request the dish to clear the interlock status.

Parameters:

task_callback (Optional[Callable], optional) – Update LRC status, defaults to None

Returns:

Task status

Return type:

Tuple[TaskStatus, str]

release_auth(task_callback=None)[source]

Release command authority.

Return type:

Tuple[TaskStatus, str]

Returns:

The result of the command execution.

reset_axes_command(task_callback=None)[source]

Reset the errors on dish structure axes.

Parameters:

task_callback (Optional[Callable], optional) – Update LRC status, defaults to None

Returns:

Task status

Return type:

Tuple[TaskStatus, str]

retake_authority(task_callback=None)[source]

Retake authority on dish structure.

Parameters:

task_callback (Optional[Callable], optional) – Update LRC status, defaults to None

Returns:

Task status

Return type:

Tuple[TaskStatus, str]

set_index_position(component_state, band, task_callback=None)[source]

Set the indexer position on the dish.

Parameters:

  • band (IndexerPosition) – The indexer position to move to.

  • task_callback (Optional[Callable], optional) – Update LRC status, defaults to None

Returns:

Task status

Return type:

Tuple[TaskStatus, str]

set_point_mode(component_state, task_callback=None)[source]

Transition the dish to POINT mode.

Parameters:

  • component_state (dict) – the current component state

  • task_callback (Optional[Callable], optional) – Update LRC status, defaults to None

Returns:

Task status

Return type:

Tuple[TaskStatus, str]

set_power_mode(component_state_callback, values, task_callback=None)[source]

Transition the dish to SetPowerMode.

Parameters:

  • values (list(float)) – (0.0 or 1.0) and limit(kW)] with 1.0 representing True

  • task_callback (Optional[Callable], optional) – Update LRC status, defaults to None

Returns:

Task status

Return type:

Tuple[TaskStatus, str]

set_standby_mode(component_state, task_callback=None)[source]

Transition the dish to STANDBY mode.

Parameters:

  • component_state (dict) – the current component state

  • task_callback (Optional[Callable], optional) – Update LRC status, defaults to None

Returns:

Task status

Return type:

Tuple[TaskStatus, str]

slew(component_state, values, speed_az, speed_el, task_callback=None)[source]

Transition the dish to SLEW.

Parameters:

  • values (list[float]) – The slew coordinates

  • speed_az (float) – Az axis speed

  • speed_el (float) – El Axis speed

  • component_state (dict) – the current component state

  • task_callback (Optional[Callable], optional) – Update LRC status, defaults to None

Returns:

Task status

Return type:

Tuple[TaskStatus, str]

stow(task_callback=None)[source]

Transition the dish to STOW immediately bypassing the LRC queue.

Parameters:

task_callback (Optional[Callable], optional) – Update LRC status, defaults to None

Returns:

Task status

Return type:

Tuple[TaskStatus, str]

take_authority(task_callback=None)[source]

Take authority on dish structure.

Parameters:

task_callback (Optional[Callable], optional) – Update LRC status, defaults to None

Returns:

Task status

Return type:

Tuple[TaskStatus, str]

track(interpolation_mode, get_current_tai_offset, task_callback=None)[source]

Transition the dish to TRACK.

Parameters:

  • interpolation_mode (TrackInterpolationMode) – The interpolation mode

  • get_current_tai_offset (Callable) – Callback to get the current tai offset

  • task_callback (Optional[Callable], optional) – Update LRC status, defaults to None

Returns:

Task status

Return type:

Tuple[TaskStatus, str]

track_load_static_off(values, task_callback=None)[source]

Load the static pointing model offsets.

Parameters:

  • values (List) – The static offset values

  • task_callback (Optional[Callable], optional) – Update LRC status, defaults to None

Returns:

Task status

Return type:

Tuple[TaskStatus, str]

track_load_table(*args: Any, **kwargs: Any) → Any[source]

Wrapper function to track updates.

Return type:

Any

track_stop(task_callback=None)[source]

Stop tracking.

Return type:

Tuple[TaskStatus, str]

unstow(component_state, task_callback=None)[source]

Request the dish to unstow.

Parameters:

  • component_state (dict) – the current component state

  • task_callback (Optional[Callable], optional) – Update LRC status, defaults to None

Returns:

Task status

Return type:

Tuple[TaskStatus, str]

update_pointing_model_params(static_on, tilt_on, amb_temp_on, band, values)[source]

Update the threshold levels for AZ/EL pointing error between actual and desired values.

Parameters:

  • static_on (bool) – Flag indicating if static pointing model correction is active.

  • tilt_on (bool) – Flag indicating if tilt correction is active.

  • amb_temp_on (bool) – Flag indicating if ambient temperature correction is active.

  • band (BandType) – The Band type for which the pointing model parameters will be applied.

  • values (list[float]) – A list containing exactly two elements: - on_source_threshold (float): The threshold value for the pointing error. - threshold_time_period (float): The time period over which the threshold is applied.

Returns:

Task status

Return type:

Tuple[TaskStatus, str]

Raises:

ValueError – If the input list does not contain exactly two elements.

update_threshold_levels(values)[source]

Update the threshold levels for AZ/EL pointing error between actual and desired values.

Parameters:

values (list[float]) – A list containing exactly two elements: - on_source_threshold (float): The threshold value for the pointing error. - threshold_time_period (float): The time period over which the threshold is applied.

Returns:

Task status

Return type:

Tuple[TaskStatus, str]

:return:TaskStatus :rtype: tuple[bool, str]

Raises:

ValueError – If the input list does not contain exactly two elements.

class CommandNotAuthorized(*args, **kwargs)[source]

Exception for incorrect command authorization.

main()[source]

For local testing.

Return type:

None

track_updates_decorator(func)[source]

Decorator to track updates with the DSC.

Return type:

Any

update_task_status(task_callback, **task_statuses)[source]

Wraps the task callback.

Return type:

None