ATCalSys

class lsst.ts.observatory.control.auxtel.ATCalSys(electrometer_index: int = 1, fiber_spectrograph_index: int = -1, domain: Optional[Domain] = None, log: Optional[Logger] = None, intended_usage: Optional[int] = None)

Bases: RemoteGroup

Implement high level ATCalSys functionality.

Parameters:

domain: `salobj.Domain`

Domain to use of the Remotes. If None, create a new domain.

electrometer_indexint

Electrometer index.

fiber_spectrograph_indexint

FiberSpectrograph index.

loglogging.Logger

Optional logging class to be used for logging operations. If None, creates a new logger. Useful to use in salobj.BaseScript and allow logging in the class use the script logging.

intended_usage: `int`

Optional integer that maps to a list of intended operations. This is used to limit the resources allocated by the class by gathering some knowledge about the usage intention. By default allocates all resources.

Attributes Summary

components	List of components names.
components_attr	List of remotes names.
electrometer
fiberspectrograph
usages	Define class usages.
valid_use_cases	Define valid usages.

Methods Summary

assert_all_enabled([message])	Check if all components are in the enabled state.
assert_liveliness()	Assert liveliness of components belonging to the group.
cancel_not_done(tasks)	Cancel all coroutines in coro_list.
check_comp_heartbeat(component)	Monitor heartbeats from the specified component and raises and exception if not.
check_component_state(component[, desired_state])	Monitor the summary state of a component and raises an exception if it is or goes to a state different than the desired state.
close()
components_to_check()	Return components for which check is enabled.
electrometer_scan(duration)	Perform an electrometer scan for the specified duration and return a large file object topic.
enable([overrides])	Enable all components.
expand_overrides([overrides])	Expand an overrides dict with entries for every component.
get_heartbeat(component)	Get last heartbeat for component.
get_identity()	Get user identity.
get_required_resources(component, intended_usage)	Return the required resources based on the intended usage of the class.
get_simulation_mode([components])	Return a list with the simulation mode for components in the group.
get_software_versions([components])	Return a list with the software versions for components in the group.
get_state(component[, ignore_timeout])	Get summary state for component.
get_work_components([components])	Parse input into a list of valid components from the group.
next_heartbeat(component)	Get next heartbeat for component.
next_state(component)	Get summary state for component.
offline()	Put all CSCs in offline.
process_as_completed(tasks)	Process tasks are they complete.
reset_checks()	Reset all checks, enabling check for all components.
set_rem_loglevel(level)	Set remotes log level.
set_state(state[, overrides, components])	Set summary state for all components.
setup_monochromator(wavelength, ...)	Setup Monochromator.
standby()	Put all CSCs in standby.
take_fiber_spectrum_after(delay, image_type, ...)	Wait, then start an acquisition with the fiber spectrograph.

Attributes Documentation

components

List of components names.

The name of the CSC follow the format used in the class constructor, e.g. CSCName or CSCName:index (for indexed components), e.g. “Hexapod:1” (for Hexapod with index=1) or “ATHexapod”.

Returns:

componentslist of str

List of CSCs names.

components_attr

List of remotes names.

The remotes names are reformatted to fit the requirements for object attributes. It will be the name of the CSC (as in components) in lowercase, replacing the colon by an underscore, e.g. “Hexapod:1” -> “hexapod_1” or “ATHexapod” -> “athexapod”.

Returns:

components_attrlist of str

List of remotes attribute names.

electrometer

fiberspectrograph

usages

Define class usages.

This property defines what remote resources are needed for each class usages. Each item in the dictionary contain a list of components, if the remotes will be used for “read only” operations an which topics will be required. The “key” of the dictionary is the usage enumeration value.

Returns:

usagesdict

Dictionary with class usages.

valid_use_cases

Define valid usages.

When subclassing, overwrite this method to return the proper enum.

Returns:

usages: enum

Methods Documentation

async assert_all_enabled(message: str = '') → None

Check if all components are in the enabled state.

Parameters:

message: `str`

Additional message to append to error.

async assert_liveliness() → None

Assert liveliness of components belonging to the group.

The assertion is done by waiting for a new heartbeat from the component. The check feature will apply to the assertion so components marked with check=False will be skipped.

Raises:

AssertionError

If cannot get heartbeat for one or more components.

async static cancel_not_done(tasks: List[Task]) → None

Cancel all coroutines in coro_list.

Remove futures from input tasks list and cancel them.

Parameters:

taskslist [futures]

A list of coroutines to cancel.

async check_comp_heartbeat(component: str) → None

Monitor heartbeats from the specified component and raises and exception if not.

This method will run forever as long as the component continues to send heartbeats. The intention is that this can run alongside an operation, to make sure the component remains responsive.

Parameters:

componentstr

Name of the component to follow. The name of the CSC follow the format CSCName or CSCName:index (for indexed components), e.g. “Hexapod:1” (for Hexapod with index=1) or “ATHexapod”.

Raises:

RuntimeError

If the component does not send heartbeats in self.fast_timeout seconds.

async check_component_state(component: str, desired_state: State = State.ENABLED) → None

Monitor the summary state of a component and raises an exception if it is or goes to a state different than the desired state.

This method will run forever as long as the summary state remains unchanged. The intention is that this can run alongside an operation that require the component to be in a certain state, when the operation is completed, the task can be canceled.

Parameters:

componentstr

Name of the component to follow. Must be one of: atmcs, atptg, ataos, atpneumatics, athexapod, atdome, atdometrajectory

desired_statesalobj.State

Desired state of the CSC.

Raises:

RuntimeError

If state is not desired_state.

KeyError

If component is not found.

async close() → None

components_to_check() → List[str]

Return components for which check is enabled.

Returns:

list of str

Components to check.

async electrometer_scan(duration: float) → BaseMsgType

Perform an electrometer scan for the specified duration and return a large file object topic.

Parameters:

durationfloat

Total duration of scan.

Returns:

lfoself.electrometer.evt_largeFileObjectAvailable.DataType

Large File Object Available event message.

async enable(overrides: Optional[Dict[str, str]] = None) → None

Enable all components.

This method will enable all group components. Users can provide overrides for the start command (in a dictionary).

Parameters:

overrides: `dict`

Dictionary with overrides to apply. If None use recommended overrides.

async expand_overrides(overrides: Optional[Dict[str, str]] = None) → Dict[str, str]

Expand an overrides dict with entries for every component.

Any components that have no specified override are set to “”.

Parameters:

overridesdict or None

A dictionary with (component name, overrides) pair or None. The component name is as it appears in the components_attr attribute, which is the name of the CSC in lowercase, replacing “:” by “_” for indexed components, e.g. “Hexapod:1” -> “hexapod_1” or “ATHexapod” -> “athexapod”.

Returns:

complete_overridesdict

Dictionary with overrides for every component in the group. Unspecifies components have override “”.

Raises:

RuntimeError

If an item in the parameter overrides dictionary is not a CSC in the group.

async get_heartbeat(component: str) → BaseMsgType

Get last heartbeat for component.

Parameters:

componentstr

Name of the component.

Returns:

heartbeat

Last component heartbeat.

get_identity() → str

Get user identity.

Returns:

str

Identity.

get_required_resources(component: str, intended_usage: Union[None, int]) → Any

Return the required resources based on the intended usage of the class.

When subclassing, overwrite this method to add the child class use cases.

Parameters:

component: `str`

Name of the component, with index as it appears in components_attr attribute (e.g. test_1 for the Test component with index 1).

intended_usage: `int` or `None`

An integer constructed from the self.valid_use_cases. Usages can be combined to enable combined operations (see base class documentation). If None, returns appropriate values to load all resources.

Returns:

resources: types.SimpleNamespace

A simple namespaces with the following attributes:

add_this: bool

Should this remote be added to the collection?

readonly: bool

Create the remote for this component in readonly mode? Some operations does not require commanding.

include: list [str]

What topics should be included?

Raises:

KeyError: If component is not in the list of components.

async get_simulation_mode(components: Optional[List[str]] = None) → Dict[str, BaseMsgType]

Return a list with the simulation mode for components in the group.

Parameters:

componentslist of str, optional

List with the name of components to get the simulation mode. If None (default) return the values for all components.

Returns:

simulation_mode: dict

Dictionary with the name of the component and the value of simulation mode.

async get_software_versions(components: Optional[List[str]] = None) → Dict[str, BaseMsgType]

Return a list with the software versions for components in the group.

Parameters:

componentslist of str, optional

List with the name of components to get the software versions. If None (default) return the values for all components.

Returns:

software_versions: dict

Dictionary with the name of the component and the value of software versions.

async get_state(component: str, ignore_timeout: bool = False) → State

Get summary state for component.

Parameters:

componentstr

Name of the component.

ignore_timeoutbool

If True will return None in case it times out getting the state. Default is False, which means raise TimeoutError.

Returns:

statesalobj.State or None

Current state of component.

Raises:

asyncio.TimeoutError

If can not get state in self.fast_timeout seconds.

get_work_components(components: Optional[List[str]] = None) → List[str]

Parse input into a list of valid components from the group.

Parameters:

componentslist of str or None

Input list of components to process or None. If None return a list with all components.

Returns:

work_componentslist of str

List of valid components.

Raises:

RuntimeError

If a component in the components input list is not part of the group.

async next_heartbeat(component: str) → BaseMsgType

Get next heartbeat for component.

Parameters:

componentstr

Name of the component.

Returns:

heartbeat

Last component heartbeat.

async next_state(component: str) → State

Get summary state for component.

Parameters:

componentstr

Name of the component.

Returns:

statesalobj.State

Current state of component.

async offline() → None

Put all CSCs in offline.

async process_as_completed(tasks: List[Task]) → Any

Process tasks are they complete.

If the first task that finishes completes successfully, it will cancel all other tasks in the list, empty the input list and return the value of the task. If the task results in an exception, it will cancel all other tasks, empty the list and raise the exception.

Parameters:

taskslist`[`asyncio.Tasks]

List of asyncio tasks to process.

Returns:

ret_valobject

Return value from the first completed task.

reset_checks() → None

Reset all checks, enabling check for all components.

set_rem_loglevel(level: int) → None

Set remotes log level.

Useful to prevent the internal salobj warnings when read queues are filling up.

Parameters:

levelint

Log level.

async set_state(state: State, overrides: Optional[Dict[str, str]] = None, components: Optional[List[str]] = None) → None

Set summary state for all components.

Parameters:

statesalobj.State

Desired state.

overridesdict or None

Settings to apply for each component.

componentslist[`str]`

List of components to set state, as they appear in self.components_attr.

Raises:

RuntimeError

• If a component in components is not part of the group.

• If it fails to transition one or more components.

async setup_monochromator(wavelength: float, entrance_slit: float, exit_slit: float, grating: int) → None

Setup Monochromator.

Parameters:

wavelengthfloat

Wavelength in nm.

entrance_slitfloat

Size of entrance slit in mm.

exit_slitfloat

Size of exist slit in mm.

gratingint

Grating to select.

async standby() → None

Put all CSCs in standby.

async take_fiber_spectrum_after(delay: float, image_type: str, integration_time: float, lamp: str, wait_for: Optional[Awaitable] = None) → BaseMsgType

Wait, then start an acquisition with the fiber spectrograph.

By default, this method will wait for delay seconds then start an acquisition with the fiber spectrograph. Optionally the user may provide a coroutine that will be awaited before the delay starts.

Parameters:

delayfloat

Seconds to wait before starting fiber spectrograph acquisition.

image_typestr

Type of each image.

integration_timefloat

Integration time for the fiber spectrum (seconds).

lampstr

Name of lamp for each image.

wait_forcoro

An awaitable that will be waited before delay and processing. If None, ignored.

Returns:

file_datafiberspectrograph.evt_largeFileObjectAvailable.DataType

Large file object available event data.