The Run class¶
Track simulation details based on token and URL.
The Run class provides a way of monitoring simulation runs by logging metrics and creating alerts based on such metrics. The recommended usage is as a context manager to ensure the run is closed upon completion.
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'] |
= online |
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]] |
= None |
callback executed when the run is aborted |
| server_token | pydantic.types.SecretStr | None |
= None |
overwrite value for server token, default is None |
| server_url | str | None |
= None |
overwrite value for server URL, default is None |
| debug | bool |
= False |
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 |
= None |
unique identifiers of the alerts to attach, by default None |
| names | list[str] | None |
= 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 |
= None |
the main executable for the command, if not specified this is taken to be the first positional 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')]] |
= None |
the script to run, note this only work if the script is not an option, if this is the case you should provide it as such and perform the upload manually, by default None |
| input_file | Optional[Annotated[pathlib._local.Path, Path(path_type='file')]] |
= None |
the input file to run, note this only work if the input file is not an option, if this is the case you should provide it as such and perform the upload manually, by default None |
| completion_callback | Optional[Callable[[int, str, str], None]] |
= None |
callback to run when process terminates (not supported on Windows) |
| completion_trigger | multiprocessing.synchronize.Event | None |
= None |
this trigger event is set when the processes completes |
| env | dict[str, str] | None |
= None |
environment variables for process |
| cwd | pathlib._local.Path | None |
= None |
working directory to execute the process within. Note that executable, input and script file paths should be 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 |
= None |
disable exception throwing instead putting Simvue into a dormant state if an error occurs |
| queue_blocking | bool | None |
= None |
block thread queues during metric/event recording |
| system_metrics_interval | Optional[Annotated[int, Gt(gt=0)]] |
= None |
frequency at which to collect resource metrics |
| enable_emission_metrics | bool | None |
= None |
enable monitoring of emission metrics |
| disable_system_metrics | bool | None |
= None |
disable monitoring of resource metrics |
| storage_id | str | None |
= None |
identifier of storage to use, by default None |
| abort_on_alert | Union[Literal['run', 'all', 'ignore'], bool, None] |
= None |
whether to abort when an alert is triggered.run - current run is aborted.terminate - script itself is terminated.ignore - alerts do 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)] |
= 1 |
frequency at which to check alert condition in seconds, by default None |
| notification | Literal['email', 'none'] |
= none |
whether to notify on triggeremail - send email to user on alert.none - send no notifications (default). |
| trigger_abort | bool |
= False |
whether this alert can trigger a run abort |
| attach_to_run | bool |
= True |
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'] |
is inside range - metric value falls within value range.is outside range - metric value falls outside of value range. |
|
| description | str | None |
= None |
description for this alert, default None |
| window | Annotated[int, Gt(gt=0)] |
= 5 |
time period in seconds over which metrics are averaged, by default 5 |
| frequency | Annotated[int, Gt(gt=0)] |
= 1 |
frequency at which to check alert condition in seconds, by default 1 |
| aggregation | Literal['average', 'sum', 'at least one', 'all'] |
= average |
method to use when aggregating metrics within time windowaverage - average across all values in window (default).sum - take the sum of all values within window.at least one - returns if at least one value in window satisfy condition.all - returns if all values in window satisfy condition. |
| notification | Literal['email', 'none'] |
= none |
whether to notify on triggeremail - send email to user on notify.none - send no notifications (default). |
| trigger_abort | bool |
= False |
whether this alert can trigger a run abort, default False |
| attach_to_run | bool |
= True |
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 threshold alert conditionsis above - value is above threshold.is below - value is below threshold. |
|
| description | str | None |
= None |
description for this alert, default None |
| window | Annotated[int, Gt(gt=0)] |
= 5 |
time period in seconds over which metrics are averaged, by default 5 |
| frequency | Annotated[int, Gt(gt=0)] |
= 1 |
frequency at which to check alert condition in seconds, by default 1 |
| aggregation | Literal['average', 'sum', 'at least one', 'all'] |
= average |
method to use when aggregating metrics within time windowaverage - average across all values in window (default).sum - take the sum of all values within window.at least one - returns if at least one value in window satisfy condition.all - returns if all values in window satisfy condition. |
| notification | Literal['email', 'none'] |
= none |
whether to notify on triggeremail - send email to user on alert.none - send no notifications (default). |
| trigger_abort | bool |
= False |
whether this alert can trigger a run abort, default False |
| attach_to_run | bool |
= True |
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 |
= None |
description for this alert, default None |
| notification | Literal['email', 'none'] |
= none |
whether to notify on triggeremail - send email to user on notify.none - send no notifications (default). |
| trigger_abort | bool |
= False |
whether this alert can trigger a run abort, default False |
| attach_to_run | bool |
= True |
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\\/\\.:]+$')])] |
= None |
the name to allocate this run, if not specified a name will be selected at random, by default None |
| metadata | dict[str, Any] |
= None |
any metadata relating to the run as key-value pairs, by default None |
| tags | list[str] | None |
= None |
a list of tags for this run, by default None |
| description | str | None |
= None |
description of the run, by default None |
| folder | Annotated[str, FieldInfo(annotation=None, required=False, default=None, metadata=[_PydanticGeneralMetadata(pattern='^/.*')])] |
= None |
folder within which to store the run, by default "/" |
| notification | Literal['none', 'all', 'error', 'lost'] |
= none |
email notification on completion settingsnone - do not notify (default).all - notify for all updates.error - notify on errors.lost - notify if run lost. |
| running | bool |
= True |
whether to set the status as running or created, the latter implying the run will be commenced at a later time. Default is True. |
| retention_period | str | None |
= None |
describer for time period to retain run, the default of None removes this constraint. |
| timeout | int | None |
= 180 |
specify the timeout of the run, if None there is no timeout |
| visibility | Union[Literal['public', 'tenant'], list[str], None] |
= 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 |
= False |
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 |
= None |
ID of alert to update, by default None |
| name | str | None |
= None |
Name of the alert to update, by default None |
| state | Literal['ok', 'critical'] |
= critical |
state to set alert took - alert is set to ok state.critical - alert is set to critical state (default). |
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 |
= 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 |
= None |
manually specify the step index for this log, by default None |
| time | float | None |
= None |
manually specify the time for this log, by default None |
| timestamp | str | None |
= 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'] |
category of file with respect to this runinput - this file is an input file.output - this file is created by the run.code - this file represents an executed script |
|
| file_type | str | None |
= None |
manually specify the MIME type for all items, by default None |
| preserve_path | bool |
= False |
_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'] |
category of file with respect to this runinput - this file is an input file.output - this file is created by the run.code - this file represents an executed script |
|
| file_type | str | None |
= None |
manually specify the MIME type for items in the directory, by default None |
| preserve_path | bool |
= False |
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 runinput - this file is an input file.output - this file is created by the run.code - this file represents an executed script |
|
| file_type | str | None |
= None |
the MIME file type else this is deduced, by default None |
| preserve_path | bool |
= False |
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\\/\\.:]+$')])]] |
= None |
name to associate with this file, by default None |
| metadata | dict[str, Any] |
= None |
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 runinput - this file is an input file.output - this file is created by the run.code - this file represents an executed script |
|
| name | Optional[Annotated[str, FieldInfo(annotation=None, required=True, metadata=[_PydanticGeneralMetadata(pattern='^[a-zA-Z0-9\\-\\_\\s\\/\\.:]+$')])]] |
= None |
name to associate with this object, by default None |
| allow_pickle | bool |
= False |
whether to allow pickling if all other serialization types fail, by default False |
| metadata | dict[str, Any] |
= None |
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 |
= None |
additional metadata to attach to this folder, by default None |
| tags | list[str] | None |
= None |
list of tags to assign to the folder, by default None |
| description | str | None |
= 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 tocompleted - run finished with zero exit status.failed - run failed to complete.terminated - run was aborted. |
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
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