Coverage for src/file_tree/template.py: 87%
675 statements
« prev ^ index » next coverage.py v7.6.9, created at 2024-12-16 14:17 +0000
1"""Define Placeholders and Template interface."""
2import itertools
3import os
4import re
5import string
6from collections import defaultdict
7from collections.abc import MutableMapping
8from functools import cmp_to_key, lru_cache
9from glob import glob
10from itertools import chain, combinations, product
11from pathlib import Path
12from typing import (
13 Any,
14 Collection,
15 Dict,
16 FrozenSet,
17 Generator,
18 Iterable,
19 Iterator,
20 List,
21 Optional,
22 Sequence,
23 Set,
24 Tuple,
25)
27import numpy as np
28import pandas as pd
29import xarray
30from parse import compile, extract_format
33def is_singular(value):
34 """Whether a value is singular or has multiple options."""
35 if isinstance(value, str):
36 return True
37 try:
38 iter(value)
39 return False
40 except TypeError:
41 return True
44class Placeholders(MutableMapping):
45 """Dictionary-like object containing the placeholder values.
47 It understands about sub-trees
48 (i.e., if "<sub_tree>/<placeholder>" does not exist it will return "<placeholder>" instead).
49 """
51 def __init__(self, *args, **kwargs):
52 """Create a new Placeholders as any dictionary."""
53 self.mapping = {}
54 self.linkages: Dict[str : FrozenSet[str]] = {}
55 self.update(dict(*args, **kwargs))
57 def copy(self) -> "Placeholders":
58 """Create copy of placeholder values."""
59 p = Placeholders()
60 p.mapping = dict(self.mapping)
61 p.linkages = dict(self.linkages)
62 return p
64 def __getitem__(self, key: str):
65 """Get placeholder values respecting sub-tree placeholders."""
66 actual_key = self.find_key(key)
67 if actual_key is None:
68 raise KeyError(f"No parameter value available for {key}")
69 if actual_key in self.linkages:
70 return self.mapping[self.linkages[actual_key]][actual_key]
71 return self.mapping[actual_key]
73 def __delitem__(self, key):
74 """Delete placeholder values represented by key."""
75 if isinstance(key, tuple):
76 key = frozenset(key)
77 del self.mapping[key]
78 if isinstance(key, frozenset):
79 for k in key:
80 del self.linkages[k]
82 def __setitem__(self, key, value):
83 """Overwrite placeholder value taking adjusting linked placeholders if needed."""
84 if isinstance(key, tuple): # create linked placeholders
85 if len(key) != len(value):
86 raise ValueError(
87 f"Attempting to set linked placeholders for {key}, "
88 + f"but {value} has a different number of elements than {key}"
89 )
90 if any([len(value[0]) != len(v) for v in value]):
91 raise ValueError(
92 f"Attempting to set linked placeholders for {key}, "
93 + f"but not all elements in {value} have the same length."
94 )
95 value = {k: v for k, v in zip(key, value)}
96 key = frozenset(key)
97 if isinstance(key, frozenset):
98 assert isinstance(value, dict)
99 for k in list(key):
100 if k in self.linkages:
101 unmatched_keys = [
102 unmatched
103 for unmatched in self.linkages[k]
104 if unmatched not in key
105 ]
106 if len(unmatched_keys) > 0:
107 raise ValueError(
108 f"Attempting to set linked placeholders for {key}, "
109 + f"but {k} is already linked to {unmatched_keys}."
110 )
111 self.mapping[key] = value
112 for k in list(key):
113 if k in self.mapping:
114 del self.mapping[k]
115 if k in self.linkages:
116 if self.linkages[k] in self.mapping and self.linkages[k] != key:
117 del self.mapping[self.linkages[k]]
118 del self.linkages[k]
119 self.linkages[k] = key
120 elif key in self.linkages:
121 old_values = self.mapping[self.linkages[key]]
122 if is_singular(value):
123 nvalue = old_values[key].count(value)
124 self.unlink(*old_values.keys())
125 if nvalue == 0:
126 for skey in old_values:
127 del self.mapping[skey]
128 self.mapping[key] = value
129 elif nvalue == 1:
130 idx = old_values[key].index(value)
131 for skey in old_values:
132 self.mapping[skey] = old_values[skey][idx]
133 else:
134 idx = [i for i, v in enumerate(old_values[key]) if v == value]
135 for skey in old_values:
136 if key == skey:
137 self.mapping[key] = value
138 else:
139 self.mapping[skey] = tuple(old_values[skey][i] for i in idx)
140 self.link(*[skey for skey in old_values if skey != key])
141 else:
142 idx = []
143 for new_v in value:
144 nfound = 0
145 for i, old_v in enumerate(old_values[key]):
146 if old_v == new_v and i not in idx:
147 idx.append(i)
148 nfound += 1
149 if nfound == 0:
150 idx.append((new_v, ))
151 for skey in old_values:
152 old_values[skey] = tuple(
153 old_values[skey][i] if isinstance(i, int) else
154 (i[0] if skey == key else None)
155 for i in idx)
156 else:
157 self.mapping[key] = value
159 def __iter__(self):
160 """Iterate over all placeholder keys that actually have values."""
161 for key in self.mapping:
162 if self.mapping[key] is not None:
163 yield key
165 def __len__(self):
166 """Return number of keys in the mapping."""
167 return len([k for k, v in self.mapping.items() if v is not None])
169 def __repr__(self):
170 """Text representation of placeholder values."""
171 return f"Placeholders({self.mapping})"
173 def find_key(self, key: str) -> Optional[str]:
174 """Find the actual key containing the value.
176 Will look for:
178 - not None value for the key itself
179 - not None value for any parent (i.e, for key "A/B", will look for "B" as well)
180 - otherwise will return None
182 Args:
183 key (str): placeholder name
185 Returns:
186 None if no value for the key is available, otherwise the key used to index the value
187 """
188 if not isinstance(key, str):
189 key = frozenset(key)
190 elif key in self.linkages:
191 return key
192 if self.mapping.get(key, None) is not None:
193 return key
194 elif "/" in key:
195 _, *parts = key.split("/")
196 new_key = "/".join(parts)
197 return self.find_key(new_key)
198 else:
199 return None
201 def missing_keys(self, all_keys: Collection[str], top_level=True) -> Set[str]:
202 """Identify any placeholder keys in `all_keys` that are not defined.
204 If `top_level` is True (default), any sub-tree information is removed from the missing keys.
205 """
206 not_defined = {key for key in all_keys if self.find_key(key) is None}
207 if not top_level:
208 return not_defined
209 return {key.split('/')[-1] for key in not_defined}
211 def split(self) -> Tuple["Placeholders", "Placeholders"]:
212 """Split all placeholders into those with a single value or those with multiple values.
214 Placeholders are considered to have multiple values if they are equivalent to 1D-arrays (lists, tuples, 1D ndarray, etc.).
215 Anything else is considered a single value (string, int, float, etc.).
217 Returns:
218 Tuple with two dictionaries:
220 1. placeholders with single values
221 2. placehodlers with multiple values
222 """
223 single_placeholders = Placeholders()
224 multi_placeholders = Placeholders()
225 for name, value in self.mapping.items():
226 if isinstance(name, frozenset) or not is_singular(value):
227 multi_placeholders[name] = value
228 else:
229 single_placeholders[name] = value
230 return single_placeholders, multi_placeholders
232 def iter_over(self, keys) -> Generator["Placeholders", None, None]:
233 """Iterate over the placeholder names.
235 Args:
236 keys (Sequence[str]): sequence of placeholder names to iterate over
238 Raises:
239 KeyError: Raised if any of the provided `keys` does not have any value.
241 Yields:
242 yield Placeholders object, where each of the listed keys only has a single possible value
243 """
244 actual_keys = [self.linkages.get(self.find_key(key), key) for key in keys]
245 unfilled = {orig for orig, key in zip(keys, actual_keys) if key is None}
246 if len(unfilled) > 0:
247 raise KeyError(f"Can not iterate over undefined placeholders: {unfilled}")
249 unique_keys = []
250 iter_values = {}
251 for key in actual_keys:
252 if key not in unique_keys:
253 if isinstance(key, frozenset): # linked placeholder
254 unique_keys.append(key)
255 iter_values[key] = [
256 {k: self[k][idx] for k in key}
257 for idx in range(len(self[list(key)[0]]))
258 ]
259 elif not is_singular(self[key]): # iterable placeholder
260 unique_keys.append(key)
261 iter_values[key] = self[key]
263 for values in product(*[iter_values[k] for k in unique_keys]):
264 new_vars = Placeholders(self)
265 for key, value in zip(unique_keys, values):
266 if isinstance(key, frozenset):
267 del new_vars[key] # break the placeholders link
268 new_vars.update(value)
269 else:
270 new_vars[key] = value
271 yield new_vars
273 def link(self, *keys):
274 """
275 Link the placeholders represented by `keys`.
277 When iterating over linked placeholders the i-th tree
278 will contain the i-th element from all linked placeholders,
279 instead of the tree containing all possible combinations of placeholder values.
281 This can be thought of using `zip` for linked variables and
282 `itertools.product` for unlinked ones.
283 """
284 actual_keys = set()
285 for key in keys:
286 if key in self.linkages:
287 actual_keys.update(self.linkages[key])
288 else:
289 actual_keys.add(key)
290 self[frozenset(actual_keys)] = {key: self[key] for key in actual_keys}
292 def unlink(self, *keys):
293 """
294 Unlink the placeholders represented by `keys`.
296 See :meth:`link` for how linking affects the iteration
297 through placeholders with multiple values.
299 Raises a ValueError if the placeholders are not actually linked.
300 """
301 if keys not in self:
302 raise ValueError(f"{keys} were not linked, so cannot unlink them")
303 new_vars = {k: self[k] for k in keys}
304 del self[keys]
305 self.update(new_vars)
307 def to_string(self, ):
308 lines = []
309 all_keys = sorted([
310 *self.linkages.keys(),
311 *[k for k in self.mapping.keys() if not isinstance(k, frozenset)]
312 ])
313 for key in sorted(all_keys):
314 value = self[key]
315 if value is None:
316 continue
317 if np.array(value).ndim == 1:
318 lines.append(
319 f"{key} = {', '.join([str(v) for v in value])}"
320 )
321 else:
322 lines.append(f"{key} = {value}")
323 for key in self.mapping.keys():
324 if isinstance(key, frozenset):
325 lines.append(f"&LINK {', '.join(sorted(key))}")
326 return "\n".join(lines)
329class MyDataArray:
330 """Wrapper around xarray.DataArray for internal usage.
332 It tries to delay creating the DataArray object as long as possible
333 (as using them for small arrays is slow...).
334 """
336 def __init__(self, data, coords=None):
337 """Create a new DataArray look-a-like."""
338 self.as_xarray = coords is None
339 if self.as_xarray:
340 assert isinstance(data, xarray.DataArray)
341 self.data_array = data
342 else:
343 self.data = data
344 self.coords = coords
346 def map(self, func) -> "MyDataArray":
347 """Apply `func` to each element of array."""
348 if self.as_xarray:
349 return MyDataArray(
350 xarray.apply_ufunc(func, self.data_array, vectorize=True)
351 )
352 else:
353 return MyDataArray(
354 np.array([func(d) for d in self.data.flat]).reshape(self.data.shape),
355 self.coords,
356 )
358 def to_xarray(
359 self,
360 ) -> xarray.DataArray:
361 """Convert to a real xarray.DataArray."""
362 if self.as_xarray:
363 return self.data_array
364 else:
365 return xarray.DataArray(
366 self.data, [_to_index(name, values) for name, values in self.coords]
367 )
369 @staticmethod
370 def concat(parts, new_index) -> "MyDataArray":
371 """Combine multiple DataArrays."""
372 if len(parts) == 0:
373 return MyDataArray(np.array([]), [])
374 to_xarray = any(p.as_xarray for p in parts) or any(
375 len(p.coords) != len(parts[0].coords)
376 or any(
377 np.all(name1 != name2)
378 for (name1, _), (name2, _) in zip(p.coords, parts[0].coords)
379 )
380 for p in parts
381 )
382 if to_xarray:
383 return MyDataArray(
384 xarray.concat([p.to_xarray() for p in parts], _to_index(*new_index))
385 )
386 else:
387 new_data = np.stack([p.data for p in parts], axis=0)
388 new_coords = list(parts[0].coords)
389 new_coords.insert(0, new_index)
390 return MyDataArray(new_data, new_coords)
393def _to_index(name, values):
394 """Convert to index for MyDataArray."""
395 if isinstance(name, str):
396 return pd.Index(values, name=name)
397 else:
398 return ("-".join(sorted(name)), pd.MultiIndex.from_tuples(values, names=name))
401class Template:
402 """Represents a single template in the FileTree."""
404 def __init__(self, parent: Optional["Template"], unique_part: str):
405 """Create a new child template in `parent` directory with `unique_part` filename."""
406 self.parent = parent
407 self.unique_part = unique_part
409 @property
410 def as_path(self) -> Path:
411 """Return the full path with no placeholders filled in."""
412 if self.parent is None:
413 return Path(self.unique_part)
414 return self.parent.as_path.joinpath(self.unique_part)
416 @property
417 def as_string(self):
418 """Return the full path with no placeholders filled in."""
419 if self.parent is None:
420 return str(self.unique_part)
421 return os.path.join(self.parent.as_string, str(self.unique_part))
423 def __str__(self):
424 """Return string representation of template."""
425 return f"Template({self.as_string})"
427 def children(self, templates: Iterable["Template"]) -> List["Template"]:
428 """Find children from a sequence of templates.
430 Args:
431 templates: sequence of possible child templates.
433 Returns:
434 list of children templates
435 """
436 res = []
437 def add_if_child(possible_child):
438 if isinstance(possible_child, DuplicateTemplate):
439 for t in possible_child.templates:
440 add_if_child(t)
441 elif possible_child.parent is self and possible_child not in res:
442 res.append(possible_child)
444 for t in templates:
445 add_if_child(t)
446 return sorted(res, key=lambda t: t.unique_part)
448 def as_multi_line(
449 self, other_templates: Dict["Template", Set[str]], indentation=4
450 ) -> str:
451 """Generate a string describing this and any child templates.
453 Args:
454 other_templates (Dict[Template, Set[str]]):
455 templates including all the child templates and itself.
456 indentation (int, optional):
457 number of spaces to use as indentation. Defaults to 4.
459 Returns:
460 str: multi-line string that can be processed by :meth:`file_tree.FileTree.read`
461 """
462 result = self._as_multi_line_helper(other_templates, indentation)
464 is_top_level = "" in other_templates[self]
465 if not is_top_level and self.parent is None:
466 return "!" + result
467 else:
468 return result
470 def _as_multi_line_helper(
471 self,
472 other_templates: Dict["Template", Set[str]],
473 indentation=4,
474 _current_indentation=0,
475 ) -> str:
476 leaves = []
477 branches = []
478 for t in sorted(
479 self.children(other_templates.keys()), key=lambda t: t.unique_part
480 ):
481 if len(t.children(other_templates.keys())) == 0:
482 leaves.append(t)
483 else:
484 branches.append(t)
486 is_top_level = "" in other_templates[self]
487 if is_top_level:
488 base_line = "."
489 assert _current_indentation == 0 and self.parent is None
490 _current_indentation = -indentation
491 else:
492 base_line = _current_indentation * " " + self.unique_part
494 all_keys = set(other_templates[self])
495 if is_top_level and all_keys == {""}:
496 lines = []
497 elif len(all_keys) == 1 and list(all_keys)[0] == self.guess_key():
498 lines = [base_line]
499 else:
500 if is_top_level:
501 all_keys.remove("")
502 lines = [base_line + f' ({",".join(sorted(all_keys))})']
504 already_done = set()
505 for t in leaves + branches:
506 if t not in already_done:
507 lines.append(
508 t._as_multi_line_helper(
509 other_templates, indentation, indentation + _current_indentation
510 )
511 )
512 already_done.add(t)
513 return "\n".join(lines)
515 @property
516 def _parts(
517 self,
518 ):
519 return TemplateParts.parse(self.as_string)
521 def placeholders(self, valid=None) -> List[str]:
522 """Return a list of the placeholder names.
524 Args:
525 valid: Collection of valid placeholder names.
526 An error is raised if any other placeholder is detected.
527 By default all placeholder names are fine.
529 Returns:
530 List[str]: placeholder names in order that they appear in the template
531 """
532 return self._parts.ordered_placeholders(valid)
534 def format_single(
535 self, placeholders: Placeholders, check=True, keep_optionals=False
536 ) -> str:
537 """Format the template with the placeholders filled in.
539 Only placeholders with a single value are considered.
541 Args:
542 placeholders (Placeholders): values to fill into the placeholder
543 check (bool): skip check for missing placeholders if set to True
544 keep_optionals: if True keep optional parameters that have not been set (will cause the check to fail)
546 Raises:
547 KeyError: if any placeholder is missing
549 Returns:
550 str: filled in template
551 """
552 single_placeholders, _ = placeholders.split()
553 template = self._parts.fill_single_placeholders(single_placeholders)
554 if not keep_optionals:
555 template = template.remove_optionals()
556 if check:
557 unfilled = template.required_placeholders()
558 if len(unfilled) > 0:
559 raise KeyError(f"Missing placeholder values for {unfilled}")
560 return str(template)
562 def format_mult(
563 self, placeholders: Placeholders, check=False, filter=False, matches=None
564 ) -> xarray.DataArray:
565 """Replace placeholders in template with the provided placeholder values.
567 Args:
568 placeholders: mapping from placeholder names to single or multiple vaalues
569 check: skip check for missing placeholders if set to True
570 filter: filter out non-existing files if set to True
571 matches: Optional pre-generated list of any matches to the template.
573 Raises:
574 KeyError: if any placeholder is missing
576 Returns:
577 xarray.DataArray: array with possible resolved paths.
578 If `filter` is set to True the non-existent paths are replaced by None
579 """
580 parts = self._parts
581 resolved = parts.resolve(placeholders)
582 if check:
583 for template in resolved.data.flatten():
584 unfilled = template.required_placeholders()
585 if len(unfilled) > 0:
586 raise KeyError(f"Missing placeholder values for {unfilled}")
587 paths = resolved.map(lambda t: str(t))
588 if not filter:
589 return paths.to_xarray()
590 placeholder_dict = dict(placeholders)
591 path_matches = [
592 str(
593 parts.fill_single_placeholders(
594 Placeholders({**placeholder_dict, **match})
595 ).remove_optionals()
596 )
597 for match in (
598 self.all_matches(placeholders) if matches is None else matches
599 )
600 ]
601 return paths.map(lambda p: p if p in path_matches else "").to_xarray()
603 def optional_placeholders(
604 self,
605 ) -> Set[str]:
606 """Find all placeholders that are only within optional blocks (i.e., they do not require a value).
608 Returns:
609 Set[str]: names of optional placeholders
610 """
611 return self._parts.optional_placeholders()
613 def required_placeholders(
614 self,
615 ) -> Set[str]:
616 """Find all placeholders that are outside of optional blocks (i.e., they do require a value).
618 Returns:
619 Set[str]: names of required placeholders
620 """
621 return self._parts.required_placeholders()
623 def guess_key(
624 self,
625 ) -> str:
626 """Propose a short name for the template.
628 The proposed short name is created by:
630 - taking the basename (i.e., last component) of the path
631 - removing the first '.' and everything beyond (to remove the extension)
633 .. warning::
635 If there are multiple dots within the path's basename,
636 this might remove far more than just the extension.
638 Returns:
639 str: proposed short name for this template (used if user does not provide one)
640 """
641 parts = self.as_path.parts
642 if len(parts) == 0:
643 return ""
644 else:
645 return parts[-1].split(".")[0]
647 def add_precursor(self, text) -> "Template":
648 """Return a new Template with any placeholder names in the unique part now preceded by `text`.
650 Used for adding sub-trees
651 """
652 parts = TemplateParts.parse(self.unique_part).parts
653 updated = "".join([str(p.add_precursor(text)) for p in parts])
654 return Template(self.parent, updated)
656 def get_all_placeholders(
657 self, placeholders: Placeholders, link=None, return_matches=False
658 ) -> Placeholders:
659 """Fill placeholders with possible values based on what is available on disk.
661 Args:
662 placeholders: New values for undefined placeholders in template.
663 link: template keys that should be linked together in the output.
664 return_matches: if True, also returns any matches to the template, which can be passed on to `format_mult`.
666 Returns:
667 Set of placeholders updated based on filed existing on disk that match this template.
668 """
669 if link is None:
670 link = []
671 elif len(link) > 0 and isinstance(link[0], str):
672 link = [link]
673 # link is now a sequence of sequence of strings
675 all_to_link = [name for single in link for name in single]
676 template_keys = {
677 *self.optional_placeholders(),
678 *self.required_placeholders(),
679 }
681 undefined = set()
682 placeholder_with_linked = placeholders.copy()
683 for name in all_to_link:
684 if placeholder_with_linked.find_key(name) is None:
685 placeholder_with_linked[name] = ""
686 undefined.add(name)
687 undefined.update(placeholders.missing_keys(template_keys))
689 matches = self.all_matches(placeholders, undefined)
691 undefined = defaultdict(set)
692 for match in matches:
693 for name, value in match.items():
694 if placeholders.find_key(name) is None and name not in all_to_link:
695 undefined[name].add(value)
697 def cmp(item1, item2):
698 if item1 is None:
699 return -1
700 if item2 is None:
701 return 1
702 if item1 < item2:
703 return -1
704 if item1 > item2:
705 return 1
706 return 0
708 res = Placeholders(
709 {k: sorted(v, key=cmp_to_key(cmp)) for k, v in undefined.items()}
710 )
711 for to_link in link:
712 res[tuple(to_link)] = list(zip(*sorted(
713 {tuple(Placeholders(match).get(key, None) for key in to_link) for match in matches}
714 )))
715 if return_matches:
716 return (res, matches)
717 return res
719 def all_matches(self, placeholders: Placeholders, keys_to_fill: Collection[str]=None) -> List[Dict[str, Any]]:
720 """Return a sequence of all possible variable values for `keys_to_fill` matching existing files on disk.
722 Only variable values matching existing placeholder values (in `placeholders`) are returned
723 (undefined placeholders are unconstrained).
724 """
725 if keys_to_fill is None:
726 keys_to_fill = placeholders.missing_keys({
727 *self.required_placeholders(),
728 *self.optional_placeholders(),
729 })
731 single_vars, multi_vars = placeholders.split()
732 res = []
734 def check_name_with_edit(match, name):
735 value = match[name]
736 if name in single_vars and single_vars.find_key(name) == name:
737 return value == single_vars[name]
738 if name in multi_vars and multi_vars.find_key(name) == name:
739 return value in multi_vars[name]
740 if name in keys_to_fill:
741 return True
742 del match[name]
743 _, *parts = name.split('/')
744 parent_name = '/'.join(parts)
745 if parent_name in match:
746 return match[parent_name] == value
747 match[parent_name] = value
748 return check_name_with_edit(match, parent_name)
750 for match in self._parts.all_matches():
751 if not all(
752 check_name_with_edit(match, name) for name in list(match.keys())
753 ):
754 continue
755 res.append(match)
756 return res
758 def rich_line(self, all_keys):
759 """Produce a line for rendering using rich."""
760 keys = all_keys[self]
761 base = self.guess_key()
762 unique_part = str(self.unique_part)
763 if base in keys:
764 keys.remove(base)
765 unique_part = str.replace(unique_part, base, f"[cyan]{base}[/cyan]")
766 if len(keys) == 0:
767 return unique_part
768 return (
769 unique_part
770 + " ("
771 + ", ".join("[cyan]" + key + "[/cyan]" for key in keys)
772 + ")"
773 )
776class DuplicateTemplate:
777 """Represents the case where a single key points to multiple templates."""
779 def __init__(self, *templates: Template):
780 self._templates = list(templates)
782 def add_template(self, template: Template):
783 """Add another conflicting template."""
784 self._templates.append(template)
786 @property
787 def templates(self, ):
788 return tuple(self._templates)
791def extract_placeholders(template, filename, known_vars=None):
792 """
793 Extract the placeholder values from the filename.
795 :param template: template matching the given filename
796 :param filename: filename
797 :param known_vars: already known placeholders
798 :return: dictionary from placeholder names to string representations
799 (unused placeholders set to None)
800 """
801 return TemplateParts.parse(template).extract_placeholders(filename, known_vars)
804class Part:
805 """
806 Individual part of a template.
808 3 subclasses are defined:
810 - :class:`Literal`:
811 piece of text
812 - :class:`Required`:
813 required placeholder to fill in
814 (between curly brackets)
815 - :class:`OptionalPart`:
816 part of text containing optional placeholders
817 (between square brackets)
818 """
820 def fill_single_placeholders(
821 self, placeholders: Placeholders, ignore_type=False
822 ) -> Sequence["Part"]:
823 """Fill in the given placeholders."""
824 return (self,)
826 def optional_placeholders(
827 self,
828 ) -> Set[str]:
829 """Return all placeholders in optional parts."""
830 return set()
832 def required_placeholders(
833 self,
834 ) -> Set[str]:
835 """Return all required placeholders."""
836 return set()
838 def contains_optionals(self, placeholders: Set["Part"] = None):
839 """Return True if this part contains the optional placeholders."""
840 return False
842 def append_placeholders(self, placeholders: List[str], valid=None):
843 """Append the placeholders in this part to the provided list in order."""
844 pass
846 def add_precursor(self, text: str) -> "Part":
847 """Prepend any placeholder names by `text`."""
848 return self
850 def for_defined(self, placeholder_names: Set[str]) -> List["Part"]:
851 """Return the template string assuming the placeholders in `placeholder_names` are defined.
853 Removes any optional parts, whose placeholders are not in `placeholder_names`.
854 """
855 return [self]
857 def remove_precursors(self, placeholders=None):
858 """Remove precursor from placeholder key."""
859 return self
862class Literal(Part):
863 """Piece of text in template without placeholders."""
865 def __init__(self, text: str):
866 """
867 Literal part is defined purely by the text it contains.
869 :param text: part of the template
870 """
871 self.text = text
873 def __str__(self):
874 """Return this part of the template as a string."""
875 return self.text
877 def __eq__(self, other):
878 """Check if text matches other `Literal`."""
879 if not isinstance(other, Literal):
880 return NotImplemented
881 return self.text == other.text
884class Required(Part):
885 """Placeholder part of template that requires a value."""
887 def __init__(self, var_name, var_formatting=None):
888 """
889 Create required part of template (between curly brackets).
891 Required placeholder part of template is defined by placeholder name and its format
893 :param var_name: name of placeholder
894 :param var_formatting: how to format the placeholder
895 """
896 self.var_name = var_name
897 self.var_formatting = var_formatting
899 def __str__(self):
900 """Return this part of the template as a string."""
901 if self.var_formatting is None or len(self.var_formatting) == 0:
902 return "{" + self.var_name + "}"
903 else:
904 return "{" + self.var_name + ":" + self.var_formatting + "}"
906 def fill_single_placeholders(self, placeholders: Placeholders, ignore_type=False):
907 """Fill placeholder values into template obeying typing."""
908 value = placeholders.get(self.var_name, None)
909 if value is None:
910 return (self,)
911 else:
912 if not ignore_type and len(self.var_formatting) > 0:
913 format_type = extract_format(self.var_formatting, [])["type"]
914 if format_type in list(r"dnbox"):
915 value = int(value)
916 elif format_type in list(r"f%eg"):
917 value = float(value)
918 elif format_type in ["t" + ft for ft in "iegachs"] and isinstance(
919 value, str
920 ):
921 from dateutil import parser
923 value = parser(value)
924 res = TemplateParts.parse(
925 format(value, "" if ignore_type else self.var_formatting)
926 )
927 if len(res.parts) == 1:
928 return res.parts
929 return res.fill_single_placeholders(
930 placeholders, ignore_type=ignore_type
931 ).parts
933 def required_placeholders(
934 self,
935 ):
936 """Return variable names."""
937 return {self.var_name}
939 def append_placeholders(self, placeholders, valid=None):
940 """Add placeholder name to list of placeholders in template."""
941 if valid is not None and self.var_name not in valid:
942 raise ValueError(f"Placeholder {self.var_name} is not defined")
943 placeholders.append(self.var_name)
945 def add_precursor(self, text: str) -> "Required":
946 """Prepend any placeholder names by `text`."""
947 return Required(text + self.var_name, self.var_formatting)
949 def remove_precursors(self, placeholders=None):
950 """Remove precursor from placeholder key."""
951 if placeholders is None:
952 new_name = self.var_name.split("/")[-1]
953 else:
954 key = placeholders.find_key(self.var_name)
955 new_name = self.var_name if key is None else key
956 return Required(new_name, self.var_formatting)
958 def __eq__(self, other):
959 """Check whether `other` placeholder matches this one."""
960 if not isinstance(other, Required):
961 return NotImplemented
962 return (self.var_name == other.var_name) & (
963 self.var_formatting == other.var_formatting
964 )
967class OptionalPart(Part):
968 """Optional part of a template (i.e., between square brackets)."""
970 def __init__(self, sub_template: "TemplateParts"):
971 """
972 Create optional part of template (between square brackets).
974 Optional part can contain literal and required parts
976 :param sub_template: part of the template within square brackets
977 """
978 self.sub_template = sub_template
980 def __str__(self):
981 """Return string representation of optional part."""
982 return "[" + str(self.sub_template) + "]"
984 def fill_single_placeholders(self, placeholders: Placeholders, ignore_type=False):
985 """Fill placeholders into text within optional part."""
986 new_opt = self.sub_template.fill_single_placeholders(
987 placeholders, ignore_type=ignore_type
988 )
989 if len(new_opt.required_placeholders()) == 0:
990 return (Literal(str(new_opt)),)
991 return (OptionalPart(new_opt),)
993 def optional_placeholders(self):
994 """Return sequence of any placeholders in the optional part of the template."""
995 return self.sub_template.required_placeholders()
997 def contains_optionals(self, placeholders=None):
998 """Check if this optional part contains any placeholders not listed in `placeholders`."""
999 if placeholders is None and len(self.optional_placeholders()) > 0:
1000 return True
1001 return len(self.optional_placeholders().intersection(placeholders)) > 0
1003 def append_placeholders(self, placeholders, valid=None):
1004 """Add any placeholders in the optional part to `placeholders` list."""
1005 try:
1006 placeholders.extend(self.sub_template.ordered_placeholders(valid=valid))
1007 except ValueError:
1008 pass
1010 def add_precursor(self, text: str) -> "OptionalPart":
1011 """Prepend precursor `text` to any placeholders in the optional part."""
1012 return OptionalPart(
1013 TemplateParts([p.add_precursor(text) for p in self.sub_template.parts])
1014 )
1016 def for_defined(self, placeholder_names: Set[str]) -> List["Part"]:
1017 """
1018 Return the template string assuming the placeholders in `placeholder_names` are defined.
1020 Removes any optional parts, whose placeholders are not in `placeholder_names`.
1021 """
1022 if len(self.optional_placeholders().difference(placeholder_names)) > 0:
1023 return []
1024 return list(self.sub_template.parts)
1026 def remove_precursors(self, placeholders=None):
1027 """Remove precursor from placeholder key."""
1028 return OptionalPart(self.sub_template.remove_precursors(placeholders))
1030 def __eq__(self, other):
1031 """Check whether two optional parts match."""
1032 if not isinstance(other, OptionalPart):
1033 return NotImplemented
1034 return self.sub_template == other.sub_template
1037class TemplateParts:
1038 """Representation of full template as sequence of `Part` objects."""
1040 optional_re = re.compile(r"(\[.*?\])")
1041 requires_re = re.compile(r"(\{.*?\})")
1043 def __init__(self, parts: Sequence[Part]):
1044 """Create new TemplateParts based on sequence."""
1045 if isinstance(parts, str):
1046 raise ValueError(
1047 "Input to Template should be a sequence of parts; "
1048 + "did you mean to call `TemplateParts.parse` instead?"
1049 )
1050 self.parts = tuple(parts)
1052 @staticmethod
1053 @lru_cache(1000)
1054 def parse(text: str) -> "TemplateParts":
1055 """Parse a template string into its constituent parts.
1057 Args:
1058 text: template as string.
1060 Raises:
1061 ValueError: raised if a parsing error is
1063 Returns:
1064 TemplateParts: object that contains the parts of the template
1065 """
1066 parts: List[Part] = []
1067 for optional_parts in TemplateParts.optional_re.split(text):
1068 if (
1069 len(optional_parts) > 0
1070 and optional_parts[0] == "["
1071 and optional_parts[-1] == "]"
1072 ):
1073 if "[" in optional_parts[1:-1] or "]" in optional_parts[1:-1]:
1074 raise ValueError(
1075 f"Can not parse {text}, because unmatching square brackets were found"
1076 )
1077 parts.append(OptionalPart(TemplateParts.parse(optional_parts[1:-1])))
1078 else:
1079 for required_parts in TemplateParts.requires_re.split(optional_parts):
1080 if (
1081 len(required_parts) > 0
1082 and required_parts[0] == "{"
1083 and required_parts[-1] == "}"
1084 ):
1085 if ":" in required_parts:
1086 var_name, var_type = required_parts[1:-1].split(":")
1087 else:
1088 var_name, var_type = required_parts[1:-1], ""
1089 parts.append(Required(var_name, var_type))
1090 else:
1091 parts.append(Literal(required_parts))
1092 return TemplateParts(parts)
1094 def __str__(self):
1095 """Return the template as a string."""
1096 return os.path.normpath("".join([str(p) for p in self.parts]))
1098 def optional_placeholders(
1099 self,
1100 ) -> Set[str]:
1101 """Set of optional placeholders."""
1102 if len(self.parts) == 0:
1103 return set()
1104 optionals = set.union(*[p.optional_placeholders() for p in self.parts])
1105 return optionals.difference(self.required_placeholders())
1107 def required_placeholders(
1108 self,
1109 ) -> Set[str]:
1110 """Set of required placeholders."""
1111 if len(self.parts) == 0:
1112 return set()
1113 return set.union(*[p.required_placeholders() for p in self.parts])
1115 def ordered_placeholders(self, valid=None) -> List[str]:
1116 """Sequence of all placeholders in order (can contain duplicates)."""
1117 ordered_vars: List[str] = []
1118 for p in self.parts:
1119 p.append_placeholders(ordered_vars, valid=valid)
1120 return ordered_vars
1122 def fill_known(self, placeholders: Placeholders, ignore_type=False) -> MyDataArray:
1123 """Fill in the known placeholders.
1125 Any optional parts, where all placeholders have been filled
1126 will be automatically replaced.
1127 """
1128 single, multi = placeholders.split()
1129 return self.remove_precursors(placeholders)._fill_known_helper(
1130 single, multi, ignore_type=ignore_type
1131 )
1133 def _fill_known_helper(
1134 self, single: Placeholders, multi: Placeholders, ignore_type=False
1135 ) -> MyDataArray:
1136 """Do work for `fill_known`."""
1137 new_template = self.fill_single_placeholders(single, ignore_type=ignore_type)
1138 for name in new_template.ordered_placeholders():
1139 use_name = multi.find_key(name)
1140 if use_name is None:
1141 continue
1142 new_multi = multi.copy()
1143 if use_name in multi.linkages:
1144 values = multi[multi.linkages[use_name]]
1145 keys = tuple(sorted(values.keys()))
1146 index = (keys, zip(*[values[k] for k in keys]))
1147 del new_multi[new_multi.linkages[use_name]]
1148 else:
1149 values = {use_name: list(multi[name])}
1150 index = (use_name, values[use_name])
1151 del new_multi[use_name]
1152 assert use_name is not None
1154 parts = []
1155 new_single = single.copy()
1156 for idx in range(len(values[use_name])):
1157 new_vals = {n: v[idx] for n, v in values.items()}
1158 new_single.mapping.update(new_vals)
1159 parts.append(
1160 new_template._fill_known_helper(
1161 new_single, new_multi, ignore_type=ignore_type
1162 )
1163 )
1165 return MyDataArray.concat(parts, index)
1166 return MyDataArray(np.array(new_template), [])
1168 def fill_single_placeholders(
1169 self, placeholders: Placeholders, ignore_type=False
1170 ) -> "TemplateParts":
1171 """
1172 Fill in placeholders with singular values.
1174 Assumes that all placeholders are in fact singular.
1175 """
1176 res = [
1177 p.fill_single_placeholders(placeholders, ignore_type=ignore_type)
1178 for p in self.parts
1179 ]
1180 return TemplateParts(list(chain(*res)))
1182 def remove_optionals(self, optionals=None) -> "TemplateParts":
1183 """
1184 Remove any optionals containing the provided placeholders.
1186 By default all optionals are removed.
1187 """
1188 return TemplateParts(
1189 [p for p in self.parts if not p.contains_optionals(optionals)]
1190 )
1192 def all_matches(
1193 self,
1194 ) -> List[Dict[str, Any]]:
1195 """Find all potential matches to existing templates.
1197 Returns a list with the possible combination of values for the placeholders.
1198 """
1199 required = self.required_placeholders()
1200 optional = self.optional_placeholders()
1201 matches = []
1202 already_globbed = {}
1203 for defined_optionals in [
1204 c for n in range(len(optional) + 1) for c in combinations(optional, n)
1205 ]:
1206 glob_placeholders = Placeholders(
1207 **{req: "*" for req in required},
1208 **{opt: "*" for opt in defined_optionals},
1209 )
1210 new_glob = str(
1211 self.fill_single_placeholders(
1212 glob_placeholders, ignore_type=True
1213 ).remove_optionals()
1214 )
1215 while "**" in new_glob:
1216 new_glob = new_glob.replace("**", "*")
1217 if new_glob not in already_globbed:
1218 already_globbed[new_glob] = glob(new_glob)
1219 res = []
1220 vars = required.union(defined_optionals)
1221 for p in self.parts:
1222 res.extend(p.for_defined(vars))
1223 parser = TemplateParts(res).get_parser()
1224 for fn in already_globbed[new_glob]:
1225 try:
1226 placeholders = parser(fn)
1227 except ValueError:
1228 continue
1229 for var_name in optional:
1230 if var_name not in placeholders:
1231 placeholders[var_name] = None
1232 matches.append(placeholders)
1233 return matches
1235 def resolve(self, placeholders, ignore_type=False) -> MyDataArray:
1236 """
1237 Resolve the template given a set of placeholders.
1239 :param placeholders: mapping of placeholder names to values
1240 :param ignore_type: if True, ignore the type formatting when
1241 filling in placeholders
1242 :return: cleaned string
1243 """
1244 return self.fill_known(placeholders, ignore_type=ignore_type).map(
1245 lambda t: t.remove_optionals()
1246 )
1248 def optional_subsets(
1249 self,
1250 ) -> Iterator["TemplateParts"]:
1251 """Yield template sub-sets with every combination optional placeholders."""
1252 optionals = self.optional_placeholders()
1253 for n_optional in range(len(optionals) + 1):
1254 for exclude_optional in itertools.combinations(optionals, n_optional):
1255 yield self.remove_optionals(exclude_optional)
1257 def extract_placeholders(self, filename, known_vars=None):
1258 """
1259 Extract the placeholder values from the filename.
1261 :param filename: filename
1262 :param known_vars: already known placeholders
1263 :return: dictionary from placeholder names to string representations
1264 (unused placeholders set to None)
1265 """
1266 if known_vars is not None:
1267 template = self.fill_known(known_vars)
1268 else:
1269 template = self
1270 while "//" in filename:
1271 filename = filename.replace("//", "/")
1273 required = template.required_placeholders()
1274 optional = template.optional_placeholders()
1275 results = []
1276 for to_fill in template.optional_subsets():
1277 sub_re = str(
1278 to_fill.fill_known(
1279 {var: r"(\S+)" for var in required.union(optional)},
1280 )
1281 )
1282 while "//" in sub_re:
1283 sub_re = sub_re.replace("//", "/")
1284 sub_re = sub_re.replace(".", r"\.")
1285 match = re.match(sub_re, filename)
1286 if match is None:
1287 continue
1289 extracted_value = {}
1290 ordered_vars = to_fill.ordered_placeholders()
1291 assert len(ordered_vars) == len(match.groups())
1293 failed = False
1294 for var, value in zip(ordered_vars, match.groups()):
1295 if var in extracted_value:
1296 if value != extracted_value[var]:
1297 failed = True
1298 break
1299 else:
1300 extracted_value[var] = value
1301 if failed or any("/" in value for value in extracted_value.values()):
1302 continue
1303 for name in template.optional_placeholders():
1304 if name not in extracted_value:
1305 extracted_value[name] = None
1306 if known_vars is not None:
1307 extracted_value.update(known_vars)
1308 results.append(extracted_value)
1309 if len(results) == 0:
1310 raise ValueError("{} did not match {}".format(filename, template))
1312 def score(placeholders):
1313 """
1314 Assign score to possible reconstructions of the placeholder values.
1316 The highest score is given to the set of placeholders that:
1318 1. has used the largest amount of optional placeholders
1319 2. has the shortest text within the placeholders (only used if equal at 1
1320 """
1321 number_used = len([v for v in placeholders.values() if v is not None])
1322 length_hint = sum([len(v) for v in placeholders.values() if v is not None])
1323 return number_used * 1000 - length_hint
1325 best = max(results, key=score)
1326 for var in results:
1327 if best != var and score(best) == score(var):
1328 raise KeyError(
1329 "Multiple equivalent ways found to parse {} using {}".format(
1330 filename, template
1331 )
1332 )
1333 return best
1335 def get_parser(self):
1336 """Create function that will parse a filename based on this template."""
1337 if any(isinstance(p, OptionalPart) for p in self.parts):
1338 raise ValueError(
1339 "Can not parse filename when there are optional parts in the template"
1340 )
1341 mapping = {
1342 old_key: "".join(new_key)
1343 for old_key, new_key in zip(
1344 self.required_placeholders(),
1345 itertools.product(*[string.ascii_letters] * 3),
1346 )
1347 }
1348 reverse = {new_key: old_key for old_key, new_key in mapping.items()}
1349 cleaned = TemplateParts(
1350 [
1351 Required(mapping[p.var_name], p.var_formatting)
1352 if isinstance(p, Required)
1353 else p
1354 for p in self.parts
1355 ]
1356 )
1357 parser = compile(str(cleaned), case_sensitive=True)
1359 def parse_filename(filename):
1360 """Parse filename based on template."""
1361 result = parser.parse(filename)
1362 if result is None:
1363 raise ValueError(
1364 f"template string ({str(self)}) does not mach filename ({filename})"
1365 )
1366 named = result.named
1367 if any(isinstance(value, str) and "/" in value for value in named.values()):
1368 raise ValueError("Placeholder can not span directories")
1369 return {reverse[key]: value for key, value in named.items()}
1371 return parse_filename
1373 def remove_precursors(self, placeholders=None):
1374 """Replace keys to those existing in the placeholders.
1376 If no placeholders provided all precursors are removed.
1377 """
1378 return TemplateParts([p.remove_precursors(placeholders) for p in self.parts])
1380 def __eq__(self, other):
1381 """Check whether other template matches this one."""
1382 if not isinstance(other, TemplateParts):
1383 return NotImplemented
1384 return (len(self.parts) == len(other.parts)) and all(
1385 p1 == p2 for p1, p2 in zip(self.parts, other.parts)
1386 )