Coverage for src/file_tree/file_tree.py: 87%
459 statements
« prev ^ index » next coverage.py v7.6.9, created at 2024-12-16 14:17 +0000
« prev ^ index » next coverage.py v7.6.9, created at 2024-12-16 14:17 +0000
1"""Defines the main FileTree object, which will be the main point of interaction."""
2import os
3import string
4import warnings
5from collections import defaultdict
6from difflib import get_close_matches
7from functools import cmp_to_key
8from pathlib import Path
9from shutil import copyfile
10from typing import Any, Collection, Dict, Generator, Optional, Sequence, Set, Union
11from warnings import warn
13import numpy as np
14import rich
15import xarray
17from .template import Placeholders, Template, is_singular, DuplicateTemplate
20class FileTree:
21 """Represents a structured directory.
23 The many methods can be split into 4 categories
25 1. The template interface. Each path (file or directory) is represented by a :class:`Template <file_tree.template.Template>`,
26 which defines the filename with any unknown parts (e.g., subject ID) marked by placeholders.
27 Templates are accessed based on their key.
29 - :meth:`get_template`: used to access a template based on its key.
30 - :meth:`template_keys`: used to list all the template keys.
31 - :meth:`add_template`: used to add a new template or overwrite an existing one.
32 - :meth:`add_subtree`: can be used to add all the templates from a different tree to this one.
33 - :meth:`override`: overrides some of the templates in this FileTree with that of another FileTree.
34 - :meth:`filter_templates`: reduce the filetree to a user-provided list of templates and its parents
36 2. The placeholder interface. Placeholders represent values to be filled into the placeholders.
37 Each placeholder can be either undefined, have a singular value, or have a sequence of possible values.
39 - You can access the :class:`placeholders dictionary-like object <file_tree.template.Placeholders>` directly through `FileTree.placeholders`
40 - :meth:`update`: returns a new FileTree with updated placeholders or updates the placeholders in the current one.
41 - :meth:`update_glob`: sets the placeholder values based on which files/directories exist on disk.
42 - :meth:`iter_vars`: iterate over all possible values for the selected placeholders.
43 - :meth:`iter`: iterate over all possible values for the placeholders that are part of a given template.
45 3. Getting the actual filenames based on filling the placeholder values into the templates.
47 - :meth:`get`: Returns a valid path by filling in all the placeholders in a template.
48 For this to work all placeholder values should be defined and singular.
49 - :meth:`get_mult`: Returns array of all possible valid paths by filling in the placeholders in a template.
50 Placeholder values can be singular or a sequence of possible values.
51 - :meth:`get_mult_glob`: Returns array with existing paths on disk.
52 Placeholder values can be singular, a sequence of possible values, or undefined.
53 In the latter case possible values for that placeholder are determined by checking the disk.
54 - :meth:`fill`: Returns new FileTree with any singular values filled into the templates and removed from the placeholder dict.
56 4. Input/output
58 - :meth:`report`: create a pretty overview of the filetree
59 - :meth:`run_app`: opens a terminal-based App to explore the filetree interactively
60 - :meth:`empty`: creates empty FileTree with no templates or placeholder values.
61 - :meth:`read`: reads a new FileTree from a file.
62 - :meth:`from_string`: reads a new FileTree from a string.
63 - :meth:`write`: writes a FileTree to a file.
64 - :meth:`to_string`: writes a FileTree to a string.
65 """
67 def __init__(
68 self,
69 templates: Dict[str, Template],
70 placeholders: Union[Dict[str, Any], Placeholders],
71 return_path=False,
72 ):
73 """Create a new FileTree with provided templates/placeholders."""
74 self._templates = templates
75 self.placeholders = Placeholders(placeholders)
76 self.return_path = return_path
78 # create new FileTree objects
79 @classmethod
80 def empty(
81 cls, top_level: Union[str, Template] = ".", return_path=False
82 ) -> "FileTree":
83 """Create a new empty FileTree containing only a top-level directory.
85 Args:
86 top_level: Top-level directory that other templates will use as a reference. Defaults to current directory.
87 return_path: if True, returns filenames as Path objects rather than strings.
89 Returns:
90 empty FileTree
91 """
92 if not isinstance(top_level, Template):
93 top_level = Template(None, top_level)
94 return cls({"": top_level}, {}, return_path=return_path)
96 @classmethod
97 def read(
98 cls,
99 name: str,
100 top_level: Union[str, Template] = ".",
101 return_path=False,
102 **placeholders,
103 ) -> "FileTree":
104 """Read a filetree based on the given name.
106 # noqa DAR101
108 Args:
109 name: name of the filetree. Interpreted as:
111 - a filename containing the tree definition if "name" or "name.tree" exist on disk
112 - one of the trees in `tree_directories` if one of those contains "name" or "name.tree"
113 - one of the tree in the plugin FileTree modules
115 top_level: top-level directory name. Defaults to current directory. Set to parent template for sub-trees.
116 return_path: if True, returns filenames as Path objects rather than strings.
117 placeholders: maps placeholder names to their values
119 Raises:
120 ValueError: if FileTree is not found.
122 Returns:
123 FileTree: tree matching the definition in the file
124 """
125 from . import parse_tree
127 if "directory" in placeholders:
128 warnings.warn(
129 f"Setting the 'directory' placeholder to {placeholders['directory']}. "
130 + "This differs from the behaviour of the old filetree in fslpy, which used the `directory` keyword to set the top-level directory. "
131 + "If you want to do that, please use the new `top_level` keyword instead of `directory`."
132 )
133 found_tree = parse_tree.search_tree(name)
134 if isinstance(found_tree, Path):
135 with open(found_tree, "r") as f:
136 text = f.read()
137 with parse_tree.extra_tree_dirs([found_tree.parent]):
138 return cls.from_string(
139 text, top_level, return_path=return_path, **placeholders
140 )
141 elif isinstance(found_tree, str):
142 return cls.from_string(
143 found_tree, top_level, return_path=return_path, **placeholders
144 )
145 elif isinstance(found_tree, FileTree):
146 new_tree = cls.empty(top_level, return_path)
147 new_tree.add_subtree(found_tree, fill=False)
148 return new_tree.update(**placeholders)
149 raise ValueError(
150 f"Type of object ({type(found_tree)}) returned when searching for FileTree named '{name}' was not recognised"
151 )
153 @classmethod
154 def from_string(
155 cls,
156 definition: str,
157 top_level: Union[str, Template] = ".",
158 return_path=False,
159 **placeholders,
160 ) -> "FileTree":
161 """Create a FileTree based on the given definition.
163 Args:
164 definition: A FileTree definition describing a structured directory
165 top_level: top-level directory name. Defaults to current directory. Set to parent template for sub-trees.
166 return_path: if True, returns filenames as Path objects rather than strings.
167 placeholders: key->value pairs setting initial value for any placeholders.
169 Returns:
170 FileTree: tree matching the definition in the file
171 """
172 from . import parse_tree
174 res = parse_tree.read_file_tree_text(definition.splitlines(), top_level).update(
175 inplace=True, **placeholders
176 )
177 res.return_path = return_path
178 return res
180 def copy(
181 self,
182 ) -> "FileTree":
183 """Create a copy of the tree.
185 The dictionaries (templates, placeholders) are copied, but the values within them are not.
187 Returns:
188 FileTree: new tree object with identical templates, sub-trees and placeholders
189 """
190 new_tree = type(self)(
191 dict(self._templates), Placeholders(self.placeholders), self.return_path
192 )
193 return new_tree
195 # template interface
196 def get_template(self, key: str, error_duplicate=True) -> Template:
197 """Return the template corresponding to `key`.
199 Raises:
200 KeyError: if no template with that identifier is available # noqa DAR402
201 ValueError: if multiple templates with that identifier are available (suppress using `error_duplicate=False`)
203 Args:
204 key (str): key identifying the template.
205 error_duplicate (bool): set to False to return a `DuplicateTemplate` object rather than raising an error
207 Returns:
208 Template: description of pathname with placeholders not filled in
209 """
210 try:
211 value = self._templates[key]
212 if error_duplicate and isinstance(value, DuplicateTemplate):
213 templates = ", ".join([str(t.as_path) for t in value.templates])
214 raise ValueError(f"There are multiple templates matching key '{key}': {templates}")
215 return value
216 except KeyError:
217 pass
218 matches = get_close_matches(key, self.template_keys())
219 if len(matches) == 0:
220 raise KeyError(f"Template key '{key}' not found in FileTree.")
221 else:
222 raise KeyError(
223 f"Template key '{key}' not found in FileTree; did you mean {' or '.join(sorted(matches))}?"
224 )
226 @property
227 def top_level(
228 self,
229 ):
230 """Top-level directory.
232 Within the template dictionary this top-level directory is represented with an empty string
233 """
234 as_string = self.get_template("").unique_part
235 if self.return_path:
236 return Path(as_string)
237 return str(as_string)
239 @top_level.setter
240 def top_level(self, value: str):
241 self.get_template("").unique_part = str(value)
243 def add_template(
244 self,
245 template_path: str,
246 key: Optional[Union[str, Sequence[str]]] = None,
247 parent: Optional[str] = "",
248 overwrite=False,
249 ) -> Template:
250 """Update the FileTree with the new template.
252 Args:
253 template_path: path name with respect to the parent (or top-level if no parent provided)
254 key: key(s) to access this template in the future. Defaults to result from :meth:`Template.guess_key <file_tree.template.Template.guess_key>`
255 (i.e., the path basename without the extension).
256 parent: if defined, `template_path` will be interpreted as relative to this template.
257 By default the top-level template is used as reference.
258 To create a template unaffiliated with the rest of the tree, set `parent` to None.
259 Such a template should be an absolute path or relative to the current directory and can be used as parent for other templates
260 overwrite: if True, overwrites any existing template rather than raising a ValueError. Defaults to False.
262 Returns:
263 Template: the newly added template object
264 """
265 if parent is None:
266 parent_template = None
267 elif isinstance(parent, Template):
268 parent_template = parent
269 else:
270 parent_template = self.get_template(parent)
271 new_template = Template(parent_template, template_path)
272 if (
273 parent_template is not None
274 and new_template.as_path == parent_template.as_path
275 ):
276 new_template = parent_template
277 return self._add_actual_template(new_template, key, overwrite=overwrite)
279 def _add_actual_template(
280 self,
281 template: Template,
282 keys: Optional[Union[str, Sequence[str]]] = None,
283 overwrite=False,
284 ):
285 if keys is None:
286 keys = template.guess_key()
287 if isinstance(keys, Path):
288 keys = str(keys)
289 if isinstance(keys, str):
290 keys = [keys]
291 for key in keys:
292 if key in self._templates:
293 old_template = self.get_template(key, error_duplicate=overwrite)
294 if not overwrite:
295 if isinstance(old_template, DuplicateTemplate):
296 old_template.add_template(template)
297 else:
298 self._templates[key] = DuplicateTemplate(old_template, template)
299 continue
301 for potential_child in self._templates.values():
302 if potential_child.parent is old_template:
303 potential_child.parent = template
304 self._templates[key] = template
305 return template
307 def override(self, new_filetree: "FileTree", required: Collection[str]=[], optional: Collection[str]=[]):
308 """Overide some templates and all placeholders in this filetree with templates from `new_filetree`.
310 A new `FileTree` is returned with all the template keys in `required` replaced or added.
311 Template keys in `optional` will also be replaced or added if they are present in `new_filetree`.
313 Any placeholders defined in `new_filetree` will be transfered as well.
315 Without supplying any keys to `required` or `optional` the new `FileTree` will be identical to this one.
316 """
317 if isinstance(required, str):
318 required = [required]
319 if isinstance(optional, str):
320 optional = [optional]
321 all_keys = set(required).union(optional)
323 old_duplicate_keys = self.template_keys(skip_duplicates=False).difference(self.template_keys(skip_duplicates=True))
324 duplicates = [key for key in all_keys if key in old_duplicate_keys]
325 if len(duplicates) > 0:
326 raise ValueError("Some of the keys to be replaced in the original FileTree are duplicates: %s", ", ".join(duplicates))
328 new_available_keys = new_filetree.template_keys(skip_duplicates=False)
329 undefined = [key for key in required if key not in new_available_keys]
330 if len(undefined) > 0:
331 raise ValueError("Some required keys are missing from the input FileTree: %s", ", ".join(undefined))
333 new_duplicate_keys = new_available_keys.difference(new_filetree.template_keys(skip_duplicates=True))
334 duplicates = [key for key in all_keys if key in new_duplicate_keys]
335 if len(duplicates) > 0:
336 raise ValueError("Some of the keys to be used in the input FileTree are duplicates: %s", ", ".join(duplicates))
338 res_filetree = self.copy()
339 res_filetree.placeholders.update(new_filetree.placeholders)
340 for key in all_keys:
341 if key in new_available_keys:
342 res_filetree._add_actual_template(new_filetree.get_template(key), key, overwrite=True)
343 return res_filetree
345 @property
346 def _iter_templates(self, ) -> Dict[Template, Set[str]]:
347 result = defaultdict(set)
348 def add_parent(t:Template):
349 if t.parent is None or t.parent in result:
350 return
351 result[t.parent]
352 add_parent(t.parent)
354 for (key, possible) in self._templates.items():
355 if isinstance(possible, DuplicateTemplate):
356 for template in possible.templates:
357 result[template].add(key)
358 add_parent(template)
359 else:
360 result[possible].add(key)
361 add_parent(possible)
362 return dict(result)
364 def template_keys(self, only_leaves=False, skip_duplicates=True):
365 """Return the keys of all the templates in the FileTree.
367 Each key will be returned for templates with multiple keys.
369 Args
370 only_leaves (bool, optional): set to True to only return templates that do not have any children.
371 skip_duplicates (bool, optional): set to False to include keys that point to multiple templates.
372 """
373 if skip_duplicates:
374 keys = {k for (k, v) in self._templates.items() if isinstance(v, Template)}
375 else:
376 keys = set(self._templates.keys())
377 if not only_leaves:
378 return keys
379 elif not skip_duplicates:
380 raise ValueError("Cannot select only leaves when not skipping duplicates.")
382 parents = {t.parent for t in self._iter_templates.keys() if t.parent is not None}
383 return {
384 key for key in keys if self.get_template(key) not in parents
385 }
387 def add_subtree(
388 self,
389 sub_tree: "FileTree",
390 precursor: Union[Optional[str], Sequence[Optional[str]]] = (None,),
391 parent: Optional[Union[str, Template]] = "",
392 fill=None,
393 ) -> None:
394 """Update the templates and the placeholders in place with those in sub_tree.
396 The top-level directory of the sub-tree will be replaced by the `parent` (unless set to None).
397 The sub-tree templates will be available with the key "<precursor>/<original_key>",
398 unless the precursor is None in which case they will be unchanged (which can easily lead to errors due to naming conflicts).
400 What happens with the placeholder values of the sub-tree depends on whether the precursor is None or not:
402 - if the precursor is None, any singular values are directly filled into the sub-tree templates.
403 Any placeholders with multiple values will be added to the top-level variable list (error is raised in case of conflicts).
404 - if the precursor is a string, the templates are updated to look for "<precursor>/<original_placeholder>" and
405 all sub-tree placeholder values are also prepended with this precursor.
406 Any template values with "<precursor>/<key>" will first look for that full key, but if that is undefined
407 they will fall back to "<key>" (see :class:`Placeholders <file_tree.template.Placeholders>`).
409 The net effect of either of these procedures is that the sub-tree placeholder values will be used in that sub-tree,
410 but will not affect templates defined elsewhere in the parent tree.
411 If a placeholder is undefined in a sub-tree, it will be taken from the parent placeholder values (if available).
413 Args:
414 sub_tree: tree to be added to the current one
415 precursor: name(s) of the sub-tree. Defaults to just adding the sub-tree to the main tree without precursor
416 parent: key of the template used as top-level directory for the sub tree.
417 Defaults to top-level directory of the main tree.
418 Can be set to None for an independent tree.
419 fill: whether any defined placeholders should be filled in before adding the sub-tree. By default this is True if there is no precursor and false otherwise
421 Raises:
422 ValueError: if there is a conflict in the template names.
423 """
424 if isinstance(precursor, str) or precursor is None:
425 precursor = [precursor]
426 for name in precursor:
427 sub_tree_fill = sub_tree
428 if name is None:
429 add_string = ""
430 if (fill is None) or fill:
431 sub_tree_fill = sub_tree.fill()
432 else:
433 add_string = name + "/"
434 if fill:
435 sub_tree_fill = sub_tree.fill()
437 to_assign = dict(sub_tree_fill._iter_templates)
438 sub_top_level = [k for (k, v) in to_assign.items() if "" in v][0]
440 if parent is None:
441 new_top_level = Template(None, sub_top_level.unique_part)
442 elif isinstance(parent, Template):
443 new_top_level = parent
444 else:
445 new_top_level = self.get_template(parent)
446 if name is None:
447 if parent is None:
448 for letter in string.ascii_letters:
449 label = f"tree_top_{letter}"
450 if (
451 label not in sub_tree_fill.template_keys()
452 and label not in self.template_keys()
453 ):
454 self._add_actual_template(new_top_level, label)
455 break
456 else:
457 self._add_actual_template(new_top_level, add_string)
459 been_assigned = {sub_top_level: new_top_level}
460 del to_assign[sub_top_level]
461 while len(to_assign) > 0:
462 for old_template, keys in list(to_assign.items()):
463 if old_template.parent is None:
464 parent_template = None
465 elif old_template.parent in been_assigned:
466 parent_template = been_assigned[old_template.parent]
467 else:
468 continue
469 new_template = Template(
470 parent_template, old_template.unique_part
471 ).add_precursor(add_string)
472 for key in keys:
473 self._add_actual_template(new_template, add_string + key)
474 been_assigned[old_template] = new_template
475 del to_assign[old_template]
477 if name is None:
478 conflict = {
479 key
480 for key in sub_tree_fill.placeholders.keys()
481 if key in self.placeholders
482 }
483 if len(conflict) > 0:
484 raise ValueError(
485 f"Sub-tree placeholder values for {conflict} conflict with those set in the parent tree."
486 )
487 for old_key, old_value in sub_tree_fill.placeholders.items():
488 if isinstance(old_key, str):
489 self.placeholders[add_string + old_key] = old_value
490 else:
491 self.placeholders[frozenset(add_string + k for k in old_key)] = {
492 add_string + k: v for k, v in old_value.items()
493 }
495 def filter_templates(
496 self, template_names: Collection[str], check=True
497 ) -> "FileTree":
498 """Create new FileTree containing just the templates in `template_names` and their parents.
500 Args:
501 template_names: names of the templates to keep.
502 check: if True, check whether all template names are actually part of the FileTree
504 Raises:
505 KeyError: if any of the template names are not in the FileTree (unless `check` is set to False).
507 Returns:
508 FileTree containing requested subset of templates.
509 """
510 all_keys = self.template_keys()
511 if check:
512 undefined = {name for name in template_names if name not in all_keys}
513 if len(undefined) > 0:
514 raise KeyError("Undefined template names found in filter: ", undefined)
516 new_filetree = FileTree(
517 {}, self.placeholders.copy(), return_path=self.return_path
518 )
520 already_added = set()
522 def add_template(template: Template):
523 if template in already_added:
524 return
525 if template.parent is not None:
526 add_template(template.parent)
527 new_filetree._add_actual_template(template, self._iter_templates[template])
528 already_added.add(template)
530 for name in template_names:
531 if name not in all_keys:
532 continue
533 add_template(self.get_template(name))
535 return new_filetree
537 # placeholders interface
538 def update(self, inplace=False, **placeholders) -> "FileTree":
539 """Update the placeholder values to be filled into the templates.
541 Args:
542 inplace (bool): if True change the placeholders in-place (and return the FileTree itself);
543 by default a new FileTree is returned with the updated values without altering this one.
544 **placeholders (Dict[str, Any]): maps placeholder names to their new values (None to mark placeholder as undefined)
546 Returns:
547 FileTree: Tree with updated placeholders (same tree as the current one if inplace is True)
548 """
549 new_tree = self if inplace else self.copy()
550 new_tree.placeholders.update(placeholders)
551 return new_tree
553 def update_glob(
554 self, template_key: Union[str, Sequence[str]], inplace=False,
555 link: Union[None, Sequence[str], Sequence[Sequence[str]]]=None
556 ) -> "FileTree":
557 """Update any undefined placeholders based on which files exist on disk for template.
559 Args:
560 template_key (str or sequence of str): key(s) of the template(s) to use
561 inplace (bool): if True change the placeholders in-place (and return the FileTree itself);
562 by default a new FileTree is returned with the updated values without altering this one.
563 link (sequences of str): template keys that should be linked together in the output.
565 Returns:
566 FileTree: Tree with updated placeholders (same tree as the current one if inplace is True)
567 """
568 if link is None:
569 link = []
570 elif len(link) > 0 and isinstance(link[0], str):
571 link = [link]
572 link_as_frozenset = [frozenset(l) for l in link]
574 if isinstance(template_key, str):
575 template_key = [template_key]
576 new_placeholders: Dict[str, Set[str]] = defaultdict(set)
577 new_links = [set() for _ in range(len(link))]
578 for key in template_key:
579 template = self.get_template(key)
580 from_template = template.get_all_placeholders(self.placeholders, link=link)
581 for name, values in from_template.items():
582 if isinstance(name, frozenset):
583 index = link_as_frozenset.index(name)
584 values_as_tuples = zip(*[values[key] for key in link[index]])
585 new_links[index].update(values_as_tuples)
586 else:
587 new_placeholders[name] = new_placeholders[name].union(values)
589 def cmp(item1, item2):
590 if item1 is None:
591 return -1
592 if item2 is None:
593 return 1
594 if item1 < item2:
595 return -1
596 if item1 > item2:
597 return 1
598 return 0
600 new_tree = self if inplace else self.copy()
601 new_tree.placeholders.update(
602 {k: sorted(v, key=cmp_to_key(cmp)) for k, v in new_placeholders.items()},
603 )
604 for key, value in zip(link, new_links):
605 new_tree.placeholders[tuple(key)] = list(zip(*sorted(value)))
606 return new_tree
608 # Extract paths
609 def get(self, key: str, make_dir=False) -> Union[str, Path]:
610 """Return template with placeholder values filled in.
612 Args:
613 key (str): identifier for the template
614 make_dir (bool, optional): If set to True, create the parent directory of the returned path.
616 Returns:
617 Path: Filled in template as Path object.
618 Returned as a `pathlib.Path` object if `FileTree.return_path` is True.
619 Otherwise a string is returned.
620 """
621 path = self.get_template(key).format_single(self.placeholders)
622 if make_dir:
623 Path(path).parent.mkdir(parents=True, exist_ok=True)
624 if self.return_path:
625 return Path(path)
626 return path
628 def get_mult(
629 self, key: Union[str, Sequence[str]], filter=False, make_dir=False
630 ) -> Union[xarray.DataArray, xarray.Dataset]:
631 """Return array of paths with all possible values filled in for the placeholders.
633 Singular placeholder values are filled into the template directly.
634 For each placeholder with multiple values a dimension is added to the output array.
635 This dimension will have the name of the placeholder and labels corresponding to the possible values (see http://xarray.pydata.org/en/stable/).
636 The precense of required, undefined placeholders will lead to an error
637 (see :meth:`get_mult_glob` or :meth:`update_glob` to set these placeholders based on which files exist on disk).
639 Args:
640 key (str, Sequence[str]): identifier(s) for the template.
641 filter (bool, optional): If Set to True, will filter out any non-existent files.
642 If the return type is strings, non-existent entries will be empty strings.
643 If the return type is Path objects, non-existent entries will be None.
644 Note that the default behaviour is opposite from :meth:`get_mult_glob`.
645 make_dir (bool, optional): If set to True, create the parent directory for each returned path.
647 Returns:
648 xarray.DataArray, xarray.Dataset: For a single key returns all possible paths in an xarray DataArray.
649 For multiple keys it returns the combination of them in an xarray Dataset.
650 Each element of in the xarray is a `pathlib.Path` object if `FileTree.return_path` is True.
651 Otherwise the xarray will contain the paths as strings.
652 """
653 if isinstance(key, str):
654 paths = self.get_template(key).format_mult(self.placeholders, filter=filter)
655 paths.name = key
656 if make_dir:
657 for path in paths.data.flat:
658 if path is not None:
659 Path(path).parent.mkdir(parents=True, exist_ok=True)
660 if self.return_path:
661 return xarray.apply_ufunc(
662 lambda p: None if p == "" else Path(p), paths, vectorize=True
663 )
664 return paths
665 else:
666 return xarray.merge(
667 [self.get_mult(k, filter=filter, make_dir=make_dir) for k in key],
668 join="exact",
669 )
671 def get_mult_glob(
672 self, key: Union[str, Sequence[str]]
673 ) -> Union[xarray.DataArray, xarray.Dataset]:
674 """Return array of paths with all possible values filled in for the placeholders.
676 Singular placeholder values are filled into the template directly.
677 For each placeholder with multiple values a dimension is added to the output array.
678 This dimension will have the name of the placeholder and labels corresponding to the possible values (see http://xarray.pydata.org/en/stable/).
679 The possible values for undefined placeholders will be determined by which files actually exist on disk.
681 The same result can be obtained by calling `self.update_glob(key).get_mult(key, filter=True)`.
682 However calling this method is more efficient, because it only has to check the disk for which files exist once.
684 Args:
685 key (str, Sequence[str]): identifier(s) for the template.
687 Returns:
688 xarray.DataArray, xarray.Dataset: For a single key returns all possible paths in an xarray DataArray.
689 For multiple keys it returns the combination of them in an xarray Dataset.
690 Each element of in the xarray is a `pathlib.Path` object if `FileTree.return_path` is True.
691 Otherwise the xarray will contain the paths as strings.
692 """
693 if isinstance(key, str):
694 template = self.get_template(key)
695 matches = template.all_matches(self.placeholders)
697 new_placeholders = Placeholders(self.placeholders)
698 updates, matches = template.get_all_placeholders(self.placeholders, return_matches=True)
699 new_placeholders.update(updates)
701 paths = template.format_mult(new_placeholders, filter=True, matches=matches)
702 paths.name = key
703 if self.return_path:
704 return paths
705 res = xarray.apply_ufunc(
706 lambda p: "" if p is None else str(p), paths, vectorize=True
707 )
708 return res
709 else:
710 return xarray.merge(
711 [self.get_mult_glob(k) for k in key],
712 join="outer",
713 fill_value=None if self.return_path else "",
714 )
716 def fill(self, keep_optionals=True) -> "FileTree":
717 """Fill in singular placeholder values.
719 Args:
720 keep_optionals: if True keep optional parameters that have not been set
722 Returns:
723 FileTree: new tree with singular placeholder values filled into the templates and removed from the placeholder dict
724 """
725 new_tree = FileTree({}, self.placeholders.split()[1], self.return_path)
726 to_assign = dict(self._iter_templates)
727 template_mappings = {None: None}
728 while len(to_assign) > 0:
729 for old_template, keys in list(to_assign.items()):
730 if old_template.parent in template_mappings:
731 new_parent = template_mappings[old_template.parent]
732 else:
733 continue
734 new_template = Template(
735 new_parent,
736 str(
737 Template(None, old_template.unique_part).format_single(
738 self.placeholders,
739 check=False,
740 keep_optionals=keep_optionals,
741 )
742 ),
743 )
744 template_mappings[old_template] = new_template
745 new_tree._add_actual_template(new_template, keys)
746 del to_assign[old_template]
747 return new_tree
749 # iteration
750 def iter_vars(
751 self, placeholders: Sequence[str]
752 ) -> Generator["FileTree", None, None]:
753 """Iterate over the user-provided placeholders.
755 A single file-tree is yielded for each possible value of the placeholders.
757 Args:
758 placeholders (Sequence[str]): sequence of placeholder names to iterate over
760 Yields:
761 FileTrees, where each placeholder only has a single possible value
762 """
763 for sub_placeholders in self.placeholders.iter_over(placeholders):
764 yield FileTree(self._templates, sub_placeholders, self.return_path)
766 def iter(
767 self, template: str, check_exists: bool = False
768 ) -> Generator["FileTree", None, None]:
769 """Iterate over trees containng all possible values for template.
771 Args:
772 template (str): short name identifier of the template
773 check_exists (bool): set to True to only return trees for which the template actually exists
775 Yields:
776 FileTrees, where each placeholder in given template only has a single possible value
777 """
778 placeholders = self.get_template(template).placeholders(
779 valid=self.placeholders.keys()
780 )
781 for tree in self.iter_vars(placeholders):
782 if not check_exists or Path(tree.get(template)).exists:
783 yield tree
785 # convert to string
786 def to_string(self, indentation=4) -> str:
787 """Convert FileTree into a valid filetree definition.
789 An identical FileTree can be created by running :meth:`from_string` on the resulting string.
791 Args:
792 indentation (int, optional): Number of spaces to use for indendation. Defaults to 4.
794 Returns:
795 String representation of FileTree.
796 """
797 lines = [self.placeholders.to_string()]
799 top_level = sorted(
800 [
801 template
802 for template in self._iter_templates.keys()
803 if template.parent is None
804 ],
805 key=lambda k: ",".join(self._iter_templates[k]),
806 )
807 already_done = set()
808 for t in top_level:
809 if t not in already_done:
810 lines.append(t.as_multi_line(self._iter_templates, indentation=indentation))
811 already_done.add(t)
812 return "\n\n".join(lines)
814 def write(self, filename, indentation=4):
815 """Write the FileTree to a disk as a text file.
817 The first few lines will contain the placeholders.
818 The remaining lines will contain the actual FileTree with all the templates (including sub-trees).
819 The top-level directory is not stored in the file and hence will need to be provided when reading the tree from the file.
821 Args:
822 filename (str or Path): where to store the file (directory should exist already)
823 indentation (int, optional): Number of spaces to use in indendation. Defaults to 4.
824 """
825 with open(filename, "w") as f:
826 f.write(self.to_string(indentation=indentation))
828 def report(self, fill=True, pager=False):
829 """Print a formatted report of the filetree to the console.
831 Prints a report of the file-tree to the terminal with:
832 - table with placeholders and their values
833 - tree of templates with template keys marked in cyan
835 Args:
836 fill (bool, optional): by default any fixed placeholders are filled in before printing the tree (using :meth:`fill`). Set to False to disable this.
837 pager (bool, optional): if set to True, the report will be filed into a pager (recommended if the output is very large)
838 """
839 if fill:
840 self = self.fill()
842 if pager:
843 from rich.console import Console
845 printer = Console()
846 with printer.pager():
847 for part in self._generate_rich_report():
848 printer.print(part)
849 else:
850 for part in self._generate_rich_report():
851 rich.print(part)
853 def _generate_rich_report(self):
854 """Generate a sequence of Rich renderables to produce report."""
855 from rich.table import Table
856 from rich.tree import Tree
858 single_vars = {}
859 multi_vars = {}
860 linked_vars = []
861 for key, value in self.placeholders.items():
862 if value is None:
863 continue
864 if isinstance(key, frozenset):
865 linked_vars.append(sorted(key))
866 for linked_key, linked_value in value.items():
867 multi_vars[linked_key] = linked_value
868 elif np.array(value).ndim == 1:
869 multi_vars[key] = value
870 else:
871 single_vars[key] = value
872 if len(single_vars) > 0:
873 single_var_table = Table("name", "value", title="Defined placeholders")
874 for key in sorted(single_vars.keys()):
875 single_var_table.add_row(key, single_vars[key])
876 yield single_var_table
877 if len(multi_vars) > 0:
878 multi_var_table = Table(
879 "name", "value", title="Placeholders with multiple options"
880 )
881 for key in sorted(multi_vars.keys()):
882 multi_var_table.add_row(key, ", ".join(str(v) for v in multi_vars[key]))
883 yield multi_var_table
885 if len(linked_vars) > 0:
886 yield "Linked variables:\n" + (
887 "\n".join([", ".join(v) for v in linked_vars])
888 )
891 def add_children(t: Template, tree: Optional[Tree]):
892 for child in sorted(t.children(self._iter_templates.keys()), key=lambda t: t.as_string):
893 child_tree = tree.add(child.rich_line(self._iter_templates))
894 add_children(child, child_tree)
896 top_level = sorted(
897 [
898 template
899 for template in self._iter_templates.keys()
900 if template.parent is None
901 ],
902 key=lambda t: ",".join(self._iter_templates[t]),
903 )
904 already_done = set()
905 for t in top_level:
906 if t not in already_done:
907 base_tree = Tree(t.rich_line(self._iter_templates))
908 add_children(t, base_tree)
909 yield base_tree
911 def run_app(
912 self,
913 ):
914 """
915 Open a terminal-based App to explore the filetree interactively.
917 The resulting app runs directly in the terminal,
918 so it should work when ssh'ing to some remote cluster.
920 There will be two panels:
922 - The left panel will show all the templates in a tree format.
923 Template keys are shown in cyan.
924 For each template the number of files that exist on disc out of the total number is shown
925 colour coded based on completeness (red: no files; yellow: some files; blue: all files).
926 Templates can be selected by hovering over them.
927 Clicking on directories with hide/show their content.
928 - The right panel will show for the selected template the complete template string
929 and a table showing for which combination of placeholders the file is present/absent
930 (rows for absent files are colour-coded red).
931 """
932 from . import app
933 app.FileTreeViewer(self).run()
936def convert(
937 src_tree: FileTree,
938 target_tree: Optional[FileTree] = None,
939 keys=None,
940 symlink=False,
941 overwrite=False,
942 glob_placeholders=None,
943):
944 """
945 Copy or link files defined in `keys` from the `src_tree` to the `target_tree`.
947 Given two example trees
949 - source::
951 subject = A,B
953 sub-{subject}
954 data
955 T1w.nii.gz
956 FLAIR.nii.gz
958 - target::
960 subject = A,B
962 data
963 sub-{subject}
964 {subject}-T1w.nii.gz (T1w)
965 {subject}-T2w.nii.gz (T2w)
967 And given pre-existing data matching the source tree::
969 .
970 ├── sub-A
971 │ └── data
972 │ ├── FLAIR.nii.gz
973 │ └── T1w.nii.gz
974 └── sub-B
975 └── data
976 ├── FLAIR.nii.gz
977 └── T1w.nii.gz
979 We can do the following conversions:
981 - `convert(source, target)`:
982 copies all matching keys from `source` to `target`.
983 This will only copy the "T1w.nii.gz" files, because they are the only match in the template keys.
984 Note that the `data` template key also matches between the two trees, but this template is not a leaf, so is ignored.
985 - `convert(source, target, keys=['T1w', ('FLAIR', 'T2w')])`:
986 copies the "T1w.nii.gz" files from `source` to `target` and
987 copies the "FLAIR.nii.gz" in `source` to "T2w..nii.gz" in `target`.
988 - `convert(source.update(subject='B'), source.update(subject='C'))`:
989 creates a new "data/sub-C" directory and
990 copies all the data from "data/sub-B" into that directory.
991 - `convert(source, keys=[('FLAIR', 'T1w')], overwrite=True)`:
992 copies the "FLAIR.nii.gz" into the "T1w.nii.gz" files overwriting the originals.
994 Warnings are raised in two cases:
996 - if a source file is missing
997 - if a target file already exists and `overwrite` is False
999 Args:
1000 src_tree: prepopulated filetree with the source files
1001 target_tree: filetree that will be populated. Defaults to same as `src_tree`.
1002 keys: collection of template keys to transfer from `src_tree` to `target_tree`. Defaults to all templates keys shared between `src_tree` and `target_tree`.
1003 symlink: if set to true links the files rather than copying them
1004 overwrite: if set to True overwrite any existing files
1005 glob_placeholders: Placeholders that should be treated as wildcards. This is meant for placeholders that have different values for each filename.
1007 Raises:
1008 ValueError: if the conversion can not be carried out. If raised no data will be copied/linked.
1009 """
1010 if target_tree is None and keys is None:
1011 raise ValueError("Conversion requires either `target_tree` or `keys` to be set")
1012 src_tree = src_tree.copy()
1013 if target_tree is None:
1014 target_tree = src_tree
1015 target_tree = target_tree.copy()
1016 if keys is None:
1017 keys = set(src_tree.template_keys(only_leaves=True)).intersection(
1018 target_tree.template_keys(only_leaves=True)
1019 )
1020 if glob_placeholders is None:
1021 glob_placeholders = set()
1022 for p in glob_placeholders:
1023 if p in src_tree.placeholders:
1024 raise ValueError(
1025 f"Placeholder {p} has been selected for globbing, however values were set for it in source tree"
1026 )
1027 if p in target_tree.placeholders:
1028 raise ValueError(
1029 f"Placeholder {p} has been selected for globbing, however values were set for it in target tree"
1030 )
1032 full_keys = {
1033 (key_definition, key_definition)
1034 if isinstance(key_definition, str)
1035 else key_definition
1036 for key_definition in keys
1037 }
1038 for src_key, target_key in full_keys:
1039 # ensure template placeholders are consistent between source and target tree
1040 for placeholder in (
1041 src_tree.get_template(src_key).placeholders()
1042 + target_tree.get_template(target_key).placeholders()
1043 ):
1044 if placeholder in glob_placeholders:
1045 continue
1046 if placeholder not in src_tree.placeholders:
1047 if placeholder not in target_tree.placeholders:
1048 raise ValueError(
1049 f"Can not convert template {src_key}, because no values have been set for {placeholder}"
1050 )
1051 src_tree.placeholders[placeholder] = target_tree.placeholders[
1052 placeholder
1053 ]
1054 elif placeholder not in target_tree.placeholders:
1055 target_tree.placeholders[placeholder] = src_tree.placeholders[
1056 placeholder
1057 ]
1058 nsrc = (
1059 -1
1060 if is_singular(src_tree.placeholders[placeholder])
1061 else len(src_tree.placeholders[placeholder])
1062 )
1063 ntarget = (
1064 -1
1065 if is_singular(target_tree.placeholders[placeholder])
1066 else len(target_tree.placeholders[placeholder])
1067 )
1068 if nsrc != ntarget:
1069 raise ValueError(
1070 f"Number of possible values for {placeholder} do not match between source and target tree"
1071 )
1073 # ensure non-singular placeholders match
1074 src_non_singular = {
1075 p
1076 for p in src_tree.get_template(src_key).placeholders()
1077 if p not in glob_placeholders and not is_singular(src_tree.placeholders[p])
1078 }
1079 target_non_singular = {
1080 p
1081 for p in target_tree.get_template(target_key).placeholders()
1082 if p not in glob_placeholders
1083 and not is_singular(target_tree.placeholders[p])
1084 }
1086 diff = src_non_singular.difference(target_non_singular).difference(
1087 glob_placeholders
1088 )
1089 if len(diff) > 0:
1090 raise ValueError(
1091 f"Placeholders {diff} in source template {src_key} has no equivalent in target template {target_key}"
1092 )
1093 diff = target_non_singular.difference(src_non_singular)
1094 if len(diff) > 0:
1095 raise ValueError(
1096 f"Placeholders {diff} in target template {target_key} has no equivalent in source template {src_key}"
1097 )
1099 # all checks have passed; let's get to work
1100 to_warn_about = ([], [])
1102 transfer_filenames = []
1104 for src_key, target_key in sorted(full_keys):
1105 iter_placeholders = sorted(
1106 {
1107 p
1108 for p in src_tree.get_template(src_key).placeholders()
1109 if p not in glob_placeholders
1110 and not is_singular(src_tree.placeholders[p])
1111 }
1112 )
1113 for single_src_tree, single_target_tree in zip(
1114 src_tree.iter_vars(iter_placeholders),
1115 target_tree.iter_vars(iter_placeholders),
1116 ):
1117 if len(glob_placeholders) == 0:
1118 src_fn = single_src_tree.get(src_key)
1119 target_fn = single_target_tree.get(target_key)
1120 else:
1121 try:
1122 src_trees = list(single_src_tree.update_glob(src_key).iter(src_key))
1123 except ValueError:
1124 to_warn_about[0].append(
1125 single_src_tree.get_template(src_key).format_single(
1126 single_src_tree.placeholders, check=False
1127 )
1128 )
1129 continue
1130 if len(src_trees) > 1:
1131 raise ValueError(
1132 f"Multiple matching filenames were found when globbing {src_key} ({single_src_tree.get(src_key)})"
1133 )
1135 keys = {
1136 key: src_trees[0].placeholders[key]
1137 for key in glob_placeholders
1138 if key in src_trees[0].placeholders
1139 }
1140 src_fn = single_src_tree.update(**keys).get(src_key)
1141 target_fn = single_target_tree.update(**keys).get(src_key)
1143 transfer_filenames.append((Path(src_fn), Path(target_fn)))
1144 if len(to_warn_about[0]) > 0:
1145 warn(
1146 f"Following source files were not found during FileTree conversion: {to_warn_about[0]}"
1147 )
1148 if len(to_warn_about[1]) > 0:
1149 warn(
1150 f"Following target files already existed during FileTree conversion: {to_warn_about[1]}"
1151 )
1152 for fn1, fn2 in transfer_filenames:
1153 _convert_file(
1154 fn1,
1155 fn2,
1156 to_warn_about,
1157 symlink=symlink,
1158 overwrite=overwrite,
1159 )
1162def _convert_file(
1163 source: Path, target: Path, to_warn_about, symlink=False, overwrite=False
1164):
1165 """
1166 Copy or link `source` file to `target` file.
1168 Helper function for :func:`convert`
1169 """
1170 if not source.exists():
1171 to_warn_about[0].append(str(source))
1172 return
1173 if target.exists():
1174 if not overwrite:
1175 to_warn_about[1].append(str(target))
1176 return
1177 os.remove(target)
1178 target.parent.mkdir(parents=True, exist_ok=True)
1179 if not symlink:
1180 copyfile(source, target, follow_symlinks=False)
1181 elif source.is_absolute():
1182 target.symlink_to(source)
1183 else:
1184 target.symlink_to(os.path.relpath(source, target.parent))