Coverage for sparkle/CLI/cancel.py: 47%

1#!/usr/bin/env python3 

2"""Command to cancel async jobs.""" 

3import sys 

4import argparse 

5import pytermgui as ptg 

6 

7from runrunner.base import Status 

8 

9from sparkle.CLI.help import logging 

10from sparkle.CLI.help import argparse_custom as ac 

11from sparkle.CLI.help import jobs as jobs_help 

12from sparkle.CLI.help import global_variables as gv 

13 

14 

15def parser_function() -> argparse.ArgumentParser: 

16 """Create parser for the cancel command.""" 

17 parser = argparse.ArgumentParser(description="Command to cancel running jobs.") 

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

19 parser.add_argument(*ac.AllJobsArgument.names, **ac.AllJobsArgument.kwargs) 

20 return parser 

21 

22 

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

24 """Main function of the cancel command.""" 

25 # Log command call 

26 logging.log_command(sys.argv) 

27 

28 # Define command line arguments 

29 parser = parser_function() 

30 

31 # Process command line arguments 

32 args = parser.parse_args(argv) 

33 

34 # Filter jobs on relevant status 

35 path = gv.settings().DEFAULT_log_output 

36 jobs = [run for run in jobs_help.get_runs_from_file(path) 

37 if run.status == Status.WAITING or run.status == Status.RUNNING] 

38 if args.all or args.job_ids: 

39 killed_jobs = [] 

40 for j in jobs: 

41 if args.all or j.run_id in args.job_ids: 

42 j.kill() 

43 killed_jobs.append(j) 

44 if len(killed_jobs) == 0: 

45 if args.all: 

46 print("No jobs to cancel.") 

47 sys.exit(0) 

48 else: 

49 print(f"ERROR: No jobs with ids {args.job_ids} to cancel.") 

50 # NOTE: Should we raise an error here instead? 

51 sys.exit(-1) 

52 print(f"Canceled {len(killed_jobs)} jobs with IDs: " 

53 f"{', '.join([j.run_id for j in killed_jobs])}.") 

54 elif len(jobs) == 0: 

55 print("No jobs available to cancel.") 

56 else: 

57 jobs = sorted(jobs, key=lambda j: j.run_id) 

58 ptg.Button.chars = {"delimiter": ["", ""]} # Disable padding around buttons 

59 

60 def cancel_jobs(self: ptg.Button) -> None: 

61 """Cancel jobs based on a button click.""" 

62 job_id = self.label.split("|")[1].strip() 

63 for job in jobs: 

64 if job.run_id == job_id: 

65 job.kill() 

66 # Replace button with label 

67 for iw, widget in enumerate(self.parent._widgets): 

68 if isinstance(widget, ptg.Button) and \ 

69 widget.label == self.label: 

70 self.parent._widgets[iw] = ptg.Label(self.label) 

71 break 

72 refresh_data(self.parent) 

73 

74 def refresh_data(self: ptg.Window) -> None: 

75 """Refresh the table.""" 

76 data_table = jobs_help.create_jobs_table(jobs, markup=True).splitlines() 

77 for iw, widget in enumerate(self._widgets): 

78 self._widgets[iw] = type(widget)(data_table[iw], onclick=cancel_jobs) 

79 self._widgets[iw].parent = self 

80 

81 table = jobs_help.create_jobs_table(jobs, markup=True).splitlines() 

82 with ptg.WindowManager() as manager: 

83 window = ( 

84 ptg.Window( 

85 width=len(table[0]), 

86 box="EMPTY", 

87 ) 

88 .set_title("Running Sparkle Jobs") 

89 ) 

90 for row in table: 

91 if "|" not in row or not row.split("|")[1].strip().isnumeric(): 

92 window._add_widget(ptg.Label(row)) 

93 else: 

94 window._add_widget(ptg.Button(label=row, onclick=cancel_jobs)) 

95 

96 manager.add(window) 

97 

98 sys.exit(0) 

99 

100 

101if __name__ == "__main__": 

102 main(sys.argv[1:])