Coverage for moptipyapps/prodsched/runtime_summary.py: 15%

1"""Create data for runtime distribution plots.""" 

2 

3import argparse 

4from typing import Final 

5 

6from pycommons.io.csv import COMMENT_START 

7from pycommons.io.csv import CSV_SEPARATOR as __CS 

8from pycommons.io.path import Path 

9from pycommons.strings.string_conv import num_to_str, str_to_num 

10from pycommons.types import type_error 

11 

12from moptipyapps.utils.shared import moptipyapps_argparser 

13 

14 

15def find_runtimes( 

16 source: str, runtimes: dict[str, list[float | int]]) -> None: 

17 """ 

18 Load an ROP multi-simulation summary from a log file. 

19 

20 :param source: the source path 

21 :param runtimes: the destination for runtimes 

22 """ 

23 if not isinstance(runtimes, dict): 

24 raise type_error(runtimes, "runtimes", dict) 

25 spath: Final[Path] = Path(source) 

26 

27 if spath.is_dir(): 

28 for subfile in spath.list_dir(): 

29 find_runtimes(subfile, runtimes) 

30 return 

31 if not spath.is_file(): 

32 return 

33 

34 computer_id: str = "" 

35 

36 # collect the raw data 

37 computer_key: Final[str] = "session.node: " 

38 simtime_key: Final[str] = "time.s: " 

39 dest: list[float | int] | None = None 

40 

41 with spath.open_for_read() as stream: 

42 for srow in stream: 

43 row = str.strip(srow) 

44 if (str.__len__(row) <= 0) or row.startswith(COMMENT_START): 

45 continue 

46 if str.startswith(row, computer_key): 

47 if (dest is not None) or (str.__len__(computer_id) > 0): 

48 raise ValueError("Illegal State, already got computer id " 

49 f"{computer_id!r}, now found {row!r}.") 

50 computer_id = str.strip(row[str.__len__(computer_key):]) 

51 if computer_id in runtimes: 

52 dest = runtimes[computer_id] 

53 else: 

54 dest = [] 

55 runtimes[computer_id] = dest 

56 elif str.startswith(row, simtime_key): 

57 if dest is None: 

58 raise ValueError("Don't have computer id yet!") 

59 dest.append(str_to_num(row[str.__len__(simtime_key):])) 

60 

61 

62def runtime_summary( 

63 source: str, 

64 dest: str) -> None: 

65 """ 

66 Convert one or multiple files from a source to a destination. 

67 

68 :param source: the source file or directory 

69 :param dest: the destination file 

70 """ 

71 src: Final[Path] = Path(source) 

72 dst_file: Final[Path] = Path(dest) 

73 

74 runtimes: dict[str, list[int | float]] = {} 

75 find_runtimes(src, runtimes) 

76 

77 data: list[list[str]] = [] 

78 max_len: int = -1 

79 for key in sorted(runtimes.keys()): 

80 lst: list[str] = [key] 

81 keydata = runtimes[key] 

82 keydata.sort() 

83 lst.extend(map(num_to_str, keydata)) 

84 max_len = max(max_len, list.__len__(lst)) 

85 data.append(lst) 

86 

87 row: list[str] = [] 

88 with dst_file.open_for_write() as stream: 

89 for i in range(max_len): 

90 row.clear() 

91 for col in data: 

92 if i == 0: 

93 row.extend((f"{col[0]}_x", f"{col[0]}_rt")) 

94 else: 

95 ll = list.__len__(col) 

96 if i < ll: 

97 row.extend((num_to_str((i - 1) / (ll - 1)), col[i])) 

98 else: 

99 row.extend(("", "")) 

100 stream.write(__CS.join(row)) 

101 stream.write("\n") 

102 

103 

104# Run to parse all log files and to create csv 

105if __name__ == "__main__": 

106 parser: Final[argparse.ArgumentParser] = moptipyapps_argparser( 

107 __file__, 

108 "Postprocess Runtimes", 

109 "Create postprocessing results.") 

110 parser.add_argument( 

111 "source", nargs="?", default="./results", 

112 help="the location of the experimental results, i.e., the root folder " 

113 "under which to search for log files", type=Path) 

114 parser.add_argument( 

115 "dest", help="the path to the destination file to be created", 

116 type=Path, nargs="?", default="./evaluation/runtimes.txt") 

117 

118 args: Final[argparse.Namespace] = parser.parse_args() 

119 runtime_summary(args.source, args.dest)