Coverage for ase / calculators / exciting / runner.py: 58.82%

1# fmt: off 

2 

3"""Binary runner and results class.""" 

4import os 

5import subprocess 

6import time 

7from pathlib import Path 

8 

9 

10class SubprocessRunResults: 

11 """Results returned from subprocess.run().""" 

12 

13 def __init__( 

14 self, stdout, stderr, return_code: int, 

15 process_time: float | None = None): 

16 self.stdout = stdout 

17 self.stderr = stderr 

18 self.return_code = return_code 

19 self.success = return_code == 0 

20 self.process_time = process_time 

21 

22 

23class SimpleBinaryRunner: 

24 """Class to execute a subprocess.""" 

25 path_type = str | Path 

26 

27 def __init__(self, 

28 binary, 

29 run_argv: list[str], 

30 omp_num_threads: int, 

31 directory: path_type = './', 

32 args=None) -> None: 

33 """Initialise class. 

34 

35 :param binary: Binary name prepended by full path, or just binary name 

36 (if present in $PATH). 

37 :param run_argv: Run commands sequentially as a list of str. 

38 For example: 

39 * For serial: ['./'] or [''] 

40 * For MPI: ['mpirun', '-np', '2'] 

41 :param omp_num_threads: Number of OMP threads. 

42 :param args: Optional arguments for the binary. 

43 """ 

44 if args is None: 

45 args = [] 

46 self.binary = binary 

47 self.directory = directory 

48 

49 self.run_argv = run_argv 

50 

51 self.omp_num_threads = omp_num_threads 

52 self.args = args 

53 

54 if directory is not None and not Path(directory).is_dir(): 

55 raise OSError(f"Run directory does not exist: {directory}") 

56 

57 if omp_num_threads <= 0: 

58 raise ValueError("Number of OMP threads must be > 0") 

59 

60 def compose_execution_list(self) -> list: 

61 """Generate a complete list of strings to pass to subprocess.run(). 

62 

63 This is done to execute the calculation. 

64 

65 For example, given: 

66 ['mpirun', '-np, '2'] + ['binary.exe'] + ['>', 'std.out'] 

67 

68 return ['mpirun', '-np, '2', 'binary.exe', '>', 'std.out'] 

69 """ 

70 return self.run_argv + [self.binary] + self.args 

71 

72 def run(self) -> SubprocessRunResults: 

73 """Run a binary.""" 

74 execution_list = self.compose_execution_list() 

75 my_env = {**os.environ} 

76 

77 time_start: float = time.time() 

78 result = subprocess.run(execution_list, 

79 env=my_env, 

80 capture_output=True, 

81 cwd=self.directory, check=False) 

82 total_time = time.time() - time_start 

83 return SubprocessRunResults( 

84 result.stdout, result.stderr, result.returncode, total_time)