CscCommander¶

class lsst.ts.salobj.CscCommander(name: str, index: int | None = 0, enable: bool = False, exclude: Sequence[str] | None = None, exclude_commands: Sequence[str] = (), fields_to_ignore: Sequence[str] = ('ignored', 'value'), telemetry_fields_to_not_compare: Sequence[str] = ('timestamp',), telemetry_fields_compare_digits: dict[str, int] | None = None)¶

Bases: object

Command a CSC from the command line.

Parameters:

namestr: SAL component name of CSC.
indexint, optional: SAL index of CSC.
enablebool, optional: Enable the CSC (when the commander starts up)? Note: amain always supplies this argument.
excludeList [str] or None, optional: Names of telemetry or event topics to not print. If None or empty then no topics are excluded.
exclude_commandsList [str], optional: Names of commands to exclude.
fields_to_ignoreList [str], optional: SAL topic fields names to ignore when specifying command parameters, and when printing events and telemetry.
telemetry_fields_to_not_compareList [str], optional: Telemetry field names to ignore when checking if a telemetry topic has changed enough to print the latest sample. These fields are printed.
telemetry_fields_compare_digitsdict [str, int]: Dict of telemetry field name: number of digits to round the value to before comparing to see if the value has changed. Silently ignored for non-float fields and fields that appear in telemetry_fields_to_not_compare.

Attributes:

domainDomain: DDS domain.
remoteRemote: Remote for the CSC being commanded.
help_dictdict: Dict of command_name: documentation. You should add one entry for every do_command method you define. Each documentation string should start with a list of argument names (the command name will be prepended for you) optionally followed by # and a brief description. The documentation string should a single line, if practical.

Notes

Warning: use with caution. Running a commander may interfere with telescope operations! If you provide a command-line script in bin/ that runs a commander, consider picking a name that is obscure and not easily confused with the script that runs the CSC.

Make a subclass of CscCommander if you want to add extra commands or provide special handling for some commands, events or telemetry. Subclasses may provide overrides as follows:

To override handling of a standard command (one defined in the XML): define an async do_<command_name> method. The method receives one argument: a list of str arguments. You should provide a custom handler for, or hide, any command with array parameters, because the command parser only accepts scalars.
To add an additional command (one not defined in the XML):
- Define an async do_<command_name> to handle the command. The method receives one argument: a list of str arguments.
- Add an entry to help_dict. The key is the command name and the value is a brief (preferably only one line) help string that lists the arguments first and a brief description after. Here is an example:
  self.help_dict["sine"] = \ "start_position amplitude # track one cycle of a sine wave"
To override handling of an event or telemetry topic: define async method evt_<event_name>_callback or tel_<event_name>_callback, respectively. It receives one argument: the message data. This can be especially useful if the default behavior is too chatty for one or more telemetry topics.

Examples

You can do a lot with just the base class. This command-line script will control the Test CSC pretty well:

import asyncio
from lsst.ts import salobj
asyncio.run(salobj.CscCommander.amain(name="Test", index=True))

However, this does not handle array-valued command parameters. See TestCscCommander for an example that handles arrays. TestCscCommander is run as follows:

import asyncio
from lsst.ts import salobj
asyncio.run(salobj.TestCscCommander.amain(index=True))

For an example with extra commands and special telemetry handling, see RotatorCommander in the ts_rotator package.

Unit Testing

To unit test a commander:

Set commander.testing = True. This appends all output messages to output queue commander.output_queue, a collections.deque. The messages are also printed to stdout, as usual.
Call commander.run_command to run a command.
Check the output using the output queue.
Clear the output queue with commander.output_queue.clear() whenever you like, e.g. before running a command whose output you want to check.

Methods Summary

`add_arguments`(parser)	Add arguments to the parser created by `make_from_cmd_line`.
`add_kwargs_from_args`(args, kwargs)	Add constructor keyword arguments based on parsed arguments.
`amain`(, index, *kwargs)	Construct the commander and run it.
`check_arguments`(args, *names)	Check that the required arguments are provided.
`close`()	Close the commander, prior to quitting.
`do_start`(args)	Allow the start command to have no arguments.
`event_callback`(data, name)	Generic callback for events.
`evt_logMessage_callback`(data)	Abbreviate the log output by omitting less-interesting fields.
`evt_summaryState_callback`(data)
`field_is_public`(name)	Return True if the specified field name is public, False otherwise.
`format_data`(data)	Format the public fields of an event or telemetry sample, for printing.
`format_dict`(data)	Format a dict for printing.
`format_item`(key, value)	Format one event or telemetry field for printing.
`get_commands_help`()	Get help for each command, as a list of strings.
`get_public_data`(data)	Get a dict of field_name: value for public fields of a message.
`get_rounded_public_data`(data[, digits])	Get a dict of field_name: value for public fields of a DDS sample with float values rounded.
`get_telemetry_comparison_dict`(public_dict, ...)	Get a dict of field name: rounded data, for comparing new telemetry to old.
`make_from_cmd_line`(index, **kwargs)	Construct a SAL-related class from command line arguments.
`output`(msg)	Print a message to output, appending a final newline.
`print_help`()	Print help.
`run_command`(cmd)	Run the specified command string and wait for it to finish.
`run_command_topic`(command_name, args)	Run a command that has an associated salobj RemoteCommand topic.
`start`()	Start asynchonous processes.
`telemetry_callback`(data, name[, digits])	Default callback for telemetry topics.

Methods Documentation

classmethod add_arguments(parser: ArgumentParser) → None¶

Add arguments to the parser created by make_from_cmd_line.

Parameters:

parserargparse.ArgumentParser: The argument parser.

Notes

If you override this method then you should almost certainly override add_kwargs_from_args as well.

classmethod add_kwargs_from_args(args: Namespace, kwargs: dict[str, Any]) → None¶

Add constructor keyword arguments based on parsed arguments.

Parameters:

argsargparse.namespace: Parsed command.
kwargsdict: Keyword argument dict for the constructor. Update this based on args. The index argument will already be present if relevant.

Notes

If you override this method then you should almost certainly override add_arguments as well.

async classmethod amain(*, index: int | IntEnum | bool | None, **kwargs: Any) → None¶

Construct the commander and run it.

Parse the command line to construct the commander, then parse and execute commands until the exit is seen.

Parameters:

indexint, enum.IntEnum, True, or None

If the CSC is indexed, do one of the following:

Specify True to make index a required command-line argument that accepts any nonzero index.
Specify an enum.IntEnum class to make index a required command-line argument that only accepts the enum values.
Specify a non-zero integer to use that index. This is rare; if the CSC is indexed then the user should usually be allowed to specify the index.

If the CSC is not indexed, specify None or 0.

**kwargsdict, optional

Additional keyword arguments for your CSC’s constructor.

check_arguments(args: Sequence[str], *names: str | tuple[str, collections.abc.Callable[[str], Any]]) → dict[str, Any]¶

Check that the required arguments are provided. and return them as a keyword argument dict with cast values.

Parameters:

argsList [str]

Command arguments, as strings.

*namesList [str or tuple]

Argument name and optional cast function. Each element is either:

An argument name, in which case the argument is cast to a float
A tuple of (name, cast function), in which case the argument is cast using the cast function. The cast function takes one str argument and returns the cast value.

async close() → None¶: Close the commander, prior to quitting.

async do_start(args: Sequence[str]) → None¶: Allow the start command to have no arguments.

async event_callback(data: BaseMsgType, name: str) → None¶

Generic callback for events.

You may provide evt_<event_name> methods to override printing of specific events.

async evt_logMessage_callback(data: BaseMsgType) → None¶

Abbreviate the log output by omitting less-interesting fields.

Omit traceback unless it is non-blank. Always omit filePath, functionName, lineNumber, process, and timestamp.

async evt_summaryState_callback(data: BaseMsgType) → None¶

field_is_public(name: str) → bool¶: Return True if the specified field name is public, False otherwise.

format_data(data: BaseMsgType) → str¶: Format the public fields of an event or telemetry sample, for printing.

format_dict(data: dict[str, Any]) → str¶

Format a dict for printing.

Unlike format_data, this requires a dict and formats all fields.

format_item(key: str, value: Any) → str¶: Format one event or telemetry field for printing.

get_commands_help() → list[str]¶

Get help for each command, as a list of strings.

End with “Other Commands:” and any commands in help_dict that are not in command_dict.

get_public_data(data: BaseMsgType) → dict[str, Any]¶

Get a dict of field_name: value for public fields of a message.

Parameters:

dataBaseMsgType: Message.

get_rounded_public_data(data: BaseMsgType, digits: int = 2) → dict[str, Any]¶: Get a dict of field_name: value for public fields of a DDS sample with float values rounded.

get_telemetry_comparison_dict(public_dict: dict[str, Any], digits: int) → dict[str, Any]¶

Get a dict of field name: rounded data, for comparing new telemetry to old.

Parameters:

public_datadict [str, Any]: Dict of field_name: value containing public fields that are to be compared.
digitsint: The default number of digits to which to round float values. May be overridden for specific fields using constructor argument telemetry_fields_compare_digits.

classmethod make_from_cmd_line(index: int | IntEnum | bool | None, **kwargs: Any) → CscCommander¶

Construct a SAL-related class from command line arguments.

Parameters:

indexint, enum.IntEnum, True, or None

If the CSC is indexed, do one of the following:

Specify True to make index a required command-line argument that accepts any nonzero index.
Specify an enum.IntEnum class to make index a required command-line argument that only accepts the enum values.
Specify a non-zero integer to use that index. This is rare; if the CSC is indexed then the user should usually be allowed to specify the index.

If the CSC is not indexed, specify None or 0.

**kwargsdict, optional

Additional keyword arguments for your class’s constructor.

Returns:

instancecls: The constructed instance.

Notes

To add additional command-line arguments, override add_arguments and add_kwargs_from_args.

output(msg: str) → None¶

Print a message to output, appending a final newline.

Please call this instead of print to support unit tests.

If self.testing is True then append str to self.output_queue instead of printing it. Use this mode for unit testing a CSC commander.

print_help() → None¶: Print help.

async run_command(cmd: str) → None¶

Run the specified command string and wait for it to finish.

Parameters:

cmdstr: Command string (command name and arguments). Note: does not handle the “exit” command.

async run_command_topic(command_name: str, args: Sequence[str]) → None¶

Run a command that has an associated salobj RemoteCommand topic.

Parameters:

command_namestr: Command name, e.g. Enable
argsList [str]: String arguments for the command. There must be exactly one argument per public fields.

Notes

This method works for command topics that take scalar arguments. To support command topics with more exotic arguments you must provide a do_<command> method that parses the arguments and add an entry to self.help_dict.

async start() → None¶: Start asynchonous processes.

async telemetry_callback(data: BaseMsgType, name: str, digits: int = 2) → None¶

Default callback for telemetry topics.

Print the telemetry information if it has changed enough (as specified by digits).

Parameters:

datatype_hints.BaseMsgType: The message data.
namestr: The name to print. Typically the basic topic name with no prefix.
digitsint: The default number of digits to which to round float values before comparing if fields have changed enough to be worth printing the new data. May be overridden for specific fields using constructor argument telemetry_fields_compare_digits.

Notes

This method cannot be directly assigned as a topic callback. Use functools.partial or similar to specify name, and, if you like, digits.
This method can also be used for events, if you only want The event published when the data changes significantly. One good candidate is the generic clockOffset event.

Navigation

CscCommander¶