Coverage for src/fsl_pipe_gui/all_panes.py: 43%

1"""Defines enumeration of all available panes for user to select the next one.""" 

2from enum import Enum 

3 

4from file_tree import FileTree 

5from fsl_pipe import Pipeline 

6from fsl_pipe.job import FileTarget, JobList, RunMethod, get_target 

7 

8 

9class AllPanes(Enum): 

10 """ 

11 Enumeration of all the possible GUI panes. 

12 

13 Panes should return one of these to indicate what the next pane will be. 

14 If None is returned by a pane instead, the application will quit. 

15 """ 

16 

17 PLACEHOLDER = 1 

18 SELECTOR = 2 

19 SUMMARY = 3 

20 RUN = 4 

21 

22 

23class PipelineSelector: 

24 """ 

25 Define which parts of the pipeline will be run. 

26 

27 This information will be updated by each pane in the pipeline. 

28 """ 

29 

30 def __init__( 

31 self, 

32 pipeline: Pipeline, 

33 file_tree: FileTree, 

34 overwrite_dependencies=False, 

35 run_method=None, 

36 ): 

37 """Create new pipeline selector to be edited in GUI.""" 

38 self.file_tree = file_tree 

39 all_templates = file_tree.template_keys() 

40 templates = { 

41 elem 

42 for piped_function in pipeline.scripts 

43 for elem in set.union( 

44 piped_function.filter_templates(False, all_templates), 

45 piped_function.filter_templates(True, all_templates), 

46 ) 

47 } 

48 self.partial_tree = file_tree.filter_templates(templates) 

49 self.pipeline = pipeline 

50 self.full_job_list = self.pipeline.generate_jobs(self.file_tree) 

51 self.selected_files = set() 

52 self.all_jobs = ([], set()) 

53 

54 self._overwrite_dependencies = overwrite_dependencies 

55 if run_method is None: 

56 run_method = RunMethod.default() 

57 self.run_method = run_method 

58 

59 @property 

60 def overwrite_dependencies( 

61 self, 

62 ): 

63 """Whether to override any already existing dependencies.""" 

64 return self._overwrite_dependencies 

65 

66 @overwrite_dependencies.setter 

67 def overwrite_dependencies(self, value): 

68 """Ensure jobs are reset when overwrite_dependencies is updated.""" 

69 self._overwrite_dependencies = value 

70 self.reset_all_jobs() 

71 

72 def get_file_targets(self, path) -> FileTarget: 

73 """Return file target corresponding to path.""" 

74 return get_target(path, self.full_job_list.targets) 

75 

76 def update_placeholders(self, **new_placeholders): 

77 """Update full job list for if placeholder values got changed.""" 

78 self.file_tree.placeholders.update(**new_placeholders) 

79 self.partial_tree.placeholders.update(**new_placeholders) 

80 self.full_job_list = self.pipeline.generate_jobs(self.file_tree) 

81 self.reset_all_jobs() 

82 

83 def add_selected_file(self, file): 

84 """Add a specific file to the list to be produced.""" 

85 self.selected_files.add(file) 

86 target = self.get_file_targets(file) 

87 target.producer.add_to_jobs( 

88 self.all_jobs, 

89 overwrite=True, 

90 overwrite_dependencies=self.overwrite_dependencies, 

91 ) 

92 

93 def remove_selected_file(self, file): 

94 """Add a specific file to the list to be produced.""" 

95 self.selected_files.remove(file) 

96 self.reset_all_jobs() 

97 

98 def reset_all_jobs( 

99 self, 

100 ): 

101 """Force a recalculation of all_jobs.""" 

102 self.all_jobs = ([], set()) 

103 for file in self.selected_files: 

104 if ( 

105 self.get_file_targets(file).producer is not None 

106 and self.get_file_targets(file).producer.expected() 

107 ): 

108 self.add_selected_file(file) 

109 

110 @property 

111 def job_list( 

112 self, 

113 ) -> JobList: 

114 """Collection of jobs that will be run.""" 

115 return JobList(self.file_tree, self.all_jobs[0], self.full_job_list.targets) 

116 

117 def run(self): 

118 """Run the pipeline.""" 

119 return self.job_list.run(self.run_method)