The Run class¶
Properties¶
duration¶
Return current run duration
executor¶
Return the executor for this run
id¶
Return the unique id of the run
name¶
Return the name of the run
processes¶
Create an array containing a list of processes
time_stamp¶
Return current timestamp
uid¶
Return the local unique identifier of the run
Methods¶
__init__(...)¶
Initialise a new Simvue run
If abort_callback is provided the first argument must be this Run instance
Parameters
| mode | Literal['online', 'offline', 'disabled'] |
mode of runningonline - objects sent directly to Simvue serveroffline - everything is written to disk for later dispatchdisabled - disable monitoring completely |
| abort_callback | Optional[Callable[[Self], None]] |
callback executed when the run is aborted |
| server_token | pydantic.types.SecretStr | None |
overwrite value for server token, default is None |
| server_url | str | None |
overwrite value for server URL, default is None |
| debug | bool |
run in debug mode, default is False |
add_alerts(...)¶
Add a set of existing alerts to this run by name or id
Parameters
| ids | list[str] | None |
unique identifiers of the alerts to attach, by default None |
| names | list[str] | None |
names of alerts to attach, by default None |
Returns
bool
returns True if successful
add_process(...)¶
Add a process to be executed to the executor. This process can take many forms, for example a be a set of positional arguments:
Provide explicitly the components of the command:executor.add_process("my_process", executable="bash", debug=True, c="return 1")
executor.add_process("my_process", executable="bash", script="my_script.sh", input="parameters.dat")
flag=True can be used to set the option and
for options taking values option=value.
When the process has completed if a function has been provided for the completion_callback argument
this will be called, this callback is expected to take the following form:
Note completion_callback is not supported on Windows operating systems.
Alternatively you can use completion_trigger to create a multiprocessing event which will be set
when the process has completed.
Parameters
| identifier | str |
A unique identifier for this process |
| executable | str | pathlib._local.Path | None |
the main executable for the command, if not specified this is taken to be the firstpositional argument, by default None |
| positional_arguments | Any, ..., optional |
all other positional arguments are taken to be part of the command to execute |
| script | Optional[Annotated[pathlib._local.Path, Path(path_type='file')]] |
the script to run, note this only work if the script is not an option, if this is the caseyou should provide it as such and perform the upload manually, by default None |
| input_file | Optional[Annotated[pathlib._local.Path, Path(path_type='file')]] |
the input file to run, note this only work if the input file is not an option, if this is the caseyou should provide it as such and perform the upload manually, by default None |
| completion_callback | Optional[Callable[[int, str, str], None]] |
callback to run when process terminates (not supported on Windows) |
| completion_trigger | multiprocessing.synchronize.Event | None |
this trigger event is set when the processes completes |
| env | dict[str, str] | None |
environment variables for process |
| cwd | pathlib._local.Path | None |
working directory to execute the process within. Note that executable, input and script file paths shouldbe absolute or relative to the directory where this method is called, not relative to the new working directory. |
| kwargs | Any, ..., optional |
all other keyword arguments are interpreted as options to the command |
close()¶
Close the run
Returns
bool
whether close was successful
config(...)¶
Optional configuration
Parameters
| suppress_errors | bool | None |
disable exception throwing instead putting Simvue into adormant state if an error occurs |
| queue_blocking | bool | None |
block thread queues during metric/event recording |
| resources_metrics_interval | Optional[Annotated[int, Gt(gt=0)]] |
frequency at which to collect resource metrics |
| enable_emission_metrics | bool | None |
enable monitoring of emission metrics |
| disable_resources_metrics | bool | None |
disable monitoring of resource metrics |
| storage_id | str | None |
identifier of storage to use, by default None |
| abort_on_alert | Union[Literal['run', 'all', 'ignore'], bool, None] |
whether to abort when an alert is triggered.If 'run' then the current run is aborted.If 'terminate' then the script itself is terminated.If 'ignore' then alerts will not affect this run |
Returns
bool
if configuration was successful
create_event_alert(...)¶
Creates an events alert with the specified name (if it doesn't exist) and applies it to the current run. If alert already exists it will not be duplicated.
Parameters
| name | str |
name of alert |
| pattern | str |
for event based alerts pattern to look for, by default None |
| frequency | Annotated[int, Gt(gt=0)] |
frequency at which to check alert condition in seconds, by default None |
| notification | Literal['email', 'none'] |
whether to notify on trigger, by default "none" |
| trigger_abort | bool |
whether this alert can trigger a run abort |
| attach_to_run | bool |
whether to attach this alert to the current run, default True |
Returns
str | None
returns the created alert ID if successful
create_metric_range_alert(...)¶
Creates a metric range alert with the specified name (if it doesn't exist) and applies it to the current run. If alert already exists it will not be duplicated.
Parameters
| name | Annotated[str, FieldInfo(annotation=None, required=True, metadata=[_PydanticGeneralMetadata(pattern='^[a-zA-Z0-9\\-\\_\\s\\/\\.:]+$')])] |
name of alert |
| metric | str |
metric to monitor |
| range_low | float |
the lower bound value |
| range_high | float |
the upper bound value |
| rule | Literal['is inside range', 'is outside range'] |
rule defining range alert conditions |
| description | str | None |
description for this alert, default None |
| window | Annotated[int, Gt(gt=0)] |
time period in seconds over which metrics are averaged, by default 5 |
| frequency | Annotated[int, Gt(gt=0)] |
frequency at which to check alert condition in seconds, by default 1 |
| aggregation | Literal['average', 'sum', 'at least one', 'all'] |
method to use when aggregating metrics within time window, default 'average'. |
| notification | Literal['email', 'none'] |
whether to notify on trigger, by default "none" |
| trigger_abort | bool |
whether this alert can trigger a run abort, default False |
| attach_to_run | bool |
whether to attach this alert to the current run, default True |
Returns
str | None
returns the created alert ID if successful
create_metric_threshold_alert(...)¶
Creates a metric threshold alert with the specified name (if it doesn't exist) and applies it to the current run. If alert already exists it will not be duplicated.
Parameters
| name | Annotated[str, FieldInfo(annotation=None, required=True, metadata=[_PydanticGeneralMetadata(pattern='^[a-zA-Z0-9\\-\\_\\s\\/\\.:]+$')])] |
name of alert |
| metric | str |
metric to monitor |
| threshold | float |
the threshold value |
| rule | Literal['is above', 'is below'] |
rule defining range alert conditions |
| description | str | None |
description for this alert, default None |
| window | Annotated[int, Gt(gt=0)] |
time period in seconds over which metrics are averaged, by default 5 |
| frequency | Annotated[int, Gt(gt=0)] |
frequency at which to check alert condition in seconds, by default 1 |
| aggregation | Literal['average', 'sum', 'at least one', 'all'] |
method to use when aggregating metrics within time window, default 'average'. |
| notification | Literal['email', 'none'] |
whether to notify on trigger, by default "none" |
| trigger_abort | bool |
whether this alert can trigger a run abort, default False |
| attach_to_run | bool |
whether to attach this alert to the current run, default True |
Returns
str | None
returns the created alert ID if successful
create_user_alert(...)¶
Creates a user alert with the specified name (if it doesn't exist) and applies it to the current run. If alert already exists it will not be duplicated.
Parameters
| name | Annotated[str, FieldInfo(annotation=None, required=True, metadata=[_PydanticGeneralMetadata(pattern='^[a-zA-Z0-9\\-\\_\\s\\/\\.:]+$')])] |
name of alert |
| description | str | None |
description for this alert, default None |
| notification | Literal['email', 'none'] |
whether to notify on trigger, by default "none" |
| trigger_abort | bool |
whether this alert can trigger a run abort, default False |
| attach_to_run | bool |
whether to attach this alert to the current run, default True |
Returns
str | None
returns the created alert ID if successful
init(...)¶
Initialise a Simvue run
Parameters
| name | Annotated[str | None, FieldInfo(annotation=None, required=True, metadata=[_PydanticGeneralMetadata(pattern='^[a-zA-Z0-9\\-\\_\\s\\/\\.:]+$')])] |
the name to allocate this run, if not specified a name will beselected at random, by default None |
| metadata | dict[str, Any] |
any metadata relating to the run as key-value pairs, by default None |
| tags | list[str] | None |
a list of tags for this run, by default None |
| description | str | None |
description of the run, by default None |
| folder | Annotated[str, FieldInfo(annotation=None, required=False, default=None, metadata=[_PydanticGeneralMetadata(pattern='^/.*')])] |
folder within which to store the run, by default "/" |
| notification | Literal['none', 'all', 'error', 'lost'] |
whether to notify the user by email upon completion of the run ifthe run is in the specified state, by default "none" |
| running | bool |
whether to set the status as running or created, the latter implyingthe run will be commenced at a later time. Default is True. |
| retention_period | str | None |
describer for time period to retain run, the default of Noneremoves this constraint. |
| timeout | int | None |
specify the timeout of the run, if None there is no timeout |
| visibility | Union[Literal['public', 'tenant'], list[str], None] |
set visibility options for this run, either: public - run viewable to all. tenant - run viewable to all within the current tenant.* A list of usernames with which to share this run |
| no_color | bool |
disable terminal colors. Default False. |
Returns
bool
whether the initialisation was successful
kill_all_processes()¶
Kill all currently running processes.
kill_process(...)¶
Kill a running process by ID
Parameters
| process_id | str |
the unique identifier for the added process |
log_alert(...)¶
Set the state of an alert - either specify the alert by ID or name.
Parameters
| identifier | str | None |
ID of alert to update, by default None |
| name | str | None |
Name of the alert to update, by default None |
| state | Literal['ok', 'critical'] |
state to set alert to, by default 'critical' |
Returns
bool
whether alert state update was successful
log_event(...)¶
Log event to the server
Parameters
| message | str |
event message to log |
| timestamp | str | None |
manually specify the time stamp for this log, by default None |
Returns
bool
whether event log was successful
log_metrics(...)¶
Log metrics to Simvue server
Parameters
| metrics | dict[Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern='^[a-zA-Z0-9\\-\\_\\s\\/\\.:=><]+$')], int | float] |
set of metrics to upload to server for this run |
| step | int | None |
manually specify the step index for this log, by default None |
| time | float | None |
manually specify the time for this log, by default None |
| timestamp | str | None |
manually specify the timestamp for this log, by default None |
Returns
bool
if the metric log was succcessful
reconnect(...)¶
Reconnect to a run in the created state
Parameters
| run_id | str |
identifier of run to connect to |
Returns
bool
whether reconnection succeeded
save_all(...)¶
Save a set of files and directories
Parameters
| items | list[Union[Annotated[pathlib._local.Path, Path(path_type='file')], Annotated[pathlib._local.Path, Path(path_type='dir')]]] |
list of file paths and directories to save |
| category | Literal['input', 'output', 'code'] |
the category to assign to the saved objects |
| file_type | str | None |
manually specify the MIME type for all items, by default None |
| preserve_path | bool |
_preserve the full path, by default False |
Returns
bool
whether the save was successful
save_directory(...)¶
Upload files from a whole directory
Parameters
| directory | Annotated[pathlib._local.Path, Path(path_type='dir')] |
the directory to save to the run |
| category | Literal['output', 'input', 'code'] |
the category to assign to the saved objects within this directory |
| file_type | str | None |
manually specify the MIME type for items in the directory, by default None |
| preserve_path | bool |
preserve the full path, by default False |
Returns
bool
if the directory save was successful
save_file(...)¶
Upload file to the server
Parameters
| file_path | Annotated[pathlib._local.Path, Path(path_type='file')] |
path to the file to upload |
| category | Literal['input', 'output', 'code'] |
category of file with respect to this run |
| file_type | str | None |
the MIME file type else this is deduced, by default None |
| preserve_path | bool |
whether to preserve the path during storage, by default False |
| name | Optional[Annotated[str, FieldInfo(annotation=None, required=True, metadata=[_PydanticGeneralMetadata(pattern='^[a-zA-Z0-9\\-\\_\\s\\/\\.:]+$')])]] |
name to associate with this file, by default None |
| metadata | dict[str, Any] |
any metadata to attach to the artifact |
Returns
bool
whether the upload was successful
save_object(...)¶
Save an object to the Simvue server
Parameters
| obj | Any |
object to serialize and send to the server |
| category | Literal['input', 'output', 'code'] |
category of file with respect to this run |
| name | Optional[Annotated[str, FieldInfo(annotation=None, required=True, metadata=[_PydanticGeneralMetadata(pattern='^[a-zA-Z0-9\\-\\_\\s\\/\\.:]+$')])]] |
name to associate with this object, by default None |
| allow_pickle | bool |
whether to allow pickling if all other serialization types fail, by default False |
| metadata | dict[str, Any] |
any metadata to attach to the artifact |
Returns
bool
whether object upload was successful
set_folder_details(...)¶
Add metadata to the specified folder
Parameters
| metadata | dict[str, int | str | float] | None |
additional metadata to attach to this folder, by default None |
| tags | list[str] | None |
list of tags to assign to the folder, by default None |
| description | str | None |
description to assign to this folder, by default None |
Returns
bool
returns True if update was successful
set_pid(...)¶
Set pid of process to be monitored
Parameters
| pid | int |
PID of the process to be monitored |
set_status(...)¶
Set run status status to assign to this run once finished
Parameters
| status | Literal['completed', 'failed', 'terminated'] |
status to set the run to |
Returns
bool
if status update was successful
set_tags(...)¶
Set tags for this run
Parameters
| tags | list[str] |
new set of tags to assign |
Returns
bool
whether the update was successful
update_metadata(...)¶
Update metadata for this run
Parameters
| metadata | dict[str, Any] |
set run metadata |
Returns
bool
if the update was successful
update_tags(...)¶
Add additional tags to this run without duplication
Parameters
| tags | list[str] |
new set of tags to attach |
Returns
bool
whether the update was successful