Skip to content

Commit

Permalink
Rework test_converter
Browse files Browse the repository at this point in the history
  • Loading branch information
@oerc0122
oerc0122 committed Feb 13, 2025
1 parent 2d59112 commit 326efdd
Showing 1 changed file with 101 additions and 186 deletions.
287 changes: 101 additions & 186 deletions MDANSE/Tests/UnitTests/test_converter.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,41 +12,60 @@

TEST_DIR = Path(__file__).parent
CONV_DIR = TEST_DIR / "Converted"
file_wd = os.path.dirname(os.path.realpath(__file__))
conv_dir = os.path.join(
os.path.dirname(os.path.realpath(__file__)),
"Converted",
)
DATA_DIR = TEST_DIR / "Data"

lammps_config = DATA_DIR / "lammps_test.config"
lammps_lammps = DATA_DIR / "lammps_test.lammps"
lammps_moly = DATA_DIR / "structure_moly.lammps"
lammps_custom = DATA_DIR / "lammps_moly_custom.txt"
lammps_xyz = DATA_DIR / "lammps_moly_xyz.txt"
lammps_h5md = CONV_DIR / "lammps_moly_h5md.h5"
vasp_xdatcar = DATA_DIR / "XDATCAR_version5"
discover_his = DATA_DIR / "sushi.his"
discover_xtd = DATA_DIR / "sushi.xtd"
cp2k_pos = DATA_DIR / "CO2GAS-pos-1.xyz"
cp2k_vel = DATA_DIR / "CO2GAS-vel-1.xyz"
cp2k_cell = DATA_DIR / "CO2GAS-1.cell"
hem_cam_pdb = DATA_DIR / "hem-cam.pdb"
hem_cam_dcd = DATA_DIR / "hem-cam.dcd"
ase_traj = DATA_DIR / "Cu_5steps_ASEformat.traj"
xyz_traj = DATA_DIR / "traj-100K-npt-1000-res.xyz"
dlp_field_v2 = DATA_DIR / "FIELD_Water"
dlp_history_v2 = DATA_DIR / "HISTORY_Water"
dlp_field_v4 = DATA_DIR / "FIELD4"
dlp_history_v4 = DATA_DIR / "HISTORY4"
dlp_field_with_grad = DATA_DIR / "FIELD_methanol_short"
dlp_history_with_grad = DATA_DIR / "HISTORY_methanol_short"
apoferritin_dcd = DATA_DIR / "apoferritin.dcd"
apoferritin_pdb = DATA_DIR / "apoferritin.pdb"
pbanew_md = DATA_DIR / "PBAnew.md"
h2o_trj = DATA_DIR / "H2O.trj"
h2o_xtd = DATA_DIR / "H2O.xtd"
md_pdb = DATA_DIR / "md.pdb"
md_xtc = DATA_DIR / "md.xtc"

def _converter_test(tmp_path, typ, result, compare, parameters, compression):
temp_name = tmp_path / "output"
out_name = temp_name.with_suffix(".mdt")
log_name = temp_name.with_suffix(".log")
result_name = CONV_DIR / result

parameters["output_files"] = (temp_name, 64, 128, compression, "INFO")

converter = Converter.create(typ)
converter.run(parameters, status=True)

traj_conf = HDFTrajectoryConfigurator("trajectory")
traj_conf.configure(out_name)
traj_conf.get_information()
traj_conf["hdf_trajectory"].close()

lammps_config = os.path.join(file_wd, "Data", "lammps_test.config")
lammps_lammps = os.path.join(file_wd, "Data", "lammps_test.lammps")
lammps_moly = os.path.join(file_wd, "Data", "structure_moly.lammps")
lammps_custom = os.path.join(file_wd, "Data", "lammps_moly_custom.txt")
lammps_xyz = os.path.join(file_wd, "Data", "lammps_moly_xyz.txt")
lammps_h5md = os.path.join(file_wd, "Converted", "lammps_moly_h5md.h5")
vasp_xdatcar = os.path.join(file_wd, "Data", "XDATCAR_version5")
discover_his = os.path.join(file_wd, "Data", "sushi.his")
discover_xtd = os.path.join(file_wd, "Data", "sushi.xtd")
cp2k_pos = os.path.join(file_wd, "Data", "CO2GAS-pos-1.xyz")
cp2k_vel = os.path.join(file_wd, "Data", "CO2GAS-vel-1.xyz")
cp2k_cell = os.path.join(file_wd, "Data", "CO2GAS-1.cell")
hem_cam_pdb = os.path.join(file_wd, "Data", "hem-cam.pdb")
hem_cam_dcd = os.path.join(file_wd, "Data", "hem-cam.dcd")
ase_traj = os.path.join(file_wd, "Data", "Cu_5steps_ASEformat.traj")
xyz_traj = os.path.join(file_wd, "Data", "traj-100K-npt-1000-res.xyz")
dlp_field_v2 = os.path.join(file_wd, "Data", "FIELD_Water")
dlp_history_v2 = os.path.join(file_wd, "Data", "HISTORY_Water")
dlp_field_v4 = os.path.join(file_wd, "Data", "FIELD4")
dlp_history_v4 = os.path.join(file_wd, "Data", "HISTORY4")
dlp_field_with_grad = os.path.join(file_wd, "Data", "FIELD_methanol_short")
dlp_history_with_grad = os.path.join(file_wd, "Data", "HISTORY_methanol_short")
apoferritin_dcd = os.path.join(file_wd, "Data", "apoferritin.dcd")
apoferritin_pdb = os.path.join(file_wd, "Data", "apoferritin.pdb")
pbanew_md = os.path.join(file_wd, "Data", "PBAnew.md")
h2o_trj = os.path.join(file_wd, "Data", "H2O.trj")
h2o_xtd = os.path.join(file_wd, "Data", "H2O.xtd")
md_pdb = os.path.join(file_wd, "Data", "md.pdb")
md_xtc = os.path.join(file_wd, "Data", "md.xtc")
with h5py.File(out_name) as actual, h5py.File(result_name) as desired:
for prop in compare:
np.testing.assert_array_almost_equal(actual[prop], desired[prop])

assert out_name.is_file()
assert log_name.is_file()

@pytest.mark.parametrize("typ,result,compare,parameters", (
("LAMMPS", "lammps.mdt",
Expand Down Expand Up @@ -97,6 +116,15 @@
"n_steps": 0,
"time_step": 50.0,
"time_unit": "fs"}),
("improvedase", "ase.mdt",
(),
{"trajectory_file": (lammps_lammps, "lammps-dump-text"),
"configuration_file": (lammps_config, "lammps-data"),
"fold": False,
"n_steps": 0,
"elements_from_mass": True,
"time_step": 50.0,
"time_unit": "fs"}),
("DL_POLY", "dlp_v2.mdt",
("/configuration/coordinates", "/unit_cell", "/time", "/charge"),
{"atom_aliases": "{}",
Expand Down Expand Up @@ -148,116 +176,62 @@
("MDAnalysis", "md.mdt",
("/configuration/coordinates", "/unit_cell", "/time"),
{"topology_file": (md_pdb, "AUTO"),
"coordinate_files": ([md_xtc], "XTC")}),
"coordinate_files": ([str(md_xtc)], "XTC")}), # Does not work with Path
("MDTraj", "hem_cam.mdt",
("/configuration/coordinates", "/unit_cell", "/time"),
{"topology_file": hem_cam_pdb,
"coordinate_files": [hem_cam_dcd],
"coordinate_files": [str(hem_cam_dcd)], # Does not work with Path
"time_step": 1.0}),
))
@pytest.mark.parametrize("compression", ["none", "gzip", "lzf"])
def test_build_mdt_file_and_load(tmp_path, typ, result, compare, parameters, compression):
temp_name = tmp_path / "output"
out_name = temp_name.with_suffix(".mdt")
log_name = temp_name.with_suffix(".log")
result_name = CONV_DIR / result

parameters["output_files"] = (temp_name, 64, 128, compression, "INFO")

converter = Converter.create(typ)
converter.run(parameters, status=True)

traj_conf = HDFTrajectoryConfigurator("trajectory")
traj_conf.configure(out_name)
traj_conf.get_information()
traj_conf["hdf_trajectory"].close()

with h5py.File(out_name) as actual, h5py.File(result_name) as desired:
for prop in compare:
np.testing.assert_array_almost_equal(actual[prop], desired[prop])

assert out_name.is_file()
assert log_name.is_file()

_converter_test(tmp_path, typ, result, compare, parameters, compression)

@pytest.mark.parametrize(
"unit_system", ["real", "metal", "si", "cgs", "electron", "micro", "nano"]
)
def test_lammps_mdt_conversion_unit_system(unit_system):
temp_name = tempfile.mktemp()

parameters = {}
parameters["config_file"] = lammps_config
parameters["mass_tolerance"] = 0.05
parameters["n_steps"] = 0
parameters["output_files"] = (temp_name, 64, 128, "gzip", "INFO")
parameters["smart_mass_association"] = True
parameters["time_step"] = 1.0
parameters["trajectory_file"] = lammps_lammps
parameters["lammps_units"] = unit_system

lammps = Converter.create("LAMMPS")
lammps.run(parameters, status=True)

traj_conf = HDFTrajectoryConfigurator("trajectory")
traj_conf.configure(temp_name + ".mdt")
traj_conf.get_information()
traj_conf["hdf_trajectory"].close()

result_file = os.path.join(conv_dir, f"lammps_{unit_system}.mdt")
with h5py.File(temp_name + ".mdt") as actual, h5py.File(result_file) as desired:
np.testing.assert_array_almost_equal(actual["/configuration/coordinates"], desired["/configuration/coordinates"])
np.testing.assert_array_almost_equal(actual["/unit_cell"], desired["/unit_cell"])
np.testing.assert_array_almost_equal(actual["/time"], desired["/time"])
np.testing.assert_array_almost_equal(actual["/charge"], desired["/charge"])

assert os.path.exists(temp_name + ".mdt")
assert os.path.isfile(temp_name + ".mdt")
os.remove(temp_name + ".mdt")
assert os.path.exists(temp_name + ".log")
assert os.path.isfile(temp_name + ".log")
os.remove(temp_name + ".log")

def test_lammps_mdt_conversion_unit_system(tmp_path, unit_system):
_converter_test(tmp_path, "LAMMPS", f"lammps_{unit_system}.mdt",
("/configuration/coordinates", "/unit_cell", "/time", "/charge"),
{"config_file": lammps_config,
"mass_tolerance": 0.05,
"n_steps": 0,
"smart_mass_association": True,
"time_step": 1.0,
"trajectory_file": lammps_lammps,
"lammps_units": unit_system},
"gzip")

@pytest.mark.parametrize(
"trajectory_file,trajectory_format",
[(lammps_custom, "custom"), (lammps_xyz, "xyz"), (lammps_h5md, "h5md")],
)
def test_lammps_mdt_conversion_trajectory_format(trajectory_file, trajectory_format):
temp_name = tempfile.mktemp()

parameters = {}
parameters["config_file"] = lammps_moly
parameters["mass_tolerance"] = 0.05
parameters["n_steps"] = 0
parameters["output_files"] = (temp_name, 64, 128, "gzip", "INFO")
parameters["smart_mass_association"] = True
parameters["time_step"] = 1.0
parameters["trajectory_file"] = trajectory_file
parameters["trajectory_format"] = trajectory_format
parameters["lammps_units"] = "electron"

lammps = Converter.create("LAMMPS")
lammps.run(parameters, status=True)

traj_conf = HDFTrajectoryConfigurator("trajectory")
traj_conf.configure(temp_name + ".mdt")
traj_conf.get_information()
traj_conf["hdf_trajectory"].close()

result_file = os.path.join(conv_dir, f"lammps_moly_{trajectory_format}.mdt")
with h5py.File(temp_name + ".mdt") as actual, h5py.File(result_file) as desired:
np.testing.assert_array_almost_equal(actual["/configuration/coordinates"], desired["/configuration/coordinates"])
np.testing.assert_array_almost_equal(actual["/unit_cell"], desired["/unit_cell"])
np.testing.assert_array_almost_equal(actual["/time"], desired["/time"])

assert os.path.exists(temp_name + ".mdt")
assert os.path.isfile(temp_name + ".mdt")
os.remove(temp_name + ".mdt")
assert os.path.exists(temp_name + ".log")
assert os.path.isfile(temp_name + ".log")
os.remove(temp_name + ".log")
def test_lammps_mdt_conversion_trajectory_format(tmp_path, trajectory_file, trajectory_format):
_converter_test(tmp_path, "LAMMPS", f"lammps_moly_{trajectory_format}.mdt",
("/configuration/coordinates", "/unit_cell", "/time", "/charge"),
{"config_file": lammps_moly,
"mass_tolerance": 0.05,
"n_steps": 0,
"smart_mass_association": True,
"time_step": 1.0,
"trajectory_file": trajectory_file,
"trajectory_format": trajectory_format,
"lammps_units": "electron"},
"gzip")

@pytest.mark.parametrize(
"trajectory",
(ase_traj, xyz_traj, vasp_xdatcar),
)
def test_improvedase_mdt_conversion_file_exists_and_loads_up_successfully(tmp_path, trajectory):
_converter_test(tmp_path, "improvedase", "ase.mdt", # Dummy
(),
{"trajectory_file": str(trajectory), # Does not work with Path
"fold": False,
"n_steps": 0,
"time_step": 50.0,
"time_unit": "fs"},
"gzip")

def test_lammps_mdt_conversion_raise_exception_with_incorrect_format():
temp_name = tempfile.mktemp()
Expand All @@ -274,62 +248,3 @@ def test_lammps_mdt_conversion_raise_exception_with_incorrect_format():
lammps = Converter.create("LAMMPS")
with pytest.raises(JobError):
lammps.run(parameters, status=True)


@pytest.mark.parametrize("trajectory", [ase_traj, xyz_traj, vasp_xdatcar])
def test_improvedase_mdt_conversion_file_exists_and_loads_up_successfully(trajectory):
temp_name = tempfile.mktemp()

parameters = {
"trajectory_file": trajectory,
"fold": False,
"output_files": (temp_name, 64, 128, "gzip", "INFO"),
"n_steps": 0,
"time_step": 50.0,
"time_unit": "fs",
}

ase_conv = Converter.create("improvedase")
ase_conv.run(parameters, status=True)

traj_conf = HDFTrajectoryConfigurator("trajectory")
traj_conf.configure(temp_name + ".mdt")
traj_conf.get_information()
traj_conf["hdf_trajectory"].close()

assert os.path.exists(temp_name + ".mdt")
assert os.path.isfile(temp_name + ".mdt")
os.remove(temp_name + ".mdt")
assert os.path.exists(temp_name + ".log")
assert os.path.isfile(temp_name + ".log")
os.remove(temp_name + ".log")


def test_improvedase_lammps_two_files():
temp_name = tempfile.mktemp()

parameters = {
"trajectory_file": (lammps_lammps, "lammps-dump-text"),
"configuration_file": (lammps_config, "lammps-data"),
"fold": False,
"output_files": (temp_name, 64, 128, "gzip", "INFO"),
"n_steps": 0,
"elements_from_mass": True,
"time_step": 50.0,
"time_unit": "fs",
}

ase_conv = Converter.create("improvedase")
ase_conv.run(parameters, status=True)

traj_conf = HDFTrajectoryConfigurator("trajectory")
traj_conf.configure(temp_name + ".mdt")
traj_conf.get_information()
traj_conf["hdf_trajectory"].close()

assert os.path.exists(temp_name + ".mdt")
assert os.path.isfile(temp_name + ".mdt")
os.remove(temp_name + ".mdt")
assert os.path.exists(temp_name + ".log")
assert os.path.isfile(temp_name + ".log")
os.remove(temp_name + ".log")

0 comments on commit 326efdd

Please sign in to comment.