Coverage for sparkle/CLI/status.py: 76%

1#!/usr/bin/env python3 

2"""Command to display the status of the platform.""" 

3 

4import sys 

5import argparse 

6from pathlib import Path 

7 

8from sparkle.structures import FeatureDataFrame, PerformanceDataFrame 

9 

10from sparkle.CLI.initialise import check_for_initialise 

11from sparkle.CLI.help import global_variables as gv 

12from sparkle.CLI.help import logging as sl 

13from sparkle.platform import Settings 

14 

15 

16def parser_function() -> argparse.ArgumentParser: 

17 """Define the command line arguments.""" 

18 parser = argparse.ArgumentParser(description="Display the status of the platform.") 

19 parser.add_argument( 

20 *Settings.OPTION_verbosity.args, **Settings.OPTION_verbosity.kwargs 

21 ) 

22 return parser 

23 

24 

25def print_objects_list(objects: list[any], type: str, details: bool = False) -> None: 

26 """Print a list of sparkle objects. 

27 

28 Args: 

29 objects: The objects to print 

30 type: The name of the object type 

31 details: Indicating if output should be detailed 

32 """ 

33 print( 

34 f"Currently Sparkle has {len(objects)} {type}(s)" 

35 + (":" if details and objects else "") 

36 ) 

37 

38 if details: 

39 for i, object in enumerate(objects): 

40 print(f"\t[{i + 1}] {Path(object).name}") 

41 

42 

43def print_feature_computation_jobs( 

44 feature_data_csv: Path, verbose: bool = False 

45) -> None: 

46 """Print a list of remaining feature computation jobs. 

47 

48 Args: 

49 feature_data_csv: Path to the feature data csv 

50 verbose: Indicating, if output should be verbose 

51 """ 

52 if not feature_data_csv.exists(): 

53 print("\nNo feature data found, cannot determine remaining jobs.") 

54 

55 feature_data = FeatureDataFrame(feature_data_csv) 

56 jobs = feature_data.remaining_jobs() 

57 

58 print( 

59 f"Currently Sparkle has {len(jobs)} remaining feature computation " 

60 "jobs that need to be performed before creating an algorithm selector" 

61 + (":" if verbose and jobs else "") 

62 ) 

63 

64 if verbose: 

65 for i, job in enumerate(jobs): 

66 print( 

67 f"[{i + 1}]: Extractor: {Path(job[1]).name}, Group: {job[2]}, " 

68 f"Instance: {Path(job[0]).name}" 

69 ) 

70 

71 

72def print_performance_computation_jobs( 

73 performance_data: PerformanceDataFrame, verbose: bool = False 

74) -> None: 

75 """Print a list of remaining performance computation jobs. 

76 

77 Args: 

78 performance_data: The Performance data 

79 verbose: Indicating, if output should be verbose 

80 """ 

81 jobs = performance_data.get_job_list() 

82 

83 print( 

84 f"Currently Sparkle has {len(jobs)} remaining performance computation" 

85 " jobs that need to be performed before creating an algorithm selector" 

86 + (":" if verbose else "") 

87 ) 

88 

89 if verbose: 

90 i = 0 

91 for solver, config, instance, run in jobs: 

92 print( 

93 f"[{i + 1}]: Solver: " 

94 f"{Path(solver).name} ({config}), Instance: " 

95 f"{Path(instance).name}" 

96 ) 

97 i += 1 

98 

99 

100def main(argv: list[str]) -> None: 

101 """Main function of the status command.""" 

102 # Log command call 

103 sl.log_command(sys.argv, gv.settings().random_state) 

104 check_for_initialise() 

105 

106 # Define command line arguments 

107 parser = parser_function() 

108 

109 # Process command line arguments 

110 args = parser.parse_args(argv) 

111 

112 performance_data = PerformanceDataFrame(gv.settings().DEFAULT_performance_data_path) 

113 

114 print("========Sparkle System Status========\n") 

115 print_objects_list( 

116 [s for s in gv.settings().DEFAULT_solver_dir.iterdir()], 

117 "Solver", 

118 args.verbosity, 

119 ) 

120 if args.verbosity: 

121 print() 

122 print_objects_list( 

123 [e for e in gv.settings().DEFAULT_extractor_dir.iterdir()], 

124 "Extractor", 

125 args.verbosity, 

126 ) 

127 if args.verbosity: 

128 print() 

129 print_objects_list( 

130 [i for i in gv.settings().DEFAULT_instance_dir.iterdir()], 

131 "Instance Set", 

132 args.verbosity, 

133 ) 

134 

135 print() 

136 print_feature_computation_jobs( 

137 gv.settings().DEFAULT_feature_data_path, args.verbosity 

138 ) 

139 print_performance_computation_jobs(performance_data, args.verbosity) 

140 

141 if args.verbosity: 

142 print("\nThe Performance Data overview:") 

143 print(performance_data) 

144 

145 # scan configurator log files for warnings 

146 configurator = gv.settings().configurator 

147 configurator.get_status_from_logs( 

148 gv.settings().get_configurator_output_path(configurator) 

149 ) 

150 

151 sys.exit(0) 

152 

153 

154if __name__ == "__main__": 

155 main(sys.argv[1:])