This API reference is not for the latest stable Salvus version.

`salvus.flow.api`

Central SalvusFlow remote job execution API.

User interactions should almost always happen through the functions defined in this module.

Functions

`add_job_group()`

def add_job_group(
    name: str, description: str, info: Optional[Dict] = None
) -> None:
    ...

Add a new job group to the database.

Parameters

name str — Name of the job group.
description str — Description of the job group.
info Optional[Dict] — Optional info dictionary.

Returns None

`cancel_all_running_jobs()`

def cancel_all_running_jobs(site_name: str, verbosity: int = 1) -> None:
    ...

Cancel all running or pending jobs for a given site.

Parameters

site_name str — Name of the site which’s jobs to cancel.
verbosity int — Verbosity level.

Returns None

`get_all_job_groups()`

def get_all_job_groups(
    include_job_count: bool = False,
) -> List[salvus.flow.db.JobGroup]:
    ...

Get all job groups from the database.

Parameters

include_job_count bool — Count the jobs for each group. Potentially somewhat slow.

Returns List[salvus.flow.db.JobGroup] — List of all job groups.

`get_job()`

def get_job(
    site_name: str, job_name: str
) -> salvus.flow.sites.salvus_job.SalvusJob:
    ...

Get a SalvusJob object by querying the database.

Parameters

site_name str — The site this job is on.
job_name str — The name of the job.

Returns salvus.flow.sites.salvus_job.SalvusJob — The queried for job.

`get_job_array()`

def get_job_array(
    site_name: str, job_array_name: str
) -> salvus.flow.sites.salvus_job_array.SalvusJobArray:
    ...

Get a SalvusJobArray object.

Parameters

site_name str — The site this job array is on.
job_array_name str — The name of the job array.

Returns salvus.flow.sites.salvus_job_array.SalvusJobArray — The queried for job array.

`get_job_arrays()`

def get_job_arrays(
    limit: Union[int, numpy.int32, numpy.int64] = 100,
    site_name: Optional[str] = None,
    job_array_status: Optional[
        List[Union[salvus.flow.sites.types.JobStatus, str]]
    ] = None,
    order_by: Iterable[str] = ("last_updated", "desc"),
    update_job_arrays: bool = False,
) -> List[salvus.flow.sites.salvus_job_array.SalvusJobArray]:
    ...

Query for a list of SalvusJobArray objects.

Parameters

limit Union[int, numpy.int32, numpy.int64] — Maximum number of returned job arrays.
site_name Optional[str] — Only return job arrays from this site.
job_array_status Optional[List[Union[salvus.flow.sites.types.JobStatus, str]]] — Limit returned jobs to job arrays with these statuses. Will return any job if not given.
order_by Iterable[str] — Choose how to order by (database column, direction). Available database columns are "last_updated", "job_array_start_time", "job_array_end_time", and "status".
update_job_arrays bool — Update pending/running job arrays.

Returns List[salvus.flow.sites.salvus_job_array.SalvusJobArray] — List of job arrays satisfying the query criteria.

`get_jobs()`

def get_jobs(
    limit: Union[int, numpy.int32, numpy.int64] = 100,
    site_name: Optional[str] = None,
    job_status: Optional[
        List[Union[salvus.flow.sites.types.JobStatus, str]],
        salvus.flow.sites.types.JobStatus,
    ] = None,
    order_by: Iterable[str] = ("last_updated", "desc"),
    update_jobs: bool = False,
    skip_jobs_from_job_arrays: bool = True,
) -> List[salvus.flow.sites.salvus_job.SalvusJob]:
    ...

Query for a list of SalvusJob objects.

Parameters

limit Union[int, numpy.int32, numpy.int64] — Maximum number of returned jobs.
site_name Optional[str] — Only return jobs from this site.
job_status Optional[List[Union[salvus.flow.sites.types.JobStatus, str]], salvus.flow.sites.types.JobStatus] — Limit returned jobs to jobs with these statuses. Will return jobs job if not given.
order_by Iterable[str] — Choose how to order by (database column, direction). Available database columns are "last_updated", "job_start_time", "job_end_time", and "status".
update_jobs bool — Update pending/running jobs.
skip_jobs_from_job_arrays bool — Don’t return jobs that are part of a job array.

Returns List[salvus.flow.sites.salvus_job.SalvusJob] — List of jobs satisfying the query criteria.

`get_site()`

def get_site(
    site_name: str, verbosity: int = 1, skip_version_number_check: bool = False
) -> salvus.flow.sites.base_site.BaseSite:
    ...

Get an initialized site object from the chose site.

Parameters

site_name str — The name of the site.
verbosity int — Verbosity level.
skip_version_number_check bool — Skip the version number check of the local Python version vs the remote site. Useful for initializing and updating sites.

Returns salvus.flow.sites.base_site.BaseSite — The requested, initialized site.

`get_task_chain()`

def get_task_chain(
    site_name: str, job_name: str
) -> salvus.flow.sites.task_chain.task_chain.TaskChain:
    ...

Get a TaskChain object by querying the database.

Parameters

site_name str — The site this job is on.
job_name str — The name of the job of the task chain.

Returns salvus.flow.sites.task_chain.task_chain.TaskChain — The queried for task chain.

`initialize_site()`

def initialize_site(
    site_name: str,
    verbosity: int = 1,
    wait_for_initial_run_in_seconds: Union[
        int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64
    ] = 3.0,
) -> salvus.flow.sites.base_site.BaseSite:
    ...

Initialize a site using information from the config TOML file.

Parameters

site_name str — Name of the site.
verbosity int — Verbosity level.
wait_for_initial_run_in_seconds Union[int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64] — How long to wait for the job to finish before querying for the first time. 3 seconds is a good number for most cases, for testing this can be shortened.

Returns salvus.flow.sites.base_site.BaseSite

`run()`

def run(
    site_name: str,
    input_file: Union[
        str, pathlib.Path, Dict, salvus.flow.utils.deep_setters._DeepSetter
    ],
    output_folder: Union[str, pathlib.Path],
    ranks: Optional[int, numpy.int32, numpy.int64] = None,
    wall_time_in_seconds: Optional[
        int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64
    ] = None,
    ping_interval_in_seconds: Optional[
        int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64
    ] = None,
    job_groups: Optional[List[str]] = None,
    job_name: Optional[str] = None,
    get_all: bool = False,
    overwrite: bool = False,
    delete_remote_files: bool = True,
    raise_on_deletion_failure: bool = False,
    verbosity: int = 1,
) -> salvus.flow.sites.salvus_job.SalvusJob:
    ...

Run one Salvus simulation and block until it is done.

Will run the simulation, wait for it to be done, download all output files, and finally delete all remote files and the job’s entry in the Salvus internal database.

Parameters

site_name str — The site to run on.
input_file Union[str, pathlib.Path, Dict, salvus.flow.utils.deep_setters._DeepSetter] — The input file or simulation object for the run.
output_folder Union[str, pathlib.Path] — Output files will be copied here.
ranks Optional[int, numpy.int32, numpy.int64] — The number of ranks to run with. Defaults to the site’s configuration value.
wall_time_in_seconds Optional[int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64] — The wall time in seconds if the site requires one.
ping_interval_in_seconds Optional[int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64] — How often to query the remote site about the job status. Defaults to the site’s configuration value.
job_groups Optional[List[str]] — Job groups to attach to the job in the database.
job_name Optional[str] — Custom job name. Be careful with this.
get_all bool — Get all output files. Wavefield outputs are omitted by default.
overwrite bool — Delete the output folder before running if it already exists.
delete_remote_files bool — Delete remote files after the run has finished.
raise_on_deletion_failure bool — If True this function will raise if, after everything ran, there is an error in the cleanup stage. This can happen for example on networked file systems.
verbosity int — The verbosity level.

Returns salvus.flow.sites.salvus_job.SalvusJob — The salvus.flow.sites.salvus_job.SalvusJob object representing the job.

`run_async()`

def run_async(
    site_name: str,
    input_file: Union[
        str, pathlib.Path, Dict, salvus.flow.utils.deep_setters._DeepSetter
    ],
    ranks: Optional[int, numpy.int32, numpy.int64] = None,
    wall_time_in_seconds: Optional[
        int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64
    ] = None,
    job_groups: Optional[List[str]] = None,
    job_name: Optional[str] = None,
    verbosity: int = 1,
) -> salvus.flow.sites.salvus_job.SalvusJob:
    ...

Launch one Salvus simulation asynchronously.

This function will return immediately upon successful submission of the job.

Parameters

site_name str — The site to run on.
input_file Union[str, pathlib.Path, Dict, salvus.flow.utils.deep_setters._DeepSetter] — The input file or simulation object for the run.
ranks Optional[int, numpy.int32, numpy.int64] — The number of ranks to run with. Defaults to the site’s configuration value.
wall_time_in_seconds Optional[int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64] — The wall time in seconds if the site requires one.
job_groups Optional[List[str]] — Job groups to attach to the job in the database.
job_name Optional[str] — Custom job name. Be careful with this.
verbosity int — The verbosity level.

Returns salvus.flow.sites.salvus_job.SalvusJob — The salvus.flow.sites.salvus_job.SalvusJob object representing the just submitted job.

`run_many()`

def run_many(
    site_name: str,
    input_files: Iterable[
        Union[
            str, pathlib.Path, Dict, salvus.flow.utils.deep_setters._DeepSetter
        ]
    ],
    output_folder: Union[str, pathlib.Path],
    ranks_per_job: Optional[int, numpy.int32, numpy.int64] = None,
    wall_time_in_seconds_per_job: Optional[
        int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64
    ] = None,
    ping_interval_in_seconds: Optional[
        int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64
    ] = None,
    get_all: bool = False,
    overwrite: bool = False,
    copy_partial_results: bool = True,
    delete_remote_files: bool = True,
    raise_on_deletion_failure: bool = False,
    verbosity: int = 1,
) -> salvus.flow.sites.salvus_job_array.SalvusJobArray:
    ...

Run many Salvus simulations and block until they are done.

Will run the simulations, wait them to be done, download all output files, and finally delete all remote files as well as the jobs’ entry in the Salvus internal database.

Parameters

site_name str — The site to run on.
input_files Iterable[Union[str, pathlib.Path, Dict, salvus.flow.utils.deep_setters._DeepSetter]] — A list of input files or simulation objects to be run.
output_folder Union[str, pathlib.Path] — Output files will be copied here.
ranks_per_job Optional[int, numpy.int32, numpy.int64] — The number of ranks to run each job with. Defaults to the site’s configuration value.
wall_time_in_seconds_per_job Optional[int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64] — The wall time in seconds per job if the site requires one.
ping_interval_in_seconds Optional[int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64] — How often to query the remote site about the jobs’ status. Defaults to the site’s configuration value.
get_all bool — Get all output files. Wavefield outputs are omitted by default.
overwrite bool — Delete the output folder before running if it already exists.
copy_partial_results bool — If this is True (the default) partial results will be copied over. This means that as long as a few jobs in the job array succeeded, these will be copied over. The function will still raise an exception but the data will be copied. Otherwise it will just raise.
delete_remote_files bool — Delete remote files after the run has finished.
raise_on_deletion_failure bool — If True this function will raise if, after everything ran, there is an error in the cleanup stage. This can happen for example on networked file systems.
verbosity int — The verbosity level.

Returns salvus.flow.sites.salvus_job_array.SalvusJobArray

`run_many_async()`

def run_many_async(
    site_name: str,
    input_files: Iterable[
        Union[
            str, pathlib.Path, Dict, salvus.flow.utils.deep_setters._DeepSetter
        ]
    ],
    ranks_per_job: Optional[int, numpy.int32, numpy.int64] = None,
    wall_time_in_seconds_per_job: Optional[
        int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64
    ] = None,
    job_groups: Optional[List[str]] = None,
    verbosity: int = 1,
) -> salvus.flow.sites.salvus_job_array.SalvusJobArray:
    ...

Launch many Salvus simulations asynchronously.

This function will return immediately upon successful submission of the jobs.

Parameters

site_name str — The site to run on.
input_files Iterable[Union[str, pathlib.Path, Dict, salvus.flow.utils.deep_setters._DeepSetter]] — A list of input files or simulation objects to be run.
ranks_per_job Optional[int, numpy.int32, numpy.int64] — The number of ranks to run each job with. Defaults to the site’s configuration value.
wall_time_in_seconds_per_job Optional[int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64] — The wall time in seconds per job if the site requires one.
job_groups Optional[List[str]] — Job groups to attach to the job in the database.
verbosity int — The verbosity level.

Returns salvus.flow.sites.salvus_job_array.SalvusJobArray — The salvus.flow.api.salvus_job_array.SalvusJobArray object representing the just submitted job array.

`run_task_chain()`

def run_task_chain(
    site_name: str,
    tasks: List[Union[salvus.flow.utils.deep_setters._DeepSetter, Callable]],
    ranks: Union[int, numpy.int32, numpy.int64],
    wall_time_in_seconds: Optional[
        int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64
    ] = None,
    data_files_to_upload: Optional[List[Union[pathlib.Path, str]]] = None,
    ping_interval_in_seconds: Optional[
        int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64
    ] = None,
    job_name: Optional[str] = None,
    verbosity: int = 1,
    interactive_tty: bool = False,
) -> salvus.flow.sites.task_chain.task_chain.TaskChain:
    ...

Launch a new task chain and wait for it to finish.

Contrary to the run() and run_many() functions this does not automatically download the results and cleans up all remote files. This still has to be done manually. The reasons for this is that task chains usually execute more complicated things and only some outputs are, in the end, required.

Parameters

site_name str — The site to run on.
tasks List[Union[salvus.flow.utils.deep_setters._DeepSetter, Callable]] — A list of tasks. Each task can either be a Salvus simulation or a Python function.
ranks Union[int, numpy.int32, numpy.int64] — The number of ranks to run with. Will only affect the Salvus runs.
wall_time_in_seconds Optional[int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64] — The wall time in seconds if the site requires one.
data_files_to_upload Optional[List[Union[pathlib.Path, str]]] — List of optional data files to upload to the task chain directory.
ping_interval_in_seconds Optional[int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64] — How often to query the remote site about the job status. Defaults to the site’s configuration value.
job_name Optional[str] — Name of the task chain job.
verbosity int — The verbosity level.
interactive_tty bool — Flag for interactive tty.

Returns salvus.flow.sites.task_chain.task_chain.TaskChain

`run_task_chain_async()`

def run_task_chain_async(
    site_name: str,
    tasks: List[Union[salvus.flow.utils.deep_setters._DeepSetter, Callable]],
    ranks: Union[int, numpy.int32, numpy.int64],
    wall_time_in_seconds: Optional[
        int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64
    ] = None,
    data_files_to_upload: Optional[List[Union[pathlib.Path, str]]] = None,
    job_name: Optional[str] = None,
    verbosity: int = 1,
    interactive_tty: bool = False,
) -> salvus.flow.sites.task_chain.task_chain.TaskChain:
    ...

Launch a new task chain and return it.

Parameters

site_name str — The site to run on.
tasks List[Union[salvus.flow.utils.deep_setters._DeepSetter, Callable]] — A list of tasks. Each task can either be a Salvus simulation or a Python function.
ranks Union[int, numpy.int32, numpy.int64] — The number of ranks to run with. Will only affect the Salvus runs.
wall_time_in_seconds Optional[int, numpy.int32, numpy.int64, float, numpy.float32, numpy.float64] — The wall time in seconds if the site requires one.
data_files_to_upload Optional[List[Union[pathlib.Path, str]]] — List of optional data files to upload to the task chain directory.
job_name Optional[str] — Name of the task chain job.
verbosity int — The verbosity level.
interactive_tty bool — Flag for interactive tty.

Returns salvus.flow.sites.task_chain.task_chain.TaskChain

`set_config()`

def set_config(site_config: Union[pathlib.Path, Dict], db_path: str) -> None:
    ...

Set the configuration for all API calls in this Python session.

Parameters

site_config Union[pathlib.Path, Dict] — Filename or existing dictionary with the site configuration.
db_path str — Path to the sqlite database used to internally keep track of things.

Returns None

`set_default_config()`

def set_default_config() -> None:
    ...

Reset the configuration for all API calls to the defaults.

Returns None