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

1#!/usr/bin/env python3 

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

3import os 

4import sys 

5import shutil 

6import argparse 

7from pathlib import Path 

8 

9from sparkle.platform import file_help as sfh 

10from sparkle.CLI.help import global_variables as gv 

11from sparkle.structures import FeatureDataFrame 

12from sparkle.CLI.compute_features import compute_features 

13from sparkle.CLI.help import logging as sl 

14from sparkle.platform import CommandName, COMMAND_DEPENDENCIES 

15from sparkle.CLI.initialise import check_for_initialise 

16from sparkle.CLI.help import argparse_custom as ac 

17from sparkle.solver 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 parser.add_argument(*ac.ExtractorPathArgument.names, 

26 **ac.ExtractorPathArgument.kwargs) 

27 group_extractor_run = parser.add_mutually_exclusive_group() 

28 group_extractor_run.add_argument(*ac.RunExtractorNowArgument.names, 

29 **ac.RunExtractorNowArgument.kwargs) 

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

31 **ac.NicknameFeatureExtractorArgument.kwargs) 

32 return parser 

33 

34 

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

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

37 # Log command call 

38 sl.log_command(sys.argv) 

39 

40 parser = parser_function() 

41 

42 # Process command line arguments 

43 args = parser.parse_args(argv) 

44 

45 check_for_initialise(COMMAND_DEPENDENCIES[CommandName.ADD_FEATURE_EXTRACTOR]) 

46 

47 extractor_source = Path(args.extractor_path) 

48 if not extractor_source.exists(): 

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

50 sys.exit(-1) 

51 

52 nickname_str = args.nickname 

53 

54 # Start add feature extractor 

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

56 

57 if extractor_target_path.exists(): 

58 print(f"Feature extractor {extractor_source.name} already exists! " 

59 "Can not add feature extractor.") 

60 sys.exit(-1) 

61 extractor_target_path.mkdir() 

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

63 

64 # Check execution permissions for wrapper 

65 extractor_wrapper = extractor_target_path / Extractor.wrapper 

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

67 print(f"The file {extractor_wrapper} does not exist or is \ 

68 not executable.") 

69 sys.exit(-1) 

70 

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

72 extractor = Extractor(extractor_target_path) 

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

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

75 feature_dataframe.save_csv() 

76 

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

78 

79 if nickname_str is not None: 

80 sfh.add_remove_platform_item( 

81 extractor_target_path, 

82 gv.extractor_nickname_list_path, 

83 gv.file_storage_data_mapping[gv.extractor_nickname_list_path], 

84 key=nickname_str) 

85 

86 if args.run_extractor_now: 

87 print("Start computing features ...") 

88 compute_features(gv.settings().DEFAULT_feature_data_path, False) 

89 

90 # Write used settings to file 

91 gv.settings().write_used_settings() 

92 sys.exit(0) 

93 

94 

95if __name__ == "__main__": 

96 main(sys.argv[1:])