ersilia.utils package¶

Subpackages¶

Submodules¶

ersilia.utils.asyncio_utils module¶

ersilia.utils.asyncio_utils.run_coroutine(coro)[source]¶

Run a coroutine to completion from synchronous code.

Uses asyncio.run when no event loop is running. Inside a running loop (e.g. a Jupyter notebook), asyncio.run is not allowed, so the loop is made re-entrant with nest_asyncio first. This keeps the patch out of processes that never need it, instead of applying it at import time.

Parameters:: coro (coroutine) – The coroutine to run.
Returns:: The coroutine’s result.
Return type:: Any

ersilia.utils.cache module¶

class ersilia.utils.cache.SetupRedis(cache, maxmemory)[source]¶

Bases: object

A utility class to ensure a Redis server is running using Docker.

This class uses the Docker SDK for Python to manage the Redis container. It checks for the availability of the Redis image, handles container lifecycle operations (restart, create) and ensures that a Docker volume is mounted for data persistence.

Parameters:

cache (bool) – Whether to use redis cache or not
maxmemory (float) – Fraction of memory used by redis

ensure_redis_running()[source]¶

Ensure that a Redis server is running by managing the Docker container.

This method checks if the Redis image exists, whether a container with the specified name is available and running, and starts or restarts the container as needed.

Raises:: RuntimeError – If the container fails to start.

remove_redis_image_if_exists()[source]¶: Checks if the specified Docker image exists. If it does, remove it forcefully.

ersilia.utils.checks module¶

Checks of user input shared by the CLI and the Python API.

Each problem is reported with what is wrong and how to fix it. In the CLI the command then exits with code 1; in the Python API an InvalidOptionError is raised instead.

ersilia.utils.checks.check_curated_examples(model_id)[source]¶

Check that a fetched model has its own (curated) example inputs.

Parameters:: model_id (str) – The model identifier.

ersilia.utils.checks.check_fetch_folder(model, from_dir)[source]¶

Check a model folder given to fetch (--from_dir).

Parameters:

model (str) – The model the user asked for (identifier or slug).
from_dir (str) – The folder with the model repository.

Returns:

The model found in the folder.

Return type:

ModelBase

ersilia.utils.checks.check_run_arguments(input, output)[source]¶

Check the input and output of ersilia run, and prepare the input.

Parameters:

input (str) – Path of the input CSV file.
output (str) – Path of the output file (.csv or .h5).

Returns:

The input path to run, and a temporary folder to remove afterwards (when a single column was extracted from a wider file), or None.

Return type:

tuple of (str, str or None)

ersilia.utils.checks.fail(message, hint=None)[source]¶

Report an error (and a hint): exit with code 1, or raise in the Python API.

Parameters:

message (str) – What went wrong.
hint (str, optional) – What to do about it.

Raises:

InvalidOptionError – In library mode (the Python API), instead of printing and exiting.

ersilia.utils.cli_query module¶

ersilia.utils.cli_query.query_yes_no(question)[source]¶

ersilia.utils.conda module¶

class ersilia.utils.conda.BaseConda[source]¶

Bases: object

Base class for managing conda environments.

default_env()[source]¶: Get the default conda environment.

is_base()[source]¶: Check if the current environment is the base environment.

conda_prefix(is_base)[source]¶: Get the conda prefix path.

static conda_prefix(is_base)[source]¶

Get the conda prefix path.

Parameters:: is_base (bool) – Whether the current environment is the base environment.
Returns:: The conda prefix path.
Return type:: str

static default_env()[source]¶

Get the default conda environment.

Returns:: The name of the default conda environment.
Return type:: str

is_base()[source]¶

Check if the current environment is the base environment.

Returns:: True if the current environment is the base environment, False otherwise.
Return type:: bool

class ersilia.utils.conda.CondaUtils(config_json=None)[source]¶

Bases: BaseConda

Utility class for managing conda environments and packages.

Parameters:: config_json (dict, optional) – Configuration settings in JSON format. Default is None.

Examples

conda_utils = CondaUtils()
conda_utils.create("myenv", "3.8")

activate_base()[source]¶

Generate a script to activate the base conda environment.

Returns:: The script to activate the base conda environment.
Return type:: str

static checksum_from_conda_yml_file(self, env_yml, overwrite)[source]¶

Generate a checksum for a conda environment YAML file.

Parameters:

env_yml (str) – The path to the conda environment YAML file.
overwrite (bool) – Whether to overwrite the name and prefix in the YAML file with the checksum.

Returns:

The checksum of the YAML file contents.

Return type:

str

checksum_from_dockerfile(dockerfile_dir, dest=None)[source]¶

Generate a checksum for a Dockerfile.

Parameters:

dockerfile_dir (str) – The directory containing the Dockerfile.
dest (str, optional) – The destination directory for the checksum file. Default is None.

Returns:

The checksum of the Dockerfile.

Return type:

str

checksum_from_file(filename)[source]¶

Generate a checksum for the contents of a file.

Parameters:: filename (str) – The path to the file.
Returns:: The checksum of the file contents.
Return type:: str

get_base_env(path)[source]¶

Get the base environment from a JSON file.

Parameters:: path (str) – The path to the JSON file.
Returns:: The base environment.
Return type:: str

get_conda_and_pip_install_commands_from_dockerfile_if_exclusive(path, force_exclusive=True)[source]¶

Identify install commands from a Dockerfile.

Parameters:

path (str) – The path to the Dockerfile.
force_exclusive (bool, optional) – Whether to force exclusive conda and pip commands. Default is True.

Returns:

A list of install commands.

Return type:

list

specs_from_dockerfile(dockerfile_dir, dest=None, use_checksum=False, name=None)[source]¶

Generate specs from a Dockerfile.

Parameters:

dockerfile_dir (str) – The directory containing the Dockerfile.
dest (str, optional) – The destination directory for the specs file. Default is None.
use_checksum (bool, optional) – Whether to use a checksum for the specs. Default is False.
name (str, optional) – The name to use for the specs. Default is None.

Returns:

The name or checksum of the specs.

Return type:

str

specs_from_dockerfile_as_json(dockerfile_dir, dest)[source]¶

Write a JSON file with the install requirements inferred from the Dockerfile.

Parameters:

dockerfile_dir (str) – The directory containing the Dockerfile.
dest (str) – The destination directory for the JSON file.

Returns:

The path to the JSON file.

Return type:

str

class ersilia.utils.conda.SimpleConda(config_json=None)[source]¶

Bases: CondaUtils

A class to manage conda environments using simple commands.

Parameters:: config_json (dict, optional) – Configuration settings in JSON format. Default is None.

Examples

simple_conda = SimpleConda()
simple_conda.create("myenv", "3.8")

active_env()[source]¶

Get the currently active conda environment.

Returns:: The name of the currently active conda environment.
Return type:: str

clone(src_env, dst_env)[source]¶

Make an exact copy of a conda environment.

Parameters:

src_env (str) – The name of the source environment.
dst_env (str) – The name of the destination environment.

Raises:

Exception – If the source environment does not exist or the destination environment already exists.

create(environment, python_version)[source]¶

Create a new conda environment.

Parameters:

environment (str) – The name of the new environment.
python_version (str) – The Python version to use in the new environment.

Return type:

None

create_executable_bash_script(environment, commandlines, file_name)[source]¶

Create an executable bash script to run commands in a conda environment.

Parameters:

environment (str) – The name of the environment.
commandlines (str or list) – The commands to run.
file_name (str) – The name of the bash script file.

Returns:

The path to the bash script file.

Return type:

str

delete(environment)[source]¶

Delete a conda environment.

Parameters:: environment (str) – The name of the environment to delete.

delete_one(environment)[source]¶

Delete a conda environment.

Parameters:: environment (str) – The name of the environment to delete.

exists(environment)[source]¶

Check if a conda environment exists.

Parameters:: environment (str) – The name of the environment to check.
Returns:: True if the environment exists, False otherwise.
Return type:: bool

export_env_yml(environment, dest)[source]¶

Export a conda environment as an environment YAML file.

Parameters:

environment (str) – The name of the environment to export.
dest (str) – The destination directory for the YAML file.

get_python_path_env(environment)[source]¶

Get the Python path for a conda environment.

Parameters:: environment (str) – The name of the environment.
Returns:: The Python path for the environment.
Return type:: str

list_eos_environments()[source]¶

List all conda environments that start with ‘eos’ and have a length of 7 characters (i.e. an allowed model ID).

Parameters:: None
Returns:: A list of conda environments that start with ‘eos’ and have a length of 7 characters.
Return type:: list

run_commandlines(environment, commandlines)[source]¶

Run commands in a given conda environment.

Parameters:

environment (str) – The name of the environment.
commandlines (str or list) – The commands to run.

Raises:

ModelPackageInstallError – If critical errors are found in the conda logs.

startswith(environment)[source]¶

Get a list of conda environments that start with the given prefix.

Parameters:: environment (str) – The prefix to check.
Returns:: A list of conda environments that start with the given prefix.
Return type:: list

class ersilia.utils.conda.StandaloneConda[source]¶

Bases: object

A class to manage standalone conda environments.

exists(environment)[source]¶

Check if a standalone conda environment exists.

Parameters:: environment (str) – The name of the environment to check.
Returns:: True if the environment exists, False otherwise.
Return type:: bool

run_commandlines(environment, commandlines)[source]¶

Run commands in a given standalone conda environment.

Parameters:

environment (str) – The name of the environment.
commandlines (str or list) – The commands to run.

ersilia.utils.config module¶

class ersilia.utils.config.Checker[source]¶

Bases: object

A class to check and manage configuration and credentials files.

config()[source]¶: Ensure the configuration file exists.

get_development_path()[source]¶: Get the development path.

config()[source]¶

Ensure the configuration file exists.

If the configuration file does not exist, it will be created by copying from the development path or downloading from GitHub.

get_development_path()[source]¶

Get the development path.

Returns:: The development path.
Return type:: str

class ersilia.utils.config.Config(json_file=None)[source]¶

Bases: object

Config class.

An instance of this object holds config file section as attributes.

Parameters:: json_file (str, optional) – The path to the JSON configuration file. Default is None.

keys()[source]¶

Get the keys of the configuration.

Returns:: The keys of the configuration.
Return type:: dict_keys

class ersilia.utils.config.Credentials(json_file=None)[source]¶

Bases: object

A class to manage credentials.

Parameters:: json_file (str, optional) – The path to the JSON credentials file. Default is None.

keys()[source]¶

Get the keys of the credentials.

Returns:: The keys of the credentials.
Return type:: dict_keys

class ersilia.utils.config.Secrets(overwrite=True)[source]¶

Bases: object

A class to manage secrets and credentials.

Parameters:: overwrite (bool, optional) – Whether to overwrite existing files. Default is True.

fetch_from_github()[source]¶: Fetch secrets from the GitHub repository.

fetch_gdrive_secrets_from_github()[source]¶: Fetch Google Drive client secrets from the GitHub repository.

to_credentials(json_file)[source]¶

Convert secrets to a credentials file.

Parameters:: json_file (str) – The path to the credentials file.
Returns:: True if the credentials file was created successfully, False otherwise.
Return type:: bool

ersilia.utils.cron module¶

ersilia.utils.cron.del_model(model_id)[source]¶

ersilia.utils.cron.model_cleanup()[source]¶

ersilia.utils.cron.seconds_to_days(s)[source]¶

ersilia.utils.csvfile module¶

class ersilia.utils.csvfile.CsvDataLoader[source]¶

Bases: object

A class to load and process data from CSV and JSON files.

load(csv_file)[source]¶: Load data from a CSV file.

read(file_path)[source]¶: Read data from a CSV, TSV, or JSON file.

load(csv_file)[source]¶

Load data from a CSV file.

Parameters:: csv_file (str) – The path to the CSV file.

read(file_path)[source]¶

Read data from a CSV, TSV, or JSON file.

Parameters:: file_path (str) – The path to the file.
Returns:: The data read from the file.
Return type:: list or dict
Raises:: ValueError – If the file format is unsupported.

ersilia.utils.docker module¶

class ersilia.utils.docker.ContainerMetricsSampler(model_id, sampling_interval=0.01)[source]¶

Bases: object

A class to sample metrics from Docker containers.

Parameters:

model_id (str) – The model identifier.
sampling_interval (float, optional) – The interval between samples in seconds. Default is 0.01.

get_average_metrics()[source]¶

Get the average metrics from the tracked samples.

Returns:: A dictionary containing the average CPU and memory usage, and the peak CPU and memory usage.
Return type:: dict

start_tracking()[source]¶: Start tracking metrics from the Docker container.

stop_tracking()[source]¶: Stop tracking metrics from the Docker container.

class ersilia.utils.docker.SimpleDocker[source]¶

Bases: object

A class to manage Docker containers and images.

brute_delete_by_model_id(model_id)[source]¶

Brute force delete a Docker image and containers by model ID.

Parameters:: model_id (str) – The model identifier.

build(path, org, img, tag)[source]¶

Build a Docker image.

Parameters:

path (str) – The path to the Dockerfile.
org (str) – The organization name.
img (str) – The image name.
tag (str) – The image tag.

cleanup_ersilia_images()[source]¶: Remove all Ersilia-related Docker images.

container_peak(model_id)[source]¶

Get the peak memory usage of a Docker container running an Ersilia model.

Parameters:: model_id (str) – The model identifier.
Returns:: The peak memory usage in MB, or None if the container is not found or an error occurs.
Return type:: float or None

containers(only_run)[source]¶

Get a dictionary of Docker containers.

Parameters:: only_run (bool) – Whether to include only running containers.
Returns:: A dictionary of Docker containers with container names as keys and image names as values.
Return type:: dict

static cp_from_container(name, img_path, local_path, org=None, img=None, tag=None)[source]¶

Copy files from a Docker container to the local filesystem.

Parameters:

name (str) – The container name.
img_path (strcp_from_image) – The path inside the container.
local_path (str) – The local path to copy files to.
org (str, optional) – The organization name. Default is None.
img (str, optional) – The image name. Default is None.
tag (str, optional) – The image tag. Default is None.
None

async cp_from_image(img_path, local_path, org, img, tag)[source]¶

Copy files from a Docker image to the local filesystem.

Parameters:

img_path (str) – The path inside the image.
local_path (str) – The local path to copy files to.
org (str) – The organization name.
img (str) – The image name.
tag (str) – The image tag.

delete(org, img, tag)[source]¶

Delete a Docker image.

Parameters:

org (str) – The organization name.
img (str) – The image name.
tag (str) – The image tag.

exec(cmd, org, img, tag, name, kill=True)[source]¶

Execute a command in a Docker container and then kill the container.

Parameters:

cmd (str) – The command to execute.
org (str) – The organization name.
img (str) – The image name.
tag (str) – The image tag.
name (str) – The container name.

static exec_container(name, cmd)[source]¶

Execute a command in a Docker container.

Parameters:

name (str) – The container name.
cmd (str) – The command to execute.

exists(org, img, tag)[source]¶

Check if a Docker image exists.

Parameters:

org (str) – The organization name.
img (str) – The image name.
tag (str) – The image tag.

Returns:

True if the image exists, False otherwise.

Return type:

bool

images()[source]¶

Get a dictionary of Docker images.

Returns:: A dictionary of Docker images with image names as keys and image IDs as values.
Return type:: dict

is_owned_by_current_user(org, img, tag)[source]¶: Return True if the image carries the current user’s label, or has no user label (backward compat).

static kill(name)[source]¶

Kill a Docker container.

Parameters:: name (str) – The container name.

label_with_current_user(org, img, tag)[source]¶: Add the current user’s UID as a Docker label to a local image.

list_eos_images()[source]¶

Get a list of eos-related Docker images.

Returns:: A list of Ersilia-related Docker image names.
Return type:: list

static remove(name)[source]¶

Remove a Docker container.

Parameters:: name (str) – The container name.

run(org, img, tag, name, memory=None)[source]¶

Run a Docker container.

Parameters:

org (str) – The organization name.
img (str) – The image name.
tag (str) – The image tag.
name (str) – The container name.
memory (int, optional) – The memory limit for the container in GB. Default is None.

Returns:

The name of the running container.

Return type:

str

class ersilia.utils.docker.SimpleDockerfileParser(path)[source]¶

Bases: DockerfileParser

A class to parse Dockerfiles.

Parameters:: path (str) – The path to the Dockerfile or the directory containing the Dockerfile.

get_baseimage()[source]¶

Get the base image from the Dockerfile.

Returns:: The base image.
Return type:: str

get_runs()[source]¶

Get the RUN commands from the Dockerfile.

Returns:: A list of RUN commands.
Return type:: list

ersilia.utils.docker.is_docker_installed()[source]¶

ersilia.utils.docker.model_image_version_reader(dir)[source]¶: Read the requested model image version from a file.

ersilia.utils.docker.resolve_platform()[source]¶

ersilia.utils.docker.set_docker_host()[source]¶

ersilia.utils.download module¶

Download utilities

class ersilia.utils.download.GitHubDownloader(overwrite, token=None, use_eosvc=False)[source]¶

Bases: object

A class to download files and repositories from GitHub.

Parameters:

overwrite (bool) – Whether to overwrite existing files.
token (str, optional) – The GitHub token for authentication. Default is None.

clone(org, repo, destination, ungit=False)[source]¶

Clone a GitHub repository.

Parameters:

org (str) – The GitHub organization name.
repo (str) – The GitHub repository name.
destination (str) – The destination directory.
ungit (bool, optional) – Whether to remove Git-related files. Default is False.

download_single(org, repo, repo_path, destination)[source]¶

Download a single file from a GitHub repository.

Parameters:

org (str) – The GitHub organization name.
repo (str) – The GitHub repository name.
repo_path (str) – The path to the file in the repository.
destination (str) – The destination path.

Returns:

True if the file was downloaded successfully, False otherwise.

Return type:

bool

class ersilia.utils.download.GoogleDriveDownloader[source]¶

Bases: object

A class to download files from Google Drive.

download_file_from_google_drive(file_id, destination)[source]¶

Download a file from Google Drive.

Parameters:

file_id (str) – The Google Drive file ID.
destination (str) – The destination path.

fetch_zip(file_id, destination)[source]¶

Download a ZIP file from Google Drive and extract it.

Parameters:

file_id (str) – The Google Drive file ID.
destination (str) – The destination directory.

static get_confirm_token(response)[source]¶

Get the confirmation token from the response.

Parameters:: response (object) – The response object.
Returns:: The confirmation token, if available.
Return type:: str

static save_response_content(response, destination)[source]¶

Save the response content to a file.

Parameters:

response (object) – The response object.
destination (str) – The destination file path.

class ersilia.utils.download.OsfDownloader(overwrite)[source]¶

Bases: object

A class to download files from the Open Science Framework (OSF).

Parameters:: overwrite (bool) – Whether to overwrite existing files.

fetch(project_id, filename, destination, tmp_folder)[source]¶

Fetch a file from OSF.

Parameters:

project_id (str) – The OSF project ID.
filename (str) – The name of the file to fetch.
destination (str) – The destination directory.
tmp_folder (str) – The temporary folder to use.

class ersilia.utils.download.PseudoDownloader(overwrite)[source]¶

Bases: object

A class to simulate downloading by copying local directories.

Parameters:: overwrite (bool) – Whether to overwrite existing files.

fetch(src, dst)[source]¶

Copy entire directory.

Parameters:

src (str) – The source directory.
dst (str) – The destination directory.

class ersilia.utils.download.S3Downloader[source]¶

Bases: object

A class to download files from an S3 bucket.

download_from_s3(bucket_url, file_name, destination)[source]¶

Download a file from an S3 bucket.

Parameters:

bucket_url (str) – The URL of the S3 bucket.
file_name (str) – The name of the file to download.
destination (str) – The destination path.

ersilia.utils.echo module¶

class ersilia.utils.echo.Silencer[source]¶

Bases: object

A class to manage the silencing of CLI output.

silence_file¶

Path to the silence file.

Type:: str

is_silence()[source]¶: Checks if the CLI output is silenced.

speak()[source]¶: Enables CLI output.

silence()[source]¶: Disables CLI output.

is_silence()[source]¶

Checks if the CLI output is silenced.

Returns:: True if the CLI output is silenced, False otherwise.
Return type:: bool

silence()[source]¶: Disables CLI output.

speak()[source]¶: Enables CLI output.

async ersilia.utils.echo.async_spinner(text, coro, done=None)[source]¶

Await coro while showing text, then print a success line.

Parameters:

text (str) – What is happening, shown while coro runs.
coro (coroutine) – The work to await.
done (str, optional) – The success line printed afterwards. Defaults to text.

ersilia.utils.echo.confirm(question, default=False)[source]¶

Ask a yes/no question, formatted like the other lines.

Parameters:

question (str) – The question, e.g. “Continue?”.
default (bool, optional) – The answer when the user just presses Enter.

Returns:

The answer.

Return type:

bool

ersilia.utils.echo.echo(text, harmonize=True, **styles)[source]¶

Print a message in the CLI’s standard style.

The kind of message is given by fg: green is a success, yellow a warning, red an error, and anything else (or nothing) is info.

Parameters:

text (str) – The message. Newlines start indented continuation lines.
harmonize (bool, optional) – If False, print the text as is, without icon or indentation.
**styles – fg selects the kind of message; err=True prints to stderr. Other click styles are ignored when harmonizing, so every message of a kind looks the same.

Notes

Errors are printed to stderr, and so are the info lines that follow an error (its hints), until the next success, warning or running step. Everything else goes to stdout.

ersilia.utils.echo.is_quiet()[source]¶

Tell whether output is silenced.

Returns:: True if nothing should be printed.
Return type:: bool

ersilia.utils.echo.reset()[source]¶

Start a new command: output goes to stdout until the next error.

Called when each CLI command or Python API call starts, so that a previous error does not send later lines to stderr.

ersilia.utils.echo.set_quiet(quiet)[source]¶

Silence or restore all echo, spinner and progress output.

Parameters:: quiet (bool) – True to print nothing.

ersilia.utils.echo.spinner(text, func, *args, done=None, **kwargs)[source]¶

Run func while showing text, then print a success line.

Parameters:

text (str) – What is happening, shown while func runs.
func (callable) – The work to do.
done (str, optional) – The success line printed afterwards. Defaults to text.

ersilia.utils.environment module¶

Utility functions to get information about the working environment.

class ersilia.utils.environment.Environment[source]¶

Bases: object

Class to handle environment settings.

has_module(module_name)[source]¶: Check if Python module is installed.

ersilia.utils.hdf5 module¶

class ersilia.utils.hdf5.Hdf5Data(values, keys, inputs, features, dtype, dim)[source]¶

Bases: object

A class to handle HDF5 data storage.

Parameters:

values (array-like or nested array-like) – The data values. Can be a 1D list or a list-of-lists.
keys (array-like) – The keys associated with the data.
inputs (array-like) – The inputs associated with the data.
features (array-like) – The features associated with the data.
dtype ({float, int, str, None}) – If float/int/str: force all values to that type (float→np.float32, int→np.int32, str→utf-8 HDF5 string). If None: infer per your original logic.

save(filename, append=False)[source]¶

Save the data to an HDF5 file.

Parameters:

filename (str) – The path to the HDF5 file.
append (bool, optional) – If True, append Values, Keys and Inputs to the datasets of an existing file (Features are written only once). Datasets are always created resizable so that later appends are possible.

class ersilia.utils.hdf5.Hdf5DataLoader[source]¶

Bases: object

A class to load data from HDF5 files.

load(h5_file)[source]¶: Load data from an HDF5 file.

load(h5_file)[source]¶

Load data from an HDF5 file.

Parameters:: h5_file (str) – The path to the HDF5 file.

class ersilia.utils.hdf5.Hdf5DataStacker(h5_files)[source]¶

Bases: object

A class to stack multiple HDF5 files into one.

Parameters:: h5_files (list) – A list of paths to the HDF5 files to stack.

stack(h5_file)[source]¶

Stack the HDF5 files into one.

Parameters:: h5_file (str) – The path to the output HDF5 file.

ersilia.utils.import module¶

ersilia.utils.import.import_extra(mod)[source]¶: Try to import a module, if not found return None

ersilia.utils.logging module¶

class ersilia.utils.logging.Logger(*args, **kwargs)[source]¶

Bases: object

A singleton class to manage logging in Ersilia.

set_verbosity(verbose)[source]¶: Set the verbosity of the logger.

debug(text)[source]¶: Log a debug message.

info(text)[source]¶: Log an info message.

warning(text)[source]¶: Log a warning message.

error(text)[source]¶: Log an error message.

critical(text)[source]¶: Log a critical message.

success(text)[source]¶: Log a success message.

critical(text)[source]¶

Log a critical message.

Parameters:: text (str) – The message to log.

debug(text)[source]¶

Log a debug message.

Parameters:: text (str) – The message to log.

error(text)[source]¶

Log an error message.

Parameters:: text (str) – The message to log.

info(text)[source]¶

Log an info message.

Parameters:: text (str) – The message to log.

set_verbosity(verbose)[source]¶

Set the verbosity of the logger.

Parameters:: verbose (bool) – Whether to enable verbose logging.

success(text)[source]¶

Log a success message.

Parameters:: text (str) – The message to log.

warning(text)[source]¶

Log a warning message.

Parameters:: text (str) – The message to log.

ersilia.utils.logging.make_temp_dir(prefix)[source]¶

Create a temporary directory with a symbolic link in the session logs directory.

Parameters:: prefix (str) – The prefix for the temporary directory name.
Returns:: The path to the temporary directory.
Return type:: str

ersilia.utils.paths module¶

class ersilia.utils.paths.ErsiliaMetadataLoader(stream)[source]¶

Bases: SafeLoader

Custom YAML loader for Ersilia metadata.

yaml_constructors = {'tag:yaml.org,2002:binary': <function SafeConstructor.construct_yaml_binary>, 'tag:yaml.org,2002:bool': <function SafeConstructor.construct_yaml_bool>, 'tag:yaml.org,2002:float': <function SafeConstructor.construct_yaml_float>, 'tag:yaml.org,2002:int': <function SafeConstructor.construct_yaml_int>, 'tag:yaml.org,2002:map': <function metadata_constructor>, 'tag:yaml.org,2002:null': <function SafeConstructor.construct_yaml_null>, 'tag:yaml.org,2002:omap': <function SafeConstructor.construct_yaml_omap>, 'tag:yaml.org,2002:pairs': <function SafeConstructor.construct_yaml_pairs>, 'tag:yaml.org,2002:seq': <function SafeConstructor.construct_yaml_seq>, 'tag:yaml.org,2002:set': <function SafeConstructor.construct_yaml_set>, 'tag:yaml.org,2002:str': <function SafeConstructor.construct_yaml_str>, 'tag:yaml.org,2002:timestamp': <function SafeConstructor.construct_yaml_timestamp>, None: <function SafeConstructor.construct_undefined>}¶

class ersilia.utils.paths.Metadata(Identifier: str, Slug: str, Title: str, Description: str, Mode: str, Input: List[str], InputShape: str, Task: List[str], Output: List[str], OutputType: List[str], OutputShape: str, Interpretation: str, Tag: List[str], Publication: str, SourceCode: str, License: str, DockerHub: str | None = None, DockerArchitecture: List[str] | None = None, S3: str | None = None, Status: str | None = None, Contributor: str | None = None)[source]¶

Bases: object

A dataclass to represent metadata for Ersilia models.

Identifier¶

The model identifier.

Type:: str

Slug¶

The model slug.

Type:: str

Title¶

The model title.

Type:: str

Description¶

The model description.

Type:: str

Mode¶

The model mode.

Type:: str

Input¶

The model input types.

Type:: List[str]

InputShape¶

The shape of the model input.

Type:: str

Task¶

The tasks the model performs.

Type:: List[str]

Output¶

The model output types.

Type:: List[str]

OutputType¶

The types of the model output.

Type:: List[str]

OutputShape¶

The shape of the model output.

Type:: str

Interpretation¶

The interpretation of the model output.

Type:: str

Tag¶

The tags associated with the model.

Type:: List[str]

Publication¶

The publication associated with the model.

Type:: str

SourceCode¶

The source code repository for the model.

Type:: str

License¶

The license for the model.

Type:: str

DockerHub¶

The DockerHub repository for the model. Default is None.

Type:: Optional[str], optional

DockerArchitecture¶

The Docker architectures supported by the model. Default is None.

Type:: Optional[List[str]], optional

S3¶

The S3 bucket for the model. Default is None.

Type:: Optional[str], optional

Status¶

The status of the model. Default is None.

Type:: Optional[str], optional

Contributor¶

The contributor of the model. Default is None.

Type:: Optional[str], optional

Contributor: str | None = None¶

Description: str¶

DockerArchitecture: List[str] | None = None¶

DockerHub: str | None = None¶

Identifier: str¶

Input: List[str]¶

InputShape: str¶

Interpretation: str¶

License: str¶

Mode: str¶

Output: List[str]¶

OutputShape: str¶

OutputType: List[str]¶

Publication: str¶

S3: str | None = None¶

Slug: str¶

SourceCode: str¶

Status: str | None = None¶

Tag: List[str]¶

Task: List[str]¶

Title: str¶

class ersilia.utils.paths.Paths[source]¶

Bases: object

A class to handle various path-related operations in Ersilia.

model_id_from_path(path)[source]¶: Guess model identifier based on the path.

org_development_path()[source]¶: Guess generic development path.

ersilia_development_path()[source]¶: Try to guess the package development path in the local computer.

exists(path)[source]¶: Check if a path exists.

ersilia_development_path()[source]¶

Try to guess the package development path in the local computer.

Returns:: The package development path if found, otherwise None.
Return type:: str or None

static exists(path)[source]¶

Check if a path exists.

Parameters:: path (str) – The path to check.
Returns:: True if the path exists, False otherwise.
Return type:: bool

static home()[source]¶

Get home directory.

Returns:: The home directory path.
Return type:: str

model_id_from_path(path)[source]¶

Guess model identifier based on the path.

Parameters:: path (str) – The path to guess the model identifier from.
Returns:: The model identifier if found, otherwise None.
Return type:: str or None

org_development_path()[source]¶

Guess generic development path.

Returns:: The generic development path if found, otherwise None.
Return type:: str or None

ersilia.utils.paths.get_metadata_from_base_dir(path)[source]¶

Get metadata from the base directory of a model.

Parameters:: path (str) – The path to the base directory.
Returns:: The metadata dictionary.
Return type:: dict
Raises:: FileNotFoundError – If the metadata file is not found.

ersilia.utils.paths.metadata_constructor(loader, node)[source]¶

ersilia.utils.ports module¶

ersilia.utils.ports.find_free_port(preferred_port=None)[source]¶

Find a free port.

Parameters:: preferred_port (int, optional) – The preferred port number to check first. Default is None.
Returns:: A free port number.
Return type:: int

ersilia.utils.ports.is_port_in_use(port)[source]¶

Check if a port is in use.

Parameters:: port (int) – The port number to check.
Returns:: True if the port is in use, False otherwise.
Return type:: bool

ersilia.utils.ports.normalize_connect_url(url: str) → str[source]¶

ersilia.utils.remove module¶

class ersilia.utils.remove.OsfRemover(username, password)[source]¶

Bases: object

A class to remove files from the Open Science Framework (OSF).

Parameters:

username (str) – The OSF username.
password (str) – The OSF password.

remove(project_id, filename)[source]¶

Remove a file from OSF.

Parameters:

project_id (str) – The OSF project ID.
filename (str) – The name of the file to remove.

ersilia.utils.resolvers module¶

class ersilia.utils.resolvers.PackMethodResolver(model_id=None, config_json=None)[source]¶

Bases: ErsiliaBase

This class resolves the pack method for a given model after the fetch resolution. Note that this class is different from the TemplateResolver, which resolves the pack method before the fetch procedure.

resolve_pack_method()[source]¶

Resolve the packaging method for a model.

Parameters:: model_path (str) – The path to the model directory.
Returns:: The packaging method.
Return type:: str

resolve_pack_method_from_github_metadata()[source]¶

Resolve the packaging method based on metadata available from GitHub

Return type:: The packaging method (fastapi or )

resolve_pack_method_source()[source]¶

Resolve the packaging method for a model based on its source files.

Parameters:: model_path (str) – The path to the model directory.
Returns:: The packaging method if found, otherwise None.
Return type:: str or None

resolve_pack_method_tracked_at_fetch()[source]¶

Read the pack method from available from the fetch procedure (i.e. as resolved by the TemplateResolver)

Return type:: The packaging method (fastapi or )

ersilia.utils.session module¶

ersilia.utils.session.check_session_env()[source]¶

Check the session name given in ERSILIA_SESSION, if any.

Raises:: SessionNameError – If the variable is set to a name that cannot name a session.

ersilia.utils.session.container_is_running(name)[source]¶

Tell whether a Docker container is running.

Parameters:: name (str) – The container name.
Returns:: True if it is running, False if it is stopped or does not exist, and None if Docker cannot be reached (so the answer is unknown).
Return type:: bool or None

ersilia.utils.session.create_session_dir()[source]¶: Create a session directory.

ersilia.utils.session.create_session_files(session_name)[source]¶

ersilia.utils.session.deregister_model_session(model_id, session_dir=None)[source]¶

Deregister a session from a model, leaving other sessions untouched.

Parameters:

model_id (str) – The model ID.
session_dir (str, optional) – The session directory to deregister. Defaults to the current session.

ersilia.utils.session.deregister_session(session_dir)[source]¶

Deregister a session from every model it was registered for.

Parameters:: session_dir (str) – The session directory.

ersilia.utils.session.determine_orphaned_session()[source]¶

Determine orphaned sessions.

Returns:: A list of orphaned session names.
Return type:: list

ersilia.utils.session.end_named_session()[source]¶

Remove the current session’s folder if it was named with ERSILIA_SESSION.

A named session is not tied to a process, so it is never cleaned up automatically; closing it removes it. A terminal’s session is left alone (it is removed when the terminal ends).

Returns:: True if a named session was removed.
Return type:: bool

ersilia.utils.session.get_current_pid()[source]¶

Get the current process ID.

Returns:: The current process ID.
Return type:: int

ersilia.utils.session.get_live_model_sessions(model_id)[source]¶

Get the sessions that are currently serving a model.

A session counts as live when its process is running and it still holds the model’s pid file.

Parameters:: model_id (str) – The model ID.
Returns:: The live session directories.
Return type:: list

ersilia.utils.session.get_model_sessions(model_id)[source]¶

Get the sessions registered as serving a model.

Parameters:: model_id (str) – The model ID.
Returns:: The session directories, possibly including stale ones.
Return type:: list

ersilia.utils.session.get_parent_pid()[source]¶

Get the parent process ID.

Returns:: The parent process ID.
Return type:: int

ersilia.utils.session.get_session_dir()[source]¶

Get the session directory.

Returns:: The session directory path.
Return type:: str

ersilia.utils.session.get_session_id()[source]¶

Get the session ID.

The session is session_<name> when ERSILIA_SESSION is set, and session_<parent process ID> (one per terminal) otherwise.

Returns:: The session ID.
Return type:: str

ersilia.utils.session.get_session_uuid()[source]¶

Get the session UUID.

Returns:: The session UUID.
Return type:: str

ersilia.utils.session.invalid_session_name()[source]¶

Get the value of ERSILIA_SESSION if it cannot name a session.

Returns:: The invalid name, or None when the variable is unset or valid.
Return type:: str or None

ersilia.utils.session.is_named_session(session_name)[source]¶

Tell whether a session was named with ERSILIA_SESSION.

Parameters:: session_name (str) – A session name such as session_myproject, or a path ending in one.
Returns:: True for a named session, False for a terminal’s session.
Return type:: bool

ersilia.utils.session.is_session_alive(session_name)[source]¶

Check whether the process that owns a session is still running.

Parameters:: session_name (str) – A session name such as session_12345, or a path ending in one.
Returns:: True if the session’s process exists.
Return type:: bool

ersilia.utils.session.kill_process_tree(pid, timeout=5)[source]¶

Terminate a process and all of its descendants.

The model server is launched as ersilia_model_serve, which in turn spawns run_uvicorn.py as a child. Killing only the recorded PID orphans the uvicorn server, so the whole tree is terminated instead: SIGTERM first, then SIGKILL for anything still alive after timeout.

Parameters:

pid (int) – The root process ID.
timeout (float, optional) – Seconds to wait for graceful termination before force-killing.

ersilia.utils.session.models_served_elsewhere()[source]¶

List the models served in other live sessions.

Returns:: (model_id, session_dir) for each model served in a live session other than this one.
Return type:: list of tuple of (str, str)

ersilia.utils.session.models_stopped_by_cleanup()[source]¶

List the models stopped by this process’s cleanup of orphaned sessions.

A session is orphaned when its parent process has ended. That happens after every command run through a wrapper such as conda run, so a model served that way is stopped by the next command.

Returns:: The model IDs.
Return type:: list of str

ersilia.utils.session.prune_empty_session_dirs()[source]¶: Prune session directories that seem to be empty, meaning they don’t contain any serving data or logs.

ersilia.utils.session.purge_session_processes(session_name)[source]¶: Stop any leftover processes and Docker containers tracked by a session’s .pid files. Used to clean up after orphaned (terminal-killed) sessions.

ersilia.utils.session.read_pid_file(path)[source]¶

Read a model’s .pid file.

Each line is “<pid> <url> <container_name|->”, written when the model is served (AutoService.serve).

Parameters:: path (str) – The .pid file.
Returns:: The process IDs and the container names it lists.
Return type:: tuple of (list of int, list of str)

ersilia.utils.session.register_model_session(model_id, session_dir)[source]¶

Several sessions (terminals) can serve the same model at the same time, so each model maps to a list of session directories.

Parameters:

model_id (str) – The model ID.
session_dir (str) – The session directory.

ersilia.utils.session.remove_orphaned_sessions()[source]¶: Remove orphaned sessions.

ersilia.utils.session.remove_session_dir(session_name)[source]¶

ersilia.utils.session.session_name_from_env()[source]¶

Get the session name given in the ERSILIA_SESSION environment variable.

An invalid name is ignored here (the terminal’s session is used), since this runs while Ersilia is being imported; check_session_env reports it.

Returns:: The name, or None when the variable is not set or not valid.
Return type:: str or None

ersilia.utils.session.session_pid_from_name(session_name)[source]¶

Get the parent process ID encoded in a session name.

Parameters:: session_name (str) – A session name such as session_12345, or a path ending in one.
Returns:: The process ID, or None if the name is not a session name.
Return type:: int or None

ersilia.utils.session.set_write_permissions(directory)[source]¶

ersilia.utils.session.stop_containers_by_name(names)[source]¶

Stop and remove the named Docker containers, if any.

Silent no-op if docker is unavailable or a name is unknown.

ersilia.utils.spinner module¶

class ersilia.utils.spinner.Spinner(text='\nLoading...\n', spinner=None, color='cyan')[source]¶

Bases: object

A colorful and animated loader for terminal applications.

ersilia.utils.spinner.show_loader(text='\nLoading...\n', color='cyan')[source]¶

ersilia.utils.system module¶

class ersilia.utils.system.SystemChecker[source]¶

Bases: object

A class to check various system properties.

is_arm64()[source]¶: Check if the system architecture is ARM64.

is_github_action()[source]¶: Check if the code is running inside a GitHub Action.

is_inside_docker()[source]¶: Check if the code is running inside a Docker container.

get_country()[source]¶

Get the country code of the system based on the IP address.

Returns:: The country of the system and its 3-character code (Country, Code).
Return type:: tuple

is_arm64()[source]¶

Check if the system architecture is ARM64.

Returns:: True if the system architecture is ARM64, False otherwise.
Return type:: bool

is_github_action()[source]¶

Check if the code is running inside a GitHub Action.

Returns:: True if running inside a GitHub Action, False otherwise.
Return type:: bool

is_inside_docker()[source]¶

Check if the code is running inside a Docker container.

Returns:: True if running inside a Docker container, False otherwise.
Return type:: bool

ersilia.utils.system.is_inside_docker()[source]¶

Check if the current environment is inside a Docker container.

Returns:: True if inside a Docker container, False otherwise.
Return type:: bool

ersilia.utils.terminal module¶

ersilia.utils.terminal.is_quiet()[source]¶

Check if the current session is in quiet mode.

Returns:: True if the session is in quiet mode, False otherwise.
Return type:: bool

ersilia.utils.terminal.is_quoted_list(s: str) → bool[source]¶

ersilia.utils.terminal.print_serve_summary(model_id, slug, url, pid, srv, session_dir, apis, store_stat, enable_cache, tracking_enabled, tracking_use_case, version=None)[source]¶: Print a rich summary table for a served model.

ersilia.utils.terminal.raw_input_with_timeout(prompt, default_answer, timeout=5)[source]¶

Prompt the user for input with a timeout.

Parameters:

prompt (str) – The prompt message.
default_answer (str) – The default answer if the user does not respond within the timeout.
timeout (int, optional) – The timeout in seconds. Default is 5.

Returns:

The user’s input or the default answer if the timeout is reached.

Return type:

str

ersilia.utils.terminal.run_command(cmd, quiet=True)[source]¶: Run a shell command and return a named tuple with stdout, stderr, and return code. :param cmd: :type cmd: str or list The command to run. :param quiet: :type quiet: bool, optional Whether to run the command in quiet mode. Defaults to is_quiet().

ersilia.utils.terminal.run_command_check_output(cmd)[source]¶

Run a shell command and capture its output.

Parameters:: cmd (str or list) – The command to run.
Returns:: The output of the command.
Return type:: str

ersilia.utils.terminal.yes_no_input(prompt, default_answer, timeout=5)[source]¶

Prompt the user for a yes/no input with a timeout.

Parameters:

prompt (str) – The prompt message.
default_answer (str) – The default answer if the user does not respond within the timeout.
timeout (int, optional) – The timeout in seconds. Default is 5.

Returns:

True if the user’s input is ‘yes’, False otherwise.

Return type:

bool

ersilia.utils.tracking module¶

ersilia.utils.tracking.init_tracking_summary(model_id)[source]¶

ersilia.utils.tracking.update_tracking_summary(model_id, incoming_data)[source]¶

ersilia.utils.uninstall module¶

class ersilia.utils.uninstall.Uninstaller[source]¶

Bases: object

A class to manage the uninstallation of Ersilia and its dependencies.

uninstall()[source]¶: Main uninstallation method.

uninstall()[source]¶: Main uninstallation method.

ersilia.utils.upload module¶

class ersilia.utils.upload.OsfUploader(overwrite, username, password)[source]¶

Bases: object

A class to upload files to the Open Science Framework (OSF).

Parameters:

overwrite (bool) – Whether to overwrite existing files.
username (str) – The OSF username.
password (str) – The OSF password.

push(project_id, filename, destination)[source]¶: Upload a file to OSF.

push(project_id, filename, destination)[source]¶

Upload a file to OSF.

Parameters:

project_id (str) – The OSF project ID.
filename (str) – The path to the file to upload.
destination (str) – The destination path on OSF.

ersilia.utils.venv module¶

class ersilia.utils.venv.SimpleVenv(root)[source]¶

Bases: ErsiliaBase

A class to manage virtual environments for Ersilia.

Parameters:: root (str) – The root directory for the virtual environments.

create(environment)[source]¶: Create a new virtual environment.

delete(environment)[source]¶: Delete a virtual environment.

run_commandlines(environment, commandlines)[source]¶: Run command lines in a virtual environment.

create(environment)[source]¶

Create a new virtual environment.

Parameters:: environment (str) – The name of the virtual environment.
Return type:: None
Raises:: VirtualEnvironmentSetupError – If the virtual environment setup fails.

delete(environment)[source]¶

Delete a virtual environment.

Parameters:: environment (str) – The name of the virtual environment.
Return type:: None

exists(environment)[source]¶

Check if a virtual environment exists.

Parameters:: environment (str) – The name of the virtual environment.
Returns:: True if the virtual environment exists, False otherwise.
Return type:: bool

run_commandlines(environment, commandlines)[source]¶

Run command lines in a virtual environment.

Parameters:

environment (str) – The name of the virtual environment.
commandlines (str) – The command lines to run.

Return type:

None

Raises:

ModelPackageInstallError – If the command lines execution fails.

ersilia.utils.versioning module¶

class ersilia.utils.versioning.Versioner(config_json=None)[source]¶

Bases: ErsiliaBase

A class to manage versioning information for Ersilia.

python_version(py_format=False)[source]¶: Get the current Python version.

ersilia_version()[source]¶: Get the current Ersilia version.

ersilia_version_with_py()[source]¶: Get the current Ersilia version with Python version.

ersilia_version_from_path(path)[source]¶: Get the Ersilia version from a given path.

server_docker_name(tag=None, as_tuple=False)[source]¶: Get the server Docker image name.

base_conda_name(org, tag)[source]¶: Get the base Conda environment name.

reformat_py(v)[source]¶: Reformat a Python version string.

base_conda_name(org, tag)[source]¶

Get the base Conda environment name.

Parameters:

org (str) – The organization name.
tag (str) – The tag for the Conda environment.

Returns:

The base Conda environment name.

Return type:

str

ersilia_version()[source]¶

Get the current Ersilia version.

Returns:: The current Ersilia version.
Return type:: str

ersilia_version_from_path(path)[source]¶

Get the Ersilia version from a given path.

Parameters:: path (str) – The path to the Ersilia installation.
Returns:: The Ersilia version if found, otherwise None.
Return type:: str or None

ersilia_version_with_py()[source]¶

Get the current Ersilia version with Python version.

Returns:: The current Ersilia version with Python version.
Return type:: str

python_version(py_format=False)[source]¶

Get the current Python version.

Parameters:: py_format (bool, optional) – Whether to return the version in Python format (e.g., ‘py38’). Default is False.
Returns:: The current Python version.
Return type:: str

static reformat_py(v)[source]¶

Reformat a Python version string.

Parameters:: v (str) – The Python version string.
Returns:: The reformatted Python version string.
Return type:: str
Raises:: Exception – If the version string is too short.

server_docker_name(tag=None, as_tuple=False)[source]¶

Get the server Docker image name.

Parameters:

tag (str, optional) – The tag for the Docker image. Default is None.
as_tuple (bool, optional) – Whether to return the name as a tuple. Default is False.

Returns:

The server Docker image name, or a tuple of (org, img, tag) if as_tuple is True.

Return type:

str or tuple

ersilia.utils.zip module¶

class ersilia.utils.zip.Zipper(remove)[source]¶

Bases: object

A class to handle zipping and unzipping files.

Parameters:: remove (bool) – Whether to remove the original files after zipping/unzipping.

unzip(file, destination)[source]¶: Unzip a file to the specified destination.

zip(dir_name, file)[source]¶: Zip a directory into a file.

unzip(file, destination)[source]¶

Unzip a file to the specified destination.

Parameters:

file (str) – The path to the zip file.
destination (str) – The destination directory.

Return type:

None

zip(dir_name, file)[source]¶

Zip a directory into a file.

Parameters:

dir_name (str) – The path to the directory to zip.
file (str) – The path to the output zip file.

Return type:

None

Module contents¶

ersilia.utils.tmp_pid_file(model_id)[source]¶: Gets filename to store process ID