MD Sampling#

Module for sampling snapshots from ASE MD simulations.

class directupsampling.sampling.sampling.GridSampler(atoms: Atoms = None, grid: Grid | list[tuple[float, ...]] | None=None, setting: SamplingSetting | dict | None = None, snapshot_container: SnapshotContainer | None = None, comm: DummyMPI = <directupsampling.parallel.MPI object>)[source]#

Bases: object

Class for sampling snapshots on a grid.

attach(observer: Callable) None[source]#

Attach on observer to be evaluated each grid list step.

calculate_with(calc: Calculator, properties: list[str] = ['energy', 'forces'], inplace: bool = False) SnapshotContainer[source]#

Perform calculations for the snapshots in self.snapshot_container.

Parameters#

calcASE Calculator or list of ASE Calculator

Calculators to calculate with. Labeled with calc.name. If several names are the same, a number is appended.

properties: list of str (optional)

List of which properties to calculate. Defaults to energy and forces. Note: For some Calculators calculating for example energy will automatically also calculate other properties like forces.

inplace: bool (optional)

If True, changes the Calculators of self.snapshot_container.atoms. Defaul is False.

Returns#

snapshot_containerSnapshotContainer

A SnapshotContainer calculated with the provided calculator. If inplace is True, this is self.snapshot_container.

call_observers() None[source]#

Call observers.

classmethod read_folders(path: str, **kwargs: dict) GridSampler[source]#

Make GridSampler from a given folder.

Returns#

GridSampler

sample_snapshots(nsnapshots: int = 150, **kwargs: dict) SnapshotContainer[source]#

Sample nsnapshots on self.grid with self.atoms.

Parameters#

nsnapshots: int

The number of snapshots to sample.

**kwargs

Additional keyword arguments to override those in self.setting. Providing seed here will override (and set the same) seed for each grid point.

Returns#

snapshot_containerSnapshotContainer

The sampled snapshots.

write_folders(joblist_file: str = 'jobList', *, short: bool = False, **kwargs: dict) list[source]#

Write folders from sampler on the grid provided.

Parameters#

joblist_filestr (optional)

Name of the file to which to write the paths of each job. Defaults to ‘jobList’

shortbool (optional)

If True, writes only the path to each (V,T) grid point, else writes path to every snapshot. Default False

root_pathstr | Path (optional)

Where to write the folder structure. The default is the current working directory.

kwargs

Arguments sent to write_folders_from_sampler.

Returns#

joblist : list

Notes#

If a Calculator is given with the parameter ismear = -1, the parameter sigma will be set to correspond to the temperature.

class directupsampling.sampling.sampling.GridSeeder(base_seed: int, min_seed: int = 10000, max_seed: int = 10000000)[source]#

Bases: Seeder

Class for making consistent seeds based on grid.

allgather(comm: DummyMPI) None[source]#

Gather key_counts from all ranks.

make_grid_seeds(grid: Grid) list[int][source]#

Make seeds for all grid points.

Parameters#

gridGrid

The grid for which to make seeds.

Returns#

seedslist of int

A list of seed integers for each grid point in the grid.

make_seed(grid_point: GridPoint) int[source]#

Make a new seed based on grid point.

Parameters#

grid_pointtuple of float

A grid point for which to make a seed.

Returns#

seedint

A seed integer in the range [self.min_seed, self.max_seed) based on the grid point and the number of times this grid point has been used before.

scatter_grid_seeds(grid: Grid, comm: DummyMPI) list[int][source]#

Scatter seeds for all grid points over MPI ranks.

Parameters#

gridGrid

The grid for which to make seeds.

commDummyMPI

The MPI communicator to use for scattering the seeds.

Returns#

seedslist of int

A list of seed integers scattered consistently with Grid.iscatter() over the MPI ranks.

class directupsampling.sampling.sampling.Seeder(base_seed: int, min_seed: int = 10000, max_seed: int = 10000000)[source]#

Bases: object

Class for making consistent seeds.

make_seed(key: str) int[source]#

Make a new seed based on key.

Parameters#

keystr

A string key for which to make a seed. The seed will be based on this key and the number of times this key has been used before.

Returns#

seedint

A seed integer in the range [self.min_seed, self.max_seed) based on the key and the number of times this key has been used before.

Settings for MD snapshot sampling.

class directupsampling.sampling.settings.MDLogSetting(interval: 'int' = 1000, logfile: 'str' = '-', header: 'bool' = True, stress: 'bool' = False, peratom: 'bool' = True, mode: 'str' = 'a', comm: 'Any' = <directupsampling.parallel.DummyMPI object at 0x7a514b332030>)[source]#

Bases: Setting

comm: Any = <directupsampling.parallel.DummyMPI object>#
header: bool = True#
interval: int = 1000#
logfile: str = '-'#
mode: str = 'a'#
peratom: bool = True#
stress: bool = False#
class directupsampling.sampling.settings.SamplingSetting(seed: int | None = None, rng: np.random.Generator = None, sampling_offset: int = 1000, sampling_interval: int = 100, sample_atoms: bool = True, file: str | pathlib.Path | None = None, buffer_for_writing: int = 10, thermostat_setting: ThermostatSetting = <factory>, log_md: bool | int = False, mdlogger_setting: MDLogSetting = <factory>, attach: Callable | list[Callable] = <factory>)[source]#

Bases: Setting

Setting for MD snapshot sampling.

Attributes#

seedint, optional

Random seed for sampling. If None, a random seed is generated.

rngnp.random.Generator, optional

Random number generator for sampling. If None, a new generator is created using the seed.

sampling_intervalint

Interval for MD timesteps to take samples.

sampling_offsetint

Offset for MD timesteps for equilibration.

sample_atomsbool

Whether to sample the atomic positions and forces (through a copy of the ASE Atoms instance). If False, only kinetic and potential energy are sampled.

filestr or pathlib.Path, optional

File path to save the sampled snapshots. If None, snapshots are not saved to a file. If specified, snapshots are buffered and written to the file every buffer_for_writing samples. The written snapshots are discarded from the container to keep memory usage low.

buffer_for_writingint

Number of samples to buffer before writing to the file. Ignored if file is None.

thermostat_settingThermostatSetting

Setting for the ASE style thermostat used in the MD simulation.

log_mdbool or int

Whether to log the MD simulation using ASE’s MDLogger. If True, logging is done every mdlogger_setting.interval steps. If False, no logging is done. If an integer, logging is done every that many steps.

mdlogger_settingMDLogSetting

Setting for the MDLogger used to log the MD simulation.

attachCallable or list of Callable

Function(s) to attach to the MD simulation, passed through to the attach method of ASE’s MD class.

Notes#

If file is specified, but the sampled snapshots are desired after the run, the file needs to be read with SnapshotContainer.read.

attach: Callable | list[Callable]#
buffer_for_writing: int = 10#
file: str | pathlib.Path | None = None#
classmethod from_any(dct: dict | SamplingSetting | None = None, **kwargs: dict) SamplingSetting[source]#

Create a SamplingSetting from a dict, an existing instance, or None.

Parameters#

dctdict, SamplingSetting, or None

Source settings. Nested dicts for thermostat_setting and mdlogger_setting are automatically cast to their dataclass types. If already a SamplingSetting, returned as-is.

**kwargs

Additional keyword arguments merged into dct (ignored when dct is already a SamplingSetting instance).

Returns#

SamplingSetting

The created SamplingSetting instance.

log_md: bool | int = False#
mdlogger_setting: MDLogSetting#
rng: np.random.Generator = None#
sample_atoms: bool = True#
sampling_interval: int = 100#
sampling_offset: int = 1000#
seed: int | None = None#
thermostat_setting: ThermostatSetting#
class directupsampling.sampling.settings.Setting[source]#

Bases: object

replace(**changes) Setting[source]#
todict() dict[source]#
class directupsampling.sampling.settings.ThermostatSetting(class_type: 'type[MolecularDynamics]' = <class 'directupsampling.thermostats.LangevinGB'>, timestep: 'float' = 0.09822694788464063, friction: 'float' = 0.10180505671156725, fixcm: 'bool' = True)[source]#

Bases: Setting

class_type#

alias of LangevinGB

fixcm: bool = True#
friction: float = 0.10180505671156725#
get_kwargs() dict[source]#
timestep: float = 0.09822694788464063#

Module for sampling snapshots from ASE MD simulations.

class directupsampling.sampling.md.DummyMDSampler(setting: SamplingSetting, comm: DummyMPI)[source]#

Bases: object

Dummy sampler class for parallel runs, to avoid deadlock.

run(ntimesteps: int) None[source]#

Run the dummy sampler, which only performs dummy writes to avoid deadlock.

write_buffer_to_database() None[source]#

Pretentd to write the buffered snapshots and clear the buffer.

class directupsampling.sampling.md.MDHandler(atoms: Atoms, temperature: float, setting: SamplingSetting)[source]#

Bases: object

Handler class for ASE Thermostat and velocity initialization.

attach(functions: list[Callable | tuple]) None[source]#

Attach functions to the thermostat.

initialize(temperature: float) None[source]#

Initialize atoms with the Maxwell-Boltzmann distribution of ASE.

irun(steps=50) Iterator[bool][source]#

Run molecular dynamics algorithm as a generator.

Parameters#

stepsint

Number of molecular dynamics steps to be run.

Yields#

convergedbool

True if the maximum number of steps are reached.

property nsteps: int#

Number of MD steps taken so far.

run(steps: int = 50)[source]#

Run molecular dynamics algorithm.

Parameters#

stepsint

Number of molecular dynamics steps to be run.

Returns#

convergedbool

True if the maximum number of steps are reached.

set_logger(setting: MDLogSetting | None = None) None[source]#

Set the MDLogger to log the MD simulation.

set_thermostat(temperature: float, setting: ThermostatSetting | None = None) None[source]#

Set the thermostat to be used for MD.

class directupsampling.sampling.md.MDSampler(atoms: Atoms, temperature: float, setting: SamplingSetting, snapshot_container: SnapshotContainer | None = None, comm=<directupsampling.parallel.DummyMPI object>)[source]#

Bases: object

is_sampling_time() bool[source]#
run(ntimesteps: int) SnapshotContainer[source]#

Run dynamics with ASE and returns snapshots.

Returns#

snapshots

list of ase.Atoms

sample() None[source]#

Take a sample of the MD simulation and save it to the container.

write_buffer_to_database() None[source]#

Write the buffered snapshots to the database and clear the buffer.

Notes#

This method also discard the written snapshots from the container, which keeps memory usage low.