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).

pydantic model batter.api.FERecord

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").

• sim_range (tuple[int, int], optional) – (start, end) lambda range used for 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.

Fields:

• canonical_smiles (str | None)

• components (List[str])

• created_at (str)

• fe_type (str)

• ligand (str)

• method (Literal['mbar', 'ti'])

• mol_name (str)

• original_name (str | None)

• original_path (str | None)

• protocol (str)

• run_id (str)

• sim_range (tuple[int, int] | None)

• status (Literal['success', 'failed', 'unbound'])

• system_name (str)

• temperature (float)

• total_dG (float)

• total_se (float)

• windows (List[batter.runtime.fe_repo.WindowResult])

field canonical_smiles: str | None = None

field components: List[str] [Optional]

field created_at: str [Optional]

field fe_type: str [Required]

field ligand: str [Required]

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

field mol_name: str [Required]

field original_name: str | None = None

field original_path: str | None = None

field protocol: str = 'abfe'

field run_id: str [Required]

field sim_range: tuple[int, int] | None = None

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

field system_name: str [Required]

field temperature: float [Required]

field total_dG: float [Required]

field total_se: float = 0.0

field windows: List[WindowResult] [Optional]

class batter.api.FEResultsRepository(store: ArtifactStore)

Bases: object

index() → DataFrame

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', sim_range: tuple[int, int] | None = None) → None

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

pydantic model batter.api.RunConfig

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.

Config:

• extra: str = forbid

Fields:

• backend (Literal['local', 'slurm'])

• create (batter.config.run.CreateArgs)

• fe_sim (Dict[str, Any] | batter.config.run.FESimArgs | batter.config.run.MDSimArgs)

• protocol (Literal['abfe', 'asfe', 'md'])

• run (batter.config.run.RunSection)

• version (int)

Validators:

• _coerce_fe_sim_model » all fields

• _lower_backend » backend

• _lower_protocol » protocol

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

Execution backend.

Validated by:

• _coerce_fe_sim_model

• _lower_backend

field create: CreateArgs [Required]

Settings for system creation/staging.

Validated by:

• _coerce_fe_sim_model

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

Simulation parameter overrides.

Validated by:

• _coerce_fe_sim_model

field protocol: Literal['abfe', 'asfe', 'md'] = 'abfe'

High-level protocol to execute.

Validated by:

• _coerce_fe_sim_model

• _lower_protocol

field run: RunSection [Required]

Execution controls and artifact destination.

Validated by:

• _coerce_fe_sim_model

field version: int = 1

Schema version of the run configuration.

Validated by:

• _coerce_fe_sim_model

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

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

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

with_base_dir(base_dir: Path) → RunConfig

Return a copy with relative paths resolved against base_dir.

class batter.api.SimSystem(name: str, root: ~pathlib.Path, topology: ~pathlib.Path | None = None, coordinates: ~pathlib.Path | None = None, protein: ~pathlib.Path | None = None, ligands: ~typing.Tuple[~pathlib.Path, ...] = (), lipid_mol: ~typing.Tuple[str, ...] = (), other_mol: ~typing.Tuple[str, ...] = (), anchors: ~typing.Tuple[str, ...] = (), meta: ~batter.systems.core.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

pydantic model batter.api.SimulationConfig

Bases: BaseModel

Simulation configuration for ABFE/ASFE 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.

Fields:

• all_atoms (Literal['yes', 'no'])

• analysis_fe_range (Tuple[int, int] | None)

• 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_steps1 (Dict[str, int])

• dic_steps2 (Dict[str, int])

• dt (float)

• enable_mcwat (Literal['yes', 'no'])

• eq_steps (int)

• eq_steps1 (int)

• eq_steps2 (int)

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

• 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)

• n_steps_dict (Dict[str, int])

• neut (str)

• neutralize_only (Literal['yes', 'no'])

• ntpr (int)

• ntwe (int)

• ntwr (int)

• ntwx (int)

• num_equil_extends (int)

• num_fe_extends (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)

• remd_numexchg (int)

• rest (List[float])

• rng (int)

• rocklin_correction (Literal['yes', 'no'])

• sdr_dist (float | None)

• solv_shell (float | None)

• system_name (str)

• temperature (float)

• ti_points (int | None)

• unbound_threshold (float)

• water_model (Literal['SPCE', 'TIP4PEW', 'TIP3P', 'TIP3PF', 'OPC'])

Validators:

• _coerce_barostat » barostat

• _coerce_yes_no » enable_mcwat

• _coerce_yes_no » hmr

• _coerce_yes_no » neutralize_only

• _coerce_yes_no » remd

• _coerce_yes_no » rocklin_correction

• _finalize » all fields

• _lower_enums » dec_int

• _lower_enums » fe_type

• _parse_lambdas » lambdas

• _validate_anchor » p1

• _validate_anchor » p2

• _validate_anchor » p3

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

save all atoms for FE

Validated by:

• _finalize

field analysis_fe_range: Tuple[int, int] | None = (2, -1)

Optional tuple (start, end) limiting FE simulations analyzed per window.

Validated by:

• _finalize

field anion: str = 'Cl-'

Anion species

Validated by:

• _finalize

field barostat: Literal[1, 2] = 2

1=Berendsen, 2=MC barostat

Validated by:

• _coerce_barostat

• _finalize

field blocks: int = 0

MBAR blocks

Validated by:

• _finalize

field buffer_x: float = 10.0

Box buffer X (Å)

Validated by:

• _finalize

field buffer_y: float = 10.0

Box buffer Y (Å)

Validated by:

• _finalize

field buffer_z: float = 10.0

Box buffer Z (Å)

Validated by:

• _finalize

field cation: str = 'Na+'

Cation species

Validated by:

• _finalize

field component_lambdas: Dict[str, List[float]] [Optional]

Lambda schedule for each component

Validated by:

• _finalize

field component_windows: Dict[str, List[float]] [Optional]

Per-component lambda values for overrides

Validated by:

• _finalize

field components: List[str] [Optional]

List of components (v, o, z, etc.)

Validated by:

• _finalize

field cut: float = 9.0

Nonbonded cutoff (Å)

Validated by:

• _finalize

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

Integration method (mbar/ti)

Validated by:

• _finalize

• _lower_enums

field dec_method: str | None = None

Decoupling method (set for fe_type=’custom’)

Validated by:

• _finalize

field dic_steps1: Dict[str, int] [Optional]

Stage1 steps per component

Validated by:

• _finalize

field dic_steps2: Dict[str, int] [Optional]

Stage2 steps per component

Validated by:

• _finalize

field dt: float = 0.004

Time step (ps)

Validated by:

• _finalize

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

Enable MC water exchange moves during equilibration templates.

Validated by:

• _coerce_yes_no

• _finalize

field eq_steps: int = 1000000

Steps per equilibration segment (derived)

Validated by:

• _finalize

field eq_steps1: int = 500000

Equilibration stage 1 steps (legacy mirror of eq_steps)

Validated by:

• _finalize

field eq_steps2: int = 1000000

Equilibration stage 2 steps (legacy mirror of eq_steps)

Validated by:

• _finalize

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

Free-energy protocol type

Validated by:

• _finalize

• _lower_enums

field gamma_ln: float = 1.0

Langevin γ (ps^-1)

Validated by:

• _finalize

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

Hydrogen mass repartitioning

Validated by:

• _coerce_yes_no

• _finalize

field infe: bool = False

Enable NFE (infinite) equilibration when true.

Validated by:

• _finalize

field ion_conc: float = 0.15

Target salt concentration (M)

Validated by:

• _finalize

field ion_def: List[Any] [Optional]

Ion tuple [cation, anion, conc]

Validated by:

• _finalize

field l1_range: float | None = None

L1 search radius (Å)

Validated by:

• _finalize

field l1_x: float | None = None

L1 center offset X (Å)

Validated by:

• _finalize

field l1_y: float | None = None

L1 center offset Y (Å)

Validated by:

• _finalize

field l1_z: float | None = None

L1 center offset Z (Å)

Validated by:

• _finalize

field lambdas: List[float] [Optional]

default lambda values

Validated by:

• _finalize

• _parse_lambdas

field lig_angle_force: float = 0.0

Ligand angle/dihedral spring (kcal/mol/rad^2)

Validated by:

• _finalize

field lig_buffer: float = 10.0

Ligand box buffer (Å)

Validated by:

• _finalize

field lig_com_force: float = 0.0

Ligand COM spring

Validated by:

• _finalize

field lig_dihcf_force: float = 0.0

Ligand dihedral spring (kcal/mol/rad^2)

Validated by:

• _finalize

field lig_distance_force: float = 0.0

Ligand COM distance spring (kcal/mol/Å^2)

Validated by:

• _finalize

field ligand_dict: Dict[str, Any] [Optional]

Ligand dictionary

Validated by:

• _finalize

field ligand_ff: str = 'gaff2'

Ligand FF

Validated by:

• _finalize

field lipid_ff: str = 'lipid21'

Lipid FF

Validated by:

• _finalize

field lipid_mol: List[str] [Optional]

Lipid molecules

Validated by:

• _finalize

field max_adis: float | None = None

Max anchor distance (Å)

Validated by:

• _finalize

field membrane_simulation: bool = True

Whether system includes a membrane

Validated by:

• _finalize

field min_adis: float | None = None

Min anchor distance (Å)

Validated by:

• _finalize

field n_steps_dict: Dict[str, int] [Optional]

Per-component steps (keys: ‘{comp}_steps1|2’)

Validated by:

• _finalize

field neut: str = ''

Alias of neutralize_only

Validated by:

• _finalize

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

Neutralize only

Validated by:

• _coerce_yes_no

• _finalize

field ntpr: int = 1000

Print energy every ntpr steps

Validated by:

• _finalize

field ntwe: int = 0

Write energy every ntwe steps

Validated by:

• _finalize

field ntwr: int = 10000

Write restart every ntwr steps

Validated by:

• _finalize

field ntwx: int = 2500

Write trajectory every ntwx steps

Validated by:

• _finalize

field num_equil_extends: int = 0

Number of equilibration extends (derived)

Validated by:

• _finalize

field num_fe_extends: int = 10

# restarts per λ

Validated by:

• _finalize

field other_mol: List[str] [Optional]

Other co-binders

Validated by:

• _finalize

field p1: str = ''

Anchor P1 “RESID@ATOM” (e.g., “85@CA”)

Validated by:

• _finalize

• _validate_anchor

field p2: str = ''

Anchor P2 “RESID@ATOM”

Validated by:

• _finalize

• _validate_anchor

field p3: str = ''

Anchor P3 “RESID@ATOM”

Validated by:

• _finalize

• _validate_anchor

field protein_align: str = 'name CA'

Alignment selection

Validated by:

• _finalize

field rec_com_force: float = 0.0

Protein COM spring

Validated by:

• _finalize

field receptor_ff: str = 'protein.ff14SB'

Receptor FF

Validated by:

• _finalize

field receptor_segment: str | None = None

Segment to embed in membrane

Validated by:

• _finalize

field release_eq: List[float] [Optional]

Equilibration release weights (derived)

Validated by:

• _finalize

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

H-REMD toggle

Validated by:

• _coerce_yes_no

• _finalize

field remd_nstlim: int = 100

Steps per REMD segment (applied to mdin-*-remd copies).

Validated by:

• _finalize

field remd_numexchg: int = 3000

Exchange attempt interval for REMD (numexchg).

Validated by:

• _finalize

field rest: List[float] [Optional]

Packed restraint constants

Validated by:

• _finalize

field rng: int = 0

Range of release_eq

Validated by:

• _finalize

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

Rocklin correction

Validated by:

• _coerce_yes_no

• _finalize

field sdr_dist: float | None = 0.0

SDR placement distance (Å)

Validated by:

• _finalize

field solv_shell: float | None = 15.0

Initial solvent shell radius (Å)

Validated by:

• _finalize

field system_name: str [Required]

System name (required)

Validated by:

• _finalize

field temperature: float = 298.15

Temperature (K)

Validated by:

• _finalize

field ti_points: int | None = 0

1. TI points (not implemented)

Validated by:

• _finalize

field unbound_threshold: float = 8.0

Distance (Å) between ligand COMs that classifies equilibration as unbound.

Constraints:

• ge = 0.0

Validated by:

• _finalize

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

Water model

Validated by:

• _finalize

classmethod from_sections(create: CreateArgs, fe: FESimArgs, *, protocol: str | None = None, fe_type: str | None = None, slurm_header_dir: Path | 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.

Returns:

Fully merged simulation configuration ready for downstream use.

Return type:

SimulationConfig

to_dict() → Dict[str, Any]

pydantic model batter.api.WindowResult

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.

Fields:

• component (str)

• dG (float)

• dG_se (float)

• lam (float)

• meta (Dict[str, Any])

• n_samples (int)

field component: str [Required]

field dG: float [Required]

field dG_se: float = 0.0

field lam: float [Required]

field meta: Dict[str, Any] [Optional]

field n_samples: int = 0

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, sim_range, 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, *, ligand: str | None = None, components: Sequence[str] | None = None, n_workers: int | None = None, sim_range: tuple[int, int] | None = None, 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) – Identifier of the execution (e.g., run-20240101).

• 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.

• sim_range ((int, int), optional) – (start, end) range of lambda windows to analyze.

• 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.

Parameters:

• path (str or pathlib.Path) – Path to the top-level run YAML file.

• on_failure ({“prune”, “raise”, “retry”}, optional) – Override for the failure policy applied to ligand pipelines.

• run_overrides (dict, optional) – Overrides applied to the run section (e.g., only FE preparation).

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

pydantic model batter.config.run.CreateArgs

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.

Config:

• extra: str = forbid

Fields:

• anchor_atoms (list[str])

• anion (str)

• cation (str)

• extra_conformation_restraints (pathlib.Path | None)

• extra_restraint_fc (float)

• extra_restraints (str | None)

• ion_conc (float)

• l1_range (float)

• ligand_ff (str)

• ligand_input (pathlib.Path | None)

• ligand_paths (dict[str, pathlib.Path | str])

• lipid_ff (str)

• lipid_mol (list[str])

• max_adis (float)

• min_adis (float)

• neutralize_only (Literal['yes', 'no'])

• other_mol (list[str])

• overwrite (bool)

• param_charge (str)

• param_method (Literal['amber', 'openff'])

• param_outdir (pathlib.Path | None)

• protein_align (str | None)

• protein_input (pathlib.Path | None)

• receptor_ff (str)

• retain_lig_prot (bool)

• solv_shell (float)

• system_coordinate (pathlib.Path | None)

• system_input (pathlib.Path | None)

• system_name (str | None)

• water_model (str)

Validators:

• _check_extra_restraints » all fields

• _coerce_create_yes_no » neutralize_only

• _coerce_ligand_input » ligand_input

• _coerce_opt_paths » param_outdir

• _coerce_opt_paths » protein_input

• _coerce_opt_paths » system_coordinate

• _coerce_opt_paths » system_input

• _normalize_ligand_paths » ligand_paths

• _require_ligands » all fields

field anchor_atoms: list[str] [Optional]

List of anchor atom selections used for restraint placement.

Validated by:

• _check_extra_restraints

• _require_ligands

field anion: str = 'Cl-'

Anion species for ion placement.

Validated by:

• _check_extra_restraints

• _require_ligands

field cation: str = 'Na+'

Cation species for ion placement.

Validated by:

• _check_extra_restraints

• _require_ligands

field extra_conformation_restraints: Path | None = None

Path to conformational restraint JSON (used for NFE workflows).

Validated by:

• _check_extra_restraints

• _require_ligands

field extra_restraint_fc: float = 10.0

Force constant (kcal/mol/Å^2) applied to extra_restraints.

Validated by:

• _check_extra_restraints

• _require_ligands

field extra_restraints: str | None = None

Optional positional restraint specification string.

Validated by:

• _check_extra_restraints

• _require_ligands

field ion_conc: float = 0.15

Target salt concentration (M).

Validated by:

• _check_extra_restraints

• _require_ligands

field l1_range: float = 6.0

Radius (Å) for L1 search when identifying pocket positions.

Validated by:

• _check_extra_restraints

• _require_ligands

field ligand_ff: str = 'gaff2'

Ligand force field identifier passed to parameterization tools.

Validated by:

• _check_extra_restraints

• _require_ligands

field ligand_input: Path | None = None

Alternative JSON file describing ligands (dict or list).

Validated by:

• _check_extra_restraints

• _coerce_ligand_input

• _require_ligands

field ligand_paths: dict[str, Path | str] [Optional]

Mapping of ligand identifiers to structure files (relative paths are resolved at runtime).

Validated by:

• _check_extra_restraints

• _normalize_ligand_paths

• _require_ligands

field lipid_ff: str = 'lipid21'

Lipid force-field identifier.

Validated by:

• _check_extra_restraints

• _require_ligands

field lipid_mol: list[str] [Optional]

Names of lipid molecules present in the system.

Validated by:

• _check_extra_restraints

• _require_ligands

field max_adis: float = 7.0

Maximum anchor-atom distance used during pose selection (Å).

Validated by:

• _check_extra_restraints

• _require_ligands

field min_adis: float = 3.0

Minimum anchor-atom distance used during pose selection (Å).

Validated by:

• _check_extra_restraints

• _require_ligands

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

If "yes", neutralize the system without adding bulk salt.

Validated by:

• _check_extra_restraints

• _coerce_create_yes_no

• _require_ligands

field other_mol: list[str] [Optional]

Names of non-lipid cofactors or co-binders.

Validated by:

• _check_extra_restraints

• _require_ligands

field overwrite: bool = True

If true, overwrite existing artifacts in the staging directory.

Validated by:

• _check_extra_restraints

• _require_ligands

field param_charge: str = 'am1bcc'

Charge derivation method for ligands.

Validated by:

• _check_extra_restraints

• _require_ligands

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

Parameterization backend to use for ligands.

Validated by:

• _check_extra_restraints

• _require_ligands

field param_outdir: Path | None = None

Optional override for the ligand parameter output directory.

Validated by:

• _check_extra_restraints

• _coerce_opt_paths

• _require_ligands

field protein_align: str | None = 'name CA'

Selection string used to align the protein prior to staging.

Validated by:

• _check_extra_restraints

• _require_ligands

field protein_input: Path | None = None

Path to the receptor structure (PDB/mmCIF).

Validated by:

• _check_extra_restraints

• _coerce_opt_paths

• _require_ligands

field receptor_ff: str = 'protein.ff14SB'

Protein force-field identifier.

Validated by:

• _check_extra_restraints

• _require_ligands

field retain_lig_prot: bool = True

Whether to retain ligand protomers generated during staging.

Validated by:

• _check_extra_restraints

• _require_ligands

field solv_shell: float = 15.0

Initial solvent shell radius (Å).

Validated by:

• _check_extra_restraints

• _require_ligands

field system_coordinate: Path | None = None

Optional starting coordinates (e.g., INPCRD/RST7).

Validated by:

• _check_extra_restraints

• _coerce_opt_paths

• _require_ligands

field system_input: Path | None = None

Optional pre-built system topology (e.g., PRMTOP).

Validated by:

• _check_extra_restraints

• _coerce_opt_paths

• _require_ligands

field system_name: str | None = 'unnamed_system'

Logical system name; used to label outputs when not provided.

Validated by:

• _check_extra_restraints

• _require_ligands

field water_model: str = 'TIP3P'

Water model used for solvation.

Validated by:

• _check_extra_restraints

• _require_ligands

resolve_paths(base: Path) → CreateArgs

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

pydantic model batter.config.run.FESimArgs

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.

Config:

• extra: str = forbid

Fields:

• analysis_fe_range (Tuple[int, int] | None)

• 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)

• ntpr (int)

• ntwe (int)

• ntwr (int)

• ntwx (int)

• num_equil_extends (int)

• num_fe_extends (int)

• rec_com_force (float)

• remd (batter.config.remd.RemdArgs)

• remd_enable (Literal['yes', 'no'])

• rocklin_correction (Literal['yes', 'no'])

• steps1 (Dict[str, int])

• steps2 (Dict[str, int])

• temperature (float)

• unbound_threshold (float)

Validators:

• _coerce_fe_yes_no » enable_mcwat

• _coerce_fe_yes_no » hmr

• _coerce_fe_yes_no » rocklin_correction

• _coerce_remd » remd

• _coerce_remd_enable » remd_enable

• _ingest_component_lambda_fields » all fields

• _ingest_legacy_step_fields » all fields

• _ingest_remd_enable » all fields

• _validate_analysis_range » analysis_fe_range

• _validate_force_const » lig_angle_force

• _validate_force_const » lig_com_force

• _validate_force_const » lig_distance_force

• _validate_force_const » rec_com_force

• _validate_lambdas » lambdas

field analysis_fe_range: Tuple[int, int] | None = None

Optional (start, end) simulation index range to analyze per FE window.

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

• _validate_analysis_range

field barostat: int = 2

Barostat selection (1=Berendsen, 2=MC).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field blocks: int = 0

Number of MBAR blocks to use during analysis.

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field buffer_x: float = 20.0

Box padding along X (Å).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field buffer_y: float = 20.0

Box padding along Y (Å).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field buffer_z: float = 20.0

Box padding along Z (Å).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field component_lambdas: Dict[str, List[float]] [Optional]

Per-component lambda overrides (key = letter).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field cut: float = 9.0

Nonbonded cutoff (Å).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field dec_int: str = 'mbar'

Free-energy integration scheme (mbar or ti).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field dt: float = 0.004

MD timestep (ps).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

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

Enable MC water exchange moves during equilibration (1 = on).

Validated by:

• _coerce_fe_yes_no

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field eq_steps: int = 1000000

Steps per equilibration segment (applied to the initial run and each extend).

Constraints:

• gt = 0

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field gamma_ln: float = 1.0

Langevin gamma value (ps^-1).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

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

Hydrogen mass repartitioning toggle.

Validated by:

• _coerce_fe_yes_no

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field lambdas: List[float] [Optional]

Default lambda schedule when component-specific overrides are not provided.

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

• _validate_lambdas

field lig_angle_force: float = 250.0

Ligand angle restraint spring constant (kcal/mol/rad^2).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

• _validate_force_const

field lig_buffer: float = 15.0

Ligand-specific box buffer (Å) for solvation boxes.

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field lig_com_force: float = 10.0

Ligand COM restraint spring constant (kcal/mol/Å^2).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

• _validate_force_const

field lig_dihcf_force: float = 0.0

Ligand dihedral restraint spring constant (kcal/mol/rad^2).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field lig_distance_force: float = 5.0

Ligand COM distance restraint spring constant (kcal/mol/Å^2).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

• _validate_force_const

field ntpr: int = 1000

Energy print frequency.

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field ntwe: int = 0

Energy write frequency (0 disables).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field ntwr: int = 2500

Restart write frequency.

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field ntwx: int = 25000

Trajectory write frequency.

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field num_equil_extends: int = 6

Number of additional equilibration segments (all run fully released).

Constraints:

• ge = 0

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field num_fe_extends: int = 10

# restarts per λ (controls how many FE simulations run per window).

Constraints:

• ge = 1

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field rec_com_force: float = 10.0

Protein COM restraint spring constant (kcal/mol/Å^2).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

• _validate_force_const

field remd: RemdArgs [Optional]

Replica-exchange MD controls (nstlim/numexchg).

Validated by:

• _coerce_remd

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field remd_enable: Literal['yes', 'no'] = 'no'

Toggle REMD execution (yes/no).

Validated by:

• _coerce_remd_enable

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

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

Apply Rocklin correction during analysis.

Validated by:

• _coerce_fe_yes_no

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field steps1: Dict[str, int] [Optional]

Stage 1 steps per component (key = letter).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field steps2: Dict[str, int] [Optional]

Stage 2 steps per component (key = letter).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field temperature: float = 298.15

Simulation temperature (K).

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

field unbound_threshold: float = 8.0

Distance threshold (Å) used to flag ligands as unbound during equilibration analysis.

Constraints:

• ge = 0.0

Validated by:

• _ingest_component_lambda_fields

• _ingest_legacy_step_fields

• _ingest_remd_enable

pydantic model batter.config.run.MDSimArgs

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.

Config:

• extra: str = forbid

Fields:

• barostat (int)

• cut (float)

• dt (float)

• enable_mcwat (Literal['yes', 'no'])

• eq_steps (int)

• gamma_ln (float)

• hmr (Literal['yes', 'no'])

• ntpr (int)

• ntwe (int)

• ntwr (int)

• ntwx (int)

• num_equil_extends (int)

• temperature (float)

Validators:

• _coerce_hmr » enable_mcwat

• _coerce_hmr » hmr

field barostat: int = 2

Barostat selection (1=Berendsen, 2=MC).

field cut: float = 9.0

Nonbonded cutoff (Å).

field dt: float = 0.004

MD timestep (ps).

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

Enable MC water exchange moves during equilibration (1 = on).

Validated by:

• _coerce_hmr

field eq_steps: int = 100000

Steps per equilibration segment.

Constraints:

• gt = 0

field gamma_ln: float = 1.0

Langevin gamma value (ps^-1).

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

Hydrogen mass repartitioning toggle.

Validated by:

• _coerce_hmr

field ntpr: int = 1000

Energy print frequency.

field ntwe: int = 0

Energy write frequency (0 disables).

field ntwr: int = 10000

Restart write frequency.

field ntwx: int = 25000

Trajectory write frequency.

field num_equil_extends: int = 2

Number of additional equilibration segments (all run fully released).

Constraints:

• ge = 0

field temperature: float = 298.15

Simulation temperature (K).

pydantic model batter.config.run.RunConfig

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.

Config:

• extra: str = forbid

Fields:

• backend (Literal['local', 'slurm'])

• create (CreateArgs)

• fe_sim (Dict[str, Any] | FESimArgs | MDSimArgs)

• protocol (Literal['abfe', 'asfe', 'md'])

• run (RunSection)

• version (int)

Validators:

• _coerce_fe_sim_model » all fields

• _lower_backend » backend

• _lower_protocol » protocol

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

Execution backend.

Validated by:

• _coerce_fe_sim_model

• _lower_backend

field create: CreateArgs [Required]

Settings for system creation/staging.

Validated by:

• _coerce_fe_sim_model

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

Simulation parameter overrides.

Validated by:

• _coerce_fe_sim_model

field protocol: Literal['abfe', 'asfe', 'md'] = 'abfe'

High-level protocol to execute.

Validated by:

• _coerce_fe_sim_model

• _lower_protocol

field run: RunSection [Required]

Execution controls and artifact destination.

Validated by:

• _coerce_fe_sim_model

field version: int = 1

Schema version of the run configuration.

Validated by:

• _coerce_fe_sim_model

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

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

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

with_base_dir(base_dir: Path) → RunConfig

Return a copy with relative paths resolved against base_dir.

pydantic model batter.config.run.RunSection

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.

Config:

• extra: str = forbid

Fields:

• allow_run_id_mismatch (bool)

• batch_gpus (int | None)

• batch_gpus_per_task (int)

• batch_mode (bool)

• batch_srun_extra (List[str])

• dry_run (bool)

• email_on_completion (str | None)

• email_sender (str)

• max_active_jobs (int | None)

• max_workers (int | None)

• on_failure (Literal['raise', 'prune', 'retry'])

• only_fe_preparation (bool)

• output_folder (pathlib.Path)

• remd (Literal['yes', 'no'])

• run_id (str)

• slurm (batter.config.run.SlurmConfig)

• slurm_header_dir (pathlib.Path | None)

• system_type (Literal['MABFE', 'MASFE'] | None)

Validators:

• _coerce_output_folder » output_folder

• _coerce_remd » remd

• _normalize_system_type » system_type

field allow_run_id_mismatch: bool = False

When True, allow reusing an explicit run_id even if the configuration hash differs from the existing execution.

field batch_gpus: int | None = None

GPUs available to the manager process for batch_mode; auto-detected from SLURM env when omitted.

Constraints:

• ge = 0

field batch_gpus_per_task: int = 1

GPUs to assign per task when batch_mode is enabled.

Constraints:

• ge = 1

field batch_mode: bool = False

When true, run SLURM jobs inline via srun inside the manager allocation instead of submitting with sbatch.

field batch_srun_extra: List[str] [Optional]

Extra srun flags appended when launching tasks in batch_mode.

field dry_run: bool = False

Force dry-run mode regardless of YAML setting.

field email_on_completion: str | None = None

Email address that should receive a notification once the run finishes (successfully or with warnings).

field email_sender: str = 'nobody@stanford.edu'

Sender address used for completion email notifications.

field max_active_jobs: int | None = 1000

Max concurrent SLURM jobs for FE submissions (0 disables throttling).

Constraints:

• ge = 0

field max_workers: int | None = None

Parallel workers for local backend (None = auto, 0 = serial).

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

Behavior on ligand failure: ‘raise’, ‘prune’, or ‘retry’ (clear FAILED sentinels and rerun once).

field only_fe_preparation: bool = False

When true, stop the workflow after FE preparation.

field output_folder: Path [Required]

Directory where system artifacts and executions are stored.

Validated by:

• _coerce_output_folder

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

Enable REMD execution (templates are always prepared).

Validated by:

• _coerce_remd

field run_id: str = 'auto'

Run identifier to use (auto picks latest).

field slurm: SlurmConfig [Optional]

field slurm_header_dir: Path | None = None

Optional directory containing user Slurm headers (defaults to ~/.batter).

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

Optional override for the system builder. When omitted, the orchestrator chooses the builder based on the protocol.

Validated by:

• _normalize_system_type

resolve_paths(base: Path) → RunSection

Return a copy where output_folder is absolute relative to base.

pydantic model batter.config.run.SlurmConfig

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.

Config:

• extra: str = ignore

Fields:

• account (str | None)

• constraint (str | None)

• extra_sbatch (List[str])

• gres (str | None)

• mem_per_cpu (str | None)

• nodes (int | None)

• ntasks_per_node (int | None)

• partition (str | None)

• qos (str | None)

• time (str | None)

field account: str | None = None

field constraint: str | None = None

field extra_sbatch: List[str] [Optional]

field gres: str | None = None

field mem_per_cpu: str | None = None

field nodes: int | None = None

field ntasks_per_node: int | None = None

field partition: str | None = None

SLURM partition / queue

field qos: str | None = None

field time: str | None = None

Walltime, e.g. ‘04:00:00’

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

pydantic model batter.config.simulation.SimulationConfig

Bases: BaseModel

Simulation configuration for ABFE/ASFE 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.

Fields:

• all_atoms (Literal['yes', 'no'])

• analysis_fe_range (Optional[Tuple[int, 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 (Optional[str])

• dic_steps1 (Dict[str, int])

• dic_steps2 (Dict[str, int])

• dt (float)

• enable_mcwat (Literal['yes', 'no'])

• eq_steps (int)

• eq_steps1 (int)

• eq_steps2 (int)

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

• gamma_ln (float)

• hmr (Literal['yes', 'no'])

• infe (bool)

• ion_conc (float)

• ion_def (List[Any])

• l1_range (Optional[float])

• l1_x (Optional[float])

• l1_y (Optional[float])

• l1_z (Optional[float])

• 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 (Optional[float])

• membrane_simulation (bool)

• min_adis (Optional[float])

• n_steps_dict (Dict[str, int])

• neut (str)

• neutralize_only (Literal['yes', 'no'])

• ntpr (int)

• ntwe (int)

• ntwr (int)

• ntwx (int)

• num_equil_extends (int)

• num_fe_extends (int)

• other_mol (List[str])

• p1 (str)

• p2 (str)

• p3 (str)

• protein_align (str)

• rec_com_force (float)

• receptor_ff (str)

• receptor_segment (Optional[str])

• release_eq (List[float])

• remd (Literal['yes', 'no'])

• remd_nstlim (int)

• remd_numexchg (int)

• rest (List[float])

• rng (int)

• rocklin_correction (Literal['yes', 'no'])

• sdr_dist (Optional[float])

• solv_shell (Optional[float])

• system_name (str)

• temperature (float)

• ti_points (Optional[int])

• unbound_threshold (float)

• water_model (Literal['SPCE', 'TIP4PEW', 'TIP3P', 'TIP3PF', 'OPC'])

Validators:

• _coerce_barostat » barostat

• _coerce_yes_no » enable_mcwat

• _coerce_yes_no » hmr

• _coerce_yes_no » neutralize_only

• _coerce_yes_no » remd

• _coerce_yes_no » rocklin_correction

• _finalize » all fields

• _lower_enums » dec_int

• _lower_enums » fe_type

• _parse_lambdas » lambdas

• _validate_anchor » p1

• _validate_anchor » p2

• _validate_anchor » p3

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

save all atoms for FE

Validated by:

• _finalize

field analysis_fe_range: Tuple[int, int] | None = (2, -1)

Optional tuple (start, end) limiting FE simulations analyzed per window.

Validated by:

• _finalize

field anion: str = 'Cl-'

Anion species

Validated by:

• _finalize

field barostat: Literal[1, 2] = 2

1=Berendsen, 2=MC barostat

Validated by:

• _coerce_barostat

• _finalize

field blocks: int = 0

MBAR blocks

Validated by:

• _finalize

field buffer_x: float = 10.0

Box buffer X (Å)

Validated by:

• _finalize

field buffer_y: float = 10.0

Box buffer Y (Å)

Validated by:

• _finalize

field buffer_z: float = 10.0

Box buffer Z (Å)

Validated by:

• _finalize

field cation: str = 'Na+'

Cation species

Validated by:

• _finalize

field component_lambdas: Dict[str, List[float]] [Optional]

Lambda schedule for each component

Validated by:

• _finalize

field component_windows: Dict[str, List[float]] [Optional]

Per-component lambda values for overrides

Validated by:

• _finalize

field components: List[str] [Optional]

List of components (v, o, z, etc.)

Validated by:

• _finalize

field cut: float = 9.0

Nonbonded cutoff (Å)

Validated by:

• _finalize

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

Integration method (mbar/ti)

Validated by:

• _finalize

• _lower_enums

field dec_method: str | None = None

Decoupling method (set for fe_type=’custom’)

Validated by:

• _finalize

field dic_steps1: Dict[str, int] [Optional]

Stage1 steps per component

Validated by:

• _finalize

field dic_steps2: Dict[str, int] [Optional]

Stage2 steps per component

Validated by:

• _finalize

field dt: float = 0.004

Time step (ps)

Validated by:

• _finalize

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

Enable MC water exchange moves during equilibration templates.

Validated by:

• _coerce_yes_no

• _finalize

field eq_steps: int = 1000000

Steps per equilibration segment (derived)

Validated by:

• _finalize

field eq_steps1: int = 500000

Equilibration stage 1 steps (legacy mirror of eq_steps)

Validated by:

• _finalize

field eq_steps2: int = 1000000

Equilibration stage 2 steps (legacy mirror of eq_steps)

Validated by:

• _finalize

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

Free-energy protocol type

Validated by:

• _finalize

• _lower_enums

field gamma_ln: float = 1.0

Langevin γ (ps^-1)

Validated by:

• _finalize

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

Hydrogen mass repartitioning

Validated by:

• _coerce_yes_no

• _finalize

field infe: bool = False

Enable NFE (infinite) equilibration when true.

Validated by:

• _finalize

field ion_conc: float = 0.15

Target salt concentration (M)

Validated by:

• _finalize

field ion_def: List[Any] [Optional]

Ion tuple [cation, anion, conc]

Validated by:

• _finalize

field l1_range: float | None = None

L1 search radius (Å)

Validated by:

• _finalize

field l1_x: float | None = None

L1 center offset X (Å)

Validated by:

• _finalize

field l1_y: float | None = None

L1 center offset Y (Å)

Validated by:

• _finalize

field l1_z: float | None = None

L1 center offset Z (Å)

Validated by:

• _finalize

field lambdas: List[float] [Optional]

default lambda values

Validated by:

• _finalize

• _parse_lambdas

field lig_angle_force: float = 0.0

Ligand angle/dihedral spring (kcal/mol/rad^2)

Validated by:

• _finalize

field lig_buffer: float = 10.0

Ligand box buffer (Å)

Validated by:

• _finalize

field lig_com_force: float = 0.0

Ligand COM spring

Validated by:

• _finalize

field lig_dihcf_force: float = 0.0

Ligand dihedral spring (kcal/mol/rad^2)

Validated by:

• _finalize

field lig_distance_force: float = 0.0

Ligand COM distance spring (kcal/mol/Å^2)

Validated by:

• _finalize

field ligand_dict: Dict[str, Any] [Optional]

Ligand dictionary

Validated by:

• _finalize

field ligand_ff: str = 'gaff2'

Ligand FF

Validated by:

• _finalize

field lipid_ff: str = 'lipid21'

Lipid FF

Validated by:

• _finalize

field lipid_mol: List[str] [Optional]

Lipid molecules

Validated by:

• _finalize

field max_adis: float | None = None

Max anchor distance (Å)

Validated by:

• _finalize

field membrane_simulation: bool = True

Whether system includes a membrane

Validated by:

• _finalize

field min_adis: float | None = None

Min anchor distance (Å)

Validated by:

• _finalize

field n_steps_dict: Dict[str, int] [Optional]

Per-component steps (keys: ‘{comp}_steps1|2’)

Validated by:

• _finalize

field neut: str = ''

Alias of neutralize_only

Validated by:

• _finalize

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

Neutralize only

Validated by:

• _coerce_yes_no

• _finalize

field ntpr: int = 1000

Print energy every ntpr steps

Validated by:

• _finalize

field ntwe: int = 0

Write energy every ntwe steps

Validated by:

• _finalize

field ntwr: int = 10000

Write restart every ntwr steps

Validated by:

• _finalize

field ntwx: int = 2500

Write trajectory every ntwx steps

Validated by:

• _finalize

field num_equil_extends: int = 0

Number of equilibration extends (derived)

Validated by:

• _finalize

field num_fe_extends: int = 10

# restarts per λ

Validated by:

• _finalize

field other_mol: List[str] [Optional]

Other co-binders

Validated by:

• _finalize

field p1: str = ''

Anchor P1 “RESID@ATOM” (e.g., “85@CA”)

Validated by:

• _finalize

• _validate_anchor

field p2: str = ''

Anchor P2 “RESID@ATOM”

Validated by:

• _finalize

• _validate_anchor

field p3: str = ''

Anchor P3 “RESID@ATOM”

Validated by:

• _finalize

• _validate_anchor

field protein_align: str = 'name CA'

Alignment selection

Validated by:

• _finalize

field rec_com_force: float = 0.0

Protein COM spring

Validated by:

• _finalize

field receptor_ff: str = 'protein.ff14SB'

Receptor FF

Validated by:

• _finalize

field receptor_segment: str | None = None

Segment to embed in membrane

Validated by:

• _finalize

field release_eq: List[float] [Optional]

Equilibration release weights (derived)

Validated by:

• _finalize

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

H-REMD toggle

Validated by:

• _coerce_yes_no

• _finalize

field remd_nstlim: int = 100

Steps per REMD segment (applied to mdin-*-remd copies).

Validated by:

• _finalize

field remd_numexchg: int = 3000

Exchange attempt interval for REMD (numexchg).

Validated by:

• _finalize

field rest: List[float] [Optional]

Packed restraint constants

Validated by:

• _finalize

field rng: int = 0

Range of release_eq

Validated by:

• _finalize

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

Rocklin correction

Validated by:

• _coerce_yes_no

• _finalize

field sdr_dist: float | None = 0.0

SDR placement distance (Å)

Validated by:

• _finalize

field solv_shell: float | None = 15.0

Initial solvent shell radius (Å)

Validated by:

• _finalize

field system_name: str [Required]

System name (required)

Validated by:

• _finalize

field temperature: float = 298.15

Temperature (K)

Validated by:

• _finalize

field ti_points: int | None = 0

1. TI points (not implemented)

Validated by:

• _finalize

field unbound_threshold: float = 8.0

Distance (Å) between ligand COMs that classifies equilibration as unbound.

Constraints:

• ge = 0.0

Validated by:

• _finalize

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

Water model

Validated by:

• _finalize

classmethod from_sections(create: CreateArgs, fe: FESimArgs, *, protocol: str | None = None, fe_type: str | None = None, slurm_header_dir: Path | 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.

Returns:

Fully merged simulation configuration ready for downstream use.

Return type:

SimulationConfig

to_dict() → Dict[str, Any]

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

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.

Parameters:

• path (str or pathlib.Path) – Path to the top-level run YAML file.

• on_failure ({“prune”, “raise”, “retry”}, optional) – Override for the failure policy applied to ligand pipelines.

• run_overrides (dict, optional) – Overrides applied to the run section (e.g., only FE preparation).

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", "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.

• NotImplementedError – Raised for protocols that are planned but not yet available (e.g., RBFE).

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: ~typing.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 = 0, 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

Utilities for monitoring and resubmitting Slurm-managed jobs.

Notes

This module relies on fcntl and is therefore intended for POSIX systems (e.g. typical HPC clusters running Slurm).

class batter.exec.slurm_mgr.SlurmJobManager(poll_s: float = 20.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, 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)

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

Initialise the manager.

Parameters:

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

• max_retries (int, optional) – Maximum number of automatic resubmissions per job.

• resubmit_backoff_s (float, optional) – Delay between a failed job and an attempted resubmission.

• registry_file (pathlib.Path, optional) – Optional JSONL file acting as a persistent queue shared across processes.

• dry_run (bool, optional) – When True do not submit jobs; only mark that a submission would occur.

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

• submit_retry_limit (int, optional) – Number of submission retries on failure per job.

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

• max_active_jobs (int, optional) – Maximum number of active jobs allowed for the user. When None, no limit is enforced.

• batch_mode (bool, optional) – When True, batch scripts (e.g., SLURMM-BATCH) may be supplied via batch_script/submit_dir on the job specs.

• batch_gpus (int, optional) – Reserved for future inline execution modes.

• gpus_per_task (int, optional) – Reserved for future inline execution modes.

• srun_extra (Sequence[str], optional) – Reserved for future inline execution modes.

add(spec: SlurmJobSpec) → None

Queue spec for later submission.

Parameters:

spec (SlurmJobSpec) – Job specification to store. Persisted to registry_file when configured.

clear() → None

Clear in-memory specs, retry bookkeeping, and remove the on-disk queue if present.

ensure_running(spec: SlurmJobSpec) → None

Ensure the given spec is submitted or already active/done (does not register).

jobs() → List[SlurmJobSpec]

Return the union of in-memory and on-disk queued specs (dedup by workdir).

set_stage(stage: str | None) → None

Limit registry loading to stage and default new specs to this stage.

wait_all() → None

Submit/monitor all registered jobs together and block until completion.

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

Block until the number of active jobs drops below max_active_jobs.

Parameters:

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

• user (str, optional) – User name to query in squeue. Defaults to $USER.

wait_until_done(specs: Iterable[SlurmJobSpec]) → None

Submit if needed and watch the given set until done/fail (legacy interface).

class batter.exec.slurm_mgr.SlurmJobSpec(workdir: ~pathlib.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: ~pathlib.Path | None = None, header_root: ~pathlib.Path | None = None, extra_sbatch: ~typing.Sequence[str] = <factory>, extra_env: ~typing.Dict[str, str] = <factory>, batch_script: ~pathlib.Path | None = None, submit_dir: ~pathlib.Path | None = None, alt_script_names: ~typing.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 (default: "SLURMM-run").

• finished_name (str, optional) – Name of the sentinel file indicating success.

• failed_name (str, optional) – Name of the sentinel file indicating failure.

• name (str, optional) – Friendly display name used in logs.

• extra_sbatch (Sequence[str], optional) – Additional sbatch flags appended during submission.

• extra_env (dict, optional) – Additional environment variables exported before submission.

• batch_script (pathlib.Path, optional) – Optional wrapper script used when batch_mode is enabled.

• submit_dir (pathlib.Path, optional) – Working directory used when submitting (defaults to workdir).

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.

script_arg() → str

Return the workdir-relative script argument for sbatch.

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: ~typing.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: ~typing.List[str] = <factory>, artifacts: ~typing.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: ~typing.List[str] = <factory>, payload: ~typing.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]

pydantic model batter.pipeline.payloads.StepPayload

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.

Config:

• extra: str = allow

• arbitrary_types_allowed: bool = True

Fields:

• sim (batter.config.simulation.SimulationConfig | None)

• sys_params (batter.pipeline.payloads.SystemParams | None)

Validators:

• _coerce » all fields

• _coerce_nested » all fields

field sim: SimulationConfig | None = None

Validated by:

• _coerce

• _coerce_nested

field sys_params: SystemParams | None = None

Validated by:

• _coerce

• _coerce_nested

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

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]

pydantic model batter.pipeline.payloads.SystemParams

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.

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.

Config:

• extra: str = allow

• arbitrary_types_allowed: bool = True

Fields:

• anchor_atoms (tuple[str, ...])

• ligand_paths (Dict[str, pathlib.Path])

• param_outdir (pathlib.Path | None)

• protein_input (pathlib.Path | None)

• system_coordinate (pathlib.Path | None)

• system_input (pathlib.Path | None)

• system_name (str | None)

• yaml_dir (pathlib.Path | None)

Validators:

• _coerce » all fields

field anchor_atoms: tuple[str, ...] = ()

Validated by:

• _coerce

field ligand_paths: Dict[str, Path] [Optional]

Validated by:

• _coerce

field param_outdir: Path | None = None

Validated by:

• _coerce

field protein_input: Path | None = None

Validated by:

• _coerce

field system_coordinate: Path | None = None

Validated by:

• _coerce

field system_input: Path | None = None

Validated by:

• _coerce

field system_name: str | None = None

Validated by:

• _coerce

field yaml_dir: Path | None = None

Validated by:

• _coerce

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

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

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]

Runtime Modules

class batter.runtime.portable.Artifact(name: str, relpath: ~pathlib.Path, kind: ~typing.Literal['file', 'dir'] = 'file', sha256: str = '', size: int = 0, meta: ~typing.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).

pydantic model batter.runtime.fe_repo.FERecord

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").

• sim_range (tuple[int, int], optional) – (start, end) lambda range used for 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.

Fields:

• canonical_smiles (str | None)

• components (List[str])

• created_at (str)

• fe_type (str)

• ligand (str)

• method (Literal['mbar', 'ti'])

• mol_name (str)

• original_name (str | None)

• original_path (str | None)

• protocol (str)

• run_id (str)

• sim_range (tuple[int, int] | None)

• status (Literal['success', 'failed', 'unbound'])

• system_name (str)

• temperature (float)

• total_dG (float)

• total_se (float)

• windows (List[WindowResult])

field canonical_smiles: str | None = None

field components: List[str] [Optional]

field created_at: str [Optional]

field fe_type: str [Required]

field ligand: str [Required]

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

field mol_name: str [Required]

field original_name: str | None = None

field original_path: str | None = None

field protocol: str = 'abfe'

field run_id: str [Required]

field sim_range: tuple[int, int] | None = None

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

field system_name: str [Required]

field temperature: float [Required]

field total_dG: float [Required]

field total_se: float = 0.0

field windows: List[WindowResult] [Optional]

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

index() → DataFrame

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', sim_range: tuple[int, int] | None = None) → None

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

pydantic model batter.runtime.fe_repo.WindowResult

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.

Fields:

• component (str)

• dG (float)

• dG_se (float)

• lam (float)

• meta (Dict[str, Any])

• n_samples (int)

field component: str [Required]

field dG: float [Required]

field dG_se: float = 0.0

field lam: float [Required]

field meta: Dict[str, Any] [Optional]

field n_samples: int = 0

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: ~pathlib.Path, topology: ~pathlib.Path | None = None, coordinates: ~pathlib.Path | None = None, protein: ~pathlib.Path | None = None, ligands: ~typing.Tuple[~pathlib.Path, ...] = (), lipid_mol: ~typing.Tuple[str, ...] = (), other_mol: ~typing.Tuple[str, ...] = (), anchors: ~typing.Tuple[str, ...] = (), meta: ~batter.systems.core.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: ~typing.Dict[str, str] = <factory>, extras: ~typing.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).