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

1#!/usr/bin/env python3 

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

3import sys 

4import argparse 

5from pathlib import Path 

6 

7from sparkle.structures import FeatureDataFrame, PerformanceDataFrame 

8 

9from sparkle.CLI.initialise import check_for_initialise 

10from sparkle.CLI.help import global_variables as gv 

11from sparkle.CLI.help import logging as sl 

12from sparkle.CLI.help import argparse_custom as ac 

13 

14 

15def parser_function() -> argparse.ArgumentParser: 

16 """Define the command line arguments.""" 

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

18 parser.add_argument(*ac.VerboseArgument.names, 

19 **ac.VerboseArgument.kwargs) 

20 return parser 

21 

22 

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

24 """Print a list of sparkle objects. 

25 

26 Args: 

27 objects: The objects to print 

28 type: The name of the object type 

29 details: Indicating if output should be detailed 

30 """ 

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

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

33 

34 if details: 

35 for i, object in enumerate(objects): 

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

37 

38 

39def print_feature_computation_jobs(feature_data_csv: Path, 

40 verbose: bool = False) -> None: 

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

42 

43 Args: 

44 feature_data_csv: Path to the feature data csv 

45 verbose: Indicating, if output should be verbose 

46 """ 

47 if not feature_data_csv.exists(): 

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

49 

50 feature_data = FeatureDataFrame(feature_data_csv) 

51 jobs = feature_data.remaining_jobs() 

52 

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

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

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

56 

57 if verbose: 

58 for i, job in enumerate(jobs): 

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

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

61 

62 

63def print_performance_computation_jobs(performance_data: PerformanceDataFrame, 

64 verbose: bool = False) -> None: 

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

66 

67 Args: 

68 performance_data: The Performance data 

69 verbose: Indicating, if output should be verbose 

70 """ 

71 jobs = performance_data.get_job_list() 

72 

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

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

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

76 

77 if verbose: 

78 i = 0 

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

80 print(f"[{i + 1}]: Solver: " 

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

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

83 i += 1 

84 

85 

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

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

88 # Log command call 

89 sl.log_command(sys.argv) 

90 check_for_initialise() 

91 

92 # Define command line arguments 

93 parser = parser_function() 

94 

95 # Process command line arguments 

96 args = parser.parse_args(argv) 

97 

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

99 

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

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

102 "Solver", args.verbose) 

103 if args.verbose: 

104 print() 

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

106 "Extractor", args.verbose) 

107 if args.verbose: 

108 print() 

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

110 "Instance Set", args.verbose) 

111 

112 print() 

113 print_feature_computation_jobs( 

114 gv.settings().DEFAULT_feature_data_path, args.verbose 

115 ) 

116 print_performance_computation_jobs(performance_data, args.verbose) 

117 

118 if args.verbose: 

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

120 print(performance_data) 

121 

122 # scan configurator log files for warnings 

123 configurator = gv.settings().get_general_sparkle_configurator() 

124 configurator.get_status_from_logs( 

125 gv.settings().get_configurator_output_path(configurator)) 

126 

127 sys.exit(0) 

128 

129 

130if __name__ == "__main__": 

131 main(sys.argv[1:])