From 0470200d001edc73983f089473c06936b1a3d6a7 Mon Sep 17 00:00:00 2001 From: Vitor Hideyoshi Date: Tue, 24 Mar 2026 23:01:45 -0300 Subject: [PATCH] refactor: update Python version and optimize dice configuration parameters --- .python-version | 2 +- control.example.yml | 10 +- diceplayer/config/dice_config.py | 7 +- diceplayer/dice/dice_handler.py | 2 +- diceplayer/dice/dice_input.py | 342 +++++++++++++++-------------- diceplayer/dice/dice_wrapper.py | 16 +- diceplayer/environment/molecule.py | 2 +- diceplayer/player.py | 7 +- diceplayer/utils/potential.py | 11 +- tests/config/test_dice_config.py | 2 +- tests/dice/test_dice_input.py | 11 +- tests/utils/test_potential.py | 49 ++--- 12 files changed, 228 insertions(+), 233 deletions(-) diff --git a/.python-version b/.python-version index c8cfe39..e4fba21 100644 --- a/.python-version +++ b/.python-version @@ -1 +1 @@ -3.10 +3.12 diff --git a/control.example.yml b/control.example.yml index d75e9c4..d184657 100644 --- a/control.example.yml +++ b/control.example.yml @@ -10,13 +10,13 @@ diceplayer: altsteps: 2000 dice: - nprocs: 4 - nmol: [1, 1000] + nprocs: 1 + nmol: [1, 200] dens: 1.5 - nstep: [2000, 3000] - isave: 0 + nstep: [200, 300] + isave: 100 outname: 'phb' - progname: '~/.local/bin/dice' + progname: 'dice' ljname: 'phb.ljc.example' randominit: 'always' seed: 12345 diff --git a/diceplayer/config/dice_config.py b/diceplayer/config/dice_config.py index c9b0429..1e5b75a 100644 --- a/diceplayer/config/dice_config.py +++ b/diceplayer/config/dice_config.py @@ -1,9 +1,8 @@ -from pathlib import Path - from pydantic import BaseModel, ConfigDict, Field from typing_extensions import Literal import random +from pathlib import Path class DiceConfig(BaseModel): @@ -38,7 +37,7 @@ class DiceConfig(BaseModel): ) irdf: int = Field( 0, - description="Flag for calculating radial distribution functions (0: no, 1: yes)", + description="Controls the interval of Monte Carlo steps at which configurations are used at computation of radial distribution functions", ) vstep: int = Field( 5000, description="Frequency of volume change moves in NPT simulations" @@ -49,7 +48,7 @@ class DiceConfig(BaseModel): isave: int = Field(1000, description="Frequency of saving the simulation results") press: float = Field(1.0, description="Pressure of the system") temp: float = Field(300.0, description="Temperature of the system") - progname: Path = Field( + progname: str = Field( "dice", description="Name of the program to run the simulation" ) randominit: str = Field( diff --git a/diceplayer/dice/dice_handler.py b/diceplayer/dice/dice_handler.py index 0a8e878..1d60496 100644 --- a/diceplayer/dice/dice_handler.py +++ b/diceplayer/dice/dice_handler.py @@ -84,7 +84,7 @@ class DiceHandler: npt_eq_config = NPTEqConfig.from_config(state.config) dice.run(npt_eq_config) - results.append(dice.extract_results()) + results.append(dice.parse_results(state.system)) @staticmethod def _generate_phb_file(state: StateModel, proc_directory: Path) -> None: diff --git a/diceplayer/dice/dice_input.py b/diceplayer/dice/dice_input.py index 5750128..0af111a 100644 --- a/diceplayer/dice/dice_input.py +++ b/diceplayer/dice/dice_input.py @@ -1,16 +1,16 @@ from diceplayer.config import PlayerConfig from diceplayer.logger import logger +from pydantic import BaseModel, Field from typing_extensions import Self -from abc import ABC -from dataclasses import dataclass, fields +import random +from enum import StrEnum from pathlib import Path -from typing import Any, Sequence, TextIO +from typing import Annotated, Any, Literal, TextIO - -DICE_KEYWORD_ORDER = [ +_ALLOWED_DICE_KEYWORD_IN_ORDER = [ "title", "ncores", "ljname", @@ -32,220 +32,226 @@ DICE_KEYWORD_ORDER = [ ] - -@dataclass(slots=True) -class BaseConfig(ABC): - ncores: int - ljname: str - outname: str - nmol: Sequence[int] - temp: float - seed: int - isave: int - - def write(self, directory: Path, filename: str = "input") -> Path: - input_path = directory / filename - - if input_path.exists(): - logger.info( - f"Dice input file {input_path} already exists and will be overwritten" - ) - input_path.unlink() - input_path.parent.mkdir(parents=True, exist_ok=True) - - with open(input_path, "w") as io: - self.write_dice_config(io) - - return input_path - - def write_dice_config(self, io_writer: TextIO) -> None: - values = {f.name: getattr(self, f.name) for f in fields(self)} - - for key in DICE_KEYWORD_ORDER: - value = values.pop(key, None) - if value is None: - continue - io_writer.write(f"{key} = {self._serialize_value(value)}\n") - - # write any remaining fields (future extensions) - for key, value in values.items(): - if value is None: - continue - io_writer.write(f"{key} = {self._serialize_value(value)}\n") - - io_writer.write("$end\n") - - @classmethod - def from_config(cls, config: PlayerConfig, **kwargs) -> Self: - base_fields = cls._extract_base_fields(config) - return cls(**(base_fields | kwargs)) - - @staticmethod - def _extract_base_fields(config: PlayerConfig) -> dict[str, Any]: - return dict( - ncores=int(config.ncores / config.dice.nprocs), - ljname=config.dice.ljname, - outname=config.dice.outname, - nmol=config.dice.nmol, - temp=config.dice.temp, - seed=config.dice.seed, - isave=config.dice.isave, - ) - - @staticmethod - def _get_nstep(config: PlayerConfig, idx: int) -> int: - if len(config.dice.nstep) > idx: - return config.dice.nstep[idx] - return config.dice.nstep[-1] - - @staticmethod - def _serialize_value(value: Any) -> str: - if value is None: - raise ValueError("DICE configuration cannot serialize None values") - - if isinstance(value, bool): - return "yes" if value else "no" - - if isinstance(value, (list, tuple)): - return " ".join(str(v) for v in value) - - return str(value) +class DiceRoutineType(StrEnum): + NVT_TER = "nvt.ter" + NVT_EQ = "nvt.eq" + NPT_TER = "npt.ter" + NPT_EQ = "npt.eq" -# ----------------------------------------------------- -# NVT BASE -# ----------------------------------------------------- +def get_nstep(config, idx: int) -> int: + if len(config.dice.nstep) > idx: + return config.dice.nstep[idx] + return config.dice.nstep[-1] -@dataclass(slots=True) -class NVTConfig(BaseConfig): - title: str = "Diceplayer Run - NVT" - dens: float = ... - nstep: int = ... - vstep: int = 0 +def get_seed(config) -> int: + return config.dice.seed or random.randint(0, 2**32 - 1) - @classmethod - def from_config(cls, config: PlayerConfig, **kwargs) -> Self: - return super(NVTConfig, cls).from_config( - config, - dens=config.dice.dens, - nstep=cls._get_nstep(config, 0), - ) + +def get_ncores(config) -> int: + return max(1, int(config.ncores / config.dice.nprocs)) # ----------------------------------------------------- # NVT THERMALIZATION # ----------------------------------------------------- +class NVTTerConfig(BaseModel): + type: Literal[DiceRoutineType.NVT_TER] = DiceRoutineType.NVT_TER - -@dataclass(slots=True) -class NVTTerConfig(NVTConfig): - title: str = "Diceplayer Run - NVT Thermalization" - upbuf: int = 360 - init: str = "yes" + title: str = "NVT Thermalization" + ncores: int + ljname: str + outname: str + nmol: list[int] + dens: float + temp: float + seed: int + init: Literal["yes"] = "yes" + nstep: int + vstep: Literal[0] = 0 + isave: int + upbuf: int @classmethod def from_config(cls, config: PlayerConfig, **kwargs) -> Self: - return super(NVTTerConfig, cls).from_config( - config, - nstep=cls._get_nstep(config, 0), + return cls( + ncores=get_ncores(config), + ljname=str(config.dice.ljname), + outname=config.dice.outname, + nmol=config.dice.nmol, + dens=config.dice.dens, + temp=config.dice.temp, + seed=get_seed(config), + nstep=get_nstep(config, 0), + isave=config.dice.isave, upbuf=config.dice.upbuf, - vstep=0, **kwargs, ) - def write(self, directory: Path, filename: str = "nvt.ter") -> Path: - return super(NVTTerConfig, self).write(directory, filename) - # ----------------------------------------------------- # NVT PRODUCTION # ----------------------------------------------------- +class NVTEqConfig(BaseModel): + type: Literal[DiceRoutineType.NVT_EQ] = DiceRoutineType.NVT_EQ - -@dataclass(slots=True) -class NVTEqConfig(NVTConfig): - title: str = "Diceplayer Run - NVT Production" - irdf: int = 0 - init: str = "yesreadxyz" + title: str = "NVT Production" + ncores: int + ljname: str + outname: str + nmol: list[int] + dens: float + temp: float + seed: int + init: Literal["no", "yesreadxyz"] = "no" + nstep: int + vstep: int + isave: int + irdf: Literal[0] = 0 + upbuf: int @classmethod def from_config(cls, config: PlayerConfig, **kwargs) -> Self: - return super(NVTEqConfig, cls).from_config( - config, - nstep=cls._get_nstep(config, 1), - irdf=config.dice.irdf, - vstep=0, + return cls( + ncores=get_ncores(config), + ljname=str(config.dice.ljname), + outname=config.dice.outname, + nmol=config.dice.nmol, + dens=config.dice.dens, + temp=config.dice.temp, + seed=get_seed(config), + nstep=get_nstep(config, 1), + vstep=config.dice.vstep, + isave=max(1, get_nstep(config, 1) // 10), + upbuf=config.dice.upbuf, **kwargs, ) - def write(self, directory: Path, filename: str = "nvt.eq") -> Path: - return super(NVTEqConfig, self).write(directory, filename) - - -# ----------------------------------------------------- -# NPT BASE -# ----------------------------------------------------- - - -@dataclass(slots=True) -class NPTConfig(BaseConfig): - title: str = "Diceplayer Run - NPT" - nstep: int = 0 - vstep: int = 5000 - press: float = 1.0 - - @classmethod - def from_config(cls, config: PlayerConfig, **kwargs) -> Self: - return super(NPTConfig, cls).from_config( - config, - press=config.dice.press, - ) - # ----------------------------------------------------- # NPT THERMALIZATION # ----------------------------------------------------- +class NPTTerConfig(BaseModel): + type: Literal[DiceRoutineType.NPT_TER] = DiceRoutineType.NPT_TER - -@dataclass(slots=True) -class NPTTerConfig(NPTConfig): - title: str = "Diceplayer Run - NPT Thermalization" - dens: float | None = None + title: str = "NPT Thermalization" + ncores: int + ljname: str + outname: str + nmol: list[int] + dens: float + temp: float + press: float + seed: int + init: Literal["yes", "yesreadxyz"] = "yes" + nstep: int + vstep: int + isave: int + upbuf: int @classmethod def from_config(cls, config: PlayerConfig, **kwargs) -> Self: - return super(NPTTerConfig, cls).from_config( - config, + return cls( + ncores=get_ncores(config), + ljname=str(config.dice.ljname), + outname=config.dice.outname, + nmol=config.dice.nmol, dens=config.dice.dens, - nstep=cls._get_nstep(config, 1), - vstep=config.dice.vstep, + temp=config.dice.temp, + press=config.dice.press, + seed=get_seed(config), + nstep=get_nstep(config, 1), + vstep=max(1, config.dice.vstep), + isave=config.dice.isave, + upbuf=config.dice.upbuf, **kwargs, ) - def write(self, directory: Path, filename: str = "npt.ter") -> Path: - return super(NPTTerConfig, self).write(directory, filename) - # ----------------------------------------------------- # NPT PRODUCTION # ----------------------------------------------------- +class NPTEqConfig(BaseModel): + type: Literal[DiceRoutineType.NPT_EQ] = DiceRoutineType.NPT_EQ - -@dataclass(slots=True) -class NPTEqConfig(NPTConfig): - title: str = "Diceplayer Run - NPT Production" - dens: float | None = None + title: str = "NPT Production" + ncores: int + ljname: str + outname: str + nmol: list[int] + dens: float + temp: float + press: float + seed: int + init: Literal["yes", "yesreadxyz"] = "yes" + nstep: int + vstep: int + isave: int + irdf: Literal[0] = 0 + upbuf: int @classmethod def from_config(cls, config: PlayerConfig, **kwargs) -> Self: - return super(NPTEqConfig, cls).from_config( - config, + return cls( + ncores=get_ncores(config), + ljname=str(config.dice.ljname), + outname=config.dice.outname, + nmol=config.dice.nmol, dens=config.dice.dens, - nstep=cls._get_nstep(config, 2), + temp=config.dice.temp, + press=config.dice.press, + seed=get_seed(config), + nstep=get_nstep(config, 2), vstep=config.dice.vstep, + isave=max(1, get_nstep(config, 2) // 10), + upbuf=config.dice.upbuf, **kwargs, ) - def write(self, directory: Path, filename: str = "npt.eq") -> Path: - return super(NPTEqConfig, self).write(directory, filename) + +DiceInputConfig = Annotated[ + NVTTerConfig | NVTEqConfig | NPTTerConfig | NPTEqConfig, + Field(discriminator="type"), +] + + +def _serialize_value(value: Any) -> str: + if value is None: + raise ValueError("DICE configuration cannot serialize None values") + + if isinstance(value, bool): + return "yes" if value else "no" + + if isinstance(value, (list, tuple)): + return " ".join(str(v) for v in value) + + return str(value) + + +def write_dice_config(obj: DiceInputConfig, io_writer: TextIO) -> None: + values = {f: getattr(obj, f) for f in obj.__class__.model_fields} + + for key in _ALLOWED_DICE_KEYWORD_IN_ORDER: + value = values.pop(key, None) + if value is None: + continue + io_writer.write(f"{key} = {_serialize_value(value)}\n") + + io_writer.write("$end\n") + + +def write_config(config: DiceInputConfig, directory: Path) -> Path: + input_path = directory / config.type + + if input_path.exists(): + logger.info( + f"Dice input file {input_path} already exists and will be overwritten" + ) + input_path.unlink() + input_path.parent.mkdir(parents=True, exist_ok=True) + + with open(input_path, "w") as io: + write_dice_config(config, io) + + return input_path diff --git a/diceplayer/dice/dice_wrapper.py b/diceplayer/dice/dice_wrapper.py index 4ec76c5..6eb7eb9 100644 --- a/diceplayer/dice/dice_wrapper.py +++ b/diceplayer/dice/dice_wrapper.py @@ -1,5 +1,6 @@ import diceplayer.dice.dice_input as dice_input from diceplayer.config import DiceConfig +from diceplayer.environment import System import subprocess from pathlib import Path @@ -15,14 +16,13 @@ class DiceWrapper: self.dice_config = dice_config self.working_directory = working_directory - def run(self, dice_config: dice_input.BaseConfig) -> None: - input_path = dice_config.write(self.working_directory) + def run(self, dice_config: dice_input.DiceInputConfig) -> None: + input_path = dice_input.write_config(dice_config, self.working_directory) output_path = input_path.parent / (input_path.name + ".out") with open(output_path, "w") as outfile, open(input_path, "r") as infile: - bin_path = self.dice_config.progname.expanduser() exit_status = subprocess.call( - bin_path, stdin=infile, stdout=outfile, cwd=self.working_directory + self.dice_config.progname, stdin=infile, stdout=outfile, cwd=self.working_directory ) if exit_status != 0: @@ -35,5 +35,9 @@ class DiceWrapper: raise RuntimeError(f"Dice simulation failed with exit status {exit_status}") - def extract_results(self) -> dict: - return {} + def parse_results(self, system: System) -> dict: + results = {} + for output_file in sorted(self.working_directory.glob("phb*.xyz")): + ... + + return results diff --git a/diceplayer/environment/molecule.py b/diceplayer/environment/molecule.py index ea96abe..c5d970b 100644 --- a/diceplayer/environment/molecule.py +++ b/diceplayer/environment/molecule.py @@ -9,8 +9,8 @@ from diceplayer.utils.ptable import GHOST_NUMBER import numpy as np import numpy.linalg as linalg import numpy.typing as npt -from typing_extensions import List, Self, Tuple from pydantic.dataclasses import dataclass +from typing_extensions import List, Self, Tuple import math from copy import deepcopy diff --git a/diceplayer/player.py b/diceplayer/player.py index 6b3c3ec..780814e 100644 --- a/diceplayer/player.py +++ b/diceplayer/player.py @@ -3,11 +3,10 @@ from diceplayer.dice.dice_handler import DiceHandler from diceplayer.logger import logger from diceplayer.state.state_handler import StateHandler from diceplayer.state.state_model import StateModel +from diceplayer.utils.potential import read_system_from_phb from typing_extensions import TypedDict, Unpack -from diceplayer.utils.potential import read_system_from_phb - class PlayerFlags(TypedDict): continuation: bool @@ -33,9 +32,7 @@ class Player: if state is None: system = read_system_from_phb(self.config) - state = StateModel( - config=self.config, system=system - ) + state = StateModel(config=self.config, system=system) else: logger.info("Resuming from existing state.") diff --git a/diceplayer/utils/potential.py b/diceplayer/utils/potential.py index 5d1ee73..5ac39b8 100644 --- a/diceplayer/utils/potential.py +++ b/diceplayer/utils/potential.py @@ -1,11 +1,8 @@ from diceplayer.config import PlayerConfig -from diceplayer.environment import System, Molecule, Atom +from diceplayer.environment import Atom, Molecule, System from pathlib import Path -from diceplayer.logger import logger -from diceplayer.state.state_model import StateModel - def read_system_from_phb(config: PlayerConfig) -> System: phb_file = Path(config.dice.ljname) @@ -41,9 +38,7 @@ def read_system_from_phb(config: PlayerConfig) -> System: for j in range(nsites): _fields = ljc_data.pop(0).split() - mol.add_atom( - Atom(*_fields) - ) + mol.add_atom(Atom(*_fields)) sys.add_type(mol) @@ -58,5 +53,3 @@ def read_system_from_phb(config: PlayerConfig) -> System: # f.write(f"{state.config.dice.combrule}\n") # f.write(f"{len(state.system.nmols)}\n") # f.write(f"{state.config.dice.nmol}\n") - - diff --git a/tests/config/test_dice_config.py b/tests/config/test_dice_config.py index 6765c24..e3af435 100644 --- a/tests/config/test_dice_config.py +++ b/tests/config/test_dice_config.py @@ -6,7 +6,7 @@ import pytest class TestDiceConfig: def test_class_instantiation(self): dice_dto = DiceConfig( - nprocs=1, + nprocs=1, ljname="test", outname="test", dens=1.0, diff --git a/tests/dice/test_dice_input.py b/tests/dice/test_dice_input.py index 8ed6dc0..06eb362 100644 --- a/tests/dice/test_dice_input.py +++ b/tests/dice/test_dice_input.py @@ -4,6 +4,7 @@ from diceplayer.dice.dice_input import ( NPTTerConfig, NVTEqConfig, NVTTerConfig, + write_config, ) import pytest @@ -60,11 +61,7 @@ class TestDiceInput: def test_write_dice_config(self, player_config: PlayerConfig, tmp_path: Path): dice_input = NVTTerConfig.from_config(player_config) - output_file = tmp_path / "nvt_ter.inp" + output_file = tmp_path / dice_input.type + write_config(dice_input, tmp_path) - with open(output_file, "w") as file: - dice_input.write_dice_config(file) - - assert output_file.exists() - - print(output_file.read_text()) + assert output_file.exists() diff --git a/tests/utils/test_potential.py b/tests/utils/test_potential.py index a3b984d..48b6c22 100644 --- a/tests/utils/test_potential.py +++ b/tests/utils/test_potential.py @@ -1,36 +1,35 @@ -from pathlib import Path -from typing import Any - -import pytest - from diceplayer.config import PlayerConfig from diceplayer.environment import System from diceplayer.utils.potential import read_system_from_phb +import pytest + class TestPotential: @pytest.fixture def player_config(self) -> PlayerConfig: - return PlayerConfig.model_validate({ - "type": "both", - "mem": 12, - "max_cyc": 100, - "switch_cyc": 50, - "ncores": 4, - "dice": { - "nprocs": 4, - "ljname": "phb.ljc.example", - "outname": "test", - "dens": 1.0, - "nmol": [12, 16], - "nstep": [1, 1], - }, - "gaussian": { - "level": "test", - "qmprog": "g16", - "keywords": "test", - }, - }) + return PlayerConfig.model_validate( + { + "type": "both", + "mem": 12, + "max_cyc": 100, + "switch_cyc": 50, + "ncores": 4, + "dice": { + "nprocs": 4, + "ljname": "phb.ljc.example", + "outname": "test", + "dens": 1.0, + "nmol": [12, 16], + "nstep": [1, 1], + }, + "gaussian": { + "level": "test", + "qmprog": "g16", + "keywords": "test", + }, + } + ) def test_read_phb(self, player_config: PlayerConfig): system = read_system_from_phb(player_config)