63 lines
2.1 KiB
Python
63 lines
2.1 KiB
Python
from pydantic import TypeAdapter
|
|
|
|
import diceplayer.dice.dice_input as dice_input
|
|
from diceplayer.config import DiceConfig
|
|
from diceplayer.environment import System
|
|
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Final
|
|
|
|
|
|
type DiceEnvironment = tuple[str, int, int, int]
|
|
DiceEnvironmentAdapter = TypeAdapter(DiceEnvironment)
|
|
|
|
|
|
DICE_FLAG_LINE: Final[int] = -2
|
|
DICE_END_FLAG: Final[str] = "End of simulation"
|
|
|
|
|
|
class DiceWrapper:
|
|
def __init__(self, dice_config: DiceConfig, working_directory: Path):
|
|
self.dice_config = dice_config
|
|
self.working_directory = 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:
|
|
exit_status = subprocess.call(
|
|
self.dice_config.progname, stdin=infile, stdout=outfile, cwd=self.working_directory
|
|
)
|
|
|
|
if exit_status != 0:
|
|
raise RuntimeError(f"Dice simulation failed with exit status {exit_status}")
|
|
|
|
with open(output_path, "r") as outfile:
|
|
line = outfile.readlines()[DICE_FLAG_LINE]
|
|
if line.strip() == DICE_END_FLAG:
|
|
return
|
|
|
|
raise RuntimeError(f"Dice simulation failed with exit status {exit_status}")
|
|
|
|
def parse_results(self, system: System) -> list[DiceEnvironment]:
|
|
NUMBER_OF_HEADER_LINES = 2
|
|
NUMBER_OF_PRIMARY_ATOMS = len(system.molecule[0].atom)
|
|
|
|
results = []
|
|
for output_file in sorted(self.working_directory.glob("phb*.xyz")):
|
|
with open(output_file, "r") as f:
|
|
for _ in range(NUMBER_OF_HEADER_LINES + NUMBER_OF_PRIMARY_ATOMS):
|
|
next(f, None)
|
|
|
|
for line in f:
|
|
if line.strip() == "":
|
|
break
|
|
|
|
results.append(
|
|
DiceEnvironmentAdapter.validate_python(line.split())
|
|
)
|
|
|
|
return results
|