5 Commits

5 changed files with 35 additions and 4 deletions
+14 -2
View File
@@ -102,8 +102,20 @@ def checksum(*args, csum=None, _seen=None):
_seen.add(obj_id)
if hasattr(arg, "__checksum__"):
logger.debug("Checksum via __checksum__: %s", str(arg))
csum.update(str(arg.__checksum__()).encode())
method = getattr(arg, "__checksum__")
if callable(method) and not isinstance(arg, type):
logger.debug("Checksum via __checksum__: %s", str(arg))
csum.update(str(method()).encode())
elif isinstance(arg, type):
try:
src = inspect.getsource(arg)
csum.update(strip_comments(src).encode())
logger.debug("Checksum via class source for %s", arg.__name__)
except (OSError, TypeError):
csum.update(arg.__name__.encode())
logger.debug("Checksum via class name for %s", arg.__name__)
else:
logger.debug("Skipping unbound __checksum__ on %s", type(arg))
elif isinstance(arg, bytes):
csum.update(arg)
elif isinstance(arg, str):
-1
View File
@@ -413,7 +413,6 @@ def center_of_masses(
) -> NDArray:
if atom_indices is None:
atom_indices = list(range(len(frame)))
print(type(frame))
res_ids = frame.residue_ids[atom_indices]
masses = frame.masses[atom_indices]
if shear:
+2 -1
View File
@@ -147,7 +147,8 @@ def shifted_correlation(
num_frames = int(len(frames) * window)
ls = np.logspace(0, np.log10(num_frames + 1), num=points)
idx = np.unique(np.int_(ls) - 1)
t = np.array([frames[i].time for i in idx]) - frames[0].time
dt = round(frames[1].time - frames[0].time, 6) # round to avoid bad floats
t = idx * dt
result = np.array(
[
+14
View File
@@ -23,6 +23,8 @@ from .logging_util import logger
from . import atoms
from .coordinates import Coordinates
from unittest.mock import MagicMock
CSR_ATTRS = ("data", "indices", "indptr")
NOJUMP_MAGIC = 2016
Group_RE = re.compile(r"\[ ([-+\w]+) \]")
@@ -240,7 +242,18 @@ def generate_nojump_matrices(trajectory: Coordinates):
save_nojump_matrices(trajectory.frames)
def _ensure_xdr(reader: BaseReader):
"""Patch missing _xdr attribute for non-XDR readers (e.g. LAMMPS DumpReader)
with a stable mock so checksums are consistent across runs."""
if not hasattr(reader.rd, '_xdr'):
mock_xdr = MagicMock()
mock_xdr.offsets = np.arange(len(reader))
print(f"Adding mock _xdr attribute for to reader of length {len(reader)}.")
reader.rd._xdr = mock_xdr
def save_nojump_matrices(reader: BaseReader, matrices: npt.ArrayLike = None):
_ensure_xdr(reader)
if matrices is None:
matrices = reader.nojump_matrices
data = {"checksum": checksum(NOJUMP_MAGIC, checksum(reader))}
@@ -253,6 +266,7 @@ def save_nojump_matrices(reader: BaseReader, matrices: npt.ArrayLike = None):
def load_nojump_matrices(reader: BaseReader):
_ensure_xdr(reader)
zipname = nojump_load_filename(reader)
try:
data = np.load(zipname, allow_pickle=True)
+5
View File
@@ -334,6 +334,11 @@ def quick1etau(t: ArrayLike, C: ArrayLike, n: int = 7) -> float:
C is C(t) the correlation function
n is the minimum number of points around 1/e required
"""
# norm, if t=0 provided
if t[0] == 0:
C /= C[0]
C, t = C[t>0], t[t>0] # make sure t=0 is dropped
# first rough estimate, the closest time. This is returned if the interpolation fails!
tau_est = t[np.argmin(np.fabs(C - np.exp(-1)))]
# reduce the data to points around 1/e