Skip to main content

Scripting API Guide

This document provides the information necessary to write test procedures using the COSMOS scripting API. Scripting in COSMOS is designed to be simple and intuitive. The code completion ability for command and telemetry mnemonics makes Script Runner the ideal place to write your procedures, however any text editor will do. If there is functionality that you don't see here or perhaps an easier syntax for doing something, please submit a ticket.

Concepts​

Programming Languages​

COSMOS scripting is implemented using either Ruby or Python. Ruby and Python are very similar scripting languages and in many cases the COSMOS APIs are identical between the two. This guide is written to support both with additional language specific information found in the Script Writing Guide.

Using Script Runner​

Script Runner is a graphical application that provides the ideal environment for running and implementing your test procedures. The Script Runner tool is broken into 4 main sections. At the top of the tool is a menu bar that allows you to do such things as open and save files, perform a syntax check, and execute your script.

Next is a tool bar that displays the currently executing script and three buttons, "Start/Go", "Pause/Retry", and "Stop". The Start/Go button is used to start the script and continue past errors or waits. The Pause/Retry button will pause the executing script. If an error is encountered the Pause button changes to Retry to re-execute the errored line. Finally, the Stop button will stop the executing script at any time.

Third is the display of the actual script. While the script is not running, you may edit and compose scripts in this area. A handy code completion feature is provided that will list out the available commands or telemetry points as you are writing your script. Simply begin writing a cmd( or tlm( line to bring up code completion. This feature greatly reduces typos in command and telemetry mnemonics.

Finally, the bottom of the display is the log messages. All commands that are sent, errors that occur, and user print statements appear in this area.

Telemetry Types​

There are four different ways that telemetry values can be retrieved in COSMOS. The following chart explains their differences.

Telemetry TypeDescription
RawRaw telemetry is exactly as it is in the telemetry packet before any conversions. All telemetry items will have a raw value except for Derived telemetry points which have no real location in a packet. Requesting raw telemetry on a derived item will return nil.
ConvertedConverted telemetry is raw telemetry that has gone through a conversion factor such as a state conversion or a polynomial conversion. If a telemetry item does not have a conversion defined, then converted telemetry will be the same as raw telemetry. This is the most common type of telemety used in scripts.
FormattedFormatted telemetry is converted telemetry that has gone through a printf style conversion into a string with units appended. Formatted telemetry will always have a string representation. If no format string or units are defined for a telemetry point, then formatted telemetry will be the same as converted telemetry except represented as string.

Script Runner API​

The following methods are designed to be used in Script Runner procedures. Many can also be used in custom built COSMOS tools. Please see the COSMOS Tool API section for methods that are more efficient to use in custom tools.

Including APIs

When writing a script for a microservice, interface, or something run within COSMOS you should include the openc3/api library. When writing a script that connects to COSMOS from outside of the COSMOS cluster you should include the openc3/script library. For more information see API vs Script

Migration from COSMOS v6 to v7​

See the Migrating From COSMOS 6 to COSMOS 7 guide for other changes including password hashing, JSON API session tokens, and configuration changes.

The following API methods have been removed from COSMOS v7. Since WITH_UNITS were removed in COSMOS 7 those APIs are deprecated and simply return the formatted result.

MethodAPIStatus
tlm_with_unitstlm_apiDeprecated, use tlm_formatted
check_with_unitsapi_sharedDeprecated, use check_formatted
tlm_variabletlm_apiRemoved, use tlm and pass type
check_tolerance_rawapi_sharedRemoved, use check_tolerance and pass type
wait_rawapi_sharedRemoved, use wait and pass type
wait_check_rawapi_sharedRemoved, use wait_check and pass type
wait_tolerance_rawapi_sharedRemoved, use wait_tolerance and pass type
wait_check_tolerance_rawapi_sharedRemoved, use wait_check_tolerance and pass type

The following API methods now return COSMOS__CANCEL instead of Cancel when the Cancel button is pushed in Script Runner: ask, ask_string, message_box, vertical_message_box, combo_box, check_box, prompt, prompt_for_hazardous, prompt_for_critical_cmd, metadata_input, open_file_dialog, open_files_dialog, open_bucket_dialog. Unless you are explicitly checking the return value for the word 'Cancel' there are no changes required.

Migration from COSMOS v5 to v6​

See the Migrating From COSMOS 5 to COSMOS 6 guide for other changes including migrating to Vue 3 and Vuetify 3.

The following API methods have been removed from COSMOS v6. Most of the deprecated API methods still remain for backwards compatibility.

MethodAPIStatus
get_all_target_infotarget_apiRemoved, use get_target_interfaces
play_wav_fileapi_sharedRemoved
status_barapi_sharedRemoved

Migration from COSMOS v4 to v5​

The following API methods are removed (will not be ported to COSMOS 5):

MethodToolStatus
clearTelemetry ViewerRemoved, use clear_screen
clear_allTelemetry ViewerRemoved, use clear_all_screens
close_local_screensTelemetry ViewerRemoved, use clear_screen
clear_disconnected_targetsScript RunnerRemoved
cmd_tlm_clear_countersCommand and Telemetry ServerRemoved
cmd_tlm_reloadCommand and Telemetry ServerRemoved
displayTelemetry ViewerRemoved, use display_screen
get_all_packet_logger_infoCommand and Telemetry ServerRemoved
get_all_target_infoCommand and Telemetry ServerRemoved, use get_target_interfaces
get_background_tasksCommand and Telemetry ServerRemoved
get_all_cmd_infoCommand and Telemetry ServerRemoved, use get_all_cmds
get_all_tlm_infoCommand and Telemetry ServerRemoved, use get_all_tlm
get_cmd_listCommand and Telemetry ServerRemoved, use get_all_cmds
get_cmd_log_filenameCommand and Telemetry ServerRemoved
get_cmd_param_listCommand and Telemetry ServerRemoved, use get_cmd
get_cmd_tlm_disconnectScript RunnerRemoved
get_disconnected_targetsScript RunnerRemoved
get_interface_infoCommand and Telemetry ServerRemoved, use get_interface
get_interface_targetsCommand and Telemetry ServerRemoved
get_output_logs_filenamesCommand and Telemetry ServerRemoved
get_packetCommand and Telemetry ServerRemoved, use get_packets
get_packet_dataCommand and Telemetry ServerRemoved, use get_packets
get_packet_logger_infoCommand and Telemetry ServerRemoved
get_packet_loggersCommand and Telemetry ServerRemoved
get_replay_modeReplayRemoved
get_router_infoCommand and Telemetry ServerRemoved, use get_router
get_scriptrunner_message_log_filenameCommand and Telemetry ServerRemoved
get_server_messageCommand and Telemetry ServerRemoved
get_server_message_log_filenameCommand and Telemetry ServerRemoved
get_server_statusCommand and Telemetry ServerRemoved
get_staleCommand and Telemetry ServerRemoved
get_target_ignored_itemsCommand and Telemetry ServerRemoved, use get_target
get_target_ignored_parametersCommand and Telemetry ServerRemoved, use get_target
get_target_infoCommand and Telemetry ServerRemoved, use get_target
get_target_listCommand and Telemetry ServerRemoved, use get_target_names
get_tlm_detailsCommand and Telemetry ServerRemoved
get_tlm_item_listCommand and Telemetry ServerRemoved
get_tlm_listCommand and Telemetry ServerRemoved
get_tlm_log_filenameCommand and Telemetry ServerRemoved
interface_stateCommand and Telemetry ServerRemoved, use get_interface
override_tlm_rawCommand and Telemetry ServerRemoved, use override_tlm
open_directory_dialogScript RunnerRemoved
play_wav_fileScript RunnerRemoved
replay_move_endReplayRemoved
replay_move_indexReplayRemoved
replay_move_startReplayRemoved
replay_playReplayRemoved
replay_reverse_playReplayRemoved
replay_select_fileReplayRemoved
replay_set_playback_delayReplayRemoved
replay_statusReplayRemoved
replay_step_backReplayRemoved
replay_step_forwardReplayRemoved
replay_stopReplayRemoved
require_utilityScript RunnerDeprecated, use load_utility
router_stateCommand and Telemetry ServerRemoved, use get_router
save_file_dialogScript RunnerRemoved, use put_target_file
save_settingCommand and Telemetry ServerDeprecated, use set_setting
set_cmd_tlm_disconnectScript RunnerRemoved, use disconnect_script
set_disconnected_targetsScript RunnerRemoved
set_replay_modeReplayRemoved
set_stdout_max_linesScript RunnerRemoved
set_tlm_rawScript RunnerRemoved, use set_tlm
show_backtraceScript RunnerRemoved, backtrace always shown
status_barScript RunnerRemoved
shutdown_cmd_tlmCommand and Telemetry ServerRemoved
start_cmd_logCommand and Telemetry ServerRemoved
start_loggingCommand and Telemetry ServerRemoved
start_new_scriptrunner_message_logCommand and Telemetry ServerRemoved
start_new_server_message_logCommand and Telemetry ServerRemoved
start_tlm_logCommand and Telemetry ServerRemoved
stop_background_taskCommand and Telemetry ServerRemoved
stop_cmd_logCommand and Telemetry ServerRemoved
stop_loggingCommand and Telemetry ServerRemoved
stop_tlm_logCommand and Telemetry ServerRemoved
subscribe_limits_eventsCommand and Telemetry ServerRemoved
subscribe_packet_dataCommand and Telemetry ServerRemoved, use subscribe_packets
subscribe_server_messagesCommand and Telemetry ServerRemoved
tlm_variableScript RunnerDeprecated, use tlm and pass type
unsubscribe_limits_eventsCommand and Telemetry ServerRemoved
unsubscribe_packet_dataCommand and Telemetry ServerRemoved
unsubscribe_server_messagesCommand and Telemetry ServerRemoved
wait_rawScript RunnerDeprecated, use wait and pass type
wait_check_rawScript RunnerDeprecated, use wait_check and pass type
wait_tolerance_rawScript RunnerDeprecated, use wait_tolerance and pass type
wait_check_tolerance_rawScript RunnerDeprecated, use wait_check_tolerance and pass type

Retrieving User Input​

These methods allow the user to enter values that are needed by the script.

ask​

Since 5.0.0

Prompts the user for input with a question. User input is automatically converted from a string to the appropriate data type. For example if the user enters "1", the number 1 as an integer will be returned.

ask("<question>", <Blank or Default>, <Password>)
ParameterDescription
questionQuestion to prompt the user with.
Blank or DefaultWhether or not to allow empty responses (optional - defaults to false). If a non-boolean value is passed it is used as a default value.
PasswordWhether to treat the entry as a password which is displayed with dots and not logged. Default is false.
value = ask("Enter an integer")
value = ask("Enter a value or nothing", True)
value = ask("Enter a value", 10)
password = ask("Enter your password", False, True)

ask_string​

Since 5.0.0

Prompts the user for input with a question. User input is always returned as a string. For example if the user enters "1", the string "1" will be returned.

ask_string("<question>", <Blank or Default>, <Password>)
ParameterDescription
questionQuestion to prompt the user with.
Blank or DefaultWhether or not to allow empty responses (optional - defaults to false). If a non-boolean value is passed it is used as a default value.
PasswordWhether to treat the entry as a password which is displayed with dots and not logged. Default is false.
string = ask_string("Enter a String")
string = ask_string("Enter a value or nothing", True)
string = ask_string("Enter a value", "test")
password = ask_string("Enter your password", False, True)

message_box​

Since 5.0.0

vertical_message_box​

Since 5.0.0

combo_box​

Since 5.0.0

check_box​

Since 7.0.0

The message_box, vertical_message_box, combo_box and check_box methods create a message box with buttons / selections / checkboxes that the user can click. The text of the button / selection / checkbox is returned.

message_box("<Message>", "<button text 1>", ...)
vertical_message_box("<Message>", "<button text 1>", ...)
combo_box("<Message>", "<selection text 1>", ...)
check_box("<Message>", "<checkbox text 1>", ...)
ParameterDescription
MessageMessage to prompt the user with.
Button/Selection/Checkbox TextText for a button or selection
informativeNamed parameter to add additional info to the dialog
detailsNamed parameter to add additional details to the dialog
multipleNamed parameter to make the combo_box multi-select
value = message_box("Select the sensor number", 'One', 'Two', informative="Smaller informative font")
value = vertical_message_box("Select the sensor number", 'One', 'Two', details="Regular details")
value = combo_box("Select the sensor number", 'One', 'Two')
match value:
case 'One':
print('Sensor One')
case 'Two':
print('Sensor Two')
values = combo_box("Select sensors to enable", 'One', 'Two', 'Three', multiple=True)
for value in values:
print(f"Enabling {value}")
values = check_box("Select sensors to enable", 'One', 'Two', 'Three')
for value in values:
print(f"Enabling {value}")

open_file_dialog​

Since 5.0.0

open_files_dialog​

Since 5.0.0

The open_file_dialog and open_files_dialog methods create a file dialog box so the user can select a single or multiple files. The selected file(s) is returned.

Note: COSMOS 5 has deprecated the save_file_dialog and open_directory_dialog methods. save_file_dialog can be replaced by put_target_file if you want to write a file back to the target. open_directory_dialog doesn't make sense in new architecture so you must request individual files.

open_file_dialog("<Title>", "<Message>", filter="<filter>")
open_files_dialog("<Title>", "<Message>", filter="<filter>")
ParameterDescription
TitleThe title to put on the dialog. Required.
MessageThe message to display in the dialog box. Optional parameter.
filterNamed parameter to filter allowed file types. Optional parameter, specified as comma delimited file types, e.g. ".txt,.doc". See https://developer.mozilla.org/en-US/docs/Web/HTML/Element/input/file#accept for more information.
file = open_file_dialog("Open a single file", "Choose something interesting", filter=".txt")
print(file)
print(file.read())
file.close()

files = open_files_dialog("Open multiple files") # message is optional
print(files) # Array of File objects (even if you select only one)
for file in files:
print(file)
print(file.read())
file.close()

open_bucket_dialog​

Since 7.0.0

The open_bucket_dialog method creates a dialog box that allows the user to browse S3 bucket files and select one. It presents the available buckets (similar to Bucket Explorer) and allows navigating directories within the selected bucket. The selected file is downloaded and returned as a file object, similar to open_file_dialog.

open_bucket_dialog("<Title>", "<Message>", default_path=<default_path>, filter=<filter>)
ParameterDescription
TitleThe title to put on the dialog. Required.
MessageThe message to display in the dialog box. Optional parameter.
default_path Since 7.1.1Path to preselect inside the dialog in the form bucket/path/to/file.ext (or bucket/path/to/folder/ for a folder with a trailing slash /). The user can still navigate elsewhere. Path traversal segments (..) are rejected. Per-scope bucket permissions are still enforced server-side. Optional parameter, defaults to nil.
filter Since 7.1.1Comma-separated list of file suffixes used to restrict the listing (e.g. ".txt" or ".txt,.json"). Folders are always shown so the user can navigate. Optional parameter, defaults to nil.
file = open_bucket_dialog("Select a File", "Choose a file from a bucket")
print(file.filename()) # The name of the selected file, note filename is a method
print(file.read())
file.close()

# Preselect the procedures folder and restrict to Python files
file = open_bucket_dialog(
"Select a Procedure",
"Choose a procedure",
# trailing slash indicates folder, otherwise assumes a file
default_path="config/DEFAULT/targets/INST2/procedures/",
filter=".py",
)
print(file.filename())
file.close()

File Manipulation​

These methods provide capability to interact with files in the target directory.

get_target_file​

Since 5.0.0

Return a file handle to a file in the target directory. Returns None (Python) or nil (Ruby) if the file is not found.

get_target_file("<File Path>", original=False)
ParameterDescription
File PathThe path to the file in the target directory. Should assume to start with a TARGET name, e.g. INST/procedures/proc.rb
originalWhether to get the original file from the plug-in, or any modifications to the file. Default is false which means to grab the modified file. If the modified file does not exist the API will automatically try to pull the original.
from openc3.utilities.string import formatted

file = get_target_file("INST/data/attitude.bin")
print(formatted(file.read())) # format a binary file
file.close() # delete file
file = get_target_file("INST/procedures/checks.rb", original=True)
print(file.read())
file.close() # delete file

put_target_file​

Since 5.0.0

Writes a file to the target directory

put_target_file("<File Path>", "IO or String")
ParameterDescription
File PathThe path to the file in the target directory. Should assume to start with a TARGET name, e.g. INST/procedures/proc.rb. The file can previously exist or not. Note: The original file from the plug-in will not be modified, however existing modified files will be overwritten.
IO or StringThe data can be an IO object or String
from io import BytesIO

put_target_file("INST/test1.txt", "this is a string test")
file = tempfile.NamedTemporaryFile(mode="w+t")
file.write("this is a Io test")
file.seek(0)
put_target_file("INST/test2.txt", file)
put_target_file("INST/test3.bin", b"\x00\x01\x02\x03\xFF\xEE\xDD\xCC") # binary

buf = BytesIO()
buf.write(b'\xDE\xAD\xBE\xEF')
buf.seek(0)
put_target_file("INST/test4.bin", buf)

delete_target_file​

Since 5.0.0

Delete a file in the target directory

delete_target_file("<File Path>")
ParameterDescription
File PathThe path to the file in the target directory. Should assume to start with a TARGET name, e.g. INST/procedures/proc.rb. Note: Only files created with put_target_file can be deleted. Original files from the plugin installation will remain.
put_target_file("INST/delete_me.txt", "to be deleted")
delete_target_file("INST/delete_me.txt")

download_file​

Since 5.8.0

Prompts the user to download a file from the OpenC3 system to their local machine.

download_file("<File Path>")
ParameterDescription
File PathFile Path to the file to download within the OpenC3 system.
download_file("targets/INST/procedures/my_data.csv")

Providing information to the user​

These methods notify the user that something has occurred.

prompt​

Since 5.0.0

Displays a message to the user and waits for them to press an ok button. If the user clicks Cancel instead, the script is paused but remains at the prompt line. Pressing Go re-displays the prompt.

prompt("<Message>")
ParameterDescription
MessageMessage to prompt the user with.
prompt("Press OK to continue")

Commands​

These methods provide capability to send commands to a target and receive information about commands in the system.

cmd​

Since 5.0.0

Sends a specified command.

cmd("<Target Name> <Command Name> with <Param #1 Name> <Param #1 Value>, <Param #2 Name> <Param #2 Value>, ...")
cmd("<Target Name>", "<Command Name>", {"Param #1 Name": <Param #1 Value>, "Param #2 Name": <Param #2 Value>, ...})
ParameterDescription
Target NameName of the target this command is associated with.
Command NameName of this command. Also referred to as its mnemonic.
Param #x NameName of a command parameter. If there are no parameters then the 'with' keyword should not be given.
Param #x ValueValue of the command parameter. Values are automatically converted to the appropriate type.
timeoutOptional named parameter to change the default timeout value of 5 seconds
log_messageOptional named parameter to prevent logging of the command
validateOptional named parameter to enable/disable validation (default is True)
cmd("INST COLLECT with DURATION 10, TYPE NORMAL")
cmd("INST", "COLLECT", { "DURATION": 10, "TYPE": "NORMAL" })
cmd("INST ABORT", timeout=10, log_message=False, validate=False)

cmd_no_range_check​

Since 5.0.0

Sends a specified command without performing range checking on its parameters. This should only be used when it is necessary to intentionally send a bad command parameter to test a target.

cmd_no_range_check("<Target Name> <Command Name> with <Param #1 Name> <Param #1 Value>, <Param #2 Name> <Param #2 Value>, ...")
cmd_no_range_check("<Target Name>", "<Command Name>", {"Param #1 Name": <Param #1 Value>, "Param #2 Name": <Param #2 Value>, ...})
ParameterDescription
Target NameName of the target this command is associated with.
Command NameName of this command. Also referred to as its mnemonic.
Param #x NameName of a command parameter. If there are no parameters then the 'with' keyword should not be given.
Param #x ValueValue of the command parameter. Values are automatically converted to the appropriate type.
timeoutOptional named parameter to change the default timeout value of 5 seconds
log_messageOptional named parameter to prevent logging of the command
cmd_no_range_check("INST COLLECT with DURATION 11, TYPE NORMAL")
cmd_no_range_check("INST", "COLLECT", {"DURATION": 11, "TYPE": "NORMAL"})

cmd_no_hazardous_check​

Since 5.0.0

Sends a specified command without performing the notification if it is a hazardous command. This should only be used when it is necessary to fully automate testing involving hazardous commands.

cmd_no_hazardous_check("<Target Name> <Command Name> with <Param #1 Name> <Param #1 Value>, <Param #2 Name> <Param #2 Value>, ...")
cmd_no_hazardous_check("<Target Name>", "<Command Name>", {"Param #1 Name": <Param #1 Value>, "Param #2 Name": <Param #2 Value>, ...})
ParameterDescription
Target NameName of the target this command is associated with.
Command NameName of this command. Also referred to as its mnemonic.
Param #x NameName of a command parameter. If there are no parameters then the 'with' keyword should not be given.
Param #x ValueValue of the command parameter. Values are automatically converted to the appropriate type.
timeoutOptional named parameter to change the default timeout value of 5 seconds
log_messageOptional named parameter to prevent logging of the command
cmd_no_hazardous_check("INST CLEAR")
cmd_no_hazardous_check("INST", "CLEAR")

cmd_no_checks​

Since 5.0.0

Sends a specified command without performing the parameter range checks or notification if it is a hazardous command. This should only be used when it is necessary to fully automate testing involving hazardous commands that intentionally have invalid parameters.

cmd_no_checks("<Target Name> <Command Name> with <Param #1 Name> <Param #1 Value>, <Param #2 Name> <Param #2 Value>, ...")
cmd_no_checks("<Target Name>", "<Command Name>", {"Param #1 Name": <Param #1 Value>, "Param #2 Name": <Param #2 Value>, ...})
ParameterDescription
Target NameName of the target this command is associated with.
Command NameName of this command. Also referred to as its mnemonic.
Param #x NameName of a command parameter. If there are no parameters then the 'with' keyword should not be given.
Param #x ValueValue of the command parameter. Values are automatically converted to the appropriate type.
timeoutOptional named parameter to change the default timeout value of 5 seconds
log_messageOptional named parameter to prevent logging of the command
cmd_no_checks("INST COLLECT with DURATION 11, TYPE SPECIAL")
cmd_no_checks("INST", "COLLECT", {"DURATION": 11, "TYPE": "SPECIAL"})

cmd_raw​

Since 5.0.0

Sends a specified command without running conversions.

cmd_raw("<Target Name> <Command Name> with <Param #1 Name> <Param #1 Value>, <Param #2 Name> <Param #2 Value>, ...")
cmd_raw("<Target Name>", "<Command Name>", {"<Param #1 Name>": <Param #1 Value>, "<Param #2 Name>": <Param #2 Value>, ...})
ParameterDescription
Target NameName of the target this command is associated with.
Command NameName of this command. Also referred to as its mnemonic.
Param #x NameName of a command parameter. If there are no parameters then the 'with' keyword should not be given.
Param #x ValueValue of the command parameter. Values are automatically converted to the appropriate type.
timeoutOptional named parameter to change the default timeout value of 5 seconds
log_messageOptional named parameter to prevent logging of the command
cmd_raw("INST COLLECT with DURATION 10, TYPE 0")
cmd_raw("INST", "COLLECT", {"DURATION": 10, "TYPE": 0})

cmd_raw_no_range_check​

Since 5.0.0

Sends a specified command without running conversions or performing range checking on its parameters. This should only be used when it is necessary to intentionally send a bad command parameter to test a target.

cmd_raw_no_range_check("<Target Name> <Command Name> with <Param #1 Name> <Param #1 Value>, <Param #2 Name> <Param #2 Value>, ...")
cmd_raw_no_range_check("<Target Name>", "<Command Name>", {"<Param #1 Name>": <Param #1 Value>, "<Param #2 Name>": <Param #2 Value>, ...})
ParameterDescription
Target NameName of the target this command is associated with.
Command NameName of this command. Also referred to as its mnemonic.
Param #x NameName of a command parameter. If there are no parameters then the 'with' keyword should not be given.
Param #x ValueValue of the command parameter. Values are automatically converted to the appropriate type.
timeoutOptional named parameter to change the default timeout value of 5 seconds
log_messageOptional named parameter to prevent logging of the command
cmd_raw_no_range_check("INST COLLECT with DURATION 11, TYPE 0")
cmd_raw_no_range_check("INST", "COLLECT", {"DURATION": 11, "TYPE": 0})

cmd_raw_no_hazardous_check​

Since 5.0.0

Sends a specified command without running conversions or performing the notification if it is a hazardous command. This should only be used when it is necessary to fully automate testing involving hazardous commands.

cmd_raw_no_hazardous_check("<Target Name> <Command Name> with <Param #1 Name> <Param #1 Value>, <Param #2 Name> <Param #2 Value>, ...")
cmd_raw_no_hazardous_check("<Target Name>", "<Command Name>", {"<Param #1 Name>": <Param #1 Value>, "<Param #2 Name>": <Param #2 Value>, ...})
ParameterDescription
Target NameName of the target this command is associated with.
Command NameName of this command. Also referred to as its mnemonic.
Param #x NameName of a command parameter. If there are no parameters then the 'with' keyword should not be given.
Param #x ValueValue of the command parameter. Values are automatically converted to the appropriate type.
timeoutOptional named parameter to change the default timeout value of 5 seconds
log_messageOptional named parameter to prevent logging of the command
cmd_raw_no_hazardous_check("INST CLEAR")
cmd_raw_no_hazardous_check("INST", "CLEAR")

cmd_raw_no_checks​

Since 5.0.0

Sends a specified command without running conversions or performing the parameter range checks or notification if it is a hazardous command. This should only be used when it is necessary to fully automate testing involving hazardous commands that intentionally have invalid parameters.

cmd_raw_no_checks("<Target Name> <Command Name> with <Param #1 Name> <Param #1 Value>, <Param #2 Name> <Param #2 Value>, ...")
cmd_raw_no_checks("<Target Name>", "<Command Name>", {"<Param #1 Name>": <Param #1 Value>, "<Param #2 Name>": <Param #2 Value>, ...})
ParameterDescription
Target NameName of the target this command is associated with.
Command NameName of this command. Also referred to as its mnemonic.
Param #x NameName of a command parameter. If there are no parameters then the 'with' keyword should not be given.
Param #x ValueValue of the command parameter. Values are automatically converted to the appropriate type.
timeoutOptional named parameter to change the default timeout value of 5 seconds
log_messageOptional named parameter to prevent logging of the command
cmd_raw_no_checks("INST COLLECT with DURATION 11, TYPE 1")
cmd_raw_no_checks("INST", "COLLECT", {"DURATION": 11, "TYPE": 1})

build_cmd​

Since 5.8.0

Builds a command binary string so you can see the raw bytes for a given command. Previously build_command. Use the get_cmd to get information about a command like endianness, description, items, etc.

build_cmd(<ARGS>, range_check=True, raw=False)
ParameterDescription
ARGSCommand parameters (see cmd)
range_checkWhether to perform range checking on the command. Default is true.
rawWhether to write the command arguments as RAW or CONVERTED value. Default is CONVERTED.
x = build_cmd("INST COLLECT with DURATION 10, TYPE NORMAL")
print(x) # => {'id': '1697298167748-0', 'result': 'SUCCESS', 'time': '1697298167749155717', 'received_time': '1697298167749155717', 'target_name': 'INST', 'packet_name': 'COLLECT', 'received_count': '2', 'buffer': bytearray(b'\x13\xe7\xc0\x00\x00\x0c\x00\x01\x00\x00A \x00\x00\xab\x00\x00\x00\x00')}

enable_cmd​

Since 5.15.1

Enables a disabled command. Sending a disabled command raises DisabledError with a message like 'INST ABORT is Disabled'.

buffer = enable_cmd("<Target Name> <Command Name>")
buffer = enable_cmd("<Target Name>", "<Command Name>")
ParameterDescription
Target NameName of the target.
Packet NameName of the command (packet).
enable_cmd("INST ABORT")

disable_cmd​

Since 5.15.1

Disables a command. Sending a disabled command raises DisabledError with a message like 'INST ABORT is Disabled'.

buffer = disable_cmd("<Target Name> <Command Name>")
buffer = disable_cmd("<Target Name>", "<Command Name>")
ParameterDescription
Target NameName of the target.
Packet NameName of the command (packet).
disable_cmd("INST ABORT")

send_raw​

Since 5.0.0

Sends raw data on an interface.

send_raw(<Interface Name>, <Data>)
ParameterDescription
Interface NameName of the interface to send the raw data on.
DataRaw ruby string of data to send.
send_raw("INST_INT", data)

get_all_cmds​

Since 5.13.0

Returns an array of the commands that are available for a particular target. The returned array is an array of hashes / list of dicts which fully describe the command packet. Previously get_all_commands.

get_all_cmds("<Target Name>")
ParameterDescription
Target NameName of the target.
cmd_list = get_all_cmds("INST")
print(cmd_list) # =>
# [{'target_name': 'INST',
# 'packet_name': 'ABORT',
# 'endianness': 'BIG_ENDIAN',
# 'description': 'Aborts a collect on the INST instrument',
# 'items': [{'name': 'CCSDSVER', 'bit_offset': 0, 'bit_size': 3, ... }]
# ...
# }]

get_all_cmd_names​

Since 5.13.0

Returns an array of the command names for a particular target. Previously get_all_command_names.

get_all_cmd_names("<Target Name>")
ParameterDescription
Target NameName of the target
cmd_list = get_all_cmd_names("INST")
print(cmd_list) # => ['ABORT', 'ARYCMD', 'ASCIICMD', ...]

get_cmd​

Since 5.13.0

Returns a command hash which fully describes the command packet. Previously get_command. To get the binary buffer of an as-built command use build_cmd.

get_cmd("<Target Name> <Packet Name>")
get_cmd("<Target Name>", "<Packet Name>")
ParameterDescription
Target NameName of the target.
Packet NameName of the packet.
abort_cmd = get_cmd("INST ABORT")
print(abort_cmd) # =>
# [{'target_name': 'INST',
# 'packet_name': 'ABORT',
# 'endianness': 'BIG_ENDIAN',
# 'description': 'Aborts a collect on the INST instrument',
# 'items': [{'name': 'CCSDSVER', 'bit_offset': 0, 'bit_size': 3, ... }]
# ...
# }]

get_param​

Since 5.13.0

Returns a hash of the given command parameter. Previously get_parameter.

get_param("<Target Name> <Command Name> <Parameter Name>")
get_param("<Target Name>", "<Command Name>", "<Parameter Name>")
ParameterDescription
Target NameName of the target.
Command NameName of the command.
Parameter NameName of the parameter.
param = get_param("INST COLLECT TYPE")
print(param) # =>
# {'name': 'TYPE', 'bit_offset': 64, 'bit_size': 16, 'data_type': 'UINT',
# 'description': 'Collect type which can be normal or special', 'default': 0,
# 'minimum': 0, 'maximum': 65535, 'endianness': 'BIG_ENDIAN', 'required': True, 'overflow': 'ERROR',
# 'states': {'NORMAL': {'value': 0}, 'SPECIAL': {'value': 1, 'hazardous': ''}}, 'limits': {}}

get_cmd_buffer​

Since 5.0.0

Returns a packet hash (similar to get_cmd) along with the raw packet buffer as a Ruby string.

buffer = get_cmd_buffer("<Target Name> <Packet Name>")['buffer']
buffer = get_cmd_buffer("<Target Name>", "<Packet Name>")['buffer']
ParameterDescription
Target NameName of the target.
Packet NameName of the packet.
packet = get_cmd_buffer("INST COLLECT")
print(packet) # =>
# {'time': '1697298923745982470', 'received_time': '1697298923745982470',
# 'target_name': 'INST', 'packet_name': 'COLLECT', 'received_count': '21', 'stored': 'false',
# 'buffer': bytearray(b'\x13\xe7\xc0\x00\x00\x0c\x00\x01\x00\x00@\xe0\x00\x00\xab\x00\x00\x00\x00')}

get_cmd_hazardous​

Since 5.0.0

Returns true/false indicating whether a particular command is flagged as hazardous.

get_cmd_hazardous("<Target Name>", "<Command Name>", <Command Params - optional>)
ParameterDescription
Target NameName of the target.
Command NameName of the command.
Command ParamsHash of the parameters given to the command (optional). Note that some commands are only hazardous based on parameter states.
hazardous = get_cmd_hazardous("INST", "COLLECT", {'TYPE': 'SPECIAL'})
print(hazardous) # => True

get_cmd_value​

Since 5.0.0

Returns reads a value from the most recently sent command packet. The pseudo-parameters 'PACKET_TIMESECONDS', 'PACKET_TIMEFORMATTED', 'RECEIVED_COUNT', 'RECEIVED_TIMEFORMATTED', and 'RECEIVED_TIMESECONDS' are also supported.

get_cmd_value("<Target Name>", "<Command Name>", "<Parameter Name>", <Value Type - optional>)
ParameterDescription
Target NameName of the target.
Command NameName of the command.
Parameter NameName of the command parameter.
Value TypeValue Type to read. RAW, CONVERTED or FORMATTED. NOTE: Symbol in Ruby and str in Python
value = get_cmd_value("INST", "COLLECT", "TEMP", "RAW")
print(value) # => 0.0

get_cmd_time​

Since 5.0.0

Returns the time of the most recent command sent.

get_cmd_time("<Target Name - optional>", "<Command Name - optional>")
ParameterDescription
Target NameName of the target. If not given, then the most recent command time to any target will be returned
Command NameName of the command. If not given, then the most recent command time to the given target will be returned
target_name, command_name, time = get_cmd_time() # Name of the most recent command sent to any target and time
target_name, command_name, time = get_cmd_time("INST") # Name of the most recent command sent to the INST target and time
target_name, command_name, time = get_cmd_time("INST", "COLLECT") # Name of the most recent INST COLLECT command and time

get_cmd_cnt​

Since 5.0.0

Returns the number of times a specified command has been sent.

get_cmd_cnt("<Target Name> <Command Name>")
get_cmd_cnt("<Target Name>", "<Command Name>")
ParameterDescription
Target NameName of the target.
Command NameName of the command.
cmd_cnt = get_cmd_cnt("INST COLLECT") # Number of times the INST COLLECT command has been sent

get_cmd_cnts​

Since 5.0.0

Returns the number of times the specified commands have been sent.

get_cmd_cnts([["<Target Name>", "<Command Name>"], ["<Target Name>", "<Command Name>"], ...])
ParameterDescription
Target NameName of the target.
Command NameName of the command.
cmd_cnt = get_cmd_cnts([['INST', 'COLLECT'], ['INST', 'ABORT']]) # Number of times the INST COLLECT and INST ABORT commands have been sent

critical_cmd_status​

Since 5.20.0

Returns the status of a critical command. One of APPROVED, REJECTED, or WAITING.

critical_cmd_status(uuid)
ParameterDescription
uuidUUID for the critical command (displayed in the COSMOS GUI)
status = critical_cmd_status("2fa14183-3148-4399-9a74-a130257118f9") # => WAITING

critical_cmd_approve​

Since 5.20.0

Approve the critical command as the current user.

critical_cmd_approve(uuid)
ParameterDescription
uuidUUID for the critical command (displayed in the COSMOS GUI)
critical_cmd_approve("2fa14183-3148-4399-9a74-a130257118f9")

critical_cmd_reject​

Since 5.20.0

Reject the critical command as the current user.

critical_cmd_reject(uuid)
ParameterDescription
uuidUUID for the critical command (displayed in the COSMOS GUI)
critical_cmd_reject("2fa14183-3148-4399-9a74-a130257118f9")

critical_cmd_can_approve​

Since 5.20.0

Returns whether or not the current user can approve the critical command.

critical_cmd_can_approve(uuid)
ParameterDescription
uuidUUID for the critical command (displayed in the COSMOS GUI)
status = critical_cmd_can_approve("2fa14183-3148-4399-9a74-a130257118f9") # => True / False

Command Queues​

Command queues can be used to store commands prior to "releasing" them for execution. Commands are sent to queues via the queue keyword argument to the various cmd API calls. Queues can be in three different states: HOLD, RELEASE, and DISABLE. HOLD builds up a FIFO of commands as they are generated. RELEASE sends the commands to the interface for execution. DISABLE rejects all commands that are sent via the cmd API calls. Sending a command while a queue is in the DISABLE state will raise an exception and stop the script.

By setting the environment variable OPENC3_DEFAULT_QUEUE in the .env file you can create a "default" queue that all commands will go to unless explicitly excluded by passing false/False to the queue keyword argument. This queue is created during COSMOS initialization and is created in RELEASE mode so normal commanding will still work. Note that this will introduce delay in the command chain because all commands will flow through the queue instead of being directly executed by the interface.

Here is an example of sending a command to a queue:

cmd("INST ABORT", queue="TEST")
# Do NOT queue the command (useful if OPENC3_DEFAULT_QUEUE is set)
cmd("INST ABORT", queue=False)

queue_create​

Since 6.8.0

Create a command queue. Queues are initially created in HOLD state.

queue_create('<Queue Name>')
ParameterDescription
Queue NameCase sensitive name of the queue
queue_create('TEST')

queue_get​

Since 6.8.0

Get information about a command queue including what state it is in.

queue_get('<Queue Name>')
ParameterDescription
Queue NameCase sensitive name of the queue
queue = queue_get('TEST')
print(queue) # =>

queue_all​

Since 6.8.0

Get information about all command queues.

all = queue_all()
print(all) # =>

queue_list​

Since 6.8.0

List all the commands in the queue including the username, the creation time, and the command itself.

queue_list('<Queue Name>')
ParameterDescription
Queue NameCase sensitive name of the queue
cmds = queue_list('TEST')
print(cmds) # =>

queue_hold​

Since 6.8.0

Puts the specified queue into HOLD mode which means that commands are queued up and not released to the interface.

queue_hold('<Queue Name>')
ParameterDescription
Queue NameCase sensitive name of the queue
queue_hold('TEST')

queue_release​

Since 6.8.0

Puts the specified queue into RELEASE mode which means that commands in the queue are released to the interface. Any new commands added to the queue are immediately released to the interface.

queue_release('<Queue Name>')
ParameterDescription
Queue NameCase sensitive name of the queue
queue_release('TEST')

queue_disable​

Since 6.8.0

Puts the specified queue into DISABLE mode which means that commands are rejected and not added to the queue. Commands added to a queue in DISABLE mode via script will raise an exception and stop.

queue_disable('<Queue Name>')
ParameterDescription
Queue NameCase sensitive name of the queue
queue_disable('TEST')

queue_exec​

Since 6.8.0

Removes a command from the queue and executes it.

queue_exec('<Queue Name>', '<Optional Index>')
ParameterDescription
Queue NameCase sensitive name of the queue
IndexRemove and execute at the specified index. If the index is not given the first command is removed and executed.
queue_exec('TEST') # Removes and executes the first command in the queue
queue_exec('TEST', index=5) # Removes and executes the command at index 5

queue_remove​

Since 6.9.1

Removes a command from the queue.

queue_remove('<Queue Name>', '<Optional Index>')
ParameterDescription
Queue NameCase sensitive name of the queue
IndexRemove at the specified index. If the index is not given the first command is removed.
queue_remove('TEST') # Removes the first command in the queue
queue_remove('TEST', index=5) # Removes the command at index 5

queue_delete​

Since 6.8.0

Deletes the specified queue and any commands that it holds.

queue_delete('<Queue Name>')
ParameterDescription
Queue NameCase sensitive name of the queue
queue_delete('TEST')

Handling Telemetry​

These methods allow the user to interact with telemetry items.

check, check_raw, check_formatted​

Since 5.0.0

Performs a verification of a telemetry item using its specified telemetry type. If the verification fails then the script will be paused with an error. If no comparison is given to check then the telemetry item is simply printed to the script output. Note: In most cases using wait_check is a better choice than using check.

check("<Target Name> <Packet Name> <Item Name> <Comparison - optional>")
ParameterDescription
Target NameName of the target of the telemetry item.
Packet NameName of the telemetry packet of the telemetry item.
Item NameName of the telemetry item.
ComparisonA comparison to perform against the telemetry item. If a comparison is not given then the telemetry item will just be printed into the script log.
Supported Comparisons

A comparison is a single operator followed by a literal value. The supported operators are ==, !=, >, >=, <, <= and in. in requires a list operand, e.g. in [1, 2, 3], whose elements follow the same rules as any other value, e.g. in ['ON', 'OFF']. Python also accepts a tuple or set, e.g. in (1, 2). Compound expressions, e.g. TIMEUS & 0x0001 == 0x0000, are not supported - use check_expression instead.

String Comparisons

When comparing against string or state values, the value must be quoted (e.g., == 'ON'). An unquoted value is rejected with Uninitialized constant ON. Did you mean 'ON' as a string?. Quoted values follow the string literal rules of the script language, so escape sequences are processed in Ruby double quoted strings and in all Python strings. The Ruby control and meta escapes \c, \C- and \M- are rejected rather than silently changed, as is string interpolation (Ruby "#{...}", Python f-strings) because the comparison is not evaluated as code. Interpolate in the script itself instead: Ruby check("INST HEALTH_STATUS TYPE == '#{expected}'") or Python check(f"INST HEALTH_STATUS TYPE == '{expected}'").

check("INST HEALTH_STATUS COLLECTS > 1")
check("INST HEALTH_STATUS TYPE == 'NORMAL'") # Strings must be quoted
check_raw("INST HEALTH_STATUS COLLECTS > 1")
check_formatted("INST HEALTH_STATUS COLLECTS > 1")
# Python passes type as string
check("INST HEALTH_STATUS COLLECTS > 1", type='RAW')

check_tolerance​

Since 5.0.0

Checks a converted telemetry item against an expected value with a tolerance. If the verification fails then the script will be paused with an error. Note: In most cases using wait_check_tolerance is a better choice than using check_tolerance.

check_tolerance("<Target Name> <Packet Name> <Item Name>", <Expected Value>, <Tolerance>)
ParameterDescription
Target NameName of the target of the telemetry item.
Packet NameName of the telemetry packet of the telemetry item.
Item NameName of the telemetry item.
Expected ValueExpected value of the telemetry item.
Tolerance± Tolerance on the expected value.
typeCONVERTED (default) or RAW (Ruby symbol, Python string)
check_tolerance("INST HEALTH_STATUS COLLECTS", 10.0, 5.0)
check_tolerance("INST HEALTH_STATUS TEMP1", 50000, 20000, type='RAW')

check_expression​

Since 5.0.0

Evaluates an expression. If the expression evaluates to false the script will be paused with an error. This method can be used to perform more complicated comparisons than using check as shown in the example. Note: In most cases using wait_check_expression is a better choice than using check_expression.

Remember that everything inside the check_expression string will be evaluated directly and thus must be valid syntax. A common mistake is to check a variable like so (Ruby variable interpolation):

check_expression("#{answer} == 'yes'") # where answer contains 'yes'

This evaluates to yes == 'yes' which is not valid syntax because the variable yes is not defined (usually). The correct way to write this expression is as follows:

check_expression("'#{answer}' == 'yes'") # where answer contains 'yes'

Now this evaluates to 'yes' == 'yes' which is true so the check passes.

check_expression(exp_to_eval, globals=None, locals=None)
ParameterDescription
exp_to_evalAn expression to evaluate.
context (ruby only)The context to call eval with. Defaults to nil. Context in Ruby is typically binding() and is usually not needed.
globals (python only)The globals to call eval with. Defaults to None. Note that to use COSMOS APIs like tlm() you must pass globals().
locals (python only)The locals to call eval with. Defaults to None. Note that if you're using local variables in a method you must pass locals().
def check(value):
# Here we using both tlm() and a local 'value' so we need to pass globals() and locals()
check_expression("tlm('INST HEALTH_STATUS COLLECTS') > value", 5, 0.25, globals(), locals())
check(5)

check_exception​

Since 5.0.0

Executes a method and expects an exception to be raised. If the method does not raise an exception, a CheckError is raised.

check_exception("<Method Name>", "<Method Params - optional>")
ParameterDescription
Method NameThe COSMOS scripting method to execute, e.g. 'cmd', etc.
Method ParamsParameters for the method
check_exception("cmd", "INST", "COLLECT", {"TYPE": "NORMAL"})

tlm, tlm_raw, tlm_formatted​

Since 5.0.0

Reads the specified form of a telemetry item.

tlm("<Target Name> <Packet Name> <Item Name>")
tlm("<Target Name>", "<Packet Name>", "<Item Name>")
ParameterDescription
Target NameName of the target of the telemetry item.
Packet NameName of the telemetry packet of the telemetry item.
Item NameName of the telemetry item.
typeNamed parameter specifying the type. RAW, CONVERTED (default) or FORMATTED (Ruby symbol, Python string).
value = tlm("INST HEALTH_STATUS COLLECTS")
value = tlm("INST", "HEALTH_STATUS", "COLLECTS")
value = tlm_raw("INST HEALTH_STATUS COLLECTS")
value = tlm_formatted("INST HEALTH_STATUS COLLECTS")
# Equivalent to tlm_raw
raw_value = tlm("INST HEALTH_STATUS COLLECTS", type='RAW')

get_tlm_buffer​

Since 5.0.0

Returns a packet hash (similar to get_tlm) along with the raw packet buffer.

buffer = get_tlm_buffer("<Target Name> <Packet Name>")['buffer']
buffer = get_tlm_buffer("<Target Name>", "<Packet Name>")['buffer']
ParameterDescription
Target NameName of the target.
Packet NameName of the packet.
packet = get_tlm_buffer("INST HEALTH_STATUS")
packet['buffer']

get_tlm_packet​

Since 5.0.0

Returns the names, values, and limits states of all telemetry items in a specified packet. The value is returned as an array of arrays with each entry containing [item_name, item_value, limits_state].

get_tlm_packet("<Target Name> <Packet Name>", <type>)
get_tlm_packet("<Target Name>", "<Packet Name>", <type>)
ParameterDescription
Target NameName of the target.
Packet NameName of the packet.
typeNamed parameter specifying the type. RAW, CONVERTED (default) or FORMATTED (Ruby symbol, Python string).
names_values_and_limits_states = get_tlm_packet("INST HEALTH_STATUS", type='FORMATTED')

get_tlm_available​

Since 6.7.0

Returns the actual items available based on the specified set of telemetry items. For example, if you request INST__HEALTH_STATUS__CCSDSVER__FORMATTED the method will return INST__HEALTH_STATUS__CCSDSVER__RAW for that item because it does not have formatting or conversions so only the RAW value is available. This must be called before calling get_tlm_values when passing a start_time / end_time as it ensures a correct request of historical data.

actual = get_tlm_available(<Items>)
ParameterDescription
ItemsArray of strings of the form ['TGT__PKT__ITEM__TYPE', ... ]
values = get_tlm_available(["INST__HEALTH_STATUS__CCSDSVER__FORMATTED", "INST__HEALTH_STATUS__TEMP1__FORMATTED"])
print(values) # ["INST__HEALTH_STATUS__CCSDSVER__RAW", "INST__HEALTH_STATUS__TEMP1__FORMATTED"]

get_tlm_values​

Since 5.0.0

Returns the values and current limits state for a specified set of telemetry items. Items can be in any telemetry packet in the system. They can all be retrieved using the same value type or a specific value type can be specified for each item.

values, limits_states, limits_settings, limits_set = get_tlm_values(<Items>)
ParameterDescription
ItemsArray of strings of the form ['TGT__PKT__ITEM__TYPE', ... ]
values = get_tlm_values(["INST__HEALTH_STATUS__TEMP1__CONVERTED", "INST__HEALTH_STATUS__TEMP2__RAW"])
print(values) # [[-100.0, :RED_LOW], [0, :RED_LOW]]

get_all_tlm​

Since 5.13.0

Returns an array of all target packet hashes. Previously get_all_telemetry.

get_all_tlm("<Target Name>")
ParameterDescription
Target NameName of the target.
packets = get_all_tlm("INST")
print(packets)
#[{"target_name"=>"INST",
# "packet_name"=>"ADCS",
# "endianness"=>"BIG_ENDIAN",
# "description"=>"Position and attitude data",
# "stale"=>true,
# "items"=>
# [{"name"=>"CCSDSVER",
# "bit_offset"=>0,
# "bit_size"=>3,
# ...

get_all_tlm_names​

Since 5.13.0

Returns an array of all target packet names. Previously get_all_telemetry_names.

get_all_tlm_names("<Target Name>")
ParameterDescription
Target NameName of the target
get_all_tlm_names("INST") # => ["ADCS", "HEALTH_STATUS", ...]

get_all_tlm_item_names​

Since 6.3.0

Returns all the item names for every packet in a target

get_all_tlm_item_names("<Target Name>")
ParameterDescription
Target NameName of the target
get_all_tlm_item_names("INST") # => ["ARY", "ARY2", "ASCIICMD", "ATTPROGRESS", ...]

get_tlm​

Since 5.13.0

Returns a packet hash. Previously get_telemetry.

get_tlm("<Target Name> <Packet Name>")
get_tlm("<Target Name>", "<Packet Name>")
ParameterDescription
Target NameName of the target.
Packet NameName of the packet.
packet = get_tlm("INST HEALTH_STATUS")
print(packet)
#{"target_name"=>"INST",
# "packet_name"=>"HEALTH_STATUS",
# "endianness"=>"BIG_ENDIAN",
# "description"=>"Health and status from the instrument",
# "stale"=>true,
# "processors"=>
# [{"name"=>"TEMP1STAT",
# "class"=>"OpenC3::StatisticsProcessor",
# "params"=>["TEMP1", 100, "CONVERTED"]},
# {"name"=>"TEMP1WATER",
# "class"=>"OpenC3::WatermarkProcessor",
# "params"=>["TEMP1", "CONVERTED"]}],
# "items"=>
# [{"name"=>"CCSDSVER",
# "bit_offset"=>0,
# "bit_size"=>3,
# ...

get_item​

Since 5.0.0

Returns an item hash.

get_item("<Target Name> <Packet Name> <Item Name>")
get_item("<Target Name>", "<Packet Name>", "<Item Name>")
ParameterDescription
Target NameName of the target.
Packet NameName of the packet.
Item NameName of the item.
item = get_item("INST HEALTH_STATUS CCSDSVER")
print(item)
#{"name"=>"CCSDSVER",
# "bit_offset"=>0,
# "bit_size"=>3,
# "data_type"=>"UINT",
# "description"=>"CCSDS packet version number (See CCSDS 133.0-B-1)",
# "endianness"=>"BIG_ENDIAN",
# "required"=>false,
# "overflow"=>"ERROR"}

get_tlm_cnt​

Since 5.0.0

Returns the number of times a specified telemetry packet has been received.

get_tlm_cnt("<Target Name> <Packet Name>")
get_tlm_cnt("<Target Name>", "<Packet Name>")
ParameterDescription
Target NameName of the target.
Packet NameName of the telemetry packet.
tlm_cnt = get_tlm_cnt("INST HEALTH_STATUS") # Number of times the INST HEALTH_STATUS telemetry packet has been received.

set_tlm​

Since 5.0.0

Sets a telemetry item value in the Command and Telemetry Server. This value will be overwritten if a new packet is received from an interface. For that reason this method is most useful if interfaces are disconnected or for testing via the Script Runner disconnect mode. Manually setting telemetry values allows for the execution of many logical paths in scripts.

set_tlm("<Target> <Packet> <Item> = <Value>", <type>)
ParameterDescription
TargetTarget name
PacketPacket name
ItemItem name
ValueValue to set
typeValue type RAW, CONVERTED (default) or FORMATTED (Ruby symbol, Python string)
set_tlm("INST HEALTH_STATUS COLLECTS = 5") # type is CONVERTED by default
check("INST HEALTH_STATUS COLLECTS == 5")
set_tlm("INST HEALTH_STATUS COLLECTS = 10", type='RAW')
check("INST HEALTH_STATUS COLLECTS == 10", type='RAW')

inject_tlm​

Since 5.0.0

Injects a packet into the system as if it was received from an interface.

inject_tlm("<target_name>", "<packet_name>", <item_hash>, type=<type>, stored=<stored>)
ParameterDescription
TargetTarget name
PacketPacket name
Item HashHash of item name/value for each item. If an item is not specified in the hash, the current value table value will be used. Optional parameter, defaults to nil.
typeType of values in the item hash, RAW, CONVERTED (default) or FORMATTED (Ruby symbol, Python string)
stored Since 7.1.0Whether the packet should be marked as stored (default false). Stored packets are written to the telemetry stream but do not update the current value table (CVT).
inject_tlm("INST", "PARAMS", {'VALUE1': 5.0, 'VALUE2': 7.0})
inject_tlm("INST", "PARAMS", {'VALUE1': 5.0, 'VALUE2': 7.0}, stored=True)

override_tlm​

Since 5.0.0

Sets the converted value for a telmetry point in the Command and Telemetry Server. This value will be maintained even if a new packet is received on the interface unless the override is canceled with the normalize_tlm method.

override_tlm("<Target> <Packet> <Item> = <Value>", <type>)
ParameterDescription
TargetTarget name
PacketPacket name
ItemItem name
ValueValue to set
typeType to override, ALL (default), RAW, CONVERTED or FORMATTED (Ruby symbol, Python string)
override_tlm("INST HEALTH_STATUS TEMP1 = 5") # All requests for TEMP1 return 5
override_tlm("INST HEALTH_STATUS TEMP2 = 0", type='RAW') # Only RAW tlm set to 0

normalize_tlm​

Since 5.0.0

Clears the override of a telmetry point in the Command and Telemetry Server.

normalize_tlm("<Target> <Packet> <Item>", <type>)
ParameterDescription
TargetTarget name
PacketPacket name
ItemItem name
typeType to normalize, ALL (default), RAW, CONVERTED or FORMATTED (Ruby symbol, Python string)
normalize_tlm("INST HEALTH_STATUS TEMP1") # clear all overrides
normalize_tlm("INST HEALTH_STATUS TEMP1", type='RAW') # clear only the RAW override

get_overrides​

Since 5.4.2

Returns an array of the the currently overridden values set by override_tlm. NOTE: This returns all the value types that are overridden which by default is all 4 values types when using override_tlm.

get_overrides()
override_tlm("INST HEALTH_STATUS TEMP1 = 5")
print(get_overrides()) # =>
# [ {'target_name': 'INST', 'packet_name': 'HEALTH_STATUS', 'item_name': 'TEMP1', 'value_type': 'RAW', 'value': 5},
# {'target_name': 'INST', 'packet_name': 'HEALTH_STATUS', 'item_name': 'TEMP1', 'value_type': 'CONVERTED', 'value': 5},
# {'target_name': 'INST', 'packet_name': 'HEALTH_STATUS', 'item_name': 'TEMP1', 'value_type': 'FORMATTED', 'value': '5'} ]

Packet Data Subscriptions​

APIs for subscribing to specific packets of data. This provides an interface to ensure that each telemetry packet is received and handled rather than relying on polling where some data may be missed.

subscribe_packets​

Since 5.0.0

Gets the current Redis stream offsets (IDs) for the given packets. These offsets are used to collect packet data from this point in time by passing them to get_packets.

This method is called subscribe_packets for historical reasons; no actual subscription is created, thus there is no need to unsubscribe.

subscribe_packets(packets)
ParameterDescription
packetsNested array of target name/packet name pairs that the user wishes to subscribe to.

Returns a unique id string which should be passed to get_packets to retrieve the subscribed data.

id = subscribe_packets([['INST', 'HEALTH_STATUS'], ['INST', 'ADCS']])

get_packets​

Since 5.0.0

Streams packet data from a previous subscription. Returns an updated id and an array of packet hashes/dictionaries. The updated id should be passed to the next call to get_packets to continue streaming.

id, packets = get_packets(id, block=None, count=1000)
ParameterDescription
idUnique id returned by subscribe_packets or the previous call to get_packets
blockNumber of seconds to block while waiting for packets from ANY stream, default nil / None (do not block)
countMaximum number of packets to return from EACH packet stream
Packet Ordering

Packets returned by get_packets are ordered within each subscribed packet stream (target/packet pair) but are NOT interleaved by time across streams. Packets from one stream are appended to the returned array, then packets from the next stream, and so on. If you require strict chronological order across packets from different streams, sort the returned array by the time (or received_time) field. Note that sorting only orders the current batch — packets across separate get_packets calls may still arrive out of order relative to each other, so subscribers needing global ordering must buffer and merge across calls.

id, packets = get_packets(id)
packets.sort(key=lambda p: int(p['time']))
id, packets = get_packets(id)
packets.sort_by! { |p| p['time'].to_i }

Returns a two element array containing the updated id and an array of packet hashes/dictionaries. Each packet hash/dictionary contains the following keys:

Metadata keys:

KeyTypeDescription
target_nameStringTarget name, e.g. "INST"
packet_nameStringPacket name, e.g. "HEALTH_STATUS"
timeIntegerPacket time as nanoseconds since the Unix epoch
received_timeIntegerTime the packet was received as nanoseconds since the Unix epoch
received_countIntegerRunning count of packets received
storedString"true" or "false" indicating if the packet was stored (not realtime)

Telemetry item keys:

Each telemetry item in the packet is included with up to four keys using the item name with different suffixes:

KeyDescription
ITEM_NAMERaw value of the telemetry item
ITEM_NAME__CConverted value (present if the item has states or a read conversion)
ITEM_NAME__FFormatted value (present if the item has a format string or units)
ITEM_NAME__LLimits state, e.g. "GREEN", "YELLOW", "RED" (present if the item has limits set)

For example, if the HEALTH_STATUS packet has a TEMP1 item with a conversion and limits, the packet hash/dictionary would contain TEMP1 (raw), TEMP1__C (converted), TEMP1__F (formatted), and TEMP1__L (limits state).

id = subscribe_packets([['INST', 'HEALTH_STATUS'], ['INST', 'ADCS']])
wait(0.1)
id, packets = get_packets(id)
for packet in packets:
print(f"{packet['PACKET_TIMESECONDS']}: {packet['target_name']} {packet['packet_name']}")

# Reuse ID from last call, allow for 1s wait, only get 1 packet
id, packets = get_packets(id, block=1000, count=1)
for packet in packets:
print(f"{packet['PACKET_TIMESECONDS']}: {packet['target_name']} {packet['packet_name']}")

get_tlm_cnt​

Since 5.0.0

Get the receive count for a telemetry packet

get_tlm_cnt("<Target> <Packet>")
get_tlm_cnt("<Target>", "<Packet>")
ParameterDescription
TargetTarget name
PacketPacket name
get_tlm_cnt("INST HEALTH_STATUS") # => 10

get_tlm_cnts​

Since 5.0.0

Get the receive counts for an array of telemetry packets

get_tlm_cnts([["<Target>", "<Packet>"], ["<Target>", "<Packet>"]])
ParameterDescription
TargetTarget name
PacketPacket name
get_tlm_cnts([["INST", "ADCS"], ["INST", "HEALTH_STATUS"]]) # => [100, 10]

get_packet_derived_items​

Since 5.0.0

Get the list of derived telemetry items for a packet

get_packet_derived_items("<Target> <Packet>")
get_packet_derived_items("<Target>", "<Packet>")
ParameterDescription
TargetTarget name
PacketPacket name
get_packet_derived_items("INST HEALTH_STATUS") # => ['PACKET_TIMESECONDS', 'PACKET_TIMEFORMATTED', ...]

Delays​

These methods allow the user to pause the script to wait for telemetry to change or for an amount of time to pass.

wait​

Since 5.0.0

Pauses the script for a configurable amount of time (minimum 10ms) or until a converted telemetry item meets given criteria. It supports three different syntaxes as shown. If no parameters are given then an infinite wait occurs until the user presses Go. Note that on a timeout, wait does not stop the script, usually wait_check is a better choice.

elapsed = wait() # => Returns the actual time waited as float
elapsed = wait(<Time>) # => Returns the actual time waited as float
ParameterDescription
TimeTime in Seconds to delay for.
# Returns True or False based on the whether the expression is True or False
success = wait(
"<Target Name> <Packet Name> <Item Name> <Comparison>",
<Timeout>,
<Polling Rate (optional)>,
type="CONVERTED",
quiet=False
) # => True / False
ParameterDescription
Target NameName of the target of the telemetry item.
Packet NameName of the telemetry packet of the telemetry item.
Item NameName of the telemetry item.
ComparisonA comparison to perform against the telemetry item.
TimeoutTimeout in seconds. Script will proceed if the wait statement times out waiting for the comparison to be true.
Polling RateHow often the comparison is evaluated in seconds. Defaults to 0.25 if not specified.
typeNamed parameter specifying the type. RAW, CONVERTED (default) or FORMATTED (Ruby symbol, Python string).
quietNamed parameter indicating whether to log the result. Defaults to false which means log the wait.
Supported Comparisons

A comparison is a single operator followed by a literal value. The supported operators are ==, !=, >, >=, <, <= and in. in requires a list operand, e.g. in [1, 2, 3], whose elements follow the same rules as any other value, e.g. in ['ON', 'OFF']. Python also accepts a tuple or set, e.g. in (1, 2). Compound expressions, e.g. TIMEUS & 0x0001 == 0x0000, are not supported - use wait_expression instead.

String Comparisons

When comparing against string or state values, the value must be quoted (e.g., == 'ON'). An unquoted value is rejected with Uninitialized constant ON. Did you mean 'ON' as a string?. Quoted values follow the string literal rules of the script language, so escape sequences are processed in Ruby double quoted strings and in all Python strings. The Ruby control and meta escapes \c, \C- and \M- are rejected rather than silently changed, as is string interpolation (Ruby "#{...}", Python f-strings) because the comparison is not evaluated as code. Interpolate in the script itself instead: Ruby wait("INST HEALTH_STATUS TYPE == '#{expected}'", 10) or Python wait(f"INST HEALTH_STATUS TYPE == '{expected}'", 10).

elapsed = wait()
elapsed = wait(5)
success = wait("INST HEALTH_STATUS COLLECTS == 3", 10)
success = wait("INST HEALTH_STATUS TYPE == 'NORMAL'", 10) # Strings must be quoted
success = wait("INST HEALTH_STATUS COLLECTS == 3", 10, type='RAW', quiet=True)

wait_tolerance​

Since 5.0.0

Pauses the script for a configurable amount of time or until a converted telemetry item meets equals an expected value within a tolerance. Note that on a timeout, wait_tolerance does not stop the script, usually wait_check_tolerance is a better choice.

# Returns True or False based on the whether the expression is true or false
success = wait_tolerance(
"<Target Name> <Packet Name> <Item Name>", # Can also be given as individual parameters
<Expected Value>,
<Tolerance>,
<Timeout>,
<Polling Rate (optional)>,
type="CONVERTED",
quiet=False
) # => True / False
ParameterDescription
Target NameName of the target of the telemetry item.
Packet NameName of the telemetry packet of the telemetry item.
Item NameName of the telemetry item.
Expected ValueExpected value of the telemetry item.
Tolerance± Tolerance on the expected value.
TimeoutTimeout in seconds. Script will proceed if the wait statement times out waiting for the comparison to be true.
Polling RateHow often the comparison is evaluated in seconds. Defaults to 0.25 if not specified.
typeNamed parameter specifying the type. RAW, CONVERTED (default) or FORMATTED (Ruby symbol, Python string).
quietNamed parameter indicating whether to log the result. Defaults to false which means log the wait.

Ruby Examples:

success = wait_tolerance("INST HEALTH_STATUS COLLECTS", 10.0, 5.0, 10)
success = wait_tolerance("INST HEALTH_STATUS COLLECTS", 10.0, 5.0, 10, type: :RAW, quiet: true)

Python Examples:

success = wait_tolerance("INST HEALTH_STATUS COLLECTS", 10.0, 5.0, 10)
success = wait_tolerance("INST HEALTH_STATUS COLLECTS", 10.0, 5.0, 10, type='RAW', quiet=True)

wait_expression​

Since 5.0.0

Pauses the script until an expression is evaluated to be true or a timeout occurs. If a timeout occurs the script will continue. This method can be used to perform more complicated comparisons than using wait as shown in the example. Note that on a timeout, wait_expression does not stop the script, usually wait_check_expression is a better choice.

# Return True or False based on the expression evaluation
wait_expression(
exp_to_eval,
timeout,
polling_rate=DEFAULT_TLM_POLLING_RATE,
globals=None,
locals=None,
quiet=False,
) # => True / False
ParameterDescription
expressionAn expression to evaluate.
timeoutTimeout in seconds. Script will proceed if the wait statement times out waiting for the comparison to be true.
polling_rateHow often the comparison is evaluated in seconds. Defaults to 0.25 if not specified.
context (ruby only)The context to call eval with. Defaults to nil. Context in Ruby is typically binding() and is usually not needed.
globals (python only)The globals to call eval with. Defaults to None. Note that to use COSMOS APIs like tlm() you must pass globals().
locals (python only)The locals to call eval with. Defaults to None. Note that if you're using local variables in a method you must pass locals().
quietWhether to log the result. Defaults to false which means log the wait.
def check(value):
# Here we using both tlm() and a local 'value' so we need to pass globals() and locals()
return wait_expression("tlm('INST HEALTH_STATUS COLLECTS') > value", 5, 0.25, globals(), locals(), quiet=True)
success = check(5)

wait_packet​

Since 5.0.0

Pauses the script until a certain number of packets have been received. If a timeout occurs the script will continue. Note that on a timeout, wait_packet does not stop the script, usually wait_check_packet is a better choice.

# Returns True or False based on the whether the packet was received
success = wait_packet(
"<Target>",
"<Packet>",
<Num Packets>,
<Timeout>,
<Polling Rate (optional)>,
quiet=False
) # => True / False
ParameterDescription
TargetThe target name
PacketThe packet name
Num PacketsThe number of packets to receive
TimeoutTimeout in seconds.
Polling RateHow often the comparison is evaluated in seconds. Defaults to 0.25 if not specified.
quietNamed parameter indicating whether to log the result. Defaults to false which means log the wait.
success = wait_packet('INST', 'HEALTH_STATUS', 5, 10) # Wait for 5 INST HEALTH_STATUS packets over 10s

wait_check​

Since 5.0.0

Combines the wait and check keywords into one. This pauses the script until the converted value of a telemetry item meets given criteria or times out. On a timeout the script stops.

# Returns the amount of time elapsed waiting for the expression
elapsed = wait_check(
"<Target Name> <Packet Name> <Item Name> <Comparison>",
<Timeout>,
<Polling Rate (optional)>,
type="CONVERTED"
) # => float
ParameterDescription
Target NameName of the target of the telemetry item.
Packet NameName of the telemetry packet of the telemetry item.
Item NameName of the telemetry item.
ComparisonA comparison to perform against the telemetry item.
TimeoutTimeout in seconds. Script will stop if the wait statement times out waiting for the comparison to be true.
Polling RateHow often the comparison is evaluated in seconds. Defaults to 0.25 if not specified.
typeNamed parameter specifying the type. RAW, CONVERTED (default) or FORMATTED (Ruby symbol, Python string).
Supported Comparisons

A comparison is a single operator followed by a literal value. The supported operators are ==, !=, >, >=, <, <= and in. in requires a list operand, e.g. in [1, 2, 3], whose elements follow the same rules as any other value, e.g. in ['ON', 'OFF']. Python also accepts a tuple or set, e.g. in (1, 2). Compound expressions, e.g. TIMEUS & 0x0001 == 0x0000, are not supported - use wait_check_expression instead.

String Comparisons

When comparing against string or state values, the value must be quoted (e.g., == 'ON'). An unquoted value is rejected with Uninitialized constant ON. Did you mean 'ON' as a string?. Quoted values follow the string literal rules of the script language, so escape sequences are processed in Ruby double quoted strings and in all Python strings. The Ruby control and meta escapes \c, \C- and \M- are rejected rather than silently changed, as is string interpolation (Ruby "#{...}", Python f-strings) because the comparison is not evaluated as code. Interpolate in the script itself instead: Ruby wait_check("INST HEALTH_STATUS TYPE == '#{expected}'", 10) or Python wait_check(f"INST HEALTH_STATUS TYPE == '{expected}'", 10).

elapsed = wait_check("INST HEALTH_STATUS COLLECTS > 5", 10)
elapsed = wait_check("INST HEALTH_STATUS TYPE == 'NORMAL'", 10) # Strings must be quoted
elapsed = wait_check("INST HEALTH_STATUS COLLECTS > 5", 10, type='RAW')

wait_check_tolerance​

Since 5.0.0

Pauses the script for a configurable amount of time or until a converted telemetry item equals an expected value within a tolerance. On a timeout the script stops.

# Returns the amount of time elapsed waiting for the expression
elapsed = wait_check_tolerance(
"<Target Name> <Packet Name> <Item Name>",
<Expected Value>,
<Tolerance>,
<Timeout>,
<Polling Rate (optional)>,
type="CONVERTED"
) # => float
ParameterDescription
Target NameName of the target of the telemetry item.
Packet NameName of the telemetry packet of the telemetry item.
Item NameName of the telemetry item.
Expected ValueExpected value of the telemetry item.
Tolerance± Tolerance on the expected value.
TimeoutTimeout in seconds. Script will stop if the wait statement times out waiting for the comparison to be true.
Polling RateHow often the comparison is evaluated in seconds. Defaults to 0.25 if not specified.
typeNamed parameter specifying the type. RAW, CONVERTED (default) or FORMATTED (Ruby symbol, Python string).
elapsed = wait_check_tolerance("INST HEALTH_STATUS COLLECTS", 10.0, 5.0, 10)
elapsed = wait_check_tolerance("INST HEALTH_STATUS COLLECTS", 10.0, 5.0, 10, type='RAW')

wait_check_expression​

Since 5.0.0

Pauses the script until an expression is evaluated to be true or a timeout occurs. If a timeout occurs the script will stop. This method can be used to perform more complicated comparisons than using wait as shown in the example. Also see the syntax notes for check_expression.

# Return time spent waiting for the expression to evaluate to True
wait_check_expression(
exp_to_eval,
timeout,
polling_rate=DEFAULT_TLM_POLLING_RATE,
globals=None,
locals=None
) # => float
ParameterDescription
expressionAn expression to evaluate.
timeoutTimeout in seconds. Script will proceed if the wait statement times out waiting for the comparison to be true.
polling_rateHow often the comparison is evaluated in seconds. Defaults to 0.25 if not specified.
context (ruby only)The context to call eval with. Defaults to nil. Context in Ruby is typically binding() and is usually not needed.
globals (python only)The globals to call eval with. Defaults to None. Note that to use COSMOS APIs like tlm() you must pass globals().
locals (python only)The locals to call eval with. Defaults to None. Note that if you're using local variables in a method you must pass locals().
# Note that for Python we need to pass globals() to be able to use COSMOS API methods like tlm()
elapsed = wait_check_expression("tlm('INST HEALTH_STATUS COLLECTS') > 5 and tlm('INST HEALTH_STATUS TEMP1') > 25.0", 10, 0.25, globals())

wait_check_packet​

Since 5.0.0

Pauses the script until a certain number of packets have been received. If a timeout occurs the script will stop.

# Returns the amount of time elapsed waiting for the packets
elapsed = wait_check_packet(
"<Target>",
"<Packet>",
<Num Packets>,
<Timeout>,
<Polling Rate (optional)>,
quiet=False
) # => float
ParameterDescription
TargetThe target name
PacketThe packet name
Num PacketsThe number of packets to receive
TimeoutTimeout in seconds. Script will stop if the wait statement times out waiting specified number of packets.
Polling RateHow often the comparison is evaluated in seconds. Defaults to 0.25 if not specified.
quietNamed parameter indicating whether to log the result. Defaults to false which means log the wait.
elapsed = wait_check_packet('INST', 'HEALTH_STATUS', 5, 10) # Wait for 5 INST HEALTH_STATUS packets over 10s

Limits​

These methods deal with handling telemetry limits.

limits_enabled?, limits_enabled​

Since 5.0.0

The limits_enabled? method returns true/false depending on whether limits are enabled for a telemetry item.

limits_enabled("<Target Name> <Packet Name> <Item Name>")
ParameterDescription
Target NameName of the target of the telemetry item.
Packet NameName of the telemetry packet of the telemetry item.
Item NameName of the telemetry item.
enabled = limits_enabled("INST HEALTH_STATUS TEMP1") # => True or False

enable_limits​

Since 5.0.0

Enables limits monitoring for the specified telemetry item.

enable_limits("<Target Name> <Packet Name> <Item Name>")
ParameterDescription
Target NameName of the target of the telemetry item.
Packet NameName of the telemetry packet of the telemetry item.
Item NameName of the telemetry item.
enable_limits("INST HEALTH_STATUS TEMP1")

disable_limits​

Since 5.0.0

Disables limits monitoring for the specified telemetry item.

disable_limits("<Target Name> <Packet Name> <Item Name>")
ParameterDescription
Target NameName of the target of the telemetry item.
Packet NameName of the telemetry packet of the telemetry item.
Item NameName of the telemetry item.
disable_limits("INST HEALTH_STATUS TEMP1")

enable_limits_group​

Since 5.0.0

Enables limits monitoring on a set of telemetry items specified in a limits group.

enable_limits_group("<Limits Group Name>")
ParameterDescription
Limits Group NameName of the limits group.
enable_limits_group("SAFE_MODE")

disable_limits_group​

Since 5.0.0

Disables limits monitoring on a set of telemetry items specified in a limits group.

disable_limits_group("<Limits Group Name>")
ParameterDescription
Limits Group NameName of the limits group.
disable_limits_group("SAFE_MODE")

get_limits_groups​

Since 5.0.0

Returns the list of limits groups in the system.

limits_groups = get_limits_groups()

set_limits_set​

Since 5.0.0

Sets the current limits set. The default limits set is DEFAULT.

set_limits_set("<Limits Set Name>")
ParameterDescription
Limits Set NameName of the limits set.
set_limits_set("DEFAULT")

delete_limits_set​

Since 7.2.1

Deletes a limits set and removes it from all telemetry items. The DEFAULT limits set and the currently active limits set cannot be deleted. Use set_limits_set to change the active set before deleting it. Use get_limits_sets to get the available limit set names.

delete_limits_set("<Limits Set Name>")
ParameterDescription
Limits Set NameName of the limits set to delete.
delete_limits_set("TVAC")

get_limits_set​

Since 5.0.0

Returns the name of the current limits set. The default limits set is DEFAULT.

limits_set = get_limits_set()

get_limits_sets​

Since 5.0.0

Returns the list of limits sets in the system.

limits_sets = get_limits_sets()

get_limits​

Since 5.0.0

Returns hash / dict of all the limits settings for a telemetry point.

get_limits(<Target Name>, <Packet Name>, <Item Name>)
ParameterDescription
Target NameName of the target of the telemetry item
Packet NameName of the telemetry packet of the telemetry item
Item NameName of the telemetry item
result = get_limits('INST', 'HEALTH_STATUS', 'TEMP1')
print(result) # => {'DEFAULT'=>[-80.0, -70.0, 60.0, 80.0, -20.0, 20.0], 'TVAC'=>[-80.0, -30.0, 30.0, 80.0]}
print(result.keys()) # => dict_keys(['DEFAULT', 'TVAC'])
print(result['DEFAULT']) # => [-80.0, -70.0, 60.0, 80.0, -20.0, 20.0]

set_limits​

Since 5.0.0

The set_limits method sets limits settings for a telemetry point. Note: In most cases it would be better to update your config files or use different limits sets rather than changing limits settings in realtime.

set_limits(<Target Name>, <Packet Name>, <Item Name>, <Red Low>, <Yellow Low>, <Yellow High>, <Red High>, <Green Low (optional)>, <Green High (optional)>, <Limits Set (optional)>, <Persistence (optional)>, <Enabled (optional)>)
ParameterDescription
Target NameName of the target of the telemetry item.
Packet NameName of the telemetry packet of the telemetry item.
Item NameName of the telemetry item.
Red LowRed Low setting for this limits set. Any value below this value will be make the item red.
Yellow LowYellow Low setting for this limits set. Any value below this value but greater than Red Low will be make the item yellow.
Yellow HighYellow High setting for this limits set. Any value above this value but less than Red High will be make the item yellow.
Red HighRed High setting for this limits set. Any value above this value will be make the item red.
Green LowOptional. If given, any value greater than Green Low and less than Green_High will make the item blue indicating a good operational value.
Green HighOptional. If given, any value greater than Green Low and less than Green_High will make the item blue indicating a good operational value.
Limits SetOptional. Set the limits for a specific limits set. If not given then it defaults to setting limits for the CUSTOM limits set.
PersistenceOptional. Set the number of samples this item must be out of limits before changing limits state. Defaults to no change. Note: This affects all limits settings across limits sets.
EnabledOptional. Whether or not limits are enabled for this item. Defaults to true. Note: This affects all limits settings across limits sets.
set_limits('INST', 'HEALTH_STATUS', 'TEMP1', -10.0, 0.0, 50.0, 60.0, 30.0, 40.0, 'TVAC', 1, True)

set_state_color​

Since 7.2.1

The set_state_color method changes the color associated with a telemetry item's state in realtime. Items with states (e.g. CONNECTED, UNAVAILABLE) use a state color (GREEN, YELLOW, or RED) to determine their limits state rather than numeric red/yellow/green limits. Pass None (Python) or nil (Ruby) as the color to clear (remove) the state color. Setting a color also enables limits for the item, but clearing the color (passing None/nil) does not change whether limits are enabled. Note: In most cases it would be better to update your config files rather than changing state colors in realtime.

set_state_color(<Target Name>, <Packet Name>, <Item Name>, <State Name>, <Color>)
ParameterDescription
Target NameName of the target of the telemetry item.
Packet NameName of the telemetry packet of the telemetry item.
Item NameName of the telemetry item.
State NameName of the state to change, e.g. 'CONNECTED'.
ColorNew color for the state. Must be one of GREEN, YELLOW, or RED. Pass None/nil to clear the state color.
set_state_color('INST', 'HEALTH_STATUS', 'GROUND1STATUS', 'CONNECTED', 'RED')
set_state_color('INST', 'HEALTH_STATUS', 'GROUND1STATUS', 'CONNECTED', None) # Clear the state color

get_out_of_limits​

Since 5.0.0

Returns an array with the target_name, packet_name, item_name, and limits_state of all items that are out of their limits ranges.

out_of_limits_items = get_out_of_limits()

get_overall_limits_state​

Since 5.0.0

Returns the overall limits state for the COSMOS system. Returns 'GREEN', 'YELLOW', or 'RED'.

get_overall_limits_state(<Ignored Items> (optional))
ParameterDescription
Ignored ItemsArray of arrays with items to ignore when determining the overall limits state. [['TARGET_NAME', 'PACKET_NAME', 'ITEM_NAME'], ...]
overall_limits_state = get_overall_limits_state()
overall_limits_state = get_overall_limits_state([['INST', 'HEALTH_STATUS', 'TEMP1']])

get_limits_events​

Since 5.0.0

Returns limits events based on an offset returned from the last time it was called.

get_limits_event(<Offset>, count)
ParameterDescription
OffsetOffset returned by the previous call to get_limits_event. Default is nil for the initial call
countNamed parameter specifying the maximum number of limits events to return. Default is 100
events = get_limits_event()
print(events)
#[["1613077715557-0",
# {"type"=>"LIMITS_CHANGE",
# "target_name"=>"TGT",
# "packet_name"=>"PKT",
# "item_name"=>"ITEM",
# "old_limits_state"=>"YELLOW_LOW",
# "new_limits_state"=>"RED_LOW",
# "time_nsec"=>"1",
# "message"=>"message"}],
# ["1613077715557-1",
# {"type"=>"LIMITS_CHANGE",
# "target_name"=>"TGT",
# "packet_name"=>"PKT",
# "item_name"=>"ITEM",
# "old_limits_state"=>"RED_LOW",
# "new_limits_state"=>"YELLOW_LOW",
# "time_nsec"=>"2",
# "message"=>"message"}]]
# The last offset is the first item ([0]) in the last event ([-1])
events = get_limits_event(events[-1][0])
print(events)
#[["1613077715657-0",
# {"type"=>"LIMITS_CHANGE",
# ...

Plugins / Packages​

APIs for getting knowledge about plugins and packages.

plugin_list​

Since 5.6.0

Returns all the installed plugins.

plugin_list(default = False)
ParameterDescription
defaultWhether to include the default COSMOS plugins (all the regular applications)
plugins = plugin_list() # => ['openc3-cosmos-demo-6.0.3.pre.beta0.20250116214358.gem__20250116214539']
plugins = plugin_list(default=True) # =>
# ['openc3-cosmos-demo-6.0.3.pre.beta0.20250116214358.gem__20250116214539',
# 'openc3-cosmos-tool-admin-6.0.3.pre.beta0.20250115200004.gem__20250116211504',
# 'openc3-cosmos-tool-bucketexplorer-6.0.3.pre.beta0.20250115200008.gem__20250116211525',
# 'openc3-cosmos-tool-cmdsender-6.0.3.pre.beta0.20250115200012.gem__20250116211515',
# 'openc3-cosmos-tool-cmdtlmserver-6.0.3.pre.beta0.20250115200015.gem__20250116211512',
# 'openc3-cosmos-tool-dataextractor-6.0.3.pre.beta0.20250115200005.gem__20250116211521',
# 'openc3-cosmos-tool-dataviewer-6.0.3.pre.beta0.20250115200009.gem__20250116211522',
# 'openc3-cosmos-tool-docs-6.0.3.pre.beta0.20250117042104.gem__20250117042154',
# 'openc3-cosmos-tool-handbooks-6.0.3.pre.beta0.20250115200014.gem__20250116211523',
# 'openc3-cosmos-tool-iframe-6.0.3.pre.beta0.20250115200011.gem__20250116211503',
# 'openc3-cosmos-tool-limitsmonitor-6.0.3.pre.beta0.20250115200017.gem__20250116211514',
# 'openc3-cosmos-tool-packetviewer-6.0.3.pre.beta0.20250115200004.gem__20250116211518',
# 'openc3-cosmos-tool-scriptrunner-6.0.3.pre.beta0.20250115200012.gem__20250116211517',
# 'openc3-cosmos-tool-tablemanager-6.0.3.pre.beta0.20250115200018.gem__20250116211524',
# 'openc3-cosmos-tool-tlmgrapher-6.0.3.pre.beta0.20250115200005.gem__20250116211520',
# 'openc3-cosmos-tool-tlmviewer-6.0.3.pre.beta0.20250115200008.gem__20250116211519',
# 'openc3-tool-base-6.0.3.pre.beta0.20250115195959.gem__20250116211459']

plugin_get​

Since 5.6.0

Returns information about an installed plugin.

plugin_get(<Plugin Name>)
ParameterDescription
Plugin NameFull name of the plugin (typically taken from plugin_list())
plugin_get('openc3-cosmos-demo-6.0.3.pre.beta0.20250116214358.gem__20250116214539') # =>
# { "name"=>"openc3-cosmos-demo-6.0.3.pre.beta0.20250116214358.gem__20250116214539",
# "variables"=>{"inst_target_name"=>"INST", ...},
# "plugin_txt_lines"=>["# Note: This plugin includes 4 targets ..."],
# "needs_dependencies"=>true,
# "updated_at"=>1737063941094624764 }

package_list​

Since 5.12.0

List all the packages installed in COSMOS.

package_list() # => {'ruby': ['openc3-cosmos-demo-6.0.3.pre.beta0.20250116214358.gem', ..., 'openc3-tool-base-6.0.3.pre.beta0.20250115195959.gem'],
# 'python': ['numpy-2.1.1', 'pip-24.0', 'setuptools-65.5.0']}

Targets​

APIs for getting knowledge about targets.

get_target_names​

Since 5.9.1

Returns a list of the targets in the system in an array. Previously get_target_list.

targets = get_target_names() # => ['INST', 'INST2', 'EXAMPLE', 'TEMPLATED']

get_target​

Since 5.0.0

Returns a target hash containing all the information about the target.

get_target("<Target Name>")
ParameterDescription
Target NameName of the target.
target = get_target("INST")
print(target)
# {"name"=>"INST",
# "folder_name"=>"INST",
# "requires"=>[],
# "ignored_parameters"=>
# ["CCSDSVER",
# "CCSDSTYPE",
# "CCSDSSHF",
# "CCSDSAPID",
# "CCSDSSEQFLAGS",
# "CCSDSSEQCNT",
# "CCSDSLENGTH",
# "PKTID"],
# "ignored_items"=>
# ["CCSDSVER",
# "CCSDSTYPE",
# "CCSDSSHF",
# "CCSDSAPID",
# "CCSDSSEQFLAGS",
# "CCSDSSEQCNT",
# "CCSDSLENGTH",
# "RECEIVED_COUNT",
# "RECEIVED_TIMESECONDS",
# "RECEIVED_TIMEFORMATTED"],
# "limits_groups"=>[],
# "cmd_tlm_files"=>
# [".../targets/INST/cmd_tlm/inst_cmds.txt",
# ".../targets/INST/cmd_tlm/inst_tlm.txt"],
# "cmd_unique_id_mode"=>false,
# "tlm_unique_id_mode"=>false,
# "id"=>nil,
# "updated_at"=>1613077058266815900,
# "plugin"=>nil}

get_target_interfaces​

Since 5.0.0

Returns the interfaces for all targets. The return value is an array of arrays where each subarray contains the target name, and a String of all the interface names.

target_ints = get_target_interfaces()
for target_name, interfaces in target_ints:
print(f"Target: {target_name}, Interfaces: {interfaces}")

Interfaces​

These methods allow the user to manipulate COSMOS interfaces.

get_interface​

Since 5.0.0

Returns an interface status including the as built interface and its current status (cmd/tlm counters, etc).

get_interface("<Interface Name>")
ParameterDescription
Interface NameName of the interface.
interface = get_interface("INST_INT")
print(interface)
# {"name"=>"INST_INT",
# "config_params"=>["interface.rb"],
# "target_names"=>["INST"],
# "connect_on_startup"=>true,
# "auto_reconnect"=>true,
# "reconnect_delay"=>5.0,
# "disable_disconnect"=>false,
# "options"=>[],
# "protocols"=>[],
# "log"=>true,
# "log_raw"=>false,
# "plugin"=>nil,
# "updated_at"=>1613076213535979900,
# "state"=>"CONNECTED",
# "clients"=>0,
# "txsize"=>0,
# "rxsize"=>0,
# "txbytes"=>0,
# "rxbytes"=>0,
# "txcnt"=>0,
# "rxcnt"=>0}

get_interface_names​

Since 5.0.0

Returns a list of the interfaces in the system in an array.

interface_names = get_interface_names() # => ['INST_INT', 'INST2_INT', 'EXAMPLE_INT', 'TEMPLATED_INT']

connect_interface​

Since 5.0.0

Connects to targets associated with a COSMOS interface.

If the interface is already connected and no interface parameters are given, this method is a no-op and the existing connection is left untouched. To force a reconnect, call disconnect_interface first, or pass interface parameters which always rebuild the interface.

connect_interface("<Interface Name>", <Interface Parameters (optional)>)
ParameterDescription
Interface NameName of the interface.
Interface ParametersParameters used to initialize the interface. If none are given then the interface will use the parameters that were given in the server configuration file.
connect_interface("INT1")
connect_interface("INT1", hostname, port)

disconnect_interface​

Since 5.0.0

Disconnects from targets associated with a COSMOS interface. If the interface has already been disconnected this is a no-op. Otherwise the interface is asked to disconnect regardless of whether it reports being connected, so any resources it is still holding are released.

disconnect_interface("<Interface Name>")
ParameterDescription
Interface NameName of the interface.
disconnect_interface("INT1")

start_raw_logging_interface​

Since 5.0.0

Starts logging of raw data on one or all interfaces. This is for debugging purposes only.

start_raw_logging_interface("<Interface Name (optional)>")
ParameterDescription
Interface NameName of the Interface to command to start raw data logging. Defaults to 'ALL' which causes all interfaces that support raw data logging to start logging raw data.
start_raw_logging_interface("int1")

stop_raw_logging_interface​

Since 5.0.0

Stops logging of raw data on one or all interfaces. This is for debugging purposes only.

stop_raw_logging_interface("<Interface Name (optional)>")
ParameterDescription
Interface NameName of the Interface to command to stop raw data logging. Defaults to 'ALL' which causes all interfaces that support raw data logging to stop logging raw data.
stop_raw_logging_interface("int1")

get_all_interface_info​

Since 5.0.0

Returns information about all interfaces. The return value is an array of arrays where each subarray contains the interface name, connection state, number of connected clients, transmit queue size, receive queue size, bytes transmitted, bytes received, command count, and telemetry count.

interface_info = get_all_interface_info()
for interface in interface_info():
# [interface_name, connection_state, num_clients, tx_q_size, rx_q_size, tx_bytes, rx_bytes, cmd_count, tlm_count]
print(f"Interface: {interface[0]}, Connection state: {interface[1]}, Num connected clients: {interface[2]}")
print(f"Transmit queue size: {interface[3]}, Receive queue size: {interface[4]}, Bytes transmitted: {interface[5]}, Bytes received: {interface[6]}")
print(f"Cmd count: {interface[7]}, Tlm count: {interface[8]}")

map_target_to_interface​

Since 5.0.9

Map a target to an interface allowing target commands and telemetry to be processed by that interface. Note this will cause the interface to respawn.

map_target_to_interface("<Target Name>", "<Interface Name>", cmd_only=False, tlm_only=False, unmap_old=True)
ParameterDescription
Target NameName of the target
Interface NameName of the interface
cmd_onlyNamed parameter whether to map target commands only to the interface (default: false)
tlm_onlyNamed parameter whether to map target telemetry only to the interface (default: false)
unmap_oldNamed parameter whether remove the target from all existing interfaces (default: true)
map_target_to_interface("INST", "INST_INT", unmap_old=False)
map_target_to_interface("INST", "INST_INT", cmd_only=True)
map_target_to_interface("INST", "INST_INT", tlm_only=True)

unmap_target_from_interface​

Since 6.9.0

Unmap a target from an interface. This removes all knowledge of the target and will cause the interface to respawn.

unmap_target_from_interface("<Target Name>", "<Interface Name>", cmd_only=False, tlm_only=False)
ParameterDescription
Target NameName of the target
Interface NameName of the interface
cmd_onlyNamed parameter whether to unmap target commands only to the interface (default: false)
tlm_onlyNamed parameter whether to unmap target telemetry only to the interface (default: false)
unmap_target_from_interface("INST", "INST_INT")
unmap_target_from_interface("INST", "INST_INT", cmd_only=True)
unmap_target_from_interface("INST", "INST_INT", tlm_only=True)

interface_cmd​

Since 5.3.0

Send a command directly to an interface. This has no effect in the standard COSMOS interfaces but can be implemented by a custom interface to change behavior.

interface_cmd("<Interface Name>", "<Command Name>", "<Command Parameters>")
ParameterDescription
Interface NameName of the interface
Command NameName of the command to send
Command ParametersAny parameters to send with the command
interface_cmd("INST", "DISABLE_CRC")

interface_protocol_cmd​

Since 5.3.0

Send a command directly to an interface protocol. This has no effect in the standard COSMOS protocols but can be implemented by a custom protocol to change behavior.

interface_protocol_cmd("<Interface Name>", "<Command Name>", "<Command Parameters>")
ParameterDescription
Interface NameName of the interface
Command NameName of the command to send
Command ParametersAny parameters to send with the command
read_writeWhether command gets send to read or write protocols. Must be one of READ, WRITE, or READ_WRITE (Ruby symbols, Python strings). The default is READ_WRITE.
indexWhich protocol in the stack the command should apply to. The default is -1 which applies the command to all.
interface_protocol_cmd("INST", "DISABLE_CRC", read_write='READ_WRITE', index=-1)

interface_target_enable​

Since 6.9.0

Enable commanding and telemetry processing for a given target in an interface.

interface_target_enable("<Interface Name>", "<Target Name>", cmd_only=False, tlm_only=False)
ParameterDescription
Interface NameName of the interface
Target NameName of the target
cmd_onlyNamed parameter whether to enable target commands only to the interface (default: false)
tlm_onlyNamed parameter whether to enable target telemetry only to the interface (default: false)
interface_target_enable("INST_INT", "INST")
interface_target_enable("INST_INT", "INST", cmd_only=True)
interface_target_enable("INST_INT", "INST", tlm_only=True)

interface_target_disable​

Since 6.9.0

Disable commanding and telemetry processing for a given target in an interface.

interface_target_disable("<Interface Name>", "<Target Name>", cmd_only=False, tlm_only=False)
ParameterDescription
Interface NameName of the interface
Target NameName of the target
cmd_onlyNamed parameter whether to disable target commands only to the interface (default: false)
tlm_onlyNamed parameter whether to disable target telemetry only to the interface (default: false)
interface_target_disable("INST_INT", "INST")
interface_target_disable("INST_INT", "INST", cmd_only=True)
interface_target_disable("INST_INT", "INST", tlm_only=True)

interface_details​

Since 6.9.0

Get details on the interface and its protocols. Unlike get_interface, which returns the stored interface model, interface_details queries the running interface microservice and returns its live runtime state, including the most recent raw data read and written and the state of every read and write protocol.

interface_details("<Interface Name>")
ParameterDescription
Interface NameName of the interface
details = interface_details("INST_INT")
print(details["state"])
# CONNECTED
print(details["cmd_target_enabled"])
# {'INST': True}
print(details["hostname"], details["write_port"], details["read_port"])
# host.docker.internal 8080 8081
print([protocol["name"] for protocol in details["read_protocols"]])
# ['BurstProtocol']
print(details["read_protocols"][0]["discard_leading_bytes"])
# 0

# Raw data arrives as a json_class dict when it is not printable text
raw = details["read_raw_data"]
if isinstance(raw, dict):
raw = bytes(raw["raw"])
print(raw)
# b'\x00\x01\x02\x03'

Return Value

Returns a hash / dict of the interface state. The following keys are always present for every interface type.

KeyTypeDescription
nameStringInterface name, e.g. "INST_INT"
stateString"CONNECTED", "ATTEMPTING", or "DISCONNECTED"
clientsIntegerNumber of connected clients (server interfaces only, otherwise 0)
txsizeIntegerNumber of packets currently queued to be written
rxsizeIntegerNumber of packets currently queued to be read
txbytesIntegerRunning total of bytes written to the interface
rxbytesIntegerRunning total of bytes read from the interface
txcntIntegerRunning total of packets written to the interface
rxcntIntegerRunning total of packets read from the interface

Each interface type adds its own keys on top of the common keys above. The values mirror the interface's configuration parameters documented in the Interfaces guide.

KeyTypeDescription
hostnameStringHost the interface connects to
write_portIntegerPort used to write commands
read_portIntegerPort used to read telemetry
write_timeoutFloatSeconds to wait for a write
read_timeoutFloatSeconds to wait for a read, nil for no timeout
Timestamp Formats

All timestamp strings are ISO 8601 in UTC with microsecond precision, e.g. 2026-08-26T16:15:30.123456Z. The format is the same whether the interface is implemented in Ruby or Python.

Binary Data

The raw data and protocol data values hold the actual bytes that crossed the interface. Data which is not printable text is encoded on the wire as {"json_class": "String", "raw": [<byte values>]}. Ruby scripts receive this decoded back into a binary String. Python scripts receive the dict as-is, so convert it with bytes(value["raw"]) before using it.

Disconnect Mode

interface_details is not on the read-only allowlist used by disconnect_script, so it returns None (Python) or nil (Ruby) when running in disconnect mode.

Routers​

These methods allow the user to manipulate COSMOS routers.

connect_router​

Since 5.0.0

Connects a COSMOS router.

connect_router("<Router Name>", <Router Parameters (optional)>)
ParameterDescription
Router NameName of the router.
Router ParametersParameters used to initialize the router. If none are given then the router will use the parameters that were given in the server configuration file.
connect_router("INST_ROUTER")
connect_router("INST_ROUTER", 7779, 7779, nil, 10.0, 'PREIDENTIFIED')

disconnect_router​

Since 5.0.0

Disconnects a COSMOS router.

disconnect_router("<Router Name>")
ParameterDescription
Router NameName of the router.
disconnect_router("INT1_ROUTER")

get_router_names​

Since 5.0.0

Returns a list of the routers in the system in an array.

router_names = get_router_names() # => ['ROUTER_INT']

get_router​

Since 5.0.0

Returns a router status including the as built router and its current status (cmd/tlm counters, etc).

get_router("<Router Name>")
ParameterDescription
Router NameName of the router.
router = get_router("ROUTER_INT")
print(router)
#{"name"=>"ROUTER_INT",
# "config_params"=>["router.rb"],
# "target_names"=>["INST"],
# "connect_on_startup"=>true,
# "auto_reconnect"=>true,
# "reconnect_delay"=>5.0,
# "disable_disconnect"=>false,
# "options"=>[],
# "protocols"=>[],
# "log"=>true,
# "log_raw"=>false,
# "plugin"=>nil,
# "updated_at"=>1613076213535979900,
# "state"=>"CONNECTED",
# "clients"=>0,
# "txsize"=>0,
# "rxsize"=>0,
# "txbytes"=>0,
# "rxbytes"=>0,
# "txcnt"=>0,
# "rxcnt"=>0}

get_all_router_info​

Since 5.0.0

Returns information about all routers. The return value is an array of arrays where each subarray contains the router name, connection state, number of connected clients, transmit queue size, receive queue size, bytes transmitted, bytes received, packets received, and packets sent.

router_info = get_all_router_info()
# router_name, connection_state, num_clients, tx_q_size, rx_q_size, tx_bytes, rx_bytes, pkts_rcvd, pkts_sent
for router in router_info:
print(f"Router: {router[0]}, Connection state: {router[1]}, Num connected clients: {router[2]}")
print(f"Transmit queue size: {router[3]}, Receive queue size: {router[4]}, Bytes transmitted: {router[5]}, Bytes received: {router[6]}")
print(f"Packets received: {router[7]}, Packets sent: {router[8]}")

map_target_to_router​

Since 6.9.0

Map a target to an router allowing target commands and telemetry to be processed by that router. Note: this will cause the router to respawn.

map_target_to_router("<Target Name>", "<Router Name>", cmd_only=False, tlm_only=False, unmap_old=True)
ParameterDescription
Target NameName of the target
Router NameName of the router
cmd_onlyNamed parameter whether to map target commands only to the router (default: false)
tlm_onlyNamed parameter whether to map target telemetry only to the router (default: false)
unmap_oldNamed parameter whether remove the target from all existing routers (default: true)
map_target_to_router("INST", "INST_ROUTER", unmap_old=False)
map_target_to_router("INST", "INST_ROUTER", cmd_only=True)
map_target_to_router("INST", "INST_ROUTER", tlm_only=True)

unmap_target_from_router​

Since 6.9.0

Unmap a target from an router. This removes all knowledge of the target and will cause the router to respawn.

unmap_target_from_router("<Target Name>", "<Router Name>", cmd_only=False, tlm_only=False)
ParameterDescription
Target NameName of the target
Router NameName of the router
cmd_onlyNamed parameter whether to unmap target commands only to the router (default: false)
tlm_onlyNamed parameter whether to unmap target telemetry only to the router (default: false)
unmap_target_from_router("INST", "INST_ROUTER")
unmap_target_from_router("INST", "INST_ROUTER", cmd_only=True)
unmap_target_from_router("INST", "INST_ROUTER", tlm_only=True)

start_raw_logging_router​

Since 5.0.0

Starts logging of raw data on one or all routers. This is for debugging purposes only.

start_raw_logging_router("<Router Name (optional)>")
ParameterDescription
Router NameName of the Router to command to start raw data logging. Defaults to 'ALL' which causes all routers that support raw data logging to start logging raw data.
start_raw_logging_router("router1")

stop_raw_logging_router​

Since 5.0.0

Stops logging of raw data on one or all routers. This is for debugging purposes only.

stop_raw_logging_router("<Router Name (optional)>")
ParameterDescription
Router NameName of the Router to command to stop raw data logging. Defaults to 'ALL' which causes all routers that support raw data logging to stop logging raw data.
stop_raw_logging_router("router1")

router_cmd​

Since 5.3.0

Send a command directly to a router. This has no effect in the standard COSMOS routers but can be implemented by a custom router to change behavior.

router_cmd("<Router Name>", "<Command Name>", "<Command Parameters>")
ParameterDescription
Router NameName of the router
Command NameName of the command to send
Command ParametersAny parameters to send with the command
router_cmd("INST", "DISABLE_CRC")

router_protocol_cmd​

Since 5.3.0

Send a command directly to an router protocol. This has no effect in the standard COSMOS protocols but can be implemented by a custom protocol to change behavior.

router_protocol_cmd("<Router Name>", "<Command Name>", "<Command Parameters>", read_write, index)
ParameterDescription
Router NameName of the router
Command NameName of the command to send
Command ParametersAny parameters to send with the command
read_writeWhether command gets send to read or write protocols. Must be one of READ, WRITE, or READ_WRITE (Ruby symbols, Python strings). The default is READ_WRITE.
indexWhich protocol in the stack the command should apply to. The default is -1 which applies the command to all.
router_protocol_cmd("INST", "DISABLE_CRC", read_write='READ_WRITE', index=-1)

router_target_enable​

Since 6.9.0

Enable commanding and telemetry processing for a given target in a router.

router_target_enable("<Router Name>", "<Target Name>", cmd_only=False, tlm_only=False)
ParameterDescription
Router NameName of the router
Target NameName of the target
cmd_onlyNamed parameter whether to enable target commands only to the router (default: false)
tlm_onlyNamed parameter whether to enable target telemetry only to the router (default: false)
router_target_enable("INST_ROUTER", "INST")
router_target_enable("INST_ROUTER", "INST", cmd_only=True)
router_target_enable("INST_ROUTER", "INST", tlm_only=True)

router_target_disable​

Since 6.9.0

Disable commanding and telemetry processing for a given target in a router.

router_target_disable("<Router Name>", "<Target Name>", cmd_only=False, tlm_only=False)
ParameterDescription
Router NameName of the router
Target NameName of the target
cmd_onlyNamed parameter whether to disable target commands only to the router (default: false)
tlm_onlyNamed parameter whether to disable target telemetry only to the router (default: false)
router_target_disable("INST_ROUTER", "INST")
router_target_disable("INST_ROUTER", "INST", cmd_only=True)
router_target_disable("INST_ROUTER", "INST", tlm_only=True)

router_details​

Since 6.9.0

Get details on the router and its protocols.

router_details("<Router Name>")
ParameterDescription
Router NameName of the router
router_details("INST_ROUTER")

Tables​

These methods allow the user to script Table Manager.

table_create_binary​

Since 6.1.0

Creates a table binary based on a table definition file. You can achieve the same result in the Table Manager GUI with File->New File. Returns the path to the binary file created.

table_create_binary(<Table Definition File>, scope=OPENC3_SCOPE)
ParameterDescription
Table Definition FilePath to the table definition file, e.g. INST/tables/config/ConfigTables_def.txt
scopeScope to use, defaults to current scope
# Full example of using table_create_binary and then editing the binary
from openc3.tools.table_manager.table_config import TableConfig
# Returns a dict: {'filename': 'INST/tables/bin/ConfigTables.bin'}
table = table_create_binary("INST2/tables/config/ConfigTables_def.txt")
file = get_target_file(table['filename'])
table_binary = file.read()

# Get the definition file so we can process the binary
def_file = get_target_file("INST2/tables/config/MCConfigurationTable_def.txt")
# Access the internal TableConfig to process the definition
config = TableConfig.process_file(def_file.name)
# Grab the table by the definition name, e.g. TABLE "MC_Configuration"
table = config.table('MC_CONFIGURATION')
# Now you can read or write individual items in the table
table.write("MEMORY_SCRUBBING", "DISABLE")
# Finally write the table.buffer (the binary) back to storage
put_target_file("INST2/tables/bin/MCConfigurationTable_NoScrub.bin", table.buffer)

table_create_report​

Since 6.1.0

Creates a CSV report of the values in a table binary. You can achieve the same result in the Table Manager GUI with the Download Report button. Returns the report filename and the report contents.

The report is written into the target's storage next to the binary, replacing the binary's extension with .csv. For example a report of INST/tables/bin/ConfigTables.bin is written to INST/tables/bin/ConfigTables.csv and can be read back with get_target_file. In the Bucket Explorer it appears under DEFAULT/targets_modified/INST/tables/bin/ConfigTables.csv. Pass save as false to get the contents in the return value without writing a file.

table_create_report(<Table Binary Filename>, <Table Definition File>, table_name=None, save=True, scope=OPENC3_SCOPE)
ParameterDescription
Table Binary FilePath to the table binary file, e.g. INST/tables/bin/ConfigTables.bin
Table Definition FilePath to the table definition file, e.g. INST/tables/config/ConfigTables_def.txt
table_nameName of the table to create the report. This only applies if the Table Binary and Table Definition consist of multiple tables. By default the report consists of all tables and is named after the binary file. If the table name is given, the report is just the specified table and is named after the table.
saveWhether to write the report into the target's storage. Defaults to true. Pass false to only return the contents.
scopeScope to use, defaults to current scope
table = table_create_report("INST/tables/bin/ConfigTables.bin", "INST/tables/config/ConfigTables_def.txt")
# Read the report back out of the target
file = get_target_file(table['filename'])
print(file.read())
file.close() # delete file

table = table_create_report("INST/tables/bin/ConfigTables.bin", "INST/tables/config/ConfigTables_def.txt", table_name="MC_CONFIGURATION", save=False)
print(table['filename']) #=> INST/tables/bin/McConfiguration.csv
print(table['contents']) #=> MC_CONFIGURATION\nLabel, Value\n...

Stashing Data​

These methods allow the user to store data into COSMOS and retrieve it. The storage is implemented as a key / value storage (Ruby hash or Python dict). This can be used in scripts to store information that applies across multiple scripts or multiple runs of a single script. As a best practice, use stashes to store transient data, since duplicate keys will be overwritten by the most recent write.

stash_set​

Since 5.2.0

Sets a stash item.

stash_set("<Stash Key>", <Stash Value>)
ParameterDescription
Stash KeyName of the stash key to set
Stash ValueValue to set
stash_set('run_count', 5)
stash_set('setpoint', 23.4)

stash_get​

Since 5.2.0

Returns the specified stash item.

stash_get("<Stash Key>")
ParameterDescription
Stash KeyName of the stash key to return
stash_get('run_count') # => 5

stash_all​

Since 5.2.0

Returns all the stash items as a Ruby hash or Python dict.

stash_all() # => ['run_count': 5, 'setpoint': 23.4]

stash_keys​

Since 5.2.0

Returns all the stash keys.

stash_keys() # => ['run_count', 'setpoint']

stash_delete​

Since 5.2.0

Deletes a stash item. Note this actions is permanent!

stash_delete("<Stash Key>")
ParameterDescription
Stash KeyName of the stash key to delete
stash_delete("run_count")

Telemetry Screens​

These methods allow the user to open, close or create unique telemetry screens from within a test procedure.

display_screen​

Since 5.6.0

Opens a telemetry screen at the specified position.

display_screen("<Target Name>", "<Screen Name>", <X Position (optional)>, <Y Position (optional)>)
ParameterDescription
Target NameTelemetry screen target name
Screen NameScreen name within the specified target
X PositionX coordinate for the upper left hand corner of the screen
Y PositionY coordinate for the upper left hand corner of the screen
display_screen("INST", "ADCS", 100, 200)

clear_screen​

Since 5.6.0

Closes an open telemetry screen.

clear_screen("<Target Name>", "<Screen Name>")
ParameterDescription
Target NameTelemetry screen target name. Note for local screens the target is 'LOCAL'.
Screen NameScreen name within the specified target
clear_screen("INST", "ADCS")

clear_all_screens​

Since 5.6.0

Closes all open screens.

clear_all_screens()

delete_screen​

Since 5.6.0

Deletes an existing Telemetry Viewer screen.

delete_screen("<Target Name>", "<Screen Name>")
ParameterDescription
Target NameTelemetry screen target name
Screen NameScreen name within the specified target
delete_screen("INST", "ADCS")

get_screen_list​

Since 5.6.0

Returns a hash of available telemetry screens keyed by target name, where each value is a list of screen names for that target.

get_screen_list() # => {'INST': ['ADCS', 'COMMANDING', ...], 'INST2': [...]}

get_screen_definition​

Since 5.6.0

Returns the text file contents of a telemetry screen definition.

get_screen_definition("<Target Name>", "<Screen Name>")
ParameterDescription
Target NameTelemetry screen target name
Screen NameScreen name within the specified target
screen_definition = get_screen_definition("INST", "HS")

create_screen​

Since 5.6.0

Allows you to create a screen directly from a script. This screen is saved to Telemetry Viewer for future use in that application.

create_screen("<Target Name>", "<Screen Name>" "<Definition>")
ParameterDescription
Target NameTelemetry screen target name
Screen NameScreen name within the specified target
DefinitionThe entire screen definition as a String
screen_def = """
SCREEN AUTO AUTO 0.1
VERTICAL
TITLE "New Screen"
VERTICALBOX
LABELVALUE INST HEALTH_STATUS TEMP1
END
END
"""
# Here we pass in the screen definition as a string
create_screen("INST", "LOCAL", screen_def)

local_screen​

Since 5.6.0

Allows you to create a local screen directly from a script which is not permanently saved to the Telemetry Viewer screen list. This is useful for one off screens that help users interact with scripts.

local_screen("<Screen Name>", "<Definition>", <X Position (optional)>, <Y Position (optional)>)
ParameterDescription
Screen NameScreen name within the specified target
DefinitionThe entire screen definition as a String
X PositionX coordinate for the upper left hand corner of the screen
Y PositionY coordinate for the upper left hand corner of the screen

NOTE: It is possible to specify a X, Y location off the visible display. If you do so and try to re-create the screen it will not display (because it is already displayed). Try issuing a clear_all_screens() first to clear any screens off the visible display space.

screen_def = """
SCREEN AUTO AUTO 0.1
VERTICAL
TITLE "Local Screen"
VERTICALBOX
LABELVALUE INST HEALTH_STATUS TEMP1
END
END
"""
# Here we pass in the screen definition as a string
local_screen("TESTING", screen_def, 600, 75)
wait(5)
# The local screen 'target' is 'LOCAL'
clear_screen("LOCAL", "TESTING")

Script Runner Scripts​

These methods allow the user to control Script Runner scripts.

start​

Since 5.0.0

Starts execution of another high level test procedure. Script Runner will load the file and immediately start executing it (with line-by-line highlighting in Script Runner) before jumping back to the calling procedure. Parameters are not directly given to high level test procedures, though they can use environment variables. If parameters are necessary, consider using a subroutine.

start("<Procedure Filename>")
ParameterDescription
Procedure FilenameName of the test procedure file. These are target-relative paths, e.g. TARGET/procedures/proc.rb. The file is retrieved from the COSMOS configuration bucket.
start("INST/procedures/test1.py")

goto​

Since 6.4.0

Jumps to a specific line in either the current file or another file. Script context and local variables are retained if jumping in the same file. This method should not be used to create loops or in any other case where normal Python/Ruby functionality could perform the same function. Use of goto increases stack depth and excess use can lead to stack level too deep errors.

goto(line_number)
goto(filename, line_number)
ParameterDescription
Line Number or Procedure FilenameIf this is the only argument given then it is the line number to goto in the same script. If two arguments are given, then it is the filename to goto
Line NumberIf the second argument, it is the line number in the filename given by the first argument to goto
goto(23)
goto("TARGET/procedures/other_script.rb", 5)
goto("TARGET/procedures/other_script.py", 12)

load_utility​

Since 5.0.0

Reads in a script file that contains useful subroutines for use in your test procedure. The file is instrumented, but instrumentation is disabled while load_utility is loading it so lines are not highlighted during the initial load. Afterwards, any functions defined in the loaded file will be highlighted when called. If you want to import subroutines but do not want their lines to be highlighted at all in Script Runner, use the standard Ruby load or require statement or Python import statement instead.

load_utility("TARGET/lib/<Utility Filename>")
ParameterDescription
Utility FilenameName of the script file containing subroutines. Include the full target name and path such as TARGET/lib/utility.rb. The file extension (.rb or .py) is optional and will be appended automatically if omitted.
load_utility("INST/lib/mode_changes.py")
load_utility("INST/lib/mode_changes") # .py extension is added automatically

script_list​

Since 5.0.0

Returns all the available files in COSMOS as an array / list. This includes configuration files at every directory level to ensure the user has access to every file. You can filter the list client side to just the 'lib' and or 'procedures' directories if you wish. Note: script names do NOT include '*' to indicate modified.

scripts = script_list()
print(len(scripts))
print(list(script for script in scripts if '/lib/' in script or '/procedures/' in script)) # =>
# [EXAMPLE/lib/example_interface.rb, INST/lib/example_limits_response.rb, ...]

script_create​

Since 5.0.0

Creates a new script with the given contents.

script_create("<Script Name>", "<Script Contents>")
ParameterDescription
Script NameFull path name of the script starting with the target
Script ContentsScript contents as text
contents = 'print("Hello from Python")'
script_create("INST2/procedures/new_script.py", contents)

script_body​

Since 5.0.0

Returns the script contents.

script_body("<Script Name>")
ParameterDescription
Script NameFull path name of the script starting with the target
script = script_body("INST2/procedures/checks.py")
print(script) # => # import os\n\n# Display the environment variables ...

script_delete​

Since 5.0.0

Deletes a script from COSMOS. Note, you can only really delete TEMP scripts and modified scripts. Scripts that are part of an installed COSMOS plugin remain as they were installed.

script_delete("<Script Name>")
ParameterDescription
Script NameFull path name of the script starting with the target
script_delete("INST/procedures/checks.rb")

script_run​

Since 5.0.0

Kicks off running a script in Script Runner. The script will run in the background, while the caller continues to run, and can be opened in Script Runner by selecting "Execution Status" from the "Script" menu at the top, and then connecting to it from the "Running Scripts" tab.

Note: In Enterprise, initialize_offline_access must have been called at least once for the user who calls this method.

script_run("<Script Name>", disconnect=False, environment=None, suite_runner=None)
ParameterDescription
Script NameFull path name of the script starting with the target. If this is the path to a test suite file, the suite_runner parameter must also be provided.
disconnectBoolean indicating whether to run the script in Disconnect
environmentHash / dict of key / value items to set as script environment variables. Note: Do not use PATH as it is reserved.
suite_runnerHash / dict of suite runner configuration values. Valid keys are described below.

script_run suite_runner parameter​

KeyValue
methodValid values are "start", "setup", and "teardown". Defaults to "start" if not provided. If script is provided, this value is ignored and start is always used.
suiteRequired; the name of the suite to run. Must be a valid suite within the given file.
groupThe name of the group to run. Must be a valid group within the given suite. If script is provided, this is required.
scriptThe name of the specific script to run. Must be a valid method name within the given group.
optionsArray of strings of suite runner options to enable. Valid options are: "manual", "pauseOnError", "continueAfterError", "abortAfterError", "loop", and "breakLoopOnError". Defaults to ["continueAfterError"] if not provided.
id = script_run("INST2/procedures/checks.py", environment={ 'USER': 'JASON'})
print(id)
id = script_run("INST2/procedures/my_script_suite.py", suite_runner={ 'suite': 'MySuite', 'group': 'ExampleGroup', 'script': 'script_2', 'method': 'start' })
print(id)

script_lock​

Since 5.0.0

Locks a script for editing. Subsequent users that open this script will get a warning that the script is currently locked.

script_lock("<Script Name>")
ParameterDescription
Script NameFull path name of the script starting with the target
script_lock("INST/procedures/checks.rb")

script_unlock​

Since 5.0.0

Unlocks a script for editing. If the script was not previously locked this does nothing.

script_unlock("<Script Name>")
ParameterDescription
Script NameFull path name of the script starting with the target
script_unlock("INST/procedures/checks.rb")

script_syntax_check​

Since 5.0.0

Performs a Ruby or Python syntax check on the given script.

script_syntax_check("<Script Name>")
ParameterDescription
Script NameFull path name of the script starting with the target
result = script_syntax_check("INST2/procedures/checks.py")
print(result) # => {'title': 'Syntax Check Successful', 'description': '["Syntax OK"]', 'success': True}

script_instrumented​

Since 5.0.0

Returns the instrumented script which allows COSMOS Script Runner to monitor the execution and provide line by line visualization. This is primarily a low level debugging method used by COSMOS developers.

script_instrumented("<Script Name>")
ParameterDescription
Script NameFull path name of the script starting with the target
script = script_instrumented("INST2/procedures/checks.py")
print(script) # => while True:\ntry:\nRunningScript.instance.pre_line_instrumentation ...

script_delete_all_breakpoints​

Since 5.0.0

Delete all breakpoints associated with all scripts.

script_delete_all_breakpoints()

step_mode​

Since 5.0.0

Places ScriptRunner into step mode where Go must be hit to proceed to the next line.

step_mode()

run_mode​

Since 5.0.0

Places ScriptRunner into run mode where the next line is run automatically.

run_mode()

disconnect_script​

Since 5.0.0

Puts scripting into disconnect mode. In disconnect mode, commands are not sent to targets, checks are all successful, and waits expire instantly. Requests for telemetry (tlm()) typically return 0. Disconnect mode is useful for dry-running scripts without having connected targets.

disconnect_script()

running_script_list​

Since 5.0.0

List the currently running scripts. Note, this will also include the script which is calling this method. Thus the list will never be empty but will always contain at least 1 item. Returns an array of hashes / list of dicts (see script_get for hash / dict contents).

running_script_list(limit = <limit>, offset = <offset>)
ParameterDescription
limitMax number to return (default 10)
offsetOffset into list to return (default 0)
running_script_list(limit = 20, offset = 10) # => [{'name': '372', 'state': 'spawning', 'shard': 0, 'filename': '__TEMP__/2025_05_02_22_38_53_386_temp.py', 'current_filename': '__TEMP__/2025_05_02_22_38_53_386_temp.py', 'line_no': 0, 'start_line_no': 1, 'end_line_no': None, 'username': 'anonymous', 'user_full_name': 'Anonymous', 'start_time': '2025-05-03T14:34:11Z', 'end_time': None, 'disconnect': False, 'environment': '{}', 'suite_runner': None, 'errors': None, 'pid': None, 'log': None, 'report': None, 'updated_at': 1746282851410918174, 'scope': 'DEFAULT'}]

script_get​

Since 5.0.0

Get information on the script with the specified ID. The information returned is the script name (id), state (spawning, init, running, paused, waiting, error, breakpoint, crashed, stopped, completed, completed_errors, killed), shard, filename, current_filename, line_no, start_line_no, end_line_no, username, user_full_name, start_time, end_time, disconnect state, environment variables, suite_runner configuration, errors, pid, log file, report file, update time, and scope.

Possible script states:

  • spawning - Not running yet
  • init - Initializing
  • running - Running
  • paused - Paused
  • waiting - Waiting (either for a wait or a prompt)
  • error - Paused with an error
  • breakpoint - Paused with a breakpoing
  • crashed - Complete after crashing with an exception
  • stopped - Prematurely stopped
  • completed - Completed successfully
  • completed_errors - Completed but errors occurred while running
  • killed - Forcefully killed
script_get("<Script Id>")
ParameterDescription
Script IdScript ID returned by script_run
script_get(15) # => {'name': '372', 'state': 'spawning', 'shard': 0, 'filename': '__TEMP__/2025_05_02_22_38_53_386_temp.py', 'current_filename': '__TEMP__/2025_05_02_22_38_53_386_temp.py', 'line_no': 0, 'start_line_no': 1, 'end_line_no': None, 'username': 'anonymous', 'user_full_name': 'Anonymous', 'start_time': '2025-05-03T14:34:11Z', 'end_time': None, 'disconnect': False, 'environment': '{}', 'suite_runner': None, 'errors': None, 'pid': None, 'log': None, 'report': None, 'updated_at': 1746282851410918174, 'scope': 'DEFAULT'}

running_script_stop​

Since 5.0.0

Stop the running script with the specified ID. This is equivalent to clicking the Stop button in the Script Runner GUI.

running_script_stop("<Script Id>")
ParameterDescription
Script IdScript ID returned by script_run
running_script_stop(15)

running_script_pause​

Since 5.0.0

Pause the running script with the specified ID. This is equivalent to clicking the Pause button in the Script Runner GUI.

running_script_pause("<Script Id>")
ParameterDescription
Script IdScript ID returned by script_run
running_script_pause(15)

running_script_retry​

Since 5.0.0

Retry the current line of the running script with the specified ID. This is equivalent to clicking the Retry button in the Script Runner GUI.

running_script_retry("<Script Id>")
ParameterDescription
Script IdScript ID returned by script_run
running_script_retry(15)

running_script_go​

Since 5.0.0

Unpause the running script with the specified ID. This is equivalent to clicking the Go button in the Script Runner GUI.

running_script_go("<Script Id>")
ParameterDescription
Script IdScript ID returned by script_run
running_script_go(15)

running_script_execute_while_paused​

Since 6.4.0

Perform a goto or execute selection on a running script.

running_script_execute_while_paused("<Script Id>", "<Script File Path>", <Start Line Number>, <End Line Number>)
ParameterDescription
Script IdScript ID returned by script_run
Script File PathPath to to the script to run ie. "INST/procedures/collect.rb"
Start Line NumberLine Number to Start At
End Line NumberOptional - Line Number to End At. If given then execute selection, else goto.
# Execute Selection
running_script_execute_while_paused("23", "INST/procedures/myprocedure.rb", 2, 4)
# Goto
running_script_execute_while_paused("33", "INST2/procedures/collect.py", 7)

running_script_step​

Since 5.0.0

Step the running script with the specified ID. This is equivalent to clicking the Step button in the Script Runner GUI's Debug window.

running_script_step("<Script Id>")
ParameterDescription
Script IdScript ID returned by script_run
running_script_step(15)

running_script_delete​

Since 5.0.0

Force quit the running script with the specified ID. This is equivalent to clicking the Delete button under the Running Scripts in the Script Runner GUI's Script -> Execution Status pane. Note, the 'stop' signal is first sent to the specified script and then the script is forcibly removed. Normally you should use the running_script_stop method.

running_script_delete("<Script Id>")
ParameterDescription
Script IdScript ID returned by script_run
running_script_delete(15)

completed_script_list​

Since 5.0.0

List the completed scripts. Returns an array of hashes / list of dicts containing the id, username, script name, script log, and start time.

completed_script_list(limit = <limit>, offset = <offset>)
ParameterDescription
limitMax number to return (default 10)
offsetOffset into list to return (default 0)
completed_script_list(limit = 10, offset = 0) # => [{'id': 16, 'user': 'Anonymous', 'name': 'INST2/procedures/new_script.py', 'log': 'DEFAULT/tool_logs/sr/20250116/2025_01_16_17_46_22_sr_new_script.txt', 'start': '2025-01-16 17:46:22 +0000'}, ...]

Script Runner Settings​

These methods allow the user to control various Script Runner settings.

set_line_delay​

Since 5.0.0

This method sets the line delay in script runner.

set_line_delay(<Delay>)
ParameterDescription
DelayThe amount of time script runner will wait between lines when executing a script, in seconds. Should be ≥ 0.0
set_line_delay(0.0)

get_line_delay​

Since 5.0.0

The method gets the line delay that script runner is currently using.

curr_line_delay = get_line_delay()

set_max_output​

Since 5.1.0

This method sets the maximum number of characters to display in Script Runner output before truncating. Default is 50,000 characters.

set_max_output(<Characters>)
ParameterDescription
CharactersNumber of characters to output before truncating
set_max_output(100)

get_max_output​

Since 5.1.0

The method gets the maximum number of characters to display in Script Runner output before truncating. Default is 50,000 characters.

print(get_max_output()) # => 50000

disable_instrumentation​

Since 5.0.0

Disables instrumentation for a block of code (line highlighting and exception catching). This is especially useful for speeding up loops that are very slow if lines are instrumented. Consider breaking code like this into a separate file and using either require/load to read the file for the same effect while still allowing errors to be caught by your script.

Use with Caution

Disabling instrumentation will cause any error that occurs while disabled to cause your script to completely stop.

with disable_instrumentation():
for x in range(1000):
# Don't want this to have to highlight 1000 times
pass

Script Runner Suites​

Creating Script Runner suites utilizes APIs to add groups to the defined suites. For more information please see running script suites.

add_group, add_group_setup, add_group_teardown, add_script​

Since 5.0.0

Adds a group's methods to the suite. The add_group method adds all the group methods including setup, teardown, and all the methods starting with 'script_' or 'test_'. The add_group_setup method adds just the setup method defined in the group class. The add_group_teardown method adds just the teardown method defined in the group class. The add_script method adds an individual method to the suite. NOTE: add_script can add any method including those not named with 'script_' or 'test_'.

add_group(<Group Class>)
add_group_setup(<Group Class>)
add_group_teardown(<Group Class>)
add_script(<Group Class>, <Method>)
ParameterDescription
Group ClassName of the previously defined class which inherits from the OpenC3 Group class. The Ruby API passes a String with the name of the group. The Python API passes the Group class directly.
MethodName of the method in the OpenC3 Group class. The Ruby API passes a String with the name of the method. The Python API passes the Group class directly.
from openc3.script import *
from openc3.script.suite import Group, Suite

class ExampleGroup(Group):
def script_1(self):
# Insert test code here ...
pass
class WrapperGroup(Group):
def setup(self):
# Insert test code here ...
pass
def my_method(self):
# Insert test code here ...
pass
def teardown(self):
# Insert test code here ...
pass
class MySuite(Suite):
def __init__(self):
super().__init__()
self.add_group(ExampleGroup)
self.add_group_setup(WrapperGroup)
self.add_script(WrapperGroup, 'my_method')
self.add_group_teardown(WrapperGroup)

Timelines​

The Timelines API allows you to manipulate Calendar timelines. Calendar is a COSMOS Enterprise tool.

list_timelines​

Since 5.12.0

Returns all the timelines in an array of hashes / list of dicts.

timelihes = list_timelines() # =>
# [{'name': 'Mine', 'color': '#e67643', 'execute': True, 'shard': 0, 'scope': 'DEFAULT', 'updated_at': 1737124024123643504}]

create_timeline​

Since 5.12.0

Create a new timeline in Calendar which can hold activities.

create_timeline(name, color=None)
ParameterDescription
nameName of the timeline
colorColor of the timeline. Must be given as a hex value, e.g. #FF0000. Default is a random color.
tl = create_timeline("Other", color="#FF0000") # =>
# {'name': 'Other', 'color': '#FF0000', 'execute': True, 'shard': 0, 'scope': 'DEFAULT', 'updated_at': 1737126348971941923}

get_timeline​

Since 5.12.0

Get information about an existing timeline.

get_timeline(name)
ParameterDescription
nameName of the timeline
tl = get_timeline("Other") # =>
# {'name': 'Other', 'color': '#FF0000', 'execute': True, 'shard': 0, 'scope': 'DEFAULT', 'updated_at': 1737126348971941923}

set_timeline_color​

Since 5.12.0

Set the displayed color for an existing timeline.

set_timeline_color(name, color)
ParameterDescription
nameName of the timeline
colorColor of the timeline. Must be given as a hex value, e.g. #FF0000.
set_timeline_color("Mine", "#4287f5")

delete_timeline​

Since 5.12.0

Delete an existing timeline. Timelines with activities can only be deleted by passing force = true.

delete_timeline(name, force=False)
ParameterDescription
nameName of the timeline
forceWhether to delete the timeline if it has activities. Default is false.
delete_timeline("Other", force=True)

create_timeline_activity​

Since 5.12.0

Create an activity on an existing timeline. Activities can be one of COMMAND, SCRIPT, or RESERVE. Activities have a start and stop time and commands and scripts take data on the command or script to execute.

create_timeline_activity(name, kind, start, stop, data={})
ParameterDescription
nameName of the timeline
kindType of the activity. One of COMMAND, SCRIPT, or RESERVE.
startStart time of the activity. Time / datetime instance.
stopStop time of the activity. Time / datetime instance.
dataHash / dict of data for COMMAND or SCRIPT type. Default is empty hash / dict. Valid keys are described below.
scopeScope of the activity. Default is the OPENC3_SCOPE, usually "DEFAULT". Must be the containing scope of the timeline specified by the name parameter.

create_timeline_activity data parameter​

KeyValue
usernameUsername to display as the creator of the activity. Default is "operator".
customTitleCustom title to display for the activity. Default is empty string which results in no custom title being shown.
notesNotes to display for the activity. Default is empty string, which results in no notes being shown.
commandCommand to execute for COMMAND type activities.
scriptScript to execute for SCRIPT type activities. Should be given as the path to the script file to run, starting with the target name, e.g. "INST/procedures/collect.rb".
environmentArray of environment variable key/value pairs to set for SCRIPT type activities, e.g. [{key: "USER", value: "JASON"}]
now = datetime.now(timezone.utc)
start = now + timedelta(hours=1)
stop = start + timedelta(hours=1)
act = create_timeline_activity("PythonTL", kind="RESERVE", start=start, stop=stop) # =>
# {'name': 'PythonTL', 'updated_at': 1737129305507111708, 'start': 1737132902, 'stop': 1737136502,
# 'kind': 'reserve', 'data': {'username': 'operator'},
# 'scope': 'DEFAULT', 'fulfillment': False, 'uuid': '46328378-ed78-4719-ad70-e84951a196fd',
# 'events': [{'time': 1737129305, 'event': 'created'}], 'recurring': {}}
act = create_timeline_activity("PythonTL", kind="COMMAND", start=start, stop=stop,
data={'command': "INST COLLECT with TYPE NORMAL, DURATION 5, TEMP 10"}) # =>
# {'name': 'PythonTL', 'updated_at': 1737129508886643928, 'start': 1737133108, 'stop': 1737136708,
# 'kind': 'command', 'data': {'command': 'INST COLLECT with TYPE NORMAL, DURATION 5, TEMP 10', 'username': 'operator'},
# 'scope': 'DEFAULT', 'fulfillment': False, 'uuid': 'cddbf034-ccdd-4c36-91c2-2653a39b06a5',
# 'events': [{'time': 1737129508, 'event': 'created'}], 'recurring': {}}
start = now + timedelta(hours=2)
stop = start + timedelta(hours=1)
act = create_timeline_activity("PythonTL", kind="SCRIPT", start=start, stop=stop,
data={'environment': [{'key': "USER", 'value': "JASON"}], 'script': "INST2/procedures/checks.py"}) # =>
# {'name': 'PythonTL', 'updated_at': 1737129509288571345, 'start': 1737136708, 'stop': 1737140308,
# 'kind': 'script', 'data': {'environment': [{'key': 'USER', 'value': 'JASON'}], 'script': 'INST2/procedures/checks.py', 'username': 'operator'},
# 'scope': 'DEFAULT', 'fulfillment': False, 'uuid': '4f8d791b-b138-4383-b5ec-85c28b2bea20',
# 'events': [{'time': 1737129509, 'event': 'created'}], 'recurring': {}}

update_timeline_activity​

Update an existing timeline activity. The activity is identified by its timeline name and current start time (id). All fields (kind, start, stop, data) are replaced with the provided values.

update_timeline_activity(name, id, kind, start, stop, uuid, data={})
ParameterDescription
nameName of the timeline
idThe current start time / score of the activity to update (integer, Unix seconds).
kindType of the activity. One of COMMAND, SCRIPT, or RESERVE.
startNew start time of the activity. Time / datetime instance.
stopNew stop time of the activity. Time / datetime instance.
uuidUUID of the activity.
dataHash / dict of data for the activity. Default is empty hash / dict. Valid keys are described above.
scopeScope of the activity. Must be the containing scope of the timeline specified by the name parameter.
# First create an activity
now = datetime.now(timezone.utc)
start = now + timedelta(hours=1)
stop = start + timedelta(hours=1)
act = create_timeline_activity("PythonTL", kind="COMMAND", start=start, stop=stop,
data={'command': "INST COLLECT with TYPE NORMAL, DURATION 5"})

# Update the activity with a new custom title, notes, and extended stop time
new_stop = stop + timedelta(minutes=30)
updated = update_timeline_activity("PythonTL",
id=act['start'],
kind="COMMAND", start=start, stop=new_stop,
uuid=act['uuid'],
data={'command': "INST COLLECT with TYPE NORMAL, DURATION 10",
'customTitle': "Extended Collection",
'notes': "Duration extended per ops request"})

get_timeline_activity​

Since 5.12.0

Get an existing timeline activity.

get_timeline_activity(name, start, uuid)
ParameterDescription
nameName of the timeline
startStart time of the activity. Time / datetime instance.
uuidUUID of the activity
act = get_timeline_activity("PythonTL", 1737133108, "cddbf034-ccdd-4c36-91c2-2653a39b06a5") # =>
# {'name': 'PythonTL', 'updated_at': 1737129508886643928, 'start': 1737133108, 'stop': 1737136708,
# 'kind': 'command', 'data': {'command': 'INST COLLECT with TYPE NORMAL, DURATION 5, TEMP 10', 'username': 'operator'},
# 'scope': 'DEFAULT', 'fulfillment': False, 'uuid': 'cddbf034-ccdd-4c36-91c2-2653a39b06a5',
# 'events': [{'time': 1737129508, 'event': 'created'}], 'recurring': {}}

get_timeline_activities​

Since 5.12.0

Get a range of timeline activities between start and stop time. If called without a start / stop time it defaults to 1 week before "now" up to 1 week from "now" (2 weeks total).

get_timeline_activities(name, start=None, stop=None, limit=None)
ParameterDescription
nameName of the timeline
startStart time of the activities. Time / datetime instance. Defaults to 7 days ago.
stopStop time of the activities. Time / datetime instance. Defaults to 7 days from now.
limitMaximum number of activities to return. Default is 1 per minute of the time range.
now = datetime.now(timezone.utc)
acts = get_timeline_activities("PythonTL", start=now - timedelta(hours=2), stop=now, limit=1000) # =>
# [{ "name"=>"PythonTL", ... }, { "name"=>"PythonTL", ... }]

delete_timeline_activity​

Since 5.12.0

Delete an existing timeline activity.

delete_timeline_activity(name, start, uuid)
ParameterDescription
nameName of the timeline
startStart time of the activity. Time / datetime instance.
uuidUUID of the activity
delete_timeline_activity("PythonTL", 1737133108, "cddbf034-ccdd-4c36-91c2-2653a39b06a5")

Metadata​

Metadata allows you to mark the regular target / packet data logged in COSMOS with your own fields. This metadata can then be searched and used to filter data when using other COSMOS tools.

metadata_all​

Since 5.2.0

Returns all the metadata that was previously set

metadata_all()
ParameterDescription
limitAmount of metadata items to return. Default is 100.
metadata_all(limit='500')

metadata_get​

Since 5.2.0

Returns metadata that was previously set

metadata_get(start)
ParameterDescription
startNamed parameter, time at which to retrieve metadata as integer seconds from epoch
metadata_get(start='500')

metadata_set​

Since 5.2.0

Sets metadata which appears on the Calendar and can be retried by metadata_get.

metadata_set(<Metadata>, start, color)
ParameterDescription
MetadataHash or dict of key value pairs to store as metadata.
startNamed parameter, time at which to store metadata. Default is now.
colorNamed parameter, color to display metadata in the calendar. Default is #003784.
metadata_set({ 'key': 'value' })
metadata_set({ 'key': 'value' }, color='ff5252')

metadata_update​

Since 5.2.0

Updates metadata that was previously set

metadata_update(<Metadata>, start, color)
ParameterDescription
MetadataHash or dict of key value pairs to update as metadata.
startNamed parameter, time at which to update metadata. Default is latest metadata.
colorNamed parameter, color to display metadata in the calendar. Default is #003784.
metadata_update({ 'key': 'value' })

metadata_input​

Since 5.2.0

Prompts the user to set existing metadata values or create new a new one.

metadata_input()

Settings​

COSMOS has several settings typically accessed through the Admin Settings tab. These APIs allow programmatic access to those same settings.

list_settings​

Since 5.0.0

Return all the current COSMOS setting name. These are the names that should be used in the other APIs.

print(list_settings()) # => ['pypi_url', 'rubygems_url', 'source_url', 'version']

get_all_settings​

Since 5.0.0

Return all the current COSMOS settings along with their values.

settings = get_all_settings() # =>
# { 'version': {'name': 'version', 'data': '5.11.4-beta0', 'updated_at': 1698074299509456507},
# 'pypi_url': {'name': 'pypi_url', 'data': 'https://pypi.org/simple', 'updated_at': 1698026776574347007},
# 'rubygems_url': {'name': 'rubygems_url', 'data': 'https://rubygems.org', 'updated_at': 1698026776574105465},
# 'source_url': {'name': 'source_url', 'data': 'https://github.com/OpenC3/cosmos', 'updated_at': 1698026776573904132} }

get_setting, get_settings​

Since 5.0.0

Return the data from the given COSMOS setting. Returns nil (Ruby) or None (Python) if the setting does not exist.

get_setting(<Setting Name>)
get_settings(<Setting Name1>, <Setting Name2>, ...)
ParameterDescription
Setting NameName of the setting to return
setting = get_setting('version') # => '5.11.4-beta0'
setting = get_setting('version', 'rubygems_url') # => ['5.11.4-beta0', 'https://rubygems.org']

set_setting​

Since 5.0.0

Sets the given setting value.

Admin Password Required

This API is only accessible externally (not within Script Runner) and requires the admin password.

set_setting(<Setting Name>, <Setting Value>)
ParameterDescription
Setting NameName of the setting to change
Setting ValueSetting value to set
set_setting('pypi_url', 'https://mypypiserver')

Configuration​

Many COSMOS tools have the ability to load and save a configuration. These APIs allow you to programmatically load and save the configuration.

config_tool_names​

Since 5.12.0

List all the configuration tool names which are used as the first parameter in the other APIs.

names = config_tool_names() # => ['telemetry_grapher', 'data_viewer']

list_configs​

Since 5.0.0

List all the saved configuration names under the given tool name.

list_configs(<Tool Name>)
ParameterDescription
Tool NameName of the tool to retrieve configuration names from
configs = list_configs('telemetry_grapher') # => ['adcs', 'temps']

load_config​

Since 5.0.0

Load a particular tool configuration.

Tool Configuration

Tool configurations are not fully documented and subject to change between releases. Only modify values returned by load_config and do not change any keys.

load_config(<Tool Name>, <Configuration Name>)
ParameterDescription
Tool NameName of the tool
Configuration NameName of the configuration
config = load_config('telemetry_grapher', 'adcs') # =>
# [ {
# "items": [
# {
# "targetName": "INST",
# "packetName": "ADCS",
# "itemName": "CCSDSVER",
# ...

save_config​

Since 5.0.0

Save a particular tool configuration.

save_config(<Tool Name>, <Configuration Name>, local_mode)
ParameterDescription
Tool NameName of the tool
Configuration NameName of the configuration
local_modeWhether to save the configuration in local mode
save_config('telemetry_grapher', 'adcs', config)

delete_config​

Since 5.0.0

Delete a particular tool configuration.

delete_config(<Tool Name>, <Configuration Name>, local_mode)
ParameterDescription
Tool NameName of the tool
Configuration NameName of the configuration
local_modeWhether to delete the configuration in local mode
delete_config('telemetry_grapher', 'adcs')

Offline Access​

An offline access token is required to execute scripts in COSMOS Enterprise. These methods support client side creation, testing, and setting of the offline_access_token.

initialize_offline_access​

Since 6.3.0

Creates and sets the offline access token for the user. Note: calling this method is required before executing any api methods that require an offline access token like script_run (Enterprise Only). This method must be called OUTSIDE of ScriptRunner as it is needed in order to start a script in the first place.

In Enterprise, the OPENC3_API_USER and OPENC3_API_PASSWORD environment variables must be set for the initialize_offline_access to generate a token. This API user must also be a valid user with the respective permissions setup in Keycloak. These two variables are not in the .env by default, and should not be as they contain sensitive information.

Depending on your deployment environment, there are several ways Secrets can be managed. In a Kubernetes deployment, Secretes can be configured, or dynamically managed by a separate secrets management tool like HashiCorp Vault or AWS Secrets Manager.

# First setup environment variables. See examples/external_script.py
initialize_offline_access()
script_run("INST2/procedures/collect.py")

offline_access_needed​

Since 5.2.0

Returns true if the user needs to generate an offline access token. Note this will only be true if the user is at least authorized to view scripts, otherwise it will always be false if script_view permission is not available for the user.

result = offline_access_needed() # => False

set_offline_access​

Since 5.2.0

Sets the offline access token in the backend. Note: You probably don't need to call this method directly, as it will be called by initialize_offline_access().

set_offline_access(offline_access_token)
ParameterDescription
offline_access_tokenKeycloak generated refresh token that contains the offline_access openid scope
set_offline_access(offline_access_token)

Autonomic​

Autonomic provides automated reactions to telemetry conditions through triggers and reactions. These methods allow you to manage autonomic groups, triggers, and reactions programmatically.

autonomic_group_list​

Since 6.4.2

Returns a list of all autonomic groups.

autonomic_group_list(scope=None)
ParameterDescription
scopeScope to operate in (optional)
groups = autonomic_group_list()

autonomic_group_create​

Since 6.4.2

Creates a new autonomic group.

autonomic_group_create(name, scope=None)
ParameterDescription
nameName of the group to create
scopeScope to operate in (optional)
group = autonomic_group_create("SAFETY")

autonomic_group_show​

Since 6.4.2

Shows details about a specific autonomic group.

autonomic_group_show(name, scope=None)
ParameterDescription
nameName of the group
scopeScope to operate in (optional)
group_info = autonomic_group_show("SAFETY")

autonomic_group_destroy​

Since 6.4.2

Destroys an autonomic group.

autonomic_group_destroy(name, scope=None)
ParameterDescription
nameName of the group to destroy
scopeScope to operate in (optional)
autonomic_group_destroy("SAFETY")

autonomic_trigger_list​

Since 6.4.2

Returns a list of triggers in a specific group.

autonomic_trigger_list(group='DEFAULT', scope=None)
ParameterDescription
groupGroup to list triggers from (default: 'DEFAULT')
scopeScope to operate in (optional)
triggers = autonomic_trigger_list(group="SAFETY")

autonomic_trigger_create​

Since 6.4.2

Creates a new trigger with the specified condition.

autonomic_trigger_create(left, operator, right, group='DEFAULT', scope=None)
ParameterDescription
leftLeft side of the trigger condition
operatorComparison operator (e.g. >, <, ==, !=, >=, <=)
rightRight side of the trigger condition
groupGroup to create the trigger in (default: 'DEFAULT')
scopeScope to operate in (optional)
left = {
"type": "item",
"target": "INST",
"packet": "HEALTH_STATUS",
"item": "TEMP1",
"valueType": "CONVERTED",
}
operator = ">"
right = {
"type": "float",
"float": 0,
}
trigger = autonomic_trigger_create(left=left, operator=operator, right=right, group="SAFETY")

autonomic_trigger_show​

Since 6.4.2

Shows details about a specific trigger.

autonomic_trigger_show(name, group='DEFAULT', scope=None)
ParameterDescription
nameName of the trigger
groupGroup the trigger belongs to (default: 'DEFAULT')
scopeScope to operate in (optional)
trigger_info = autonomic_trigger_show("TRIG1", group="SAFETY")

autonomic_trigger_enable​

Since 6.4.2

Enables a trigger.

autonomic_trigger_enable(name, group='DEFAULT', scope=None)
ParameterDescription
nameName of the trigger to enable
groupGroup the trigger belongs to (default: 'DEFAULT')
scopeScope to operate in (optional)
autonomic_trigger_enable("TRIG1", group="SAFETY")

autonomic_trigger_disable​

Since 6.4.2

Disables a trigger.

autonomic_trigger_disable(name, group='DEFAULT', scope=None)
ParameterDescription
nameName of the trigger to disable
groupGroup the trigger belongs to (default: 'DEFAULT')
scopeScope to operate in (optional)
autonomic_trigger_disable("TRIG1", group="SAFETY")

autonomic_trigger_update​

Since 6.4.2

Updates an existing trigger with new parameters.

autonomic_trigger_update(name, group='DEFAULT', left=left, operator=operator, right=right, scope=None)
ParameterDescription
nameName of the trigger to update
groupGroup the trigger belongs to (default: 'DEFAULT')
leftNew left side of the trigger condition (optional)
operatorNew comparison operator (optional)
rightNew right side of the trigger condition (optional)
scopeScope to operate in (optional)
right = {
"type": "float",
"float": 100,
}
updated_trigger = autonomic_trigger_update(
"TRIG1",
group="SAFETY",
right=right
)

autonomic_trigger_destroy​

Since 6.4.2

Destroys a trigger.

autonomic_trigger_destroy(name, group='DEFAULT', scope=None)
ParameterDescription
nameName of the trigger to destroy
groupGroup the trigger belongs to (default: 'DEFAULT')
scopeScope to operate in (optional)
autonomic_trigger_destroy("TRIG1", group="SAFETY")

autonomic_reaction_list​

Since 6.4.2

Returns a list of all reactions.

autonomic_reaction_list(scope=None)
ParameterDescription
scopeScope to operate in (optional)
reactions = autonomic_reaction_list()

autonomic_reaction_create​

Since 6.4.2

Creates a new reaction that executes actions when triggers are activated.

autonomic_reaction_create(triggers, actions, trigger_level='EDGE', snooze=0, scope=None)
ParameterDescription
triggersList of trigger names that activate this reaction
actionsList of actions to perform when triggered
trigger_levelTrigger level: 'EDGE' or 'LEVEL' (default: 'EDGE')
snoozeSnooze time in seconds between activations (default: 0)
scopeScope to operate in (optional)
triggers = [{
'name': 'TRIG1',
'group': 'SAFETY',
}]
actions = [{
'type': 'command',
'value': 'INST ABORT'
}]
reaction = autonomic_reaction_create(
triggers=triggers,
actions=actions,
trigger_level='EDGE',
snooze=300
)

autonomic_reaction_show​

Since 6.4.2

Shows details about a specific reaction.

autonomic_reaction_show(name, scope=None)
ParameterDescription
nameName of the reaction
scopeScope to operate in (optional)
reaction_info = autonomic_reaction_show("REACT1")

autonomic_reaction_enable​

Since 6.4.2

Enables a reaction.

autonomic_reaction_enable(name, scope=None)
ParameterDescription
nameName of the reaction to enable
scopeScope to operate in (optional)
autonomic_reaction_enable("REACT1")

autonomic_reaction_disable​

Since 6.4.2

Disables a reaction.

autonomic_reaction_disable(name, scope=None)
ParameterDescription
nameName of the reaction to disable
scopeScope to operate in (optional)
autonomic_reaction_disable("REACT1")

autonomic_reaction_execute​

Since 6.4.2

Manually executes a reaction's actions.

autonomic_reaction_execute(name, scope=None)
ParameterDescription
nameName of the reaction to execute
scopeScope to operate in (optional)
autonomic_reaction_execute("REACT1")

autonomic_reaction_update​

Since 6.4.2

Updates an existing reaction with new parameters.

autonomic_reaction_update(name, triggers=None, actions=None, trigger_level=None, snooze=None, scope=None)
ParameterDescription
nameName of the reaction to update
triggersNew list of trigger names (optional)
actionsNew list of actions (optional)
trigger_levelNew trigger level (optional)
snoozeNew snooze time in seconds (optional)
scopeScope to operate in (optional)
updated_reaction = autonomic_reaction_update(
"REACT1",
snooze=600
)

autonomic_reaction_destroy​

Since 6.4.2

Destroys a reaction.

autonomic_reaction_destroy(name, scope=None)
ParameterDescription
nameName of the reaction to destroy
scopeScope to operate in (optional)
autonomic_reaction_destroy("REACT1")

Utilities​

Various utilities that can be used in scripts.

open_tab​

Since 7.0.0

Opens a new tab in the browser.

open_tab(url)
ParameterDescription
urlURL to open in a new tab
open_tab("https://github.com/OpenC3/cosmos")