Coverage for src/sparkle/CLI/remove_feature_extractor.py: 86%

1#!/usr/bin/env python3 

2"""Sparkle command to remove a feature extractor from the Sparkle platform.""" 

3 

4import sys 

5import argparse 

6import shutil 

7 

8from sparkle.platform import file_help as sfh 

9from sparkle.CLI.help import global_variables as gv 

10from sparkle.structures import FeatureDataFrame 

11from sparkle.CLI.help import logging as sl 

12from sparkle.CLI.initialise import check_for_initialise 

13from sparkle.CLI.help import argparse_custom as ac 

14from sparkle.CLI.help.nicknames import resolve_object_name 

15from sparkle.selector import Extractor 

16from sparkle.CLI.help import jobs as jobs_help 

17 

18 

19def parser_function() -> argparse.ArgumentParser: 

20 """Define the command line arguments.""" 

21 parser = argparse.ArgumentParser( 

22 description="Remove a feature extractor from the platform." 

23 ) 

24 parser.add_argument( 

25 *ac.ExtractorPathArgument.names, **ac.ExtractorPathArgument.kwargs 

26 ) 

27 return parser 

28 

29 

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

31 """Main function of the remove feature extractor command.""" 

32 # Log command call 

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

34 check_for_initialise() 

35 

36 # Define command line arguments 

37 parser = parser_function() 

38 

39 # Process command line arguments 

40 args = parser.parse_args(argv) 

41 extractor_nicknames = gv.file_storage_data_mapping[gv.extractor_nickname_list_path] 

42 extractor = resolve_object_name( 

43 args.extractor_path, 

44 extractor_nicknames, 

45 gv.settings().DEFAULT_extractor_dir, 

46 class_name=Extractor, 

47 ) 

48 

49 if extractor is None: 

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

51 sys.exit(-1) 

52 

53 print(f"Starting removing feature extractor {extractor.name} ...") 

54 

55 for key in extractor_nicknames: 

56 if extractor_nicknames == extractor.directory: 

57 sfh.add_remove_platform_item( 

58 None, 

59 gv.extractor_nickname_list_path, 

60 extractor_nicknames, 

61 key=key, 

62 remove=True, 

63 ) 

64 break 

65 

66 jobs_help.check_running_waiting_jobs( 

67 gv.settings().DEFAULT_log_output, 

68 ) 

69 

70 if gv.settings().DEFAULT_feature_data_path.exists(): 

71 feature_data = FeatureDataFrame(gv.settings().DEFAULT_feature_data_path) 

72 feature_data.remove_extractor(extractor.name) 

73 feature_data.save_csv() 

74 

75 # We unlink symbolics links, erase copies 

76 if extractor.directory.is_symlink(): 

77 extractor.directory.unlink() 

78 else: 

79 # Remove the directory and all its files 

80 shutil.rmtree(extractor.directory) 

81 

82 print(f"Removing feature extractor {extractor.name} done!") 

83 sys.exit(0) 

84 

85 

86if __name__ == "__main__": 

87 main(sys.argv[1:])