Coverage for src/fsl_pipe_gui/selector_tab.py: 36%

1"""Visualise the output selection part of the FSL pipeline GUI.""" 

2import itertools 

3import os.path as op 

4from functools import lru_cache 

5 

6from file_tree import FileTree, Template 

7from fsl_pipe.job import RunMethod 

8from rich.style import Style 

9from rich.text import Text 

10from textual.app import App, ComposeResult 

11from textual.binding import Binding 

12from textual.containers import Horizontal, Vertical 

13from textual.message import Message 

14from textual.widgets import Button, Checkbox, DataTable, Footer, Header, Select, Tree 

15from textual.widgets.tree import TreeNode 

16 

17from .all_panes import AllPanes, PipelineSelector 

18 

19 

20class TemplateSelect(Message, bubble=True): 

21 """Message sent when a template in the sidebar gets selected.""" 

22 

23 def __init__(self, sender, template: Template): 

24 """Create template selector.""" 

25 self.template = template 

26 super().__init__(sender) 

27 

28 

29class OutputSelector(App): 

30 """Textual application to select pipeline output templates/files.""" 

31 

32 TITLE = "FSL pipeline" 

33 CSS_PATH = "css/selector.css" 

34 

35 BINDINGS = [ 

36 Binding("r", "run", "Run pipeline", show=True), 

37 Binding("a", "select_all", "Select all for this template", show=True), 

38 ] 

39 

40 def __init__(self, selector: PipelineSelector): 

41 """Create pipeline output selector.""" 

42 super().__init__() 

43 self.selector = selector 

44 

45 @property 

46 def file_tree( 

47 self, 

48 ): 

49 """Input/output file structure.""" 

50 return self.selector.file_tree 

51 

52 @property 

53 def pipeline( 

54 self, 

55 ): 

56 """Set of pipeline recipes.""" 

57 return self.selector.pipeline 

58 

59 def compose(self) -> ComposeResult: 

60 """Create basic layout.""" 

61 yield Header() 

62 self.template_renderer = TemplateRenderer(self.selector) 

63 yield Vertical( 

64 Horizontal( 

65 TemplateTreeControl(self.selector.partial_tree, self.template_renderer), 

66 self.template_renderer, 

67 ), 

68 Horizontal( 

69 Button("Back", name="back"), 

70 Button("Run pipeline", name="run"), 

71 Select( 

72 [(e.name, e) for e in RunMethod], value=self.selector.run_method 

73 ), 

74 Checkbox( 

75 "Overwrite dependencies", value=self.selector.overwrite_dependencies 

76 ), 

77 id="control-bar", 

78 ), 

79 ) 

80 yield Footer() 

81 

82 async def handle_template_select(self, message: TemplateSelect): 

83 """User has selected a template.""" 

84 template = message.template 

85 self.app.sub_title = template.as_string 

86 await self.body.update(TemplateRenderer(template, self.file_tree)) 

87 

88 def on_button_pressed(self, event: Button.Pressed): 

89 """Continue to the next app.""" 

90 if event.button.name == "back": 

91 self.exit(AllPanes.PLACEHOLDER) 

92 elif event.button.name == "run": 

93 self.action_run() 

94 

95 def on_checkbox_changed(self, event: Checkbox.Changed): 

96 """Update overwrite dependencies.""" 

97 self.selector.overwrite_dependencies = event.checkbox.value 

98 

99 def on_select_changed(self, event: Select.Changed): 

100 """Update run method.""" 

101 self.selector.run_method = event.select.value 

102 

103 def action_run( 

104 self, 

105 ): 

106 """Run the pipeline.""" 

107 self.exit(AllPanes.SUMMARY) 

108 

109 def action_select_all(self): 

110 """Select all files matching this template.""" 

111 self.template_renderer.select_all() 

112 

113 

114class TemplateTreeControl(Tree): 

115 """Sidebar containing all template definitions in FileTree.""" 

116 

117 current_node = None 

118 BINDINGS = [ 

119 Binding("space", "toggle_node", "Collapse/Expand Node", show=True), 

120 Binding("up", "cursor_up", "Move Up", show=True), 

121 Binding("down", "cursor_down", "Move Down", show=True), 

122 ] 

123 

124 def __init__(self, file_tree: FileTree, renderer, name: str = None): 

125 """ 

126 Create a new template sidebar based on given FileTree. 

127 

128 :param file_tree: FileTree to interact with 

129 :param renderer: Panel showing the files corresponding to selected template. 

130 :param name: name of the sidebar within textual. 

131 """ 

132 self.file_tree = file_tree 

133 super().__init__("-", name=name) 

134 self.show_root = False 

135 self.find_children(self.root, self.file_tree.get_template("")) 

136 self.root.expand_all() 

137 self.renderer = renderer 

138 self.select_node(self.get_node_at_line(0)) 

139 

140 def on_mount( 

141 self, 

142 ): 

143 """Take focus on mount.""" 

144 self.focus() 

145 

146 def find_children(self, parent_node: TreeNode, template: Template): 

147 """ 

148 Find all the children of a template and add them to the node. 

149 

150 Calls itself recursively. 

151 """ 

152 all_children = template.children(self.file_tree._templates.values()) 

153 if len(all_children) == 0: 

154 parent_node.add_leaf(template.unique_part, template) 

155 else: 

156 this_node = parent_node.add(template.unique_part, template) 

157 children = set() 

158 for child in all_children: 

159 if child not in children: 

160 self.find_children(this_node, child) 

161 children.add(child) 

162 

163 def render_label(self, node: TreeNode[Template], base_style, style): 

164 """Render line in tree.""" 

165 if node.data is None: 

166 return node.label 

167 label = _render_node_helper(self.file_tree, node).copy() 

168 if node is self.cursor_node: 

169 label.stylize("reverse") 

170 if not node.is_expanded and len(node.children) > 0: 

171 label = Text("📁 ") + label 

172 return label 

173 

174 def on_tree_node_highlighted(self): 

175 """Inform other panel if template is selected.""" 

176 if self.current_node is not self.cursor_node: 

177 self.current_node = self.cursor_node 

178 self.renderer.render_template(self.current_node.data) 

179 

180 

181@lru_cache(None) 

182def _render_node_helper(tree: FileTree, node: TreeNode[Template]): 

183 meta = { 

184 "@click": f"click_label({node.id})", 

185 "tree_node": node.id, 

186 # "cursor": node.is_cursor, 

187 } 

188 paths = tree.get_mult( 

189 _get_template_key(tree, node.data), filter=True 

190 ).data.flatten() 

191 existing = [p for p in paths if p != ""] 

192 color = "blue" if len(existing) == len(paths) else "yellow" 

193 if len(existing) == 0: 

194 color = "red" 

195 counter = f" [{color}][{len(existing)}/{len(paths)}][/{color}]" 

196 res = Text.from_markup( 

197 node.data.rich_line(tree._templates) + counter, overflow="ellipsis" 

198 ) 

199 res.apply_meta(meta) 

200 return res 

201 

202 

203def _get_template_key(tree, template): 

204 """Get key representing template with file-tree.""" 

205 keys = {k for k, t in tree._templates.items() if t is template} 

206 return next(iter(keys)) 

207 

208 

209class TemplateRenderer(DataTable): 

210 """ 

211 Helper class to create a Rich rendering of a template. 

212 

213 There are two parts: 

214 

215 - a text file with the template 

216 - a table with the possible placeholder value combinations 

217 (shaded red for non-existing files) 

218 """ 

219 

220 def __init__(self, selector: PipelineSelector): 

221 """Create new renderer for template.""" 

222 super().__init__() 

223 self.selector = selector 

224 self.cursor_type = "row" 

225 

226 @property 

227 def file_tree( 

228 self, 

229 ): 

230 """Input/output file structure.""" 

231 return self.selector.file_tree 

232 

233 def on_mount(self): 

234 """Render upper-level template on mount.""" 

235 self.render_template(self.file_tree.get_template("")) 

236 

237 def render_template(self, template: Template): 

238 """Render the template as rich text.""" 

239 self.current_template = template 

240 self.clear(columns=True) 

241 xr = self.file_tree.get_mult( 

242 _get_template_key(self.file_tree, template), filter=False 

243 ) 

244 coords = sorted(xr.coords.keys()) 

245 self.add_column("", key="checkboxes") 

246 self.add_columns("", *coords, "filename") 

247 for values in itertools.product(*[xr.coords[c].data for c in coords]): 

248 path = xr.sel(**{c: v for c, v in zip(coords, values)}).item() 

249 style = Style(bgcolor="blue" if op.exists(path) else None) 

250 self.add_row( 

251 self.get_checkbox(path), 

252 *[Text(v, style=style) for v in values], 

253 Text(path, style=style), 

254 key=path, 

255 ) 

256 

257 def on_data_table_row_selected(self, message: DataTable.RowSelected): 

258 """Add or remove selected row from pipeline run.""" 

259 self.toggle_checkbox(message.row_key) 

260 

261 def select_all( 

262 self, 

263 ): 

264 """Add or remove selected row from pipeline run.""" 

265 for row in self.rows: 

266 self.toggle_checkbox(row, True) 

267 

268 def toggle_checkbox(self, row_key, new_value=None): 

269 """Toggles the checkbox in given row.""" 

270 current_checkbox = self.get_cell(row_key, "checkboxes") 

271 if current_checkbox == "☑" and new_value is not True: 

272 self.selector.remove_selected_file(row_key.value) 

273 self.update_cell(row_key, "checkboxes", self.get_checkbox(row_key.value)) 

274 elif current_checkbox in "☐⊡" and new_value is not False: 

275 self.selector.add_selected_file(row_key.value) 

276 self.update_cell(row_key, "checkboxes", "☑") 

277 

278 def get_checkbox(self, filename): 

279 """Return checkbox corresponding to filename.""" 

280 if filename in self.selector.selected_files: 

281 return "☑" 

282 target = self.selector.get_file_targets(filename) 

283 if target.producer is None: 

284 return "" 

285 elif target.producer in self.selector.all_jobs[1]: 

286 return "⊡" 

287 elif target.producer.expected(): 

288 return "☐" 

289 else: 

290 return "M"