Coverage for sparkle/CLI/check.py: 0%

1"""Command to help users check if their input solvers, datasets etc. are correct.""" 

2import sys 

3import os 

4import argparse 

5 

6from runrunner import Runner 

7 

8from sparkle.solver import Solver 

9from sparkle.instance import Instance_Set, InstanceSet 

10from sparkle.selector.extractor import Extractor 

11from sparkle.types import SolverStatus 

12 

13from sparkle.CLI.help import logging as sl 

14from sparkle.CLI.help import argparse_custom as ac 

15from sparkle.CLI.help import global_variables as gv 

16 

17 

18def parser_function() -> argparse.ArgumentParser: 

19 """Define the command line arguments.""" 

20 parser = argparse.ArgumentParser( 

21 description="Command to help users check if their input solvers, instance sets " 

22 "or feature extractors are readable by Sparkle. Specify a path to " 

23 "an instance to test calling the wrapper.") 

24 parser.add_argument(*ac.CheckTypeArgument.names, **ac.CheckTypeArgument.kwargs) 

25 parser.add_argument(*ac.CheckPathArgument.names, **ac.CheckPathArgument.kwargs) 

26 parser.add_argument(*ac.InstancePathOptional.names, 

27 **ac.InstancePathOptional.kwargs) 

28 parser.add_argument(*ac.CutOffTimeArgument.names, 

29 **ac.CutOffTimeArgument.kwargs) 

30 parser.add_argument(*ac.SeedArgument.names, 

31 **ac.SeedArgument.kwargs) 

32 return parser 

33 

34 

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

36 """Main entry point of the Check command.""" 

37 # Log command call 

38 sl.log_command(sys.argv) 

39 parser = parser_function() 

40 args = parser.parse_args(argv) 

41 type_map = { 

42 "extractor": Extractor, 

43 "feature-extractor": Extractor, 

44 "solver": Solver, 

45 "instance-set": Instance_Set, 

46 "Extractor": Extractor, 

47 "Feature-Extractor": Extractor, 

48 "Instance-Set": Instance_Set, 

49 "Solver": Solver, 

50 "FeatureExtractor": Extractor, 

51 "InstanceSet": Instance_Set} 

52 type = type_map[args.type] 

53 print(f"Checking {type.__name__} in directory {args.path} ...") 

54 object = type(args.path) 

55 print("Resolved to:") 

56 print(object.__repr__()) 

57 

58 # Conduct object specific tests 

59 if isinstance(object, Solver): 

60 if object.pcs_file: 

61 print() 

62 print(object.get_configuration_space()) 

63 if not os.access(object.wrapper_path, os.X_OK): 

64 print(f"Wrapper {object.wrapper_path} is not executable!" 

65 f"Check that wrapper execution rights are set for all.") 

66 sys.exit(-1) 

67 if args.instance_path: # Instance to test with 

68 # Test the wrapper with a dummy call 

69 print(f"\nTesting Wrapper {object.wrapper} ...") 

70 # Patchfix runsolver 

71 object.runsolver_exec = gv.settings().DEFAULT_runsolver_exec 

72 

73 objectives = gv.settings().get_general_sparkle_objectives() 

74 cutoff = args.cutoff_time if args.cutoff_time else 5 # Short call 

75 configuration = {} 

76 if object.pcs_file: 

77 print("\nSample Configuration:") 

78 sample_conf = object.get_configuration_space().sample_configuration() 

79 print(sample_conf) 

80 configuration = dict(sample_conf) 

81 result = object.run( 

82 instances=args.instance_path, 

83 objectives=objectives, 

84 seed=42, # Dummy seed 

85 cutoff_time=cutoff, 

86 configuration=configuration, 

87 log_dir=sl.caller_log_dir, 

88 run_on=Runner.LOCAL) 

89 print("Result:") 

90 for obj in objectives: # Check objectives 

91 if obj.name not in result: 

92 print(f"\tSolver output is missing objective {obj.name}") 

93 else: 

94 print(f"\t{obj.name}: {result[obj.name]}") 

95 if result["status"] == SolverStatus.UNKNOWN: 

96 print(f"Sparkle was unable to process {obj.name} output. " 

97 "Check that your wrapper is able to handle KILL SIGNALS. " 

98 "It is important to always communicate back to Sparkle " 

99 "on regular exit and termination signals for stability.") 

100 sys.exit(-1) 

101 elif isinstance(object, Extractor): 

102 if args.instance_path: # Test on an instance 

103 # Patchfix runsolver 

104 object.runsolver_exec = gv.settings().DEFAULT_runsolver_exec 

105 print(f"\nTesting Wrapper {object.wrapper} ...") 

106 # cutoff = args.cutoff_time if args.cutoff_time else 20 # Short call 

107 result = object.run(instance=args.instance_path, 

108 log_dir=sl.caller_log_dir) 

109 # Print feature results 

110 print("Feature values:") 

111 for f_group, f_name, f_value in result: 

112 print(f"\t[{f_group}] {f_name}: {f_value}") 

113 else: 

114 print("Feature names:") 

115 for f_group, fname in object.features: 

116 print(f"\t{f_group}: {fname}") 

117 elif isinstance(object, InstanceSet): 

118 print("\nList of instances:") 

119 for i_name, i_path in zip(object.instance_names, object.instance_paths): 

120 print(f"\t{i_name}: {i_path}") 

121 

122 

123if __name__ == "__main__": 

124 main(sys.argv[1:])