Looper Python API

The looper Python API provides classes for managing pipeline submissions and compute configurations.

Project

The main class for working with looper projects. Extends peppy's Project with pipeline submission capabilities.

Project

class Project(cfg: str | None = None, amendments=None, divcfg_path: str | None = None, **kwargs)

Bases: peppyProject

Looper-specific Project.

Parameters:

• cfg (str): Path to configuration file with data from which Project is to be built. (default: None)
• amendments (Iterable[str]): Name indicating amendment to use, optional. (default: None)
• divcfg_path (str): Path to an environment configuration YAML file specifying compute settings. (default: None)

cli_pifaces

@property
def cli_pifaces -> list[str] | None

Collection of pipeline interface sources specified in object constructor.

Returns:

• list[str] | None: list[str]: Collection of pipeline interface sources.

get_sample_piface

def get_sample_piface(sample_name: str) -> list | None

Get a list of pipeline interfaces associated with the specified sample.

Note that only valid pipeline interfaces will show up in the result (ones that exist on disk/remotely and validate successfully against the schema).

Parameters:

• sample_name (str): Name of the sample to retrieve list of pipeline interfaces for.

Returns:

• list | None: list[looper.PipelineInterface]: Collection of valid pipeline interfaces associated with selected sample.

get_schemas

@staticmethod
def get_schemas(pifaces, schema_key: str = INPUT_SCHEMA_KEY) -> list[str]

Get the list of unique schema paths for a list of pipeline interfaces.

Parameters:

• pifaces (str | Iterable[str]): Pipeline interfaces to search schemas for.
• schema_key (str): Where to look for schemas in the piface. (default: INPUT_SCHEMA_KEY)

Returns:

• list[str]: Iterable[str]: Unique list of schema file paths.

make_project_dirs

def make_project_dirs() -> None

Create project directory structure if it doesn't exist.

output_dir

@property
def output_dir -> str

Output directory for the project, specified in object constructor.

Returns:

• str: Path to the output directory.

pipeline_interfaces

@property
def pipeline_interfaces -> list

Flat list of all valid interface objects associated with this Project.

Note that only valid pipeline interfaces will show up in the result (ones that exist on disk/remotely and validate successfully against the schema).

Returns:

• list: list[looper.PipelineInterface]: List of pipeline interfaces.

populate_pipeline_outputs

def populate_pipeline_outputs() -> None

Populate project and sample output attributes based on output schemas that pipeline interfaces point to.

results_folder

@property
def results_folder -> str

Path to the results folder for the project.

Returns:

• str: Path to the results folder in the output folder.

selected_compute_package

@property
def selected_compute_package -> str | None

Compute package name specified in object constructor.

Returns:

• str (str | None): Compute package name.

set_sample_piface

def set_sample_piface(sample_piface: list[str] | str) -> NoReturn

Add sample pipeline interfaces variable to object.

Parameters:

• sample_piface (list | str): Sample pipeline interface.

submission_folder

@property
def submission_folder -> str

Path to the submission folder for the project.

Returns:

• str: Path to the submission in the output folder.

PipelineInterface

Parses and holds information from a pipeline interface YAML file, including resource specifications and command templates.

PipelineInterface

class PipelineInterface(config: str | Mapping, pipeline_type: str | None = None)

Bases: YAMLConfigManager

This class parses, holds, and returns information for a yaml file that specifies how to interact with each individual pipeline.

This includes both resources to request for cluster job submission, as well as arguments to be passed from the sample annotation metadata to the pipeline.

Parameters:

• config (str | Mapping): Path to file from which to parse configuration data, or pre-parsed configuration data.
• pipeline_type (str): Type of the pipeline, must be either 'sample' or 'project'. (default: None)

choose_resource_package

def choose_resource_package(namespaces: dict, file_size: float) -> dict

Select resource bundle for given input file size to given pipeline.

Parameters:

• file_size (float): Size of input data (in gigabytes).
• namespaces (Mapping[Mapping[str]]): Namespaced variables to pass as a context for fluid attributes command rendering.

Returns:

• MutableMapping (dict): Resource bundle appropriate for given pipeline, for given input file size.

Raises:

• ValueError: If indicated file size is negative, or if the file size value specified for any resource package is negative.
• InvalidResourceSpecificationException: If no default resource package specification is provided.

get_pipeline_schemas

def get_pipeline_schemas(schema_key: str = INPUT_SCHEMA_KEY) -> str | None

Get path to the pipeline schema.

Parameters:

• schema_key (str): Where to look for schemas in the pipeline iface. (default: INPUT_SCHEMA_KEY)

Returns:

• str (str | None): Absolute path to the pipeline schema file.

pipeline_name

@property
def pipeline_name -> str

render_var_templates

def render_var_templates(namespaces: dict) -> dict

Render path templates under 'var_templates' in this pipeline interface.

Parameters:

• namespaces (dict): Namespaces to use for rendering.

SubmissionConductor

Collects and submits pipeline jobs. Manages job pooling based on file size or command count limits.

SubmissionConductor

class SubmissionConductor(pipeline_interface: PipelineInterface, prj: Project, delay: float = 0, extra_args: str | None = None, extra_args_override: str | None = None, ignore_flags: bool = False, compute_variables: dict | None = None, max_cmds: int | None = None, max_size: int | float | None = None, max_jobs: int | float | None = None, automatic: bool = True, collate: bool = False)

Collects and then submits pipeline jobs.

This class holds a 'pool' of commands to submit as a single cluster job. Eager to submit a job, each instance's collection of commands expands until it reaches the 'pool' has been filled, and it's therefore time to submit the job. The pool fills as soon as a fill criteria has been reached, which can be either total input file size or the number of individual commands.

Create a job submission manager.

The most critical inputs are the pipeline interface and the pipeline key, which together determine which provide critical pipeline information like resource allocation packages and which pipeline will be overseen by this instance, respectively.

Parameters:

• pipeline_interface (PipelineInterface): Collection of important data for one or more pipelines, like resource allocation packages and option/argument specifications.
• prj (Project): Project with which each sample being considered is associated (what generated each sample).
• delay (float): Time (in seconds) to wait before submitting a job once it's ready. (default: 0)
• extra_args (str): String to pass to each job generated, for example additional pipeline arguments. (default: None)
• extra_args_override (str): String to pass to each job generated, for example additional pipeline arguments. This deactivates the 'extra' functionality that appends strings defined in Sample.command_extra and Project.looper.command_extra to the command template. (default: None)
• ignore_flags (bool): Whether to ignore flag files present in the sample folder for each sample considered for submission. (default: False)
• compute_variables (dict[str]): A dict with variables that will be made available to the compute package. For example, this should include the name of the cluster partition to which job or jobs will be submitted. (default: None)
• max_cmds (int | None): Upper bound on number of commands to include in a single job script. (default: None)
• max_size (int | float | None): Upper bound on total file size of inputs used by the commands lumped into single job script. (default: None)
• max_jobs (int | float | None): Upper bound on total number of jobs to group samples for submission. (default: None)
• automatic (bool): Whether the submission should be automatic once the pool reaches capacity. (default: True)
• collate (bool): Whether a collate job is to be submitted (runs on the project level, rather that on the sample level). (default: False)

add_sample

def add_sample(sample, rerun: bool = False) -> list

Add a sample for submission to this conductor.

Parameters:

• sample (peppy.Sample): Sample to be included with this conductor's currently growing collection of command submissions.
• rerun (bool): Whether the given sample is being rerun rather than run for the first time. (default: False)

Returns:

• list: List of skip reasons if sample was not added.

Raises:

• TypeError: If sample subtype is provided but does not extend the base Sample class.

failed_samples

@property
def failed_samples -> list[str]

is_project_submittable

def is_project_submittable(force: bool = False) -> bool

Check whether the current project has been already submitted.

Parameters:

• force (bool): Whether to force the project submission (ignore status/flags). (default: False)

Returns:

• bool: True if the project is submittable, False otherwise.

num_cmd_submissions

@property
def num_cmd_submissions -> int

Return the number of commands that this conductor has submitted.

Returns:

• int: Number of commands submitted so far.

num_job_submissions

@property
def num_job_submissions -> int

Return the number of jobs that this conductor has submitted.

Returns:

• int: Number of jobs submitted so far.

submit

def submit(force: bool = False) -> bool

Submit one or more commands as a job.

This call will submit the commands corresponding to the current pool of samples if and only if the argument to 'force' evaluates to a true value, or the pool of samples is full.

Parameters:

• force (bool): Whether submission should be done/simulated even if this conductor's pool isn't full. (default: False)

Returns:

• bool: Whether a job was submitted (or would've been if not for dry run).

write_script

def write_script(pool: list, size: float) -> str

Create the script for job submission.

Parameters:

• pool (Iterable[peppy.Sample]): Collection of sample instances.
• size (float): Cumulative size of the given pool.

Returns:

• str: Path to the job submission script created.

ComputingConfiguration

Manages compute environment settings from divvy configuration files. Handles resource packages and submission templates.

ComputingConfiguration

class ComputingConfiguration(entries=None, wait_max=None, strict_ro_locks: bool = False)

Bases: YAMLConfigManager

Represents computing configuration objects.

The ComputingConfiguration class provides a computing configuration object that is an in memory representation of a divvy computing configuration file. This object has various functions to allow a user to activate, modify, and retrieve computing configuration files, and use these values to populate job submission script templates.

Parameters:

• entries (str | Iterable[str, object] | Mapping[str, object]): Config collection of key-value pairs. (default: None)

activate_package

def activate_package(package_name: str) -> bool

Activates a compute package.

This copies the computing attributes from the configuration file into the compute attribute, where the class stores current compute settings.

Parameters:

• package_name (str): Name for non-resource compute bundle, the name of a subsection in an environment configuration file.

Returns:

• bool: Success flag for attempt to establish compute settings.

clean_start

def clean_start(package_name: str) -> bool

Clear current active settings and then activate the given package.

Parameters:

• package_name (str): Name of the resource package to activate.

Returns:

• bool: Success flag.

default_config_file

@property
def default_config_file -> str

Path to default compute environment settings file.

Returns:

• str: Path to default compute settings file.

get_active_package

def get_active_package() -> YAMLConfigManager

Returns settings for the currently active compute package.

Returns:

• YAMLConfigManager: Data defining the active compute package.

get_adapters

def get_adapters() -> YAMLConfigManager

Get current adapters, if defined.

Adapters are sourced from the 'adapters' section in the root of the divvy configuration file and updated with an active compute package-specific set of adapters, if any defined in 'adapters' section under currently active compute package.

Returns:

• YAMLConfigManager: Current adapters mapping.

list_compute_packages

def list_compute_packages() -> set[str]

Returns a list of available compute packages.

Returns:

• set[str]: set[str]: Names of available compute packages.

reset_active_settings

def reset_active_settings() -> bool

Clear out current compute settings.

Returns:

• bool: Success flag.

template

def template() -> str

Get the currently active submission template.

Returns:

• str: Submission script content template for current state.

templates_folder

@property
def templates_folder -> str

Path to folder with default submission templates.

Returns:

• str: Path to folder with default submission templates.

update_packages

def update_packages(config_file: str) -> bool

Parse data from divvy configuration file.

Given a divvy configuration file, this function will update (not overwrite) existing compute packages with existing values. It does not affect any currently active settings.

Parameters:

• config_file (str): Path to file with new divvy configuration data.

write_script

def write_script(output_path: str, extra_vars: list | None = None) -> str

Given currently active settings, populate the active template to write a submission script.

Additionally use the current adapters to adjust the select of the provided variables.

Parameters:

• output_path (str): Path to file to write as submission script.
• extra_vars (Iterable[Mapping]): A list of Dict objects with key-value pairs with which to populate template fields. These will override any values in the currently active compute package. (default: None)

Returns:

• str: Path to the submission script file.

Utility Functions

select_divvy_config

select_divvy_config

def select_divvy_config(filepath: str | None) -> str

Selects the divvy config file path to load.

This uses a priority ordering to first choose a config file path if it's given, but if not, then look in a priority list of environment variables and choose the first available file path to return. If none of these options succeed, the default config path will be returned.

Parameters:

• filepath (str | NoneType): Direct file path specification.

Returns:

• str: Path to the config file to read.