Coverage for src/sparkle/CLI/add_feature_extractor.py: 82%

1#!/usr/bin/env python3 

2"""Sparkle command to add a feature extractor to the Sparkle platform.""" 

3 

4import os 

5import stat 

6import sys 

7import shutil 

8import argparse 

9from pathlib import Path 

10 

11from sparkle.platform import file_help as sfh 

12from sparkle.CLI.help import global_variables as gv 

13from sparkle.structures import FeatureDataFrame 

14from sparkle.CLI.help import logging as sl 

15from sparkle.CLI.initialise import check_for_initialise 

16from sparkle.CLI.help import argparse_custom as ac 

17from sparkle.selector import Extractor 

18from sparkle.CLI.help import jobs as jobs_help 

19 

20 

21def parser_function() -> argparse.ArgumentParser: 

22 """Define the command line arguments.""" 

23 # Define command line arguments 

24 parser = argparse.ArgumentParser( 

25 description="Add a feature extractor to the platform." 

26 ) 

27 parser.add_argument( 

28 *ac.ExtractorPathArgument.names, **ac.ExtractorPathArgument.kwargs 

29 ) 

30 parser.add_argument( 

31 *ac.NicknameFeatureExtractorArgument.names, 

32 **ac.NicknameFeatureExtractorArgument.kwargs, 

33 ) 

34 parser.add_argument(*ac.NoCopyArgument.names, **ac.NoCopyArgument.kwargs) 

35 return parser 

36 

37 

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

39 """Main function of the add feature extractor command.""" 

40 # Log command call 

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

42 check_for_initialise() 

43 

44 parser = parser_function() 

45 

46 # Process command line arguments 

47 args = parser.parse_args(argv) 

48 

49 extractor_source_path = Path(args.extractor_path) 

50 if not extractor_source_path.exists(): 

51 print(f'Feature extractor path "{extractor_source_path}" does not exist!') 

52 sys.exit(-1) 

53 

54 nickname_str = args.nickname 

55 

56 # Start add feature extractor 

57 extractor_target_path = ( 

58 gv.settings().DEFAULT_extractor_dir / extractor_source_path.name 

59 ) 

60 

61 if extractor_target_path.exists(): 

62 print( 

63 f"Feature extractor {extractor_source_path.name} already exists! " 

64 "Can not add feature extractor." 

65 ) 

66 sys.exit(-1) 

67 

68 # Check execution permissions for wrapper 

69 extractor_source = Extractor(extractor_source_path) 

70 if extractor_source.wrapper is None: 

71 print( 

72 f"The Extractor has no wrapper in its directory; please check that the directory {extractor_source_path} contains a file with the name '{Extractor.wrapper_file_name}'!" 

73 ) 

74 sys.exit(-1) 

75 if not extractor_source.wrapper.is_file() or not os.access( 

76 extractor_source.wrapper, os.X_OK 

77 ): 

78 print( 

79 f"The file {extractor_source.wrapper} does not exist or is \ 

80 not executable." 

81 ) 

82 sys.exit(-1) 

83 

84 jobs_help.check_running_waiting_jobs( 

85 gv.settings().DEFAULT_log_output, 

86 ) 

87 

88 # Get the extractor features groups and names from the wrapper, try to add to FDF 

89 feature_dataframe = FeatureDataFrame(gv.settings().DEFAULT_feature_data_path) 

90 feature_dataframe.add_extractor(extractor_source.name, extractor_source.features) 

91 

92 if args.no_copy: 

93 print( 

94 f"Creating symbolic link from {extractor_source_path} " 

95 f"to {extractor_target_path}..." 

96 ) 

97 extractor_target_path.symlink_to(extractor_source_path.absolute()) 

98 else: 

99 print(f"Copying feature extractor {extractor_source_path.name} ...") 

100 extractor_target_path.mkdir() 

101 shutil.copytree(extractor_source_path, extractor_target_path, dirs_exist_ok=True) 

102 

103 extractor = Extractor(extractor_target_path) 

104 # Everything passed, can save FDF 

105 feature_dataframe.save_csv() 

106 

107 # Add RunSolver executable to the solver 

108 runsolver_path = gv.settings().DEFAULT_runsolver_exec 

109 if runsolver_path.name in [file.name for file in extractor_target_path.iterdir()]: 

110 print( 

111 "Warning! RunSolver executable detected in Extractor " 

112 f"{extractor.name}. This will be replaced with " 

113 f"Sparkle's version of RunSolver. ({runsolver_path})" 

114 ) 

115 

116 if runsolver_path.exists(): 

117 runsolver_target = extractor.directory / runsolver_path.name 

118 shutil.copyfile(runsolver_path, runsolver_target) 

119 runsolver_target.chmod(runsolver_target.stat().st_mode | stat.S_IEXEC) 

120 else: 

121 print("Warning! RunSolver does not exists. Falling back to PyRunSolver.") 

122 

123 print(f"Adding feature extractor {extractor_target_path.name} done!") 

124 

125 if nickname_str is not None: 

126 sfh.add_remove_platform_item( 

127 extractor_target_path, 

128 gv.extractor_nickname_list_path, 

129 gv.file_storage_data_mapping[gv.extractor_nickname_list_path], 

130 key=nickname_str, 

131 ) 

132 

133 # Write used settings to file 

134 gv.settings().write_used_settings() 

135 sys.exit(0) 

136 

137 

138if __name__ == "__main__": 

139 main(sys.argv[1:])