API Documentation

Public API

Public API for BATTER.

This module collects the stable entry points intended for external consumption. They fall into four broad categories:

• Configuration helpers – load and dump RunConfig / SimulationConfig objects.

• Execution – orchestrate complete workflows from a YAML definition.

• Portable results – inspect and copy artifacts produced by a run.

• Utilities – clone the state of an execution for reproducibility.

Typical usage

Run a workflow from a top-level YAML:

from batter.api import run_from_yaml
run_from_yaml("examples/mabfe.yaml")

Inspect FE records stored in a work directory:

from batter.api import list_fe_runs, load_fe_run
runs = list_fe_runs("work/adrb2")
latest = runs.iloc[-1]["run_id"]
# pass ``ligand`` when the run contains more than one ligand
record = load_fe_run("work/adrb2", latest, ligand="LIG1")

Run FE analysis on an existing execution:

from batter.api import run_analysis_from_execution
run_analysis_from_execution("work/adrb2", latest, ligand="LIG1")

For more examples, refer to docs/getting_started.rst and the tutorials.

class batter.api.ArtifactStore(root: Path | str, manifest_name: str = 'manifest.json')

Bases: object

Portable store with a relocatable root and JSON manifest.

Parameters:

• root (path-like) – Store root directory (e.g., a run’s work directory).

• manifest_name (str) – File name for the manifest JSON under root (default: “manifest.json”).

Examples

>>> store = ArtifactStore("work/at1r_aai")
>>> p = store.put_file(Path("results.txt"), name="fe/latest", dst_rel=Path("fe/results.txt"))
>>> store.save_manifest()
>>> # move directory to a new cluster...
>>> store2 = ArtifactStore("new_root/at1r_aai"); store2.load_manifest()
>>> store2.path("fe/latest")
new_root/at1r_aai/fe/results.txt

list_artifacts(*, prefix: str | None = None, kind: Literal['file', 'dir', None] = None) → List[Artifact]

Inspect manifest entries, optionally filtering by name or kind.

Parameters:

• prefix (str, optional) – When provided, only artifacts whose logical name starts with prefix are returned.

• kind ({‘file’, ‘dir’, None}, optional) – Restrict results to files or directories. None (default) returns both.

Returns:

Matching artifacts in alphabetical order.

Return type:

list[Artifact]

load_manifest() → None

Load the manifest JSON from root.

path(name: str) → Path

Resolve an artifact name to an absolute path under the current root.

put_dir(src_dir: Path, name: str, dst_rel: Path | None = None, overwrite_manifest_entry: bool = False) → Path

Copy a directory under the store and record it in the manifest.

Notes

• No per-file hashing; use put_file() for critical files.

put_file(src: Path, name: str, dst_rel: Path | None = None, overwrite_manifest_entry: bool = False) → Path

Copy a file under the store and record it in the manifest.

Parameters:

• src (path-like) – Source file path (must exist and be a file).

• name (str) – Logical artifact name to register under.

• dst_rel (path-like, optional) – Relative destination path. Defaults to name.replace('/', '_').

• overwrite_manifest_entry (bool) – If True, allows replacing an existing manifest entry with the same name.

Returns:

Absolute destination path.

Return type:

pathlib.Path

rebase(new_root: Path | str) → ArtifactStore

Create a new store view with the same manifest but a different root.

Parameters:

new_root (path-like) – Target root directory.

Returns:

New store pointing to new_root.

Return type:

ArtifactStore

save_manifest() → Path

Write the manifest JSON under root (atomic).

class batter.api.FERecord(*, run_id: str, ligand: str, mol_name: str, system_name: str, fe_type: str, temperature: float, method: Literal['mbar', 'ti']='mbar', total_dG: float, total_se: float = 0.0, components: List[str] = <factory>, created_at: str = <factory>, windows: List[WindowResult] = <factory>, canonical_smiles: str | None = None, original_name: str | None = None, original_path: str | None = None, protocol: str = 'abfe', analysis_start_step: int | None = None, n_bootstraps: int | None = None, status: Literal['success', 'failed', 'unbound']='success')

Bases: BaseModel

A full FE result bundle (portable, versioned).

Parameters:

• run_id (str) – Unique run identifier.

• ligand (str) – Ligand identifier.

• mol_name (str) – Molecule resname.

• system_name (str) – Logical system name.

• fe_type (str) – Protocol type (e.g., ‘uno_rest’, ‘asfe’).

• temperature (float) – Simulation temperature (K).

• method ({“mbar”,”ti”}) – Integration method.

• total_dG (float) – Total free energy (kcal/mol).

• total_se (float) – Standard error (kcal/mol).

• components (list[str]) – Active components in this run.

• created_at (str) – ISO-8601 timestamp (UTC, Z-suffix).

• windows (list[WindowResult]) – Per-window results.

• canonical_smiles (str, optional) – Canonicalised ligand SMILES captured during parameterization.

• original_name (str, optional) – Original ligand identifier or title when known.

• original_path (str, optional) – Source path of the ligand before staging.

• protocol (str) – Logical protocol used to generate the result (e.g., "abfe").

• analysis_start_step (int, optional) – First production step included in analysis.

• n_bootstraps (int, optional) – Number of MBAR bootstrap resamples used during analysis.

• status ({“success”,”failed”,”unbound”}) – Final status recorded for the ligand.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

analysis_start_step: int | None

canonical_smiles: str | None

components: List[str]

created_at: str

fe_type: str

ligand: str

method: Literal['mbar', 'ti']

model_config = {}

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

mol_name: str

n_bootstraps: int | None

original_name: str | None

original_path: str | None

protocol: str

run_id: str

status: Literal['success', 'failed', 'unbound']

system_name: str

temperature: float

total_dG: float

total_se: float

windows: List[WindowResult]

class batter.api.FEResultsRepository(store: ArtifactStore)

Bases: object

index() → DataFrame

ligand_dir(run_id: str, ligand: str) → Path

load(run_id: str, ligand: str) → FERecord

record_failure(run_id: str, ligand: str, system_name: str, temperature: float, *, status: Literal['failed', 'unbound'], reason: str | None = None, canonical_smiles: str | None = None, original_name: str | None = None, original_path: str | None = None, protocol: str = 'abfe', analysis_start_step: int | None = None, n_bootstraps: int | None = None) → None

save(rec: FERecord, copy_from: Path | None = None) → None

class batter.api.RunConfig(*, version: int = 1, protocol: Literal['abfe', 'rbfe', 'asfe', 'md']='abfe', backend: Literal['local', 'slurm']='local', create: CreateArgs, fe_sim: Dict[str, ~typing.Any] | ~batter.config.run.FESimArgs | ~batter.config.run.MDSimArgs=<factory>, run: RunSection, rbfe: RBFENetworkArgs | None = None)

Bases: BaseModel

Top-level YAML config.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

backend: Literal['local', 'slurm']

create: CreateArgs

fe_sim: Dict[str, Any] | FESimArgs | MDSimArgs

classmethod load(path: Path | str) → RunConfig

Load and validate a run configuration from disk.

Parameters:

path (str or pathlib.Path) – Location of the YAML file to parse.

Returns:

Fully validated configuration object.

Return type:

RunConfig

model_config = {'extra': 'forbid'}

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

classmethod model_validate_yaml(yaml_text: str) → RunConfig

Validate a run configuration from an in-memory YAML string.

Parameters:

yaml_text (str) – Raw YAML content describing the run configuration.

Returns:

Validated configuration model.

Return type:

RunConfig

protocol: Literal['abfe', 'rbfe', 'asfe', 'md']

rbfe: RBFENetworkArgs | None

resolved_sim_config() → SimulationConfig

Build the effective simulation configuration for this run.

Returns:

Simulation parameters derived from create and fe_sim sections.

Return type:

SimulationConfig

run: RunSection

version: int

with_base_dir(base_dir: Path) → RunConfig

Return a copy with relative paths resolved against base_dir.

class batter.api.SimSystem(name: str, root: Path, topology: Path | None = None, coordinates: Path | None = None, protein: Path | None = None, ligands: Path, ...]=(), lipid_mol: Tuple[str, ...]=(), other_mol: Tuple[str, ...]=(), anchors: Tuple[str, ...]=(), meta: SystemMeta = <factory>)

Bases: object

Immutable descriptor of a simulation system and its on-disk artifacts.

Parameters:

• name (str) – Logical system name (e.g., "AT1R_AAI").

• root (pathlib.Path) – Working directory where artifacts live. This directory is considered relocatable; other modules should store relative paths when possible.

• topology (pathlib.Path, optional) – Path to an explicit topology (e.g., AMBER PRMTOP). May be None if the builder generates it later.

• coordinates (pathlib.Path, optional) – Coordinates or restart file (e.g., RST7/INPCRD).

• protein (pathlib.Path, optional) – Input protein structure file (PDB/mmCIF).

• ligands (tuple[pathlib.Path, …]) – One or more ligand structure files.

• lipid_mol (tuple[str, …]) – Lipid names present in the system (e.g., ("POPC",)).

• other_mol (tuple[str, …]) – Other cofactor present in the system``).

• anchors (tuple[str, …]) – Anchor atoms in the form "RESID@ATOM" (e.g., "85@CA").

• meta (SystemMeta) – Free-form metadata bundle for provenance (e.g., software versions).

anchors: Tuple[str, ...]

coordinates: Path | None

ligands: Tuple[Path, ...]

lipid_mol: Tuple[str, ...]

meta: SystemMeta

name: str

other_mol: Tuple[str, ...]

path(*parts: str | Path) → Path

Join root with the provided path segments.

Parameters:

*parts (str or Path) – Relative path components appended in order.

Returns:

Absolute path pointing inside root.

Return type:

pathlib.Path

protein: Path | None

root: Path

topology: Path | None

with_artifacts(**kw) → SimSystem

Return a new SimSystem with updated artifact attributes.

Examples

>>> sys = SimSystem(name="X", root=Path("work/X"))
>>> sys2 = sys.with_artifacts(topology=Path("work/X/top.prmtop"))

with_meta(**updates: Any) → SimSystem

Return a copy of the system with merged metadata.

Parameters:

**updates – Keyword arguments forwarded to SystemMeta.merge().

Returns:

Copy of the system containing the updated metadata bundle.

Return type:

SimSystem

class batter.api.SimulationConfig(*, system_name: str, fe_type: ~typing.Literal['custom', 'rest', 'sdr', 'dd', 'sdr-rest', 'express', 'relative', 'uno', 'uno_com', 'uno_rest', 'self', 'uno_dd', 'dd-rest', 'asfe', 'md'], dec_int: ~typing.Literal['mbar', 'ti'] = 'mbar', remd: ~typing.Literal['yes', 'no'] = 'no', remd_nstlim: int = 100, slurm_header_dir: ~pathlib.Path = <factory>, infe: bool = False, p1: str = '', p2: str = '', p3: str = '', other_mol: ~typing.List[str] = <factory>, lipid_mol: ~typing.List[str] = <factory>, solv_shell: float | None = 15.0, rocklin_correction: ~typing.Literal['yes', 'no'] = 'no', release_eq: ~typing.List[float] = <factory>, ti_points: int | None = 0, lambdas: ~typing.List[float] = <factory>, component_windows: ~typing.Dict[str, ~typing.List[float]] = <factory>, sdr_dist: float | None = 0.0, dec_method: str | None = None, blocks: int = 0, unbound_threshold: ~typing.Annotated[float, ~annotated_types.Ge(ge=0)] = 8.0, analysis_start_step: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, n_bootstraps: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, lig_distance_force: float = 0.0, lig_angle_force: float = 0.0, lig_dihcf_force: float = 0.0, rec_com_force: float = 0.0, lig_com_force: float = 0.0, water_model: ~typing.Literal['SPCE', 'TIP4PEW', 'TIP3P', 'TIP3PF', 'OPC'] = 'TIP3P', buffer_x: float = 10.0, buffer_y: float = 10.0, buffer_z: float = 15.0, lig_buffer: float = 10.0, neutralize_only: ~typing.Literal['yes', 'no'] = 'no', cation: str = 'Na+', anion: str = 'Cl-', ion_conc: float = 0.15, hmr: ~typing.Literal['yes', 'no'] = 'no', enable_mcwat: ~typing.Literal['yes', 'no'] = 'yes', temperature: float = 298.15, eq_steps: int = 1000000, n_steps_dict: ~typing.Dict[str, int] = <factory>, l1_x: float | None = None, l1_y: float | None = None, l1_z: float | None = None, l1_range: float | None = None, min_adis: float | None = None, max_adis: float | None = None, ntpr: int = 100, ntwr: int = 10000, ntwe: int = 0, ntwx: int = 2500, cut: float = 9.0, gamma_ln: float = 1.0, barostat: ~typing.Literal[1, 2] = 2, dt: float = 0.004, all_atoms: ~typing.Literal['yes', 'no'] = 'no', receptor_ff: str = 'protein.ff14SB', ligand_ff: str = 'gaff2', lipid_ff: str = 'lipid21', ligand_dict: ~typing.Dict[str, ~typing.Any] = <factory>, rng: int = 0, ion_def: ~typing.List[~typing.Any] = <factory>, dic_n_steps: ~typing.Dict[str, int] = <factory>, rest: ~typing.List[float] = <factory>, neut: str = '', protein_align: str = 'name CA', receptor_segment: str | None = None, components: ~typing.List[str] = <factory>, component_lambdas: ~typing.Dict[str, ~typing.List[float]] = <factory>, membrane_simulation: bool = True)

Bases: BaseModel

Simulation configuration for ABFE/ASFE/RBFE workflows. Values are fed by RunConfig.resolved_sim_config(), which merges create: and fe_sim:.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

all_atoms: Literal['yes', 'no']

analysis_start_step: int

anion: str

barostat: Literal[1, 2]

blocks: int

buffer_x: float

buffer_y: float

buffer_z: float

cation: str

component_lambdas: Dict[str, List[float]]

component_windows: Dict[str, List[float]]

components: List[str]

cut: float

dec_int: Literal['mbar', 'ti']

dec_method: str | None

dic_n_steps: Dict[str, int]

dt: float

enable_mcwat: Literal['yes', 'no']

eq_steps: int

fe_type: Literal['custom', 'rest', 'sdr', 'dd', 'sdr-rest', 'express', 'relative', 'uno', 'uno_com', 'uno_rest', 'self', 'uno_dd', 'dd-rest', 'asfe', 'md']

classmethod from_sections(create: CreateArgs, fe: FESimArgs, *, protocol: str | None = None, fe_type: str | None = None, slurm_header_dir: Path | None = None, run_remd: str | bool | None = None) → SimulationConfig

Construct a SimulationConfig from run sections.

Parameters:

• create (CreateArgs) – System creation inputs taken from the create YAML section.

• fe (FESimArgs) – Free-energy simulation overrides from the fe_sim section.

• run_remd ({“yes”,”no”}, optional) – Whether REMD execution is enabled (controls submission only; REMD inputs are always written during preparation).

Returns:

Fully merged simulation configuration ready for downstream use.

Return type:

SimulationConfig

gamma_ln: float

hmr: Literal['yes', 'no']

infe: bool

ion_conc: float

ion_def: List[Any]

l1_range: float | None

l1_x: float | None

l1_y: float | None

l1_z: float | None

lambdas: List[float]

lig_angle_force: float

lig_buffer: float

lig_com_force: float

lig_dihcf_force: float

lig_distance_force: float

ligand_dict: Dict[str, Any]

ligand_ff: str

lipid_ff: str

lipid_mol: List[str]

max_adis: float | None

membrane_simulation: bool

min_adis: float | None

model_config = {}

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

n_bootstraps: int

n_steps_dict: Dict[str, int]

neut: str

neutralize_only: Literal['yes', 'no']

ntpr: int

ntwe: int

ntwr: int

ntwx: int

other_mol: List[str]

p1: str

p2: str

p3: str

protein_align: str

rec_com_force: float

receptor_ff: str

receptor_segment: str | None

release_eq: List[float]

remd: Literal['yes', 'no']

remd_nstlim: int

rest: List[float]

rng: int

rocklin_correction: Literal['yes', 'no']

sdr_dist: float | None

slurm_header_dir: Path

solv_shell: float | None

system_name: str

temperature: float

ti_points: int | None

to_dict() → Dict[str, Any]

unbound_threshold: float

water_model: Literal['SPCE', 'TIP4PEW', 'TIP3P', 'TIP3PF', 'OPC']

class batter.api.WindowResult(*, component: str, lam: float, dG: float, dG_se: float = 0.0, n_samples: int = 0, meta: Dict[str, ~typing.Any]=<factory>)

Bases: BaseModel

Result for a single lambda window/component.

Parameters:

• component (str) – Component key (e.g., ‘e’, ‘v’, ‘z’).

• lam (float) – Lambda value in [0, 1].

• dG (float) – Free-energy increment (kcal/mol).

• dG_se (float) – Standard error (kcal/mol).

• n_samples (int) – Samples (or effective sample size).

• meta (dict) – Extra metadata.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

component: str

dG: float

dG_se: float

lam: float

meta: Dict[str, Any]

model_config = {}

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

n_samples: int

batter.api.clone_execution(work_dir: Path, src_run_id: str, dst_run_id: str | None = None, *, dst_root: Path | None = None, mode: Literal['copy', 'hardlink', 'symlink'] = 'hardlink', only_equil: bool = True, reset_states: bool = True, overwrite: bool = False) → Path

batter.api.dump_run_config(cfg: RunConfig, path: Path | str) → None

Serialize a run configuration to YAML.

Parameters:

• cfg (RunConfig) – Configuration object to export.

• path (str or pathlib.Path) – Destination path for the YAML file.

batter.api.list_fe_runs(work_dir: str | Path) → pd.DataFrame

Return an index of FE runs contained in a portable work directory.

Parameters:

work_dir (str or Path) – Path to the root directory of a BATTER execution (portable layout).

Returns:

DataFrame with one row per stored FE run. Columns include run_id, ligand, mol_name, system_name, temperature, total_dG, total_se, canonical_smiles, original_name, original_path, protocol, analysis_start_step, n_bootstraps, status, failure_reason, and created_at.

Return type:

pandas.DataFrame

batter.api.load_fe_run(work_dir: str | Path, run_id: str, ligand: str | None = None) → FERecord

Load a single FE record by run_id from a portable work directory.

Parameters:

• work_dir (str or Path) – Root directory of the BATTER execution.

• run_id (str) – Identifier of the FE run to load (as returned by list_fe_runs()).

• ligand (str, optional) – Ligand identifier when multiple ligands were processed in the run. If omitted, the sole ligand is selected automatically or a ValueError is raised when multiple matches exist.

Returns:

Structured record containing total ΔG, standard error, components, and per-window results.

Return type:

FERecord

batter.api.load_run_config(path: Path | str) → RunConfig

Read a run-level YAML file and return a validated configuration.

Parameters:

path (str or pathlib.Path) – Location of the run YAML file.

Returns:

Parsed run configuration.

Return type:

RunConfig

batter.api.load_sim_config(path: Path | str) → SimulationConfig

Load a simulation configuration from YAML.

Parameters:

path (str or pathlib.Path) – Path to the simulation YAML file.

Returns:

Validated simulation configuration.

Return type:

SimulationConfig

batter.api.run_analysis_from_execution(work_dir: str | Path, run_id: str | None = None, *, ligand: str | None = None, components: Sequence[str] | None = None, n_workers: int | None = None, analysis_start_step: int | None = None, n_bootstraps: int | None = None, overwrite: bool = True, raise_on_error: bool = True) → None

Run FE analysis for a partially finished/finished execution.

Parameters:

• work_dir (str or Path) – Root directory containing the portable execution store.

• run_id (str, optional) – Identifier of the execution (e.g., run-20240101). When omitted, the most recently modified execution under <work_dir>/executions is used.

• ligand (str, optional) – Ligand identifier to target when only a subset should be analyzed.

• components (sequence of str, optional) – Components to include during analysis (overrides sim_cfg.components).

• n_workers (int, optional) – Number of worker processes requested for the analysis handler.

• analysis_start_step (int, optional) – First production step to include in analysis (per window); overrides config.

• n_bootstraps (int, optional) – Number of MBAR bootstrap resamples; overrides config.

• overwrite (bool, optional) – When True (default), overwrite any existing analysis results for the run_id. When False, skip ligands that already have analysis outputs.

• raise_on_error (bool, optional) – When True (default) propagate errors raised by the analysis handler. Set to False to log the failure and continue with other ligands.

batter.api.run_from_yaml(path: Path | str, on_failure: Literal['prune', 'raise', 'retry'] = None, run_overrides: Dict[str, Any] | None = None) → None

Execute a BATTER workflow described by a YAML file.

batter.api.save_sim_config(cfg: SimulationConfig, path: Path | str) → None

Write a simulation configuration to YAML.

Parameters:

• cfg (SimulationConfig) – Configuration object to serialise.

• path (str or pathlib.Path) – Output file path for the YAML representation.

Config Modules

class batter.config.run.CreateArgs(*, system_name: str | None = 'unnamed_system', protein_input: Path | None = None, system_input: Path | None = None, system_coordinate: Path | None = None, protein_align: str | None = 'name CA', ligand_paths: dict[str, ~pathlib.Path | str]=<factory>, ligand_input: Path | None = None, ligand_ff: str = 'gaff2', retain_lig_prot: bool = True, param_method: Literal['amber', 'openff']='amber', param_charge: str = 'am1bcc', param_outdir: Path | None = None, anchor_atoms: list[str] = <factory>, lipid_mol: list[str] = <factory>, other_mol: list[str] = <factory>, overwrite: bool = True, extra_restraints: str | None = None, extra_restraint_fc: float = 10.0, extra_conformation_restraints: Path | None = None, receptor_ff: str = 'protein.ff14SB', lipid_ff: str = 'lipid21', solv_shell: float = 15.0, cation: str = 'Na+', anion: str = 'Cl-', ion_conc: float = 0.15, neutralize_only: Literal['yes', 'no']='no', water_model: str = 'TIP3P', l1_range: float = 6.0, min_adis: float = 3.0, max_adis: float = 7.0)

Bases: BaseModel

Inputs for system creation and staging.

Notes

This section mirrors the create block in the run YAML file.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

anchor_atoms: list[str]

anion: str

cation: str

extra_conformation_restraints: Path | None

extra_restraint_fc: float

extra_restraints: str | None

ion_conc: float

l1_range: float

ligand_ff: str

ligand_input: Path | None

ligand_paths: dict[str, Path | str]

lipid_ff: str

lipid_mol: list[str]

max_adis: float

min_adis: float

model_config = {'extra': 'forbid'}

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

neutralize_only: Literal['yes', 'no']

other_mol: list[str]

overwrite: bool

param_charge: str

param_method: Literal['amber', 'openff']

param_outdir: Path | None

protein_align: str | None

protein_input: Path | None

receptor_ff: str

resolve_paths(base: Path) → CreateArgs

Return a copy where path fields are absolute relative to base.

retain_lig_prot: bool

solv_shell: float

system_coordinate: Path | None

system_input: Path | None

system_name: str | None

water_model: str

class batter.config.run.FESimArgs(*, dec_int: str = 'mbar', remd: RemdArgs = <factory>, rocklin_correction: Literal['yes', 'no']='no', lambdas: List[float] = <factory>, component_lambdas: Dict[str, ~typing.List[float]]=<factory>, blocks: int = 0, lig_buffer: float = 15.0, lig_distance_force: float = 5.0, lig_angle_force: float = 250.0, lig_dihcf_force: float = 0.0, rec_com_force: float = 10.0, lig_com_force: float = 10.0, buffer_x: float = 20.0, buffer_y: float = 20.0, buffer_z: float = 20.0, eq_steps: Annotated[int, ~annotated_types.Ge(ge=0)] = 1000000, n_steps: Dict[str, int]=<factory>, ntpr: int = 100, ntwr: int = 2500, ntwe: int = 0, ntwx: int = 25000, cut: float = 9.0, gamma_ln: float = 1.0, dt: float = 0.004, hmr: Literal['yes', 'no']='no', enable_mcwat: Literal['yes', 'no']='yes', temperature: float = 298.15, barostat: int = 2, unbound_threshold: Annotated[float, ~annotated_types.Ge(ge=0)] = 8.0, analysis_start_step: Annotated[int, ~annotated_types.Ge(ge=0)] = 0, n_bootstraps: Annotated[int, ~annotated_types.Ge(ge=0)] = 0)

Bases: BaseModel

Free-energy simulation knobs loaded from the fe_sim section.

The fields feed directly into batter.config.simulation.SimulationConfig overrides. fe_type is resolved internally from protocol rather than being set by users.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

analysis_start_step: int

barostat: int

blocks: int

buffer_x: float

buffer_y: float

buffer_z: float

component_lambdas: Dict[str, List[float]]

cut: float

dec_int: str

dt: float

enable_mcwat: Literal['yes', 'no']

eq_steps: int

gamma_ln: float

hmr: Literal['yes', 'no']

lambdas: List[float]

lig_angle_force: float

lig_buffer: float

lig_com_force: float

lig_dihcf_force: float

lig_distance_force: float

model_config = {'extra': 'forbid'}

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

n_bootstraps: int

n_steps: Dict[str, int]

ntpr: int

ntwe: int

ntwr: int

ntwx: int

rec_com_force: float

remd: RemdArgs

rocklin_correction: Literal['yes', 'no']

temperature: float

unbound_threshold: float

class batter.config.run.MDSimArgs(*, dt: float = 0.004, temperature: float = 298.15, eq_steps: Annotated[int, Ge(ge=0)] = 100000, ntpr: int = 100, ntwr: int = 10000, ntwe: int = 0, ntwx: int = 25000, cut: float = 9.0, gamma_ln: float = 1.0, barostat: int = 2, hmr: Literal['yes', 'no'] = 'yes', enable_mcwat: Literal['yes', 'no'] = 'yes')

Bases: BaseModel

Simulation overrides used when protocol == "md".

These runs reuse the equilibration steps from ABFE but never schedule FE windows, so only generic MD knobs are required (no lambdas, SDR restraints, etc.).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

barostat: int

cut: float

dt: float

enable_mcwat: Literal['yes', 'no']

eq_steps: int

gamma_ln: float

hmr: Literal['yes', 'no']

model_config = {'extra': 'forbid'}

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

ntpr: int

ntwe: int

ntwr: int

ntwx: int

temperature: float

class batter.config.run.RBFENetworkArgs(*, mapping: str | None = 'default', atom_mapper: Literal['kartograf', 'lomap'] = 'kartograf', konnektor_layout: str | None = None, both_directions: bool = False, mapping_file: Path | None = None)

Bases: BaseModel

RBFE network mapping controls.

Users can specify a mapping strategy by name (mapping) or provide an explicit mapping file (mapping_file).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

atom_mapper: Literal['kartograf', 'lomap']

both_directions: bool

konnektor_layout: str | None

mapping: str | None

mapping_file: Path | None

model_config = {'extra': 'forbid'}

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

resolve_paths(base: Path) → RBFENetworkArgs

class batter.config.run.RunConfig(*, version: int = 1, protocol: Literal['abfe', 'rbfe', 'asfe', 'md']='abfe', backend: Literal['local', 'slurm']='local', create: CreateArgs, fe_sim: Dict[str, ~typing.Any] | ~batter.config.run.FESimArgs | ~batter.config.run.MDSimArgs=<factory>, run: RunSection, rbfe: RBFENetworkArgs | None = None)

Bases: BaseModel

Top-level YAML config.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

backend: Literal['local', 'slurm']

create: CreateArgs

fe_sim: Dict[str, Any] | FESimArgs | MDSimArgs

classmethod load(path: Path | str) → RunConfig

Load and validate a run configuration from disk.

Parameters:

path (str or pathlib.Path) – Location of the YAML file to parse.

Returns:

Fully validated configuration object.

Return type:

RunConfig

model_config = {'extra': 'forbid'}

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

classmethod model_validate_yaml(yaml_text: str) → RunConfig

Validate a run configuration from an in-memory YAML string.

Parameters:

yaml_text (str) – Raw YAML content describing the run configuration.

Returns:

Validated configuration model.

Return type:

RunConfig

protocol: Literal['abfe', 'rbfe', 'asfe', 'md']

rbfe: RBFENetworkArgs | None

resolved_sim_config() → SimulationConfig

Build the effective simulation configuration for this run.

Returns:

Simulation parameters derived from create and fe_sim sections.

Return type:

SimulationConfig

run: RunSection

version: int

with_base_dir(base_dir: Path) → RunConfig

Return a copy with relative paths resolved against base_dir.

class batter.config.run.RunSection(*, output_folder: Path, system_type: Literal['MABFE', 'MASFE'] | None=None, only_fe_preparation: bool = False, on_failure: Literal['raise', 'prune', 'retry']='raise', max_workers: int | None = None, max_active_jobs: Annotated[int | None, ~annotated_types.Ge(ge=0)] = 1000, batch_mode: bool = False, batch_gpus: Annotated[int | None, ~annotated_types.Ge(ge=0)] = None, batch_gpus_per_task: Annotated[int, ~annotated_types.Ge(ge=1)] = 1, batch_srun_extra: List[str] = <factory>, dry_run: bool = False, clean_failures: bool = False, remd: Literal['yes', 'no']='no', run_id: str = 'auto', allow_run_id_mismatch: bool = False, slurm_header_dir: Path | None = None, email_sender: str = 'nobody@stanford.edu', email_on_completion: str | None = None, slurm: SlurmConfig = <factory>)

Bases: BaseModel

Run-related settings, including where outputs land.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

allow_run_id_mismatch: bool

batch_gpus: int | None

batch_gpus_per_task: int

batch_mode: bool

batch_srun_extra: List[str]

clean_failures: bool

dry_run: bool

email_on_completion: str | None

email_sender: str

max_active_jobs: int | None

max_workers: int | None

model_config = {'extra': 'forbid'}

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

on_failure: Literal['raise', 'prune', 'retry']

only_fe_preparation: bool

output_folder: Path

remd: Literal['yes', 'no']

resolve_paths(base: Path) → RunSection

Return a copy where output_folder is absolute relative to base.

run_id: str

slurm: SlurmConfig

slurm_header_dir: Path | None

system_type: Literal['MABFE', 'MASFE'] | None

class batter.config.run.SlurmConfig(*, partition: str | None = None, time: str | None = None, nodes: int | None = None, ntasks_per_node: int | None = None, mem_per_cpu: str | None = None, gres: str | None = None, account: str | None = None, qos: str | None = None, constraint: str | None = None, extra_sbatch: List[str] = <factory>)

Bases: BaseModel

SLURM-specific configuration.

Parameters:

• partition (str, optional) – SLURM partition/queue name.

• time (str, optional) – Walltime in the HH:MM:SS format.

• nodes (int, optional) – Number of nodes to request.

• ntasks_per_node (int, optional) – Number of tasks per node.

• mem_per_cpu (str, optional) – Memory per CPU (e.g., 16G).

• gres (str, optional) – Generic resource string (e.g., GPU spec).

• account (str, optional) – Account to charge for jobs.

• qos (str, optional) – QoS string if required by the cluster.

• constraint (str, optional) – Constraint string passed to sbatch.

• extra_sbatch (list[str]) – Additional arguments appended to the sbatch submission command.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

account: str | None

constraint: str | None

extra_sbatch: List[str]

gres: str | None

mem_per_cpu: str | None

model_config = {'extra': 'ignore'}

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

nodes: int | None

ntasks_per_node: int | None

partition: str | None

qos: str | None

time: str | None

to_sbatch_flags() → List[str]

Produce a flat list of sbatch command-line flags.

Returns:

Sequence suitable for passing to subprocess.run().

Return type:

list of str

class batter.config.simulation.SimulationConfig(*, system_name: str, fe_type: ~typing.Literal['custom', 'rest', 'sdr', 'dd', 'sdr-rest', 'express', 'relative', 'uno', 'uno_com', 'uno_rest', 'self', 'uno_dd', 'dd-rest', 'asfe', 'md'], dec_int: ~typing.Literal['mbar', 'ti'] = 'mbar', remd: ~typing.Literal['yes', 'no'] = 'no', remd_nstlim: int = 100, slurm_header_dir: ~pathlib.Path = <factory>, infe: bool = False, p1: str = '', p2: str = '', p3: str = '', other_mol: ~typing.List[str] = <factory>, lipid_mol: ~typing.List[str] = <factory>, solv_shell: float | None = 15.0, rocklin_correction: ~typing.Literal['yes', 'no'] = 'no', release_eq: ~typing.List[float] = <factory>, ti_points: int | None = 0, lambdas: ~typing.List[float] = <factory>, component_windows: ~typing.Dict[str, ~typing.List[float]] = <factory>, sdr_dist: float | None = 0.0, dec_method: str | None = None, blocks: int = 0, unbound_threshold: ~typing.Annotated[float, ~annotated_types.Ge(ge=0)] = 8.0, analysis_start_step: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, n_bootstraps: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 0, lig_distance_force: float = 0.0, lig_angle_force: float = 0.0, lig_dihcf_force: float = 0.0, rec_com_force: float = 0.0, lig_com_force: float = 0.0, water_model: ~typing.Literal['SPCE', 'TIP4PEW', 'TIP3P', 'TIP3PF', 'OPC'] = 'TIP3P', buffer_x: float = 10.0, buffer_y: float = 10.0, buffer_z: float = 15.0, lig_buffer: float = 10.0, neutralize_only: ~typing.Literal['yes', 'no'] = 'no', cation: str = 'Na+', anion: str = 'Cl-', ion_conc: float = 0.15, hmr: ~typing.Literal['yes', 'no'] = 'no', enable_mcwat: ~typing.Literal['yes', 'no'] = 'yes', temperature: float = 298.15, eq_steps: int = 1000000, n_steps_dict: ~typing.Dict[str, int] = <factory>, l1_x: float | None = None, l1_y: float | None = None, l1_z: float | None = None, l1_range: float | None = None, min_adis: float | None = None, max_adis: float | None = None, ntpr: int = 100, ntwr: int = 10000, ntwe: int = 0, ntwx: int = 2500, cut: float = 9.0, gamma_ln: float = 1.0, barostat: ~typing.Literal[1, 2] = 2, dt: float = 0.004, all_atoms: ~typing.Literal['yes', 'no'] = 'no', receptor_ff: str = 'protein.ff14SB', ligand_ff: str = 'gaff2', lipid_ff: str = 'lipid21', ligand_dict: ~typing.Dict[str, ~typing.Any] = <factory>, rng: int = 0, ion_def: ~typing.List[~typing.Any] = <factory>, dic_n_steps: ~typing.Dict[str, int] = <factory>, rest: ~typing.List[float] = <factory>, neut: str = '', protein_align: str = 'name CA', receptor_segment: str | None = None, components: ~typing.List[str] = <factory>, component_lambdas: ~typing.Dict[str, ~typing.List[float]] = <factory>, membrane_simulation: bool = True)

Bases: BaseModel

Simulation configuration for ABFE/ASFE/RBFE workflows. Values are fed by RunConfig.resolved_sim_config(), which merges create: and fe_sim:.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

all_atoms: Literal['yes', 'no']

analysis_start_step: int

anion: str

barostat: Literal[1, 2]

blocks: int

buffer_x: float

buffer_y: float

buffer_z: float

cation: str

component_lambdas: Dict[str, List[float]]

component_windows: Dict[str, List[float]]

components: List[str]

cut: float

dec_int: Literal['mbar', 'ti']

dec_method: str | None

dic_n_steps: Dict[str, int]

dt: float

enable_mcwat: Literal['yes', 'no']

eq_steps: int

fe_type: Literal['custom', 'rest', 'sdr', 'dd', 'sdr-rest', 'express', 'relative', 'uno', 'uno_com', 'uno_rest', 'self', 'uno_dd', 'dd-rest', 'asfe', 'md']

classmethod from_sections(create: CreateArgs, fe: FESimArgs, *, protocol: str | None = None, fe_type: str | None = None, slurm_header_dir: Path | None = None, run_remd: str | bool | None = None) → SimulationConfig

Construct a SimulationConfig from run sections.

Parameters:

• create (CreateArgs) – System creation inputs taken from the create YAML section.

• fe (FESimArgs) – Free-energy simulation overrides from the fe_sim section.

• run_remd ({“yes”,”no”}, optional) – Whether REMD execution is enabled (controls submission only; REMD inputs are always written during preparation).

Returns:

Fully merged simulation configuration ready for downstream use.

Return type:

SimulationConfig

gamma_ln: float

hmr: Literal['yes', 'no']

infe: bool

ion_conc: float

ion_def: List[Any]

l1_range: float | None

l1_x: float | None

l1_y: float | None

l1_z: float | None

lambdas: List[float]

lig_angle_force: float

lig_buffer: float

lig_com_force: float

lig_dihcf_force: float

lig_distance_force: float

ligand_dict: Dict[str, Any]

ligand_ff: str

lipid_ff: str

lipid_mol: List[str]

max_adis: float | None

membrane_simulation: bool

min_adis: float | None

model_config = {}

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

n_bootstraps: int

n_steps_dict: Dict[str, int]

neut: str

neutralize_only: Literal['yes', 'no']

ntpr: int

ntwe: int

ntwr: int

ntwx: int

other_mol: List[str]

p1: str

p2: str

p3: str

protein_align: str

rec_com_force: float

receptor_ff: str

receptor_segment: str | None

release_eq: List[float]

remd: Literal['yes', 'no']

remd_nstlim: int

rest: List[float]

rng: int

rocklin_correction: Literal['yes', 'no']

sdr_dist: float | None

slurm_header_dir: Path

solv_shell: float | None

system_name: str

temperature: float

ti_points: int | None

to_dict() → Dict[str, Any]

unbound_threshold: float

water_model: Literal['SPCE', 'TIP4PEW', 'TIP3P', 'TIP3PF', 'OPC']

batter.config.utils.coerce_yes_no(value: Any) → str | None

Normalize boolean-like values into "yes" or "no".

Parameters:

value – Input flag provided by the user. Supported types include bool, numeric scalars, or strings such as "true" and "0".

Returns:

"yes" or "no" when the flag can be interpreted. None is returned unchanged to preserve optional semantics.

Return type:

str or None

Raises:

ValueError – If the value cannot be coerced into a boolean switch.

batter.config.utils.expand_env_vars(data: Any, *, base_dir: Path | None = None) → Any

Recursively expand environment variables in a YAML-derived structure.

Parameters:

• data – Parsed YAML content to normalise.

• base_dir (Path, optional) – Base directory for resolving relative (./) paths.

Returns:

Structure with string values expanded.

Return type:

Any

batter.config.utils.normalize_optional_path(value: Any) → Path | None

Resolve optional path-like values into pathlib.Path objects.

Parameters:

value – Path candidate that may be None or an empty string. Strings may contain environment variables or ~.

Returns:

Expanded path when provided; None if the value is empty.

Return type:

pathlib.Path or None

batter.config.utils.sanitize_ligand_name(name: str) → str

Convert a ligand identifier into a filesystem-safe token.

Parameters:

name (str) – Original ligand identifier, often derived from filenames or keys.

Returns:

Uppercase alphanumeric token with unsafe characters replaced by underscores.

Return type:

str

batter.config.utils.sanitize_user_ligand_name(name: str) → str

Sanitize and validate a user-provided ligand identifier.

Reserved names that conflict with BATTER directory layout are rejected.

Orchestrator Modules

batter.orchestrate.run

Top-level orchestration entry for BATTER runs.

This module wires: YAML (RunConfig) → shared system build → bulk ligand staging → single param job (“param_ligands”) → per-ligand pipelines → FE record save.

batter.orchestrate.run.run_from_yaml(path: Path | str, on_failure: Literal['prune', 'raise', 'retry'] = None, run_overrides: Dict[str, Any] | None = None) → None

Execute a BATTER workflow described by a YAML file.

Selection helpers for choosing the correct pipeline implementation.

batter.orchestrate.pipeline_utils.select_pipeline(protocol: str, sim_cfg: SimulationConfig, only_fe_prep: bool, *, sys_params: SystemParams | dict | None = None, partition: str | None = None) → Pipeline

Return the protocol-specific pipeline for a run.

Parameters:

• protocol (str) – Name of the requested protocol ("abfe", "rbfe", "asfe", or "md").

• sim_cfg (SimulationConfig) – Validated simulation configuration produced by RunConfig.

• only_fe_prep (bool) – When True, truncate the pipeline after FE preparation steps.

• sys_params (SystemParams or dict, optional) – Extra parameters passed to system-level pipeline steps.

Returns:

Pipeline instance tailored to the requested protocol.

Return type:

Pipeline

Raises:

ValueError – If the protocol name is not recognised.

Utilities for configuring execution backends used by the orchestrator.

batter.orchestrate.backend.register_local_handlers(backend: LocalBackend) → None

Register built-in pipeline handlers on the local backend.

Parameters:

backend (LocalBackend) – Backend instance that should receive the default handler mapping.

Raises:

RuntimeError – If optional handler dependencies (for example openff-toolkit) are missing.

Execution Modules

Interfaces shared by execution backends.

class batter.exec.base.ExecBackend(*args, **kwargs)

Protocol implemented by execution backends.

name: str

run(step: Step, system: SimSystem, params: Dict) → ExecResult

Execute step for system.

Parameters:

• step (Step) – Step metadata as produced by the pipeline.

• system (SimSystem) – Simulation system descriptor.

• params (dict) – Backend-specific parameters, potentially including resources.

Returns:

Execution artifacts and job identifiers.

Return type:

ExecResult

class batter.exec.base.Resources(time: str | None = None, cpus: int | None = None, gpus: int | None = None, mem: str | None = None, partition: str | None = None, account: str | None = None, extra: Mapping[str, str]=<factory>)

Resource hints supplied to execution backends.

Parameters:

• time (str, optional) – Walltime (e.g., "02:00:00").

• cpus (int, optional) – CPU cores per task.

• gpus (int, optional) – Number of GPUs required.

• mem (str, optional) – Memory request (e.g., "16G").

• partition (str, optional) – Scheduler partition or queue.

• account (str, optional) – Scheduler account.

• extra (Mapping[str, str], optional) – Backend-specific SBATCH-style flags.

account: str | None

cpus: int | None

extra: Mapping[str, str]

gpus: int | None

mem: str | None

partition: str | None

time: str | None

Execution backend for running pipelines locally.

class batter.exec.local.LocalBackend(max_workers: int | None = None)

In-process execution backend with optional parallel orchestration.

Parameters:

max_workers (int, optional) – Maximum number of worker processes to use when run_parallel() is invoked. None lets the backend auto-detect resources; 0 or 1 forces serial execution.

name: str = 'local'

register(step_name: str, handler: Callable[[Step, SimSystem, Mapping], ExecResult]) → None

Register a callable to execute step_name.

Parameters:

• step_name (str) – Identifier of the step (matches batter.pipeline.step.Step.name).

• handler (Callable[[Step, SimSystem, Mapping], ExecResult]) – Function responsible for executing the step.

run(step: Step, system: SimSystem, params: Mapping) → ExecResult

Execute step for system on the local machine.

Parameters:

• step – Pipeline step metadata.

• system – Simulation system descriptor.

• params – Step parameters, typically generated by the orchestration layer.

Returns:

Artifacts and job identifiers (empty for local execution).

Return type:

ExecResult

run_parallel(pipeline: Pipeline, systems: Iterable[SimSystem], *, max_workers: int | None = None, description: str = '', batch_size: str | int = 'auto', verbose: int = 10, prefer: str = 'processes', backend: str | None = None) → Dict[str, Mapping[str, ExecResult]]

Execute pipeline for multiple systems in parallel.

Parameters:

• pipeline – Pipeline object providing the sequence of steps to execute.

• systems (Iterable[SimSystem]) – Collection of systems to process.

• max_workers (int, optional) – Override the configured worker cap; None falls back to the value provided at construction time.

• description (str, optional) – Human-readable label used in debug logging.

• batch_size, verbose, prefer, backend – Joblib configuration knobs forwarded to joblib.Parallel.

Returns:

Mapping of system.name to per-step results.

Return type:

dict

Raises:

RuntimeError – When one or more systems fail.

Execution backend that submits steps to Slurm via sbatch.

class batter.exec.slurm.SlurmBackend(*args, **kwargs)

Slurm backend that materializes lightweight job scripts.

name: str = 'slurm'

run(step: Step, system: SimSystem, params: Dict[str, Any]) → ExecResult

Submit step to Slurm.

Parameters:

• step (Step) – Pipeline step metadata.

• system (SimSystem) – Simulation system whose root directory stores scripts and logs.

• params (dict) – Backend-specific options. Recognised keys include resources, env (exported variables), and payload (shell snippet).

Returns:

Artifacts referencing the generated script and log paths together with the submitted job identifier (if available).

Return type:

ExecResult

class batter.exec.slurm_mgr.SlurmJobManager(poll_s: float = 60.0, max_retries: int = 3, resubmit_backoff_s: float = 30.0, registry_file: Path | None = None, dry_run: bool = False, sbatch_flags: Sequence[str] | None = None, submit_retry_limit: int = 3, submit_retry_delay_s: float = 60.0, max_active_jobs: int | None = None, partition: str | None = None, batch_mode: bool = False, batch_gpus: int | None = None, gpus_per_task: int = 1, srun_extra: Sequence[str] | None = None, stage: str | None = None, header_root: Path | None = None, **_ignored: Any)

Submit, monitor, and resubmit Slurm jobs for BATTER executions.

Parameters:

• poll_s (float, optional) – Poll interval (seconds) between status checks.

• max_retries (int, optional) – Maximum automatic resubmissions per workdir (excluding TIMEOUT and COMPLETED-without-sentinel).

• resubmit_backoff_s (float, optional) – Sleep before resubmitting a job after detecting termination/missing state.

• registry_file (pathlib.Path, optional) – JSONL queue file for cross-process coordination.

• dry_run (bool, optional) – When True, do not submit; record that submission would occur.

• sbatch_flags (Sequence[str], optional) – Global sbatch flags appended to every submission.

• submit_retry_limit (int, optional) – Number of retries for the submission command itself.

• submit_retry_delay_s (float, optional) – Delay between submission retries.

• max_active_jobs (int, optional) – Cap on concurrent jobs for the user (checked via one squeue -u call).

• partition (str, optional) – Partition filter used by max_active_jobs checks.

Other Parameters:

• batch_mode, batch_gpus, gpus_per_task, srun_extra, stage, header_root – Accepted for compatibility with older code paths. This manager does not implement batch execution; values are stored/ignored.

• **_ignored – Extra kwargs are accepted and ignored for compatibility.

add(spec: SlurmJobSpec) → None

Queue spec for later submission and optionally persist to registry.

Parameters:

spec (SlurmJobSpec) – Job specification.

clear() → None

Clear in-memory queue/retry book and remove on-disk registry if present.

ensure_running(spec: SlurmJobSpec) → None

Ensure the spec is submitted or already done/active.

Parameters:

spec (SlurmJobSpec) – Job spec.

Notes

This method does not register specs; it’s a one-off submit-if-needed.

jobs() → List[SlurmJobSpec]

Return merged in-memory + registry specs (dedup by workdir).

set_stage(stage: str | None) → None

Set the active stage filter for registry loading/submission.

Parameters:

stage (str or None) – Stage key such as equil, fe_equil, fe, etc. If None, stage filtering is disabled.

wait_all() → None

Submit/monitor all registered jobs and block until completion.

wait_for_slot(poll_s: float | None = None, user: str | None = None, partition: str | None = None) → None

Block until active jobs drop below max_active_jobs.

Parameters:

• poll_s (float, optional) – Polling interval in seconds (defaults to poll_s).

• user (str, optional) – Unix username (defaults to $USER).

• partition (str, optional) – Partition to filter on (defaults to manager partition).

wait_until_done(specs: Iterable[SlurmJobSpec]) → None

Legacy interface: monitor a given set until complete.

class batter.exec.slurm_mgr.SlurmJobSpec(workdir: Path, script_rel: str = 'SLURMM-run', finished_name: str = 'FINISHED', failed_name: str = 'FAILED', name: str | None = None, stage: str | None = None, body_rel: str | None = None, header_name: str | None = None, header_template: Path | None = None, header_root: Path | None = None, batch_script: Path | None = None, extra_sbatch: Sequence[str] = <factory>, extra_env: Dict[str, str]=<factory>, submit_dir: Path | None = None, alt_script_names: Sequence[str] = ('SLURMM-run', 'SLURMM-Run', 'slurmm-run', 'run.sh'))

Descriptor for a Slurm job managed by SlurmJobManager.

Parameters:

• workdir (pathlib.Path) – Working directory containing submission scripts and sentinel files.

• script_rel (str, optional) – Preferred relative submission script path.

• finished_name (str, optional) – Sentinel file name indicating success.

• failed_name (str, optional) – Sentinel file name indicating failure.

• name (str, optional) – Friendly display name.

• stage (str, optional) – Logical stage used for registry filtering.

• extra_sbatch (Sequence[str], optional) – Extra sbatch flags (job-specific).

• extra_env (dict, optional) – Extra environment variables to export (job-specific).

• submit_dir (pathlib.Path, optional) – Directory to submit from (defaults to workdir).

Notes

The remaining fields are legacy compatibility fields used by older BATTER versions and/or existing registry entries. The manager may ignore them.

alt_script_names: Sequence[str] = ('SLURMM-run', 'SLURMM-Run', 'slurmm-run', 'run.sh')

batch_script: Path | None = None

body_rel: str | None = None

extra_env: Dict[str, str]

extra_sbatch: Sequence[str]

failed_name: str = 'FAILED'

failed_path() → Path

Sentinel path signalling failure.

finished_name: str = 'FINISHED'

finished_path() → Path

Sentinel path signalling successful completion.

header_name: str | None = None

header_root: Path | None = None

header_template: Path | None = None

jobid_path() → Path

Path containing the most recent Slurm job identifier.

name: str | None = None

resolve_script_abs() → Path

Return the absolute path to the submission script.

Returns:

Existing script path if found, otherwise the preferred path.

Return type:

pathlib.Path

script_arg() → str

Return the submission-script path argument for sbatch.

Returns:

Script path relative to submit_dir when possible.

Return type:

str

script_rel: str = 'SLURMM-run'

stage: str | None = None

submit_dir: Path | None = None

workdir: Path

Helpers for constructing AMBER mdin control files.

class batter.exec.amber.mdin.AmberMdin(*, cut: float = 9.0, ioutfm: int = 1, ntb: int = 1, ntxo: int = 2)

Mutable representation of an AMBER mdin file.

Parameters:

• cut (float, optional) – Non-bonded cutoff in Å (default: 9.0).

• ioutfm (int, optional) – Output format flag (1 → NetCDF).

• ntb (int, optional) – Periodic boundary condition flag.

• ntxo (int, optional) – Restart write format.

add_block(name: str, params: Dict[str, object] | None = None) → None

Append a named control block.

add_raw(line: str) → None

Append a raw line verbatim to the output.

apply_defaults(*, cut: float = 9.0, ioutfm: int = 1, ntb: int = 1, ntxo: int = 2) → None

Initialise with a baseline cntrl block.

override_block(block_name: str, param_dict: Dict[str, object]) → None

Merge param_dict into an existing block or create the block.

save(filename: str | Path) → None

Write the mdin file to filename.

to_string() → str

Render the mdin contents as text.

update_param(block_name: str, key: str, value: object) → None

Update a single parameter within block_name.

batter.exec.amber.mdin.apply_disang(mdin: AmberMdin, *, filename: str = 'disang.rest') → None

Reference a DISANG restraint file.

batter.exec.amber.mdin.apply_membrane_npt(mdin: AmberMdin, *, temp: float = 298.15, steps: int = 50000, barostat: int = 2, dt: float = 0.004) → None

Configure semi-isotropic NPT suitable for membranes.

batter.exec.amber.mdin.apply_minimization(mdin: AmberMdin, *, steps: int = 5000) → None

Enable energy minimisation for steps iterations.

batter.exec.amber.mdin.apply_npt(mdin: AmberMdin, *, temp: float = 298.15, steps: int = 50000, barostat: int = 2, dt: float = 0.004) → None

Configure standard NPT dynamics.

batter.exec.amber.mdin.apply_restraints(mdin: AmberMdin, *, mask: str, weight: float = 50.0) → None

Add positional restraints.

batter.exec.amber.mdin.apply_ti(mdin: AmberMdin, *, lbd_val: float, timask1: str, timask2: str, scmask1: str, scmask2: str, crgmask: str) → None

Configure thermodynamic integration (TI) parameters.

batter.exec.amber.mdin.apply_wt_end(mdin: AmberMdin) → None

Append the &wt type='END' control line.

Slurm-backed equilibration handler.

batter.exec.handlers.equil.equil_handler(step: Step, system: SimSystem, params: Dict[str, Any]) → ExecResult

Submit and register the equilibration job with the Slurm manager.

Parameters:

• step (Step) – Pipeline step metadata (unused but provided for symmetry).

• system (SimSystem) – Simulation system descriptor.

• params (dict) – Raw handler payload; validated into StepPayload.

Returns:

Result containing either existing artifacts (when already finished) or the work directory to be monitored by the manager.

Return type:

ExecResult

Raises:

• FileNotFoundError – If the expected submission script is missing.

• RuntimeError – When payload['job_mgr'] is not a SlurmJobManager.

Handlers that queue free-energy equilibration and production jobs.

batter.exec.handlers.fe.fe_equil_handler(step: Step, system: SimSystem, params: Dict[str, Any]) → ExecResult

Queue equilibration jobs for each component of a ligand.

Parameters:

• step, system (ignored) – Included for parity with the handler signature.

• params (dict) – Handler payload containing the job manager and configuration values.

Returns:

Number of jobs enqueued (without waiting for completion).

Return type:

ExecResult

batter.exec.handlers.fe.fe_handler(step: Step, system: SimSystem, params: Dict[str, Any]) → ExecResult

Queue production jobs for each component/window combination.

Parameters:

• step, system (ignored) – Provided for handler API compatibility.

• params (dict) – Handler payload containing the job manager and configuration values.

Returns:

Number of jobs enqueued (without waiting for completion).

Return type:

ExecResult

Prepare complex systems (protein/ligand/membrane) for simulations.

batter.exec.handlers.system_prep.system_prep(step: Step, system: SimSystem, params: Dict[str, Any]) → ExecResult

Prepare a system by aligning components and generating reference structures.

Parameters:

• step (Step) – Pipeline metadata (unused).

• system (SimSystem) – Simulation system descriptor.

• params (dict) – Handler payload validated into StepPayload.

Returns:

Paths to generated reference structures and a metadata dictionary with anchor and membrane information.

Return type:

ExecResult

Minimal system-preparation handler for MASFE workflows.

batter.exec.handlers.system_prep_masfe.system_prep_masfe(step: Step, system: SimSystem, params: Dict[str, Any]) → ExecResult

Prepare a MASFE solvation system by staging ligands and overrides.

Parameters:

• step (Step) – Pipeline metadata (unused).

• system (SimSystem) – Simulation system descriptor.

• params (dict) – Handler payload validated into StepPayload.

Returns:

Manifest of staged ligands and paths to generated files.

Return type:

ExecResult

Parameterisation Modules

Pipeline Modules

class batter.pipeline.pipeline.Pipeline(steps: List[Step])

Bases: object

Directed acyclic pipeline of Step objects.

Parameters:

steps (list[Step]) – Steps that form a DAG. Dependencies are given by Step.requires.

Notes

• A simple topological sort is performed before execution.

• Backends must implement a run(step, system) -> ExecResult method.

adjacency() → Dict[str, List[str]]

Return the adjacency list describing the DAG.

Returns:

Mapping of each step to the steps that depend on it.

Return type:

dict[str, list[str]]

dependencies(step_name: str) → List[str]

Retrieve the declared dependencies for step_name.

Parameters:

step_name (str) – Step identifier.

Returns:

Names of prerequisite steps.

Return type:

list[str]

Raises:

KeyError – If step_name does not exist in the pipeline.

describe() → List[Dict[str, Any]]

Return a serialisable summary of the pipeline.

Returns:

Each entry contains name, requires, and payload_type keys.

Return type:

list of dict

ordered_steps() → List[Step]

Return steps in execution order.

run(backend, system) → Dict[str, ExecResult]

Execute steps in topological order.

Parameters:

• backend – Object providing run(step, system) -> ExecResult.

• system – The SimSystem descriptor.

Returns:

Mapping from step name to execution result.

Return type:

dict[str, ExecResult]

Raises:

RuntimeError – If a required dependency has not been produced.

class batter.pipeline.pipeline.PipelineState(results: Dict[str, ~batter.pipeline.step.ExecResult]=<factory>)

Bases: object

In-memory state of a pipeline execution.

Variables:

results (dict[str, ExecResult]) – Per-step execution results.

results: Dict[str, ExecResult]

class batter.pipeline.step.ExecResult(job_ids: List[str] = <factory>, artifacts: Mapping[str, ~typing.Any]=<factory>)

Execution result returned by a backend.

Parameters:

• job_ids (list[str]) – Scheduler or process identifiers (may be empty for local runs).

• artifacts (Mapping[str, Any]) – Named outputs (paths, metrics, small JSON blobs).

artifacts: Mapping[str, Any]

job_ids: List[str]

class batter.pipeline.step.Step(name: str, requires: List[str] = <factory>, payload: Any = None)

One unit of work in the pipeline.

Parameters:

• name (str) – Unique step name (e.g., "prepare_fe").

• requires (list[str]) – Names of steps that must complete before this step can run.

• payload (Any, optional) – Typed payload consumed by the backend. Typically a StepPayload.

Notes

• Steps are immutable descriptors. Execution is handled by a backend.

• The backend decides how to interpret params (e.g., templates, flags).

name: str

property params: Any

Backwards-compatible alias for payload.

payload: Any

replace(**updates: Any) → Step

Return a new Step with selected attributes updated.

Parameters:

**updates – Keyword overrides for any of the dataclass fields (name, requires, or payload).

Returns:

Fresh step instance containing the requested updates.

Return type:

Step

requires: List[str]

class batter.pipeline.payloads.StepPayload(*, sim: SimulationConfig | None = None, sys_params: SystemParams | None = None, **extra_data: Any)

Typed payload passed to pipeline step handlers.

The payload binds the SimulationConfig and SystemParams objects used by most handlers while permitting arbitrary extra values for backwards compatibility or specialised needs.

Parameters:

• sim (SimulationConfig, optional) – Resolved simulation configuration for the step.

• sys_params (SystemParams, optional) – Shared system-level parameters.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

copy_with(**updates: Any) → StepPayload

Create a new StepPayload with additional updates.

Parameters:

**updates – Keyword overrides applied to the current payload.

Returns:

New payload containing the merged data.

Return type:

StepPayload

get(item: str, default: Any = None) → Any

Safe lookup for a payload value with a default.

Parameters:

• item (str) – Key to fetch.

• default (Any, optional) – Value returned when the key is missing or None.

Returns:

Requested value or the default.

Return type:

Any

model_config = {'arbitrary_types_allowed': True, 'extra': 'allow'}

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

sim: SimulationConfig | None

sys_params: SystemParams | None

to_mapping() → Dict[str, Any]

Convert the payload (including extras) to a plain dictionary.

Returns:

Merged representation of fields and extras.

Return type:

dict[str, Any]

class batter.pipeline.payloads.SystemParams(*, param_outdir: Path | None = None, system_name: str | None = None, protein_input: Path | None = None, system_input: Path | None = None, system_coordinate: Path | None = None, ligand_paths: Dict[str, ~pathlib.Path]=<factory>, yaml_dir: Path | None = None, anchor_atoms: tuple[str, ...]=(), extra_restraints: str | None = None, extra_restraint_fc: float | None = None, extra_conformation_restraints: Path | None = None, **extra_data: Any)

System-level inputs shared by multiple pipeline steps.

This wrapper normalises common fields (paths, anchor atoms, etc.) while still allowing arbitrary extra keys. Paths are converted to pathlib.Path instances, making downstream usage safer.

Parameters:

• param_outdir (Path, optional) – Directory where ligand parameter outputs should be written.

• system_name (str, optional) – Logical system name propagated to child steps.

• protein_input, system_input, system_coordinate (Path, optional) – Paths to the protein topology/coordinate inputs if supplied.

• ligand_paths (dict[str, Path]) – Mapping of ligand identifiers to staged files.

• yaml_dir (Path, optional) – Directory containing the originating YAML (useful for resolving relatives).

• anchor_atoms (tuple[str, …]) – Anchor atom labels used for restraint placement.

• extra_restraints (str, optional) – Optional positional restraint selection string.

• extra_restraint_fc (float, optional) – Force constant (kcal/mol/Å^2) applied to extra_restraints.

• extra_conformation_restraints (Path, optional) – Path to a conformational restraint JSON file.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

anchor_atoms: tuple[str, ...]

copy_with(**updates: Any) → SystemParams

Create a new SystemParams with additional updates.

Parameters:

**updates – Keyword overrides applied atop the existing data.

Returns:

A new instance incorporating the updates.

Return type:

SystemParams

extra_conformation_restraints: Path | None

extra_restraint_fc: float | None

extra_restraints: str | None

get(item: str, default: Any = None) → Any

Safe lookup for a field or extra value with a default.

Parameters:

• item (str) – Key to fetch.

• default (Any, optional) – Value returned when the key is missing or None.

Returns:

Requested value or the default.

Return type:

Any

ligand_paths: Dict[str, Path]

model_config = {'arbitrary_types_allowed': True, 'extra': 'allow'}

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

param_outdir: Path | None

protein_input: Path | None

system_coordinate: Path | None

system_input: Path | None

system_name: str | None

to_mapping() → Dict[str, Any]

Convert the model (including extras) to a plain dictionary.

Returns:

Merged view of standard fields and extras.

Return type:

dict[str, Any]

yaml_dir: Path | None

Runtime Modules

class batter.runtime.portable.Artifact(name: str, relpath: Path, kind: Literal['file', 'dir']='file', sha256: str = '', size: int = 0, meta: Dict[str, ~typing.Any]=<factory>)

Bases: object

A single artifact tracked by the manifest.

Parameters:

• name (str) – Logical name (e.g., “fe/index” or “traj/lig1.zarr”).

• relpath (pathlib.Path) – Path relative to the store root.

• kind ({“file”,”dir”}) – File or directory artifact.

• sha256 (str) – SHA-256 of the file (empty for directories).

• size (int) – Size in bytes (files only; 0 for directories).

• meta (dict) – Free-form metadata (component, lambda, etc.).

kind: Literal['file', 'dir']

meta: Dict[str, Any]

name: str

relpath: Path

sha256: str

size: int

class batter.runtime.portable.ArtifactManifest

Bases: object

In-memory manifest for a portable artifact store.

Notes

• Paths are relative to enable rebasing the store to a new root.

• Serialize with to_dict() / from_dict().

add(art: Artifact, overwrite: bool = False) → None

exists(name: str) → bool

classmethod from_dict(d: Dict[str, Any]) → ArtifactManifest

get(name: str) → Artifact

items() → List[Artifact]

Return all registered artifacts sorted by name.

Returns:

Snapshot of the manifest contents.

Return type:

list[Artifact]

names() → List[str]

to_dict() → Dict[str, Any]

class batter.runtime.portable.ArtifactStore(root: Path | str, manifest_name: str = 'manifest.json')

Bases: object

Portable store with a relocatable root and JSON manifest.

Parameters:

• root (path-like) – Store root directory (e.g., a run’s work directory).

• manifest_name (str) – File name for the manifest JSON under root (default: “manifest.json”).

Examples

>>> store = ArtifactStore("work/at1r_aai")
>>> p = store.put_file(Path("results.txt"), name="fe/latest", dst_rel=Path("fe/results.txt"))
>>> store.save_manifest()
>>> # move directory to a new cluster...
>>> store2 = ArtifactStore("new_root/at1r_aai"); store2.load_manifest()
>>> store2.path("fe/latest")
new_root/at1r_aai/fe/results.txt

list_artifacts(*, prefix: str | None = None, kind: Literal['file', 'dir', None] = None) → List[Artifact]

Inspect manifest entries, optionally filtering by name or kind.

Parameters:

• prefix (str, optional) – When provided, only artifacts whose logical name starts with prefix are returned.

• kind ({‘file’, ‘dir’, None}, optional) – Restrict results to files or directories. None (default) returns both.

Returns:

Matching artifacts in alphabetical order.

Return type:

list[Artifact]

load_manifest() → None

Load the manifest JSON from root.

path(name: str) → Path

Resolve an artifact name to an absolute path under the current root.

put_dir(src_dir: Path, name: str, dst_rel: Path | None = None, overwrite_manifest_entry: bool = False) → Path

Copy a directory under the store and record it in the manifest.

Notes

• No per-file hashing; use put_file() for critical files.

put_file(src: Path, name: str, dst_rel: Path | None = None, overwrite_manifest_entry: bool = False) → Path

Copy a file under the store and record it in the manifest.

Parameters:

• src (path-like) – Source file path (must exist and be a file).

• name (str) – Logical artifact name to register under.

• dst_rel (path-like, optional) – Relative destination path. Defaults to name.replace('/', '_').

• overwrite_manifest_entry (bool) – If True, allows replacing an existing manifest entry with the same name.

Returns:

Absolute destination path.

Return type:

pathlib.Path

rebase(new_root: Path | str) → ArtifactStore

Create a new store view with the same manifest but a different root.

Parameters:

new_root (path-like) – Target root directory.

Returns:

New store pointing to new_root.

Return type:

ArtifactStore

save_manifest() → Path

Write the manifest JSON under root (atomic).

class batter.runtime.fe_repo.FERecord(*, run_id: str, ligand: str, mol_name: str, system_name: str, fe_type: str, temperature: float, method: Literal['mbar', 'ti']='mbar', total_dG: float, total_se: float = 0.0, components: List[str] = <factory>, created_at: str = <factory>, windows: List[WindowResult] = <factory>, canonical_smiles: str | None = None, original_name: str | None = None, original_path: str | None = None, protocol: str = 'abfe', analysis_start_step: int | None = None, n_bootstraps: int | None = None, status: Literal['success', 'failed', 'unbound']='success')

A full FE result bundle (portable, versioned).

Parameters:

• run_id (str) – Unique run identifier.

• ligand (str) – Ligand identifier.

• mol_name (str) – Molecule resname.

• system_name (str) – Logical system name.

• fe_type (str) – Protocol type (e.g., ‘uno_rest’, ‘asfe’).

• temperature (float) – Simulation temperature (K).

• method ({“mbar”,”ti”}) – Integration method.

• total_dG (float) – Total free energy (kcal/mol).

• total_se (float) – Standard error (kcal/mol).

• components (list[str]) – Active components in this run.

• created_at (str) – ISO-8601 timestamp (UTC, Z-suffix).

• windows (list[WindowResult]) – Per-window results.

• canonical_smiles (str, optional) – Canonicalised ligand SMILES captured during parameterization.

• original_name (str, optional) – Original ligand identifier or title when known.

• original_path (str, optional) – Source path of the ligand before staging.

• protocol (str) – Logical protocol used to generate the result (e.g., "abfe").

• analysis_start_step (int, optional) – First production step included in analysis.

• n_bootstraps (int, optional) – Number of MBAR bootstrap resamples used during analysis.

• status ({“success”,”failed”,”unbound”}) – Final status recorded for the ligand.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

analysis_start_step: int | None

canonical_smiles: str | None

components: List[str]

created_at: str

fe_type: str

ligand: str

method: Literal['mbar', 'ti']

model_config = {}

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

mol_name: str

n_bootstraps: int | None

original_name: str | None

original_path: str | None

protocol: str

run_id: str

status: Literal['success', 'failed', 'unbound']

system_name: str

temperature: float

total_dG: float

total_se: float

windows: List[WindowResult]

class batter.runtime.fe_repo.FEResultsRepository(store: ArtifactStore)

index() → DataFrame

ligand_dir(run_id: str, ligand: str) → Path

load(run_id: str, ligand: str) → FERecord

record_failure(run_id: str, ligand: str, system_name: str, temperature: float, *, status: Literal['failed', 'unbound'], reason: str | None = None, canonical_smiles: str | None = None, original_name: str | None = None, original_path: str | None = None, protocol: str = 'abfe', analysis_start_step: int | None = None, n_bootstraps: int | None = None) → None

save(rec: FERecord, copy_from: Path | None = None) → None

class batter.runtime.fe_repo.WindowResult(*, component: str, lam: float, dG: float, dG_se: float = 0.0, n_samples: int = 0, meta: Dict[str, ~typing.Any]=<factory>)

Result for a single lambda window/component.

Parameters:

• component (str) – Component key (e.g., ‘e’, ‘v’, ‘z’).

• lam (float) – Lambda value in [0, 1].

• dG (float) – Free-energy increment (kcal/mol).

• dG_se (float) – Standard error (kcal/mol).

• n_samples (int) – Samples (or effective sample size).

• meta (dict) – Extra metadata.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

component: str

dG: float

dG_se: float

lam: float

meta: Dict[str, Any]

model_config = {}

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

n_samples: int

Systems Modules

class batter.systems.core.CreateSystemLike(*args, **kwargs)

Structural typing interface for inputs to a system builder.

Notes

This Protocol is intentionally minimal to avoid import cycles with Pydantic models. Any object with these attributes (e.g., a Pydantic model instance) satisfies the protocol.

anchor_atoms: Sequence[str]

ligand_ff: str

ligand_paths: Sequence[Path]

lipid_mol: Sequence[str]

other_mol: Sequence[str]

overwrite: bool

protein_input: Path | None

retain_lig_prot: bool

system_coordinate: Path | None

system_name: str

system_topology: Path | None

class batter.systems.core.SimSystem(name: str, root: Path, topology: Path | None = None, coordinates: Path | None = None, protein: Path | None = None, ligands: Path, ...]=(), lipid_mol: Tuple[str, ...]=(), other_mol: Tuple[str, ...]=(), anchors: Tuple[str, ...]=(), meta: SystemMeta = <factory>)

Immutable descriptor of a simulation system and its on-disk artifacts.

Parameters:

• name (str) – Logical system name (e.g., "AT1R_AAI").

• root (pathlib.Path) – Working directory where artifacts live. This directory is considered relocatable; other modules should store relative paths when possible.

• topology (pathlib.Path, optional) – Path to an explicit topology (e.g., AMBER PRMTOP). May be None if the builder generates it later.

• coordinates (pathlib.Path, optional) – Coordinates or restart file (e.g., RST7/INPCRD).

• protein (pathlib.Path, optional) – Input protein structure file (PDB/mmCIF).

• ligands (tuple[pathlib.Path, …]) – One or more ligand structure files.

• lipid_mol (tuple[str, …]) – Lipid names present in the system (e.g., ("POPC",)).

• other_mol (tuple[str, …]) – Other cofactor present in the system``).

• anchors (tuple[str, …]) – Anchor atoms in the form "RESID@ATOM" (e.g., "85@CA").

• meta (SystemMeta) – Free-form metadata bundle for provenance (e.g., software versions).

anchors: Tuple[str, ...]

coordinates: Path | None

ligands: Tuple[Path, ...]

lipid_mol: Tuple[str, ...]

meta: SystemMeta

name: str

other_mol: Tuple[str, ...]

path(*parts: str | Path) → Path

Join root with the provided path segments.

Parameters:

*parts (str or Path) – Relative path components appended in order.

Returns:

Absolute path pointing inside root.

Return type:

pathlib.Path

protein: Path | None

root: Path

topology: Path | None

with_artifacts(**kw) → SimSystem

Return a new SimSystem with updated artifact attributes.

Examples

>>> sys = SimSystem(name="X", root=Path("work/X"))
>>> sys2 = sys.with_artifacts(topology=Path("work/X/top.prmtop"))

with_meta(**updates: Any) → SimSystem

Return a copy of the system with merged metadata.

Parameters:

**updates – Keyword arguments forwarded to SystemMeta.merge().

Returns:

Copy of the system containing the updated metadata bundle.

Return type:

SimSystem

class batter.systems.core.SystemBuilder(*args, **kwargs)

Interface for creating or updating on-disk artifacts for a system.

build(system, args)

Materialize artifacts for system using args, returning an updated SimSystem. Implementations must be idempotent: calling build twice with the same inputs must produce the same state without corrupting outputs.

build(system: SimSystem, args: CreateSystemLike) → SimSystem

class batter.systems.core.SystemMeta(ligand: str | None = None, residue_name: str | None = None, mode: str | None = None, param_dir_dict: Dict[str, str]=<factory>, extras: Dict[str, ~typing.Any]=<factory>)

Structured metadata attached to a SimSystem.

Parameters:

• ligand (str, optional) – Ligand identifier associated with the system (if applicable).

• residue_name (str, optional) – Residue name used for the ligand.

• mode (str, optional) – High-level mode indicator (e.g., "MABFE" vs "MASFE").

• param_dir_dict (dict[str, str]) – Mapping from residue names to parameter storage directories.

• extras (dict[str, Any]) – Additional context stored alongside the known fields.

extras: Dict[str, Any]

classmethod from_mapping(data: Mapping[str, Any] | None) → SystemMeta

Construct a SystemMeta from a mapping-like object.

Parameters:

data (mapping or None) – Source metadata. If already a SystemMeta, it is returned.

Returns:

Normalised metadata object.

Return type:

SystemMeta

get(key: str, default: Any = None) → Any

Retrieve a value by key with an optional default.

Parameters:

• key (str) – Metadata key.

• default (Any, optional) – Value returned when the key is missing.

Returns:

Stored value or the default.

Return type:

Any

ligand: str | None

merge(**updates: Any) → SystemMeta

Create a new SystemMeta with updated values.

Parameters:

**updates – Keyword overrides applied to the existing metadata.

Returns:

New instance containing the merged metadata.

Return type:

SystemMeta

mode: str | None

param_dir_dict: Dict[str, str]

residue_name: str | None

to_dict() → Dict[str, Any]

Convert the metadata to a plain dictionary.

Returns:

All known fields plus extra entries.

Return type:

dict[str, Any]

class batter.systems.mabfe.MABFEBuilder(*args, **kwargs)

Builder for membrane/absolute free-energy (MABFE) systems.

This builder prepares a shared working directory under system.root and, optionally, stages all ligands at once into per-ligand subfolders.

Directory layout (relative to system.root):

inputs/           # canonical copies of user-provided inputs
artifacts/        # files produced by builders (e.g., PRMTOP, RST7)
simulations/
  <LIG1>/inputs/ligand.<ext>
          artifacts/
  <LIG2>/inputs/ligand.<ext>
          artifacts/
  ...

build(system: SimSystem, args: CreateSystemLike) → SimSystem

Prepare the shared system area (stage protein/topology/coordinates/inputs).

Uses the actual suffixes from user inputs (no hard-coded extensions).

build_all_ligands(parent: SimSystem, lig_paths: Sequence[Path], overwrite: bool = False) → Dict[str, SimSystem]

Stage all ligands at once under parent.root/simulations/<NAME>/....

Ligands are copied as inputs/ligand.<ext> using each source’s suffix.

static make_child_for_ligand(parent: SimSystem, lig_name: str, lig_src: Path) → SimSystem

Create a single per-ligand child system under simulations/<NAME>/ with ligand.<ext>.

batter.systems.mabfe.make_ligand_subsystem(parent: SimSystem, lig_name: str, lig_src: Path) → SimSystem

batter.systems.mabfe.prepare_subsystems_for_ligands(parent: SimSystem, lig_paths: Iterable[Path]) → Dict[str, SimSystem]

class batter.systems.masfe.MASFEBuilder(*args, **kwargs)

Builder for membrane-free (solvation) absolute free-energy (MASFE) systems.

This builder prepares a shared working directory under system.root and, optionally, stages all ligands at once into per-ligand subfolders.

Differences vs MABFE:

• No protein/topology/coordinates are required or staged.

• The resulting SimSystem stores None for protein, topology, and coordinates.

Directory layout (relative to system.root):

inputs/           # canonical copies of user-provided ligand inputs
artifacts/        # files produced by builders
simulations/
  <LIG1>/inputs/ligand.<ext>
          artifacts/
  <LIG2>/inputs/ligand.<ext>
          artifacts/
  ...

build(system: SimSystem, args: CreateSystemLike) → SimSystem

Prepare the shared system area (stage ligand inputs).

Uses the actual suffixes from user inputs (no hard-coded extensions).

build_all_ligands(parent: SimSystem, lig_paths: Sequence[Path], overwrite: bool = False) → Dict[str, SimSystem]

Stage all ligands at once under parent.root/simulations/<NAME>/....

Ligands are copied as inputs/ligand.<ext> using each source’s suffix.

static make_child_for_ligand(parent: SimSystem, lig_name: str, lig_src: Path) → SimSystem

Create a single per-ligand child system under simulations/<NAME>/ with ligand.<ext>.

batter.systems.masfe.make_ligand_subsystem_masfe(parent: SimSystem, lig_name: str, lig_src: Path) → SimSystem

batter.systems.masfe.prepare_subsystems_for_ligands_masfe(parent: SimSystem, lig_paths: Iterable[Path]) → Dict[str, SimSystem]

Analysis Modules

Utilities for inspecting replica-exchange simulations.

class batter.analysis.remd.RemdLog(inputfile: str)

Bases: object

Read and analyse AMBER remlog files.

The parser reconstructs the replica $leftrightarrow$ state mapping at each exchange step and reports high-level metrics such as average single-pass duration and the number of round trips.

Parameters:

inputfile (str) – Path to the remlog text file produced by AMBER.

analyze() → Dict[str, float | List[float]]

Summarise the replica trajectory.

Returns:

Dictionary with the same keys as get_remd_info().

Return type:

dict

classmethod get_remd_info(inputfile: str) → Dict[str, float | List[float]]

Convenience helper that parses and analyses a remlog file.

Parameters:

inputfile (str) – Path to the remlog text file.

Returns:

Same structure as analyze().

Return type:

dict

batter.analysis.remd.plot_trajectory(replica_trajectory, figsize=(10, 6), alpha=0.8, linewidth=1.5, subplot=False, ncols=4)

Visualise the replica walk through thermodynamic states.

Parameters:

• replica_trajectory (numpy.ndarray) – Array of shape (n_replica, n_step + 1) containing state indices.

• figsize (tuple, optional) – Base figure size. When subplot=True the width/height apply to each panel instead of the aggregate.

• alpha (float, optional) – Line transparency used for individual replica traces.

• linewidth (float, optional) – Width of trajectory lines.

• subplot (bool, optional) – When True, render one subplot per replica; otherwise plot all replicas on a shared axis.

• ncols (int, optional) – Number of subplot columns when subplot=True.

Small numerical helpers used across batter.analysis.

batter.analysis.utils.MakeChunksWithSize(istart: int, istop: int, size: int) → List[List[int]]

Build index chunks covering [istart, istop) with approximately size elements.

Parameters:

• istart (int) – Starting index (inclusive).

• istop (int) – Stopping index (exclusive).

• size (int) – Target chunk size prior to merging trailing fragments.

Returns:

Collection of contiguous index lists.

Return type:

list[list[int]]

batter.analysis.utils.MakeGroupedChunks(ene: ndarray, size: int) → List[List[int]]

Merge adjacent chunks when their means are statistically indistinguishable.

Parameters:

• ene (numpy.ndarray) – One-dimensional array containing the energy trace used for grouping.

• size (int) – Requested minimum chunk size prior to the adaptive merge step.

Returns:

List of index groups representing contiguous frames with similar means.

Return type:

list[list[int]]

batter.analysis.utils.SizedChunks(lst: Iterable[int], n: int) → Generator[List[int], None, None]

Yield successive n-sized chunks from an iterable.

Parameters:

• lst (Iterable[int]) – Source iterable that should be partitioned. The iterable is consumed, so pass a sequence (e.g. range) if it needs to be reused.

• n (int) – Requested chunk size.

Yields:

list[int] – Consecutive slices of length n (the final chunk may be shorter).

batter.analysis.utils.exclude_outliers(df: DataFrame, iclam: int) → DataFrame

Remove energy spikes that would destabilise MBAR fits.

Parameters:

• df (pandas.DataFrame) – Reduced potential values with time points along the rows and lambda states in the columns.

• iclam (int) – Index of the reference lambda column. The algorithm analyses this column to decide which trajectory chunks should be discarded.

Returns:

Filtered dataframe with the same columns as df but potentially fewer rows if outliers were detected.

Return type:

pandas.DataFrame

Notes

The implementation mirrors the heuristics used in the original fe-toolkit scripts: frames are chunked into ~200-sample blocks, grouped via a Welch t-test, and discarded whenever any lambda exhibits a value more than 3σ + 1000 kcal/mol below the reference median (after correcting for mixed precision offsets).