Command Line

Pipeline CLI for running multiple qm_atlas commands in sequence.

Allows combining multiple CLI steps into a single pipeline run, configured via a YAML file. Paths for input files and output (results directory) can be provided as CLI arguments instead of being hard-coded in the config.

Pipeline Behavior:

• When a step has submit: true in its submission_config, the pipeline automatically forces wait: true so that subsequent steps only begin after the submitted jobs have finished.

• The pipeline itself can be submitted to the cluster using the top-level submission_config.submit: true option. In that case, the resolved config is dumped to a YAML file and the pipeline is re-invoked on the cluster with qm-atlas pipeline --config <resolved_config>.yaml.

Example YAML config:

steps:
  - command: read_input

  - command: fast_conformers
    submission_config:
      submit: true

  - command: add_reference_conformers

  - command: optimize_reference_conformers
    submission_config:
      submit: true

  - command: conformer_properties
    submission_config:
      submit: true
    conformer_property_options:
      calculation_tasks:
        - backend: "turbomole_single_point"

  - command: cosmotherm_properties
    submission_config:
      submit: true

Usage:

qm-atlas pipeline --config pipeline.yaml \
    --input_file ./molecules.sdf \
    --results_directory ./results

# With separate reference conformer input files:
qm-atlas pipeline --config pipeline.yaml \
    --input_file ./molecules.sdf \
    --reference_input_file ./ref_conformers.sdf \
    --results_directory ./results

# Submit the pipeline itself to the cluster:
qm-atlas pipeline --config pipeline.yaml \
    --input_file ./molecules.sdf \
    --results_directory ./results \
    --submission_config.submit true

class qm_atlas.command_line.pipeline.PipelineSubmissionConfig(**data)

UGE submission settings specific to the Pipeline CLI.

Parameters:

• submit (bool)

• engine (str)

• cores_per_task (int)

• mem_per_core (int)

• jobs_per_task (int)

• max_total_num_cores (int)

• max_time (str)

• name (str)

• user_email (str | None)

• wait (bool)

cores_per_task: int

max_time: str

name: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class qm_atlas.command_line.pipeline.PipelineConfig(**data)

Complete configuration for a pipeline run.

Parameters:

• results_directory (Path | None)

• input_file (list[Path] | None)

• reference_input_file (list[Path] | None)

• software_config (Path | None)

• submission_config (SubmissionConfig)

• verbose (bool)

• steps (list[Any])

results_directory: Path | None

input_file: list[Path] | None

reference_input_file: list[Path] | None

software_config: Path | None

submission_config: SubmissionConfig

verbose: bool

steps: list[Any]

classmethod validate_results_directory_field(v)

Validate results directory using arg_utils validation.

For read_input, the results directory is created if it doesn’t exist. Pydantic handles conversion to Path.

Return type:

Path | None

Parameters:

v (Path | None)

classmethod validate_input_file(v)

Resolve paths and validate existence and file type.

Return type:

list[Path] | None

Parameters:

v (list[Path] | None)

resolve_steps()

Inject paths, force wait, and deserialize steps to typed config objects.

This runs after all field validators, so self.results_directory etc. are already resolved. For each raw step dict the method:

1. Injects top-level pipeline paths (results_directory, input_file, …)

2. Forces wait=true on steps with submit=true

3. Deserializes the dict into the appropriate config class via STEP_REGISTRY, following the same pattern used by ConformerPropertyOptions for calculation_tasks.

Return type:

PipelineConfig

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

qm_atlas.command_line.pipeline.load_config(argv=None)

Load and validate configuration using jsonargparse + Pydantic.

Supports: - YAML config file via –config - Command-line argument overrides for paths and submission settings

Parameters:

argv (list[str] | None) – Command-line arguments. If None, uses sys.argv[1:].

Returns:

Validated configuration object with paths resolved.

Return type:

PipelineConfig

Raises:

SystemExit – On parse errors or validation failures.

qm_atlas.command_line.pipeline.run_pipeline(config)

Execute all steps in the pipeline sequentially.

Parameters:

config (PipelineConfig) – Validated pipeline configuration with paths and wait flags already resolved.

Return type:

None

qm_atlas.command_line.pipeline.submit(config)

Submit the pipeline itself to the cluster.

Dumps the fully resolved config to a YAML file, then submits qm-atlas pipeline --config <resolved_config>.yaml to the cluster.

Return type:

None

Parameters:

config (PipelineConfig)

qm_atlas.command_line.pipeline.main(argv=None, config=None, log_file=None)

Main entry point for the pipeline CLI.

Parameters:

• argv (list[str] | None) – List of command-line arguments (for testing). If None, uses sys.argv[1:].

• config (PipelineConfig | None) – Optional PipelineConfig object. If provided, overrides command-line arguments.

• log_file (Path | None) – Optional file to append logs to.

Return type:

None

Utilities for command-line argument handling and validation.

class qm_atlas.command_line.base_config.CalculationInput(**data)

Configuration for accessing existing results directories or compound directories.

Requirements: - results_directory is always required - Exactly one of compound_directories or task_control_file can be provided (mutually exclusive) - If task_id is provided, task_control_file must also be provided

Parameters:

• results_directory (Path | None)

• compound_directories (list[Path] | None)

• task_control_file (Path | None)

• task_id (int | None)

results_directory: Path | None

compound_directories: list[Path] | None

task_control_file: Path | None

task_id: int | None

classmethod validate_results_directory_field(v)

Validate results directory using arg_utils validation.

For read_input, the results directory is created if it doesn’t exist. Pydantic handles conversion to Path.

Return type:

Path | None

Parameters:

v (Path | None)

classmethod validate_compound_directories_field(v)

Convert compound_directories to list of Paths and validate.

Ensures all paths exist and are directories.

Return type:

list[Path] | None

Parameters:

v (list[Path | str] | str | Path)

classmethod validate_task_control_file(input_path)

Validate task control file if explicitly provided.

Return type:

Path | None

Parameters:

input_path (Path | None)

validate_mutual_exclusivity()

Validate that compound_directories and task_control_file are mutually exclusive.

Return type:

CalculationInput

validate_task_id_requires_task_control_file()

Validate that task_id is only provided with task_control_file.

Return type:

CalculationInput

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class qm_atlas.command_line.base_config.SubmissionConfig(**data)

Configuration for UGE (Univa Grid Engine) job submission.

This class is designed to be used across multiple CLIs with different defaults. Use the with_defaults() class method to create instances with CLI-specific defaults, or provide defaults when creating the parent config model.

Parameters:

• submit (bool)

• engine (str)

• cores_per_task (int)

• mem_per_core (int)

• jobs_per_task (int)

• max_total_num_cores (int)

• max_time (str)

• name (str)

• user_email (str | None)

• wait (bool)

submit: bool

engine: str

cores_per_task: int

mem_per_core: int

jobs_per_task: int

max_total_num_cores: int

max_time: str

name: str

user_email: str | None

wait: bool

validate_cores_configuration()

Ensure max_total_num_cores is greater than or equal to cores_per_task.

Return type:

SubmissionConfig

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class qm_atlas.command_line.base_config.BaseInterfaceConfig(**data)

Base configuration with common input and submission settings.

Parameters:

• input (CalculationInput)

• submission_config (SubmissionConfig)

• software_config (Path | None)

• n_cores (int | None)

• verbose (bool)

• force_rerun (bool)

input: CalculationInput

submission_config: SubmissionConfig

software_config: Path | None

n_cores: int | None

verbose: bool

force_rerun: bool

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

qm_atlas.command_line.base_config.apply_software_config(software_config_path)

Apply a software config file by setting the env var and reloading SOFTWARE_CONFIG.

This should be called early in each CLI main() function after config loading. If software_config_path is provided, validates the file exists, sets the qm_atlas_SOFTWARE_CONFIG_FILE environment variable, and reloads the global SOFTWARE_CONFIG singleton.

Parameters:

software_config_path (Path | None) – Path to software config YAML file, or None to skip.

Return type:

None

qm_atlas.command_line.base_config.setup_logging(verbose=False, log_file=None)

Configure logging based on config verbosity level.

Parameters:

• verbose (bool) – Enable verbose logging

• log_file (Path | None) – Optional file to append logs to.

Return type:

None

qm_atlas.command_line.base_config.validate_results_dir(results_dir, create_if_missing=True)

Validate and prepare the results directory.

Parameters:

• results_dir (Path) – Path to the results directory

• create_if_missing (bool) – If True, create the directory if it doesn’t exist. If False, raise an error if the directory doesn’t exist.

Raises:

ValueError – If parent directory doesn’t exist or results_dir validation fails

qm_atlas.command_line.base_config.get_input_worker_paths(cpd_dir)

Searches for input sdf files under cpd_dir and returns a list of their paths.

Parameters:

cpd_dir (CpdDir) – The compound directory to search under.

Returns:

A list of WorkerPath objects containing the compound directory, the input sdf file path, and the associated log file path.

Return type:

list[tuple[CpdDir, Path, Path]]

qm_atlas.command_line.base_config.get_reference_worker_paths(cpd_dir)

Searches for reference sdf files under cpd_dir and returns a list of their paths.

Parameters:

cpd_dir (CpdDir) – The compound directory to search under.

Returns:

A list of WorkerPath objects containing the compound directory, the reference sdf file path, and the associated log file path.

Return type:

list[tuple[CpdDir, Path, Path]]

qm_atlas.command_line.base_config.get_conformer_property_worker_paths(cpd_dir)

Searches for result sdf files under cpd_dir and returns a list of their paths.

Parameters:

cpd_dir (CpdDir) – The compound directory to search under.

Returns:

A list of WorkerPath objects containing the compound directory, the result sdf file path, and the associated log file path.

Return type:

list[tuple[CpdDir, Path, Path]]

qm_atlas.command_line.base_config.get_cosmo_property_worker_paths(cpd_dir)

Searches for conformer COSMO files under cpd_dir and returns a list of their paths.

Parameters:

cpd_dir (CpdDir) – The compound directory to search under.

Returns:

A list of WorkerPath objects containing the compound directory, the COSMO file path, and the associated log file path.

Return type:

list[tuple[CpdDir, Path, Path]]

qm_atlas.command_line.base_config.get_solvent_screening_worker_paths(cpd_dir)

Returns a single worker path per compound directory for solvent screening.

Solvent screening collects all neutral-state COSMO conformers in one call, so only one entry per compound directory is needed.

Parameters:

cpd_dir (CpdDir) – The compound directory to search under.

Returns:

A list containing at most one WorkerPath for the compound.

Return type:

list[tuple[CpdDir, Path, Path]]

qm_atlas.command_line.base_config.missing_reference_optimizations(cpd_dir, sdf_file, requested_configs)

Return the requested optimization configs with no matching existing result.

This is the full picture behind worker_path_completed() for reference optimization: an empty list means every requested setting is already present, a non-empty list is exactly the settings that still need to run (e.g. a newly added force constant).

Parameters:

• cpd_dir (CpdDir) – The owning compound directory.

• sdf_file (Path) – The reference input file to inspect.

• requested_configs (list[list[XtbOptions | JobexOptions | XtbTurbomoleOptions]]) – The requested optimization settings.

Returns:

The subset of requested settings still to run.

Return type:

list[list[XtbOptions | JobexOptions | XtbTurbomoleOptions]]

qm_atlas.command_line.base_config.worker_path_completed(cpd_dir, sdf_file, workflow_type, **kwargs)

Return whether sdf_file already has results for workflow_type.

Workflows without a registered completeness check are never considered done.

Parameters:

• cpd_dir (CpdDir) – The owning compound directory.

• sdf_file (Path) – The worker-path input file to check.

• workflow_type (str) – The workflow whose results define completeness.

• **kwargs – Additional keyword arguments passed to the workflow-specific checker.

Returns:

True if results already exist and the worker path can be skipped.

Return type:

bool

qm_atlas.command_line.base_config.extract_worker_paths(calculation_input, workflow_type, skip_completed=False, **kwargs)

Extract worker paths based on input configuration.

Parameters:

• calculation_input (CalculationInput) – The resolved input configuration.

• workflow_type (str) – The workflow whose worker paths are being built.

• skip_completed (bool) – When True, drop worker paths whose results already exist (used to resume a partially completed run). Only applied to the compound-directory and results-directory branches; task-control slices are assumed to have been filtered when the task array was created.

• **kwargs – Additional keyword arguments passed to the workflow-specific checker.

Returns:

The worker paths to process.

Return type:

list[tuple[CpdDir, Path, Path]]

qm_atlas.command_line.base_config.extract_cpd_dirs(calculation_input)

Return the compound directories referenced by calculation_input.

Unlike extract_worker_paths(), this resolves whole compound directories rather than per-conformer worker paths, which is what post-processing commands (collect, averaging, Boltzmann weights) operate on.

Parameters:

calculation_input (CalculationInput) – The resolved input configuration.

Returns:

The unique compound directories to process.

Return type:

list[CpdDir]