Coverage for sparkle/CLI/cleanup.py: 58%

1#!/usr/bin/env python3 

2"""Command to remove temporary files not affecting the platform state.""" 

3import sys 

4import argparse 

5import shutil 

6 

7from sparkle.CLI.help import logging as sl 

8from sparkle.CLI.help import global_variables as gv 

9from sparkle.CLI.help import argparse_custom as ac 

10from sparkle.CLI.help import snapshot_help as snh 

11from sparkle.CLI.help import jobs as jobs_help 

12 

13 

14def parser_function() -> argparse.ArgumentParser: 

15 """Define the command line arguments.""" 

16 parser = argparse.ArgumentParser(description="Command to clean files from the " 

17 "platform.") 

18 parser.add_argument(*ac.CleanupArgumentAll.names, **ac.CleanupArgumentAll.kwargs) 

19 parser.add_argument(*ac.CleanupArgumentRemove.names, 

20 **ac.CleanupArgumentRemove.kwargs) 

21 parser.add_argument(*ac.CleanUpPerformanceDataArgument.names, 

22 **ac.CleanUpPerformanceDataArgument.kwargs) 

23 return parser 

24 

25 

26def remove_temporary_files() -> None: 

27 """Remove temporary files. Only removes files not affecting the sparkle state.""" 

28 shutil.rmtree(gv.settings().DEFAULT_log_output, ignore_errors=True) 

29 gv.settings().DEFAULT_log_output.mkdir() 

30 

31 

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

33 """Main function of the cleanup command.""" 

34 # Log command call 

35 sl.log_command(sys.argv) 

36 

37 # Define command line arguments 

38 parser = parser_function() 

39 

40 # Process command line arguments 

41 args = parser.parse_args(argv) 

42 if args.all: 

43 shutil.rmtree(gv.settings().DEFAULT_output, ignore_errors=True) 

44 snh.create_working_dirs() 

45 print("Removed all output files from the platform!") 

46 elif args.remove: 

47 snh.remove_current_platform() 

48 snh.create_working_dirs() 

49 print("Cleaned platform of all files!") 

50 else: 

51 remove_temporary_files() 

52 print("Cleaned platform of temporary files!") 

53 if args.performance_data: 

54 # Check if we can cleanup the PerformanceDataFrame if necessary 

55 from runrunner.base import Status 

56 running_jobs = jobs_help.get_runs_from_file(gv.settings().DEFAULT_log_output, 

57 filter=[Status.WAITING, 

58 Status.RUNNING]) 

59 if len(running_jobs) > 0: 

60 print("WARNING: There are still running jobs! Continue cleaning? [y/n]") 

61 a = input() 

62 if a != "y": 

63 sys.exit(0) 

64 from sparkle.structures import PerformanceDataFrame 

65 performance_data = PerformanceDataFrame( 

66 gv.settings().DEFAULT_performance_data_path) 

67 index_num = len(performance_data.index) 

68 # We only clean lines that are completely empty 

69 performance_data.remove_empty_runs() 

70 performance_data.save_csv() 

71 print(f"Removed {index_num - len(performance_data.index)} rows from the " 

72 f"Performance DataFrame, leaving {len(performance_data.index)} rows.") 

73 sys.exit(0) 

74 

75 

76if __name__ == "__main__": 

77 main(sys.argv[1:])