Repository navigation
Expand file tree
/
Copy pathpop.py
More file actions
193 lines (166 loc) · 6.49 KB
/
Copy pathpop.py
File metadata and controls
193 lines (166 loc) · 6.49 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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
from __future__ import annotations
from itertools import chain
from typing import Iterable, Iterator
from refinery.lib.argformats import DelayedNumSeqArgument
from refinery.lib.meta import check_variable_name
from refinery.lib.types import Param
from refinery.units import Arg, Chunk, Unit
_MERGE_META = '@'
_CONVERSION = ':'
_CHERRYPICK = '='
class _popcount:
def __init__(self, name: str):
self.conversion = None
self.cherrypick = None
self.current = 0
if name == _MERGE_META:
self.count = 1
self.field = ...
return
try:
if isinstance(name, int):
count = name
else:
count = int(name, 0)
except Exception:
name, colon, conversion = name.partition(_CONVERSION)
name, equal, cherrypick = name.partition(_CHERRYPICK)
if equal:
self.cherrypick = cherrypick
if colon:
self.conversion = conversion
self.count = 1
self.field = check_variable_name(name)
else:
self.count = count
self.field = None
if self.count < 1:
raise ValueError(F'Popcounts must be positive integer numbers, {self.count} is invalid.')
@property
def done(self):
return self.current < 1
def reset(self):
self.current = self.count
return self
def into(self, meta: dict, chunk: Chunk):
if self.done:
return False
if self.field:
if self.field is ...:
meta.update(chunk.meta.current)
else:
chunk = chunk.meta[c] if (c := self.cherrypick) else chunk
if c := self.conversion:
chunk = DelayedNumSeqArgument(c, seed=chunk, reverse=True, typecheck=False)(chunk)
meta[self.field] = chunk
self.current -= 1
return True
class pop(Unit):
"""
Consume one or more visible chunks in the frame and store them as meta variables. Trailing
invisible chunks are made visible.
In processing order, remove visible chunks from the current frame and store their contents
in the given meta variables on all chunks that remain. The first invisible chunk in the input
stream is consequently made visible again. If pop is used at the end of a frame, variables are
made local to the parent frame. A pop instruction has the following format:
count | {_MERGE_META} | name[{_CHERRYPICK}source][{_CONVERSION}conversion]
If the instruction is an integer, it is interpreted as `count`, specifying a number of chunks
to be skipped from the frame without storing them. The letter "{_MERGE_META}" can be used to
remove a single chunk from the input and merge all of its meta data into the ones that follow.
Otherwise, the pop instruction consists of the name of the variable to be created, an optional
source variable name, and an optional conversion sequence. If no source variable is specified,
the chunk contents are used as the source. The conversion is a sequence of multibin handlers
that are applied to the source data from right to left before storing it.
For example, the argument `k:le:b64` first decodes the chunk data using base64, then converts
it to an integer in little endian format, and store the integer result in the variable `k`. The
visual aid is that the content is passed from right to left through all conversions, into the
variable `k`. Similarly, the argument k=size will store the current chunk's size in `k`.
"""
FilterEverything = True
def __init__(
self,
*names: Param[str, Arg.String(metavar='instruction', help='A sequence of instructions, see above.')]
):
if not names:
names = _MERGE_META,
super().__init__(names=[_popcount(n) for n in names])
self._tos = None
self._eof = True
def process(self, data):
return data
def finish(self) -> Iterable[Chunk]:
eof = self._eof
self._tos = None
self._eof = True
if not eof:
msg = 'Not all variables could be assigned.'
if not self.leniency:
raise ValueError(F'{msg} Increase leniency to downgrade this failure to a warning.')
self.log_warn(msg)
yield from ()
def filter(self, chunks: Iterable[Chunk]):
variables = {}
remaining: Iterator[_popcount] = iter(self.args.names)
pop = next(remaining).reset()
tos = self._tos
all_invisible = True
all_variables_assigned = False
path = None
view = None
it = iter(chunks)
for chunk in it:
if (path is None):
path = tuple(chunk.path)
if not chunk.visible:
self.log_debug('buffering invisible chunk')
if tos is not None:
yield tos
tos = chunk
continue
else:
all_invisible = False
if (view is None):
view = tuple(chunk.view)
try:
while not pop.into(variables, chunk):
pop = next(remaining).reset()
except StopIteration:
all_variables_assigned = True
if tos is not None:
yield tos
tos = chunk
break
if not all_variables_assigned and pop.done:
try:
next(remaining)
except StopIteration:
all_variables_assigned = True
if not all_variables_assigned:
if all_invisible and path and not any(path):
self._tos = tos
self._eof = False
return
else:
self._eof = True
nesting = self.args.nesting
if tos is not None:
if path and view and tos.path != path:
tos = tos.copy()
tos.path[:] = path
tos.view[:] = view
it = chain([tos], it)
for chunk in it:
meta = chunk.meta
meta.update(variables)
if nesting < 0:
for name in variables:
meta.set_scope(name, chunk.scope + nesting)
chunk.visible = True
yield chunk
if _d := pop.__doc__:
pop.__doc__ = _d = _d.format(
_MERGE_META=_MERGE_META,
_CONVERSION=_CONVERSION,
_CHERRYPICK=_CHERRYPICK,
)
__pdoc__ = dict(pop=_d)