Repository navigation
Expand file tree
/
Copy pathmin.py
More file actions
65 lines (55 loc) · 2 KB
/
Copy pathmin.py
File metadata and controls
65 lines (55 loc) · 2 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
from __future__ import annotations
from typing import Iterable
from refinery.lib.meta import metavars
from refinery.lib.types import Param
from refinery.units import Arg, Chunk, Unit
class min(Unit):
"""
Picks the minimum of all chunks in the current `refinery.lib.frame` with respect to its content
or a meta variable.
"""
def __init__(
self,
key: Param[str | None, Arg.String('key',
help='A meta variable expression to sort by instead of sorting the content.')] = None,
):
super().__init__(key=key)
def filter(self, chunks: Iterable[Chunk]):
def get_value(chunk: Chunk):
if key is None:
return chunk
return metavars(chunk).get(key)
key = self.args.key
it = iter(chunks)
for min_chunk in it:
if not min_chunk.visible:
yield min_chunk
else:
min_index = 0
min_value = get_value(min_chunk)
break
else:
return
for index, chunk in enumerate(chunks, 1):
if not chunk.visible:
yield chunk
continue
value = get_value(chunk)
try:
is_min = value < min_value
except TypeError:
if min_value is None:
self.log_info(
F'Discarding chunk {min_index} in favor of {index} because {key} was not '
F'set on the former; new minimum is {value!r}.')
is_min = True
else:
self.log_info(
F'Discarding chunk {index} because {key} had value {value!r}; it could not '
F'be compared to the current minimum {min_value!r} on chunk {min_index}.')
is_min = False
if is_min:
min_value = value
min_chunk = chunk
min_index = index
yield min_chunk