Adding Simulation and MultistateSimulation reporter

Examples/LJ_langevin.py

@@ -43,15 +43,15 @@
 # build the neighbor list from the sampler state
 nbr_list.build_from_state(sampler_state)
 
-from chiron.reporters import SimulationReporter
+from chiron.reporters import _SimulationReporter
 
 # initialize a reporter to save the simulation data
 filename = "test_lj.h5"
 import os
 
 if os.path.isfile(filename):
     os.remove(filename)
-reporter = SimulationReporter("test_lj.h5", lj_fluid.topology, 1)
+reporter = _SimulationReporter("test_lj.h5", lj_fluid.topology, 1)
 
 from chiron.integrators import LangevinIntegrator
 

Examples/LJ_mcmove.py

@@ -45,23 +45,23 @@
 # build the neighbor list from the sampler state
 nbr_list.build_from_state(sampler_state)
 
-from chiron.reporters import SimulationReporter
+from chiron.reporters import _SimulationReporter
 
 # initialize a reporter to save the simulation data
 filename = "test_lj.h5"
 import os
 
 if os.path.isfile(filename):
     os.remove(filename)
-reporter = SimulationReporter("test_mc_lj.h5", lj_fluid.topology, 1)
+reporter = _SimulationReporter("test_mc_lj.h5", lj_fluid.topology, 1)
 
 from chiron.mcmc import MetropolisDisplacementMove
 
 mc_move = MetropolisDisplacementMove(
     seed=1234,
     displacement_sigma=0.01 * unit.nanometer,
     nr_of_moves=1000,
-    simulation_reporter=reporter,
+    reporter=reporter,
 )
 
 mc_move.run(sampler_state, thermodynamic_state, nbr_list, True)

chiron/analysis.py

@@ -0,0 +1,60 @@
+import numpy as np
+
+
+class MBAREstimator:
+    def __init__(self) -> None:
+        """
+        Initialize the MBAR analysis class.
+
+        Returns:
+        - None
+        """
+        self.mbar_f_k = None
+        self.mbar = None
+
+    def initialize(self, u_kn: np.ndarray, N_k: np.ndarray):
+        """
+        Initialize the analysis object.
+
+        Parameters
+        ----------
+        u_kn: np.ndarray
+            Array of dimensionless reduced potentials for each state.
+        N_k: np.ndarray
+            Array of number of samples for each state.
+
+        """
+        from pymbar import MBAR
+        from loguru import logger as log
+
+        log.debug(f"{N_k=}")
+        self.mbar = MBAR(u_kn=u_kn, N_k=N_k)
+
+    @property
+    def f_k(self):
+        """
+        Free energy for each state.
+
+        Returns
+        -------
+        mbar.f_k.
+        """
+
+        from loguru import logger as log
+
+        log.debug(self.mbar.f_k)
+        return self.mbar.f_k
+
+    def get_free_energy_difference(self):
+        """
+        Calculate the free energy difference between the endstates.
+
+        Returns
+        -------
+        float
+        """
+        from loguru import logger as log
+
+        log.debug(self.mbar.f_k[-1])
+        self.f_k = self.mbar.f_k
+        return self.mbar_f_k[-1]

chiron/integrators.py

@@ -4,8 +4,10 @@
 from jax import random
 from openmm import unit
 from .states import SamplerState, ThermodynamicState
-from .reporters import SimulationReporter
+from .reporters import LangevinDynamicsReporter
 from typing import Optional
+from .potential import NeuralNetworkPotential
+from .neighbors import PairsBase
 
 
 class LangevinIntegrator:
@@ -25,7 +27,7 @@ def __init__(
         stepsize=1.0 * unit.femtoseconds,
         collision_rate=1.0 / unit.picoseconds,
         save_frequency: int = 100,
-        reporter: Optional[SimulationReporter] = None,
+        reporter: Optional[LangevinDynamicsReporter] = None,
         save_traj_in_memory: bool = False,
     ) -> None:
         """
@@ -41,8 +43,9 @@ def __init__(
             Frequency of saving the simulation data. Default is 100.
         reporter : SimulationReporter, optional
             Reporter object for saving the simulation data. Default is None.
-        save_traj_in_memory : bool
-            Whether to save the trajectory in memory. For debugging purposes only.
+        save_traj_in_memory: bool
+            Flag indicating whether to save the trajectory in memory.
+            Default is False. NOTE: Only for debugging purposes.
         """
         from loguru import logger as log
 
@@ -53,13 +56,13 @@ def __init__(
 
         self.stepsize = stepsize
         self.collision_rate = collision_rate
-        if reporter is not None:
-            log.info(f"Using reporter {reporter} saving to {reporter.filename}")
+        if reporter:
+            log.info(f"Using reporter {reporter} saving to {reporter.file_path}")
             self.reporter = reporter
         self.save_frequency = save_frequency
+        self.velocities = None
         self.save_traj_in_memory = save_traj_in_memory
         self.traj = []
-        self.velocities = None
 
     def set_velocities(self, vel: unit.Quantity) -> None:
         """
@@ -77,8 +80,7 @@ def run(
         sampler_state: SamplerState,
         thermodynamic_state: ThermodynamicState,
         n_steps: int = 5_000,
-        key=random.PRNGKey(0),
-        nbr_list=None,
+        nbr_list: Optional[PairsBase] = None,
         progress_bar=False,
     ):
         """
@@ -92,9 +94,7 @@ def run(
             The thermodynamic state of the system, including temperature and potential.
         n_steps : int, optional
             Number of simulation steps to perform.
-        key : jax.random.PRNGKey, optional
-            Random key for generating random numbers.
-        nbr_list : NeighborListNsqrd, optional
+        nbr_list : PairBase, optional
             Neighbor list for the system.
         progress_bar : bool, optional
             Flag indicating whether to display a progress bar during integration.
@@ -116,8 +116,11 @@ def run(
         log.debug("Running Langevin dynamics")
         log.debug(f"n_steps = {n_steps}")
         log.debug(f"temperature = {temperature}")
-        log.debug(f"Using seed: {key}")
 
+        # Initialize the random number generator
+        key = sampler_state.random_seed
+
+        # Convert to dimensionless quantities
         kbT_unitless = (self.kB * temperature).value_in_unit_system(unit.md_unit_system)
         mass_unitless = jnp.array(mass.value_in_unit_system(unit.md_unit_system))[
             :, None
@@ -127,22 +130,24 @@ def run(
         collision_rate_unitless = self.collision_rate.value_in_unit_system(
             unit.md_unit_system
         )
+        a = jnp.exp((-collision_rate_unitless * stepsize_unitless))
+        b = jnp.sqrt(1 - jnp.exp(-2 * collision_rate_unitless * stepsize_unitless))
 
         # Initialize velocities
         if self.velocities is None:
             v0 = sigma_v * random.normal(key, x0.shape)
         else:
             v0 = self.velocities.value_in_unit_system(unit.md_unit_system)
-        # Convert to dimensionless quantities
-        a = jnp.exp((-collision_rate_unitless * stepsize_unitless))
-        b = jnp.sqrt(1 - jnp.exp(-2 * collision_rate_unitless * stepsize_unitless))
 
         x = x0
         v = v0
+
         if nbr_list is not None:
             nbr_list.build_from_state(sampler_state)
 
         F = potential.compute_force(x, nbr_list)
+
+        # propagation loop
         for step in tqdm(range(n_steps)) if self.progress_bar else range(n_steps):
             key, subkey = random.split(key)
             # v
@@ -151,46 +156,80 @@ def run(
             x += (stepsize_unitless * 0.5) * v
 
             if nbr_list is not None:
-                x = nbr_list.space.wrap(x)
-                # check if we need to rebuild the neighborlist after moving the particles
-                if nbr_list.check(x):
-                    nbr_list.build(x, self.box_vectors)
+                x = self._wrap_and_rebuild_neighborlist(x, nbr_list)
             # o
             random_noise_v = random.normal(subkey, x.shape)
             v = (a * v) + (b * sigma_v * random_noise_v)
 
             x += (stepsize_unitless * 0.5) * v
             if nbr_list is not None:
-                x = nbr_list.space.wrap(x)
-                # check if we need to rebuild the neighborlist after moving the particles
-                if nbr_list.check(x):
-                    nbr_list.build(x, self.box_vectors)
+                x = self._wrap_and_rebuild_neighborlist(x, nbr_list)
 
             F = potential.compute_force(x, nbr_list)
             # v
             v += (stepsize_unitless * 0.5) * F / mass_unitless
 
             if step % self.save_frequency == 0:
-                # log.debug(f"Saving at step {step}")
-                # check if reporter is attribute of the class
-                # log.debug(f"step {step} energy {potential.compute_energy(x, nbr_list)}")
-                # log.debug(f"step {step} force {F}")
-
                 if hasattr(self, "reporter") and self.reporter is not None:
-                    d = {
-                        "traj": x,
-                        "energy": potential.compute_energy(x, nbr_list),
-                        "step": step,
-                    }
-                    if nbr_list is not None:
-                        d["box_vectors"] = nbr_list.space.box_vectors
-
-                    # log.debug(d)
-                    self.reporter.report(d)
+                    self._report(x, potential, nbr_list, step)
+
                 if self.save_traj_in_memory:
                     self.traj.append(x)
 
         log.debug("Finished running Langevin dynamics")
         # save the final state of the simulation in the sampler_state object
         sampler_state.x0 = x
+        sampler_state.v0 = v
         # self.reporter.close()
+
+    def _wrap_and_rebuild_neighborlist(self, x: jnp.array, nbr_list: PairsBase):
+        """
+        Wrap the coordinates and rebuild the neighborlist if necessary.
+
+        Parameters
+        ----------
+        x: jnp.array
+            The coordinates of the particles.
+        nbr_list: PairsBsse
+            The neighborlist object.
+        """
+
+        x = nbr_list.space.wrap(x)
+        # check if we need to rebuild the neighborlist after moving the particles
+        if nbr_list.check(x):
+            nbr_list.build(x, self.box_vectors)
+        return x
+
+    def _report(
+        self,
+        x: jnp.array,
+        potential: NeuralNetworkPotential,
+        nbr_list: PairsBase,
+        step: int,
+    ):
+        """
+        Reports the trajectory, energy, step, and box vectors (if available) to the reporter.
+
+        Parameters
+        ----------
+            x : jnp.array
+                current coordinate set
+            potential: NeuralNetworkPotential
+                potential used to compute the energy and force
+            nbr_list: PairsBase
+                The neighbor list
+            step: int
+                The current time step.
+
+        Returns:
+            None
+        """
+        d = {
+            "traj": x,
+            "energy": potential.compute_energy(x, nbr_list),
+            "step": step,
+        }
+        if nbr_list is not None:
+            d["box_vectors"] = nbr_list.space.box_vectors
+
+        self.reporter.report(d)