Coverage for sparkle/CLI/add_feature_extractor.py: 85%

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 

18 

19 

20def parser_function() -> argparse.ArgumentParser: 

21 """Define the command line arguments.""" 

22 # Define command line arguments 

23 parser = argparse.ArgumentParser( 

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

25 ) 

26 parser.add_argument( 

27 *ac.ExtractorPathArgument.names, **ac.ExtractorPathArgument.kwargs 

28 ) 

29 parser.add_argument( 

30 *ac.NicknameFeatureExtractorArgument.names, 

31 **ac.NicknameFeatureExtractorArgument.kwargs, 

32 ) 

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

34 return parser 

35 

36 

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

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

39 # Log command call 

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

41 check_for_initialise() 

42 

43 parser = parser_function() 

44 

45 # Process command line arguments 

46 args = parser.parse_args(argv) 

47 

48 extractor_source = Path(args.extractor_path) 

49 if not extractor_source.exists(): 

50 print(f'Feature extractor path "{extractor_source}" does not exist!') 

51 sys.exit(-1) 

52 

53 nickname_str = args.nickname 

54 

55 # Start add feature extractor 

56 extractor_target_path = gv.settings().DEFAULT_extractor_dir / extractor_source.name 

57 

58 if extractor_target_path.exists(): 

59 print( 

60 f"Feature extractor {extractor_source.name} already exists! " 

61 "Can not add feature extractor." 

62 ) 

63 sys.exit(-1) 

64 

65 if args.no_copy: 

66 print( 

67 f"Creating symbolic link from {extractor_source} " 

68 f"to {extractor_target_path}..." 

69 ) 

70 extractor_target_path.symlink_to(extractor_source.absolute()) 

71 else: 

72 print(f"Copying feature extractor {extractor_source.name} ...") 

73 extractor_target_path.mkdir() 

74 shutil.copytree(extractor_source, extractor_target_path, dirs_exist_ok=True) 

75 

76 # Check execution permissions for wrapper 

77 extractor_wrapper = extractor_target_path / Extractor.wrapper 

78 if not extractor_wrapper.is_file() or not os.access(extractor_wrapper, os.X_OK): 

79 print( 

80 f"The file {extractor_wrapper} does not exist or is \ 

81 not executable." 

82 ) 

83 sys.exit(-1) 

84 

85 extractor = Extractor(extractor_target_path) 

86 

87 # Add RunSolver executable to the solver 

88 runsolver_path = gv.settings().DEFAULT_runsolver_exec 

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

90 print( 

91 "Warning! RunSolver executable detected in Extractor " 

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

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

94 ) 

95 runsolver_target = extractor.directory / runsolver_path.name 

96 shutil.copyfile(runsolver_path, runsolver_target) 

97 runsolver_target.chmod(os.stat(runsolver_target).st_mode | stat.S_IEXEC) 

98 

99 # Get the extractor features groups and names from the wrapper 

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

101 feature_dataframe.add_extractor(extractor.name, extractor.features) 

102 feature_dataframe.save_csv() 

103 

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

105 

106 if nickname_str is not None: 

107 sfh.add_remove_platform_item( 

108 extractor_target_path, 

109 gv.extractor_nickname_list_path, 

110 gv.file_storage_data_mapping[gv.extractor_nickname_list_path], 

111 key=nickname_str, 

112 ) 

113 

114 # Write used settings to file 

115 gv.settings().write_used_settings() 

116 sys.exit(0) 

117 

118 

119if __name__ == "__main__": 

120 main(sys.argv[1:])