CscCommander¶
- class lsst.ts.salobj.CscCommander(name: str, index: Optional[int] = 0, enable: bool = False, exclude: Optional[Sequence[str]] = None, exclude_commands: Sequence[str] = (), fields_to_ignore: Sequence[str] = ('ignored', 'value', 'priority'), telemetry_fields_to_not_compare: Sequence[str] = ('timestamp',))¶
Bases:
object
Command a CSC from the command line.
- Parameters
- name
str
SAL component name of CSC.
- index
int
, optional SAL index of CSC.
- enable
bool
, optional Enable the CSC (when the commander starts up)? Note:
amain
always supplies this argument.- exclude
List
[str
] orNone
, optional Names of telemetry or event topics to not print. If
None
or empty then no topics are excluded.- exclude_commands
List
[str
], optional Names of commands to exclude.
- fields_to_ignore
List
[str
], optional SAL topic fields names to ignore when specifying command parameters, and when printing events and telemetry.
- telemetry_fields_to_not_compare
List
[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.
- name
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 hide unwanted commands: delete them from
self.command_dict
in your constructor. Most CSCs should hide the “abort”, “enterControl” and “setValue” commands, as follows:for command_to_ignore in ("abort", "enterControl", "setValue"): del self.command_dict[command_to_ignore]
To override handling of a standard command (one defined in the XML): define a
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 a
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 method
evt_<event_name>_callback
ortel_<event_name>_callback
, respectively. It receives one argument: the DDS sample. This can be especially useful if the default behavior is too chatty for one or more telemetry topics.
I have not found a way to write a unit test for this class. I tried running a commander in a subprocess but could not figure out how to send multiple commands (the
subprocess.communicate
method only allows sending one item of data). Instead I suggest manually running it to control the Test CSC.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 queuecommander.output_queue
, acollections.deque
. The messages are also printed tostdout
, 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.
- Attributes
- domain
Domain
DDS domain.
- remote
Remote
Remote for the CSC being commanded.
- help_dict
dict
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.
- domain
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.
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 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 DDS sample.
get_rounded_public_data
(data[, digits])Get a dict of field_name: value for public fields of a DDS sample with float values rounded.
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.
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])Generic callback for telemetry.
Methods Documentation
- classmethod add_arguments(parser: argparse.ArgumentParser) None ¶
Add arguments to the parser created by
make_from_cmd_line
.- Parameters
- parser
argparse.ArgumentParser
The argument parser.
- 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: argparse.Namespace, kwargs: Dict[str, Any]) None ¶
Add constructor keyword arguments based on parsed arguments.
- Parameters
- args
argparse.namespace
Parsed command.
- kwargs
dict
Keyword argument dict for the constructor. Update this based on
args
. The index argument will already be present if relevant.
- args
Notes
If you override this method then you should almost certainly override
add_arguments
as well.
- async classmethod amain(*, index: Optional[Union[int, enum.IntEnum, bool]], **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
- index
int
,enum.IntEnum
,True
, orNone
If the CSC is indexed, do one of the following:
Specify
True
to makeindex
a required command-line argument that accepts any nonzero index.Specify an
enum.IntEnum
class to makeindex
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.- **kwargs
dict
, optional Additional keyword arguments for your CSC’s constructor.
- index
- check_arguments(args: Sequence[str], *names: Union[str, Tuple[str, 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
- args
List
[str
] Command arguments, as strings.
- *names
List
[str
ortuple
] 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.
- args
- event_callback(data: Any, name: str) None ¶
Generic callback for events.
You may provide evt_<event_name> methods to override printing of specific events.
- evt_summaryState_callback(data: lsst.ts.salobj.type_hints.BaseMsgType) None ¶
- field_is_public(name: str) bool ¶
Return True if the specified field name is public, False otherwise.
- format_data(data: Any) str ¶
Format the public fields of an event or telemetry sample, for printing.
- format_dict(data: Any) str ¶
Format a dict for printing.
Unlike format_data, this requires a dict and formats all fields.
- 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: Any) Dict[str, Any] ¶
Get a dict of field_name: value for public fields of a DDS sample.
- Parameters
- data
dds_sample
DDS sample.
- data
- get_rounded_public_data(data: Any, digits: int = 2) Any ¶
Get a dict of field_name: value for public fields of a DDS sample with float values rounded.
- classmethod make_from_cmd_line(index: Optional[Union[int, enum.IntEnum, bool]], **kwargs: Any) lsst.ts.salobj.csc_commander.CscCommander ¶
Construct a SAL-related class from command line arguments.
- Parameters
- index
int
,enum.IntEnum
,True
, orNone
If the CSC is indexed, do one of the following:
Specify
True
to makeindex
a required command-line argument that accepts any nonzero index.Specify an
enum.IntEnum
class to makeindex
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.- **kwargs
dict
, optional Additional keyword arguments for your class’s constructor.
- index
- Returns
- instance
cls
The constructed instance.
- instance
Notes
To add additional command-line arguments, override
add_arguments
andadd_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 appendstr
toself.output_queue
instead of printing it. Use this mode for unit testing a CSC commander.
- async run_command(cmd: str) None ¶
Run the specified command string and wait for it to finish.
- Parameters
- cmd
str
Command string (command name and arguments). Note: does not handle the “exit” command.
- cmd
- async run_command_topic(command_name: str, args: Sequence[str]) None ¶
Run a command that has an associated salobj RemoteCommand topic.
- Parameters
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.