Completed
Pull Request — master (#42)
by TJ
10:39
created

ElementCallable.make()   B

Complexity

Conditions 6

Size

Total Lines 17

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 0
CRAP Score 42

Importance

Changes 2
Bugs 0 Features 0
Metric Value
cc 6
c 2
b 0
f 0
dl 0
loc 17
ccs 0
cts 0
cp 0
crap 42
rs 8
1
"""
2
Element callable.
3
4
Call a function to validate data.
5
6
TODO: Update the format here
7
8
.. code-block:: python
9
10
    from binascii import crc32
11
12
    ExampleMessage = Message('VarTest', [('x', 'B'), ('y', 'B')])
13
14
    CRCedMessage = Message('CRCedMessage', [
15
       ('data', ExampleMessage),                    # A data field that has the example message in it
16
       ('crc', 'I', crc32, ['data']),               # Crc the data, and give an error if we have something unexpected
17
       ('crc', 'I', crc32, ['data'], False),        # Crc the data, but don't give an error
18
   ])
19
20
Following creating this message, you have two options:
21
22
1. Specify a value. The function will be used to validate the value.
23
24
.. code-block:: python
25
26
    def adder(x, y):
27
        return x + y
28
29
    AdderMessage = Message('AdderMessage', [
30
        ('item_a', 'H'),
31
        ('item_b', 'B'),
32
        ('function_data', 'I', adder, ['item_a', 'item_b']),
33
    ])
34
35
    test_data = {
36
        'item_a': 2,
37
        'item_b': 5,
38
        'function_data': 7,
39
    }
40
41
    made = AdderMessage.make(test_data)
42
    assert made.item_a == 2
43
    assert made.item_b == 5
44
    assert made.function_data == 7
45
46
    # If you specify the wrong value, you'll get a ValueError
47
    test_data = {
48
        'item_a': 2,
49
        'item_b': 5,
50
        'function_data': 33,
51
    }
52
53
    try:
54
        made = AdderMessage.make(test_data)
55
    except ValueError:
56
        print('Told you so')
57
58
    # Unless you specify `False` in your original item, then
59
    # nobody will care.
60
61
2. Use the function to generate a value.
62
63
.. code-block:: python
64
65
    def adder(x, y):
66
        return x + y
67
68
    AdderMessage = Message('AdderMessage', [
69
        ('item_a', 'H'),
70
        ('item_b', 'B'),
71
        ('function_data', 'I', adder, ['item_a', 'item_b']),
72
    ])
73
74
    test_data = {
75
        'item_a': 2,
76
        'item_b': 5,
77
    }
78
79
    made = AdderMessage.make(test_data)
80
    assert made.item_a == 2
81
    assert made.item_b == 5
82
    assert made.function_data == 7
83
84 1
"""
85 1
86
import copy
87 1
import struct
88
89 1
from typing import Optional
0 ignored issues
show
Configuration introduced by
The import typing could not be resolved.

This can be caused by one of the following:

1. Missing Dependencies

This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands.

# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3
Tip: We are currently not using virtualenv to run pylint, when installing your modules make sure to use the command for the correct version.

2. Missing __init__.py files

This error could also result from missing __init__.py files in your module folders. Make sure that you place one file in each sub-folder.

Loading history...
90 1
91
from starstruct.element import register, Element
92
from starstruct.modes import Mode
93 1
94 1
95
@register
96
class ElementCallable(Element):
97
    """
98
    Initialize a StarStruct element object.
99
100
    :param field: The fields passed into the constructor of the element
101
    :param mode: The mode in which to pack the bytes
102
    :param alignment: Number of bytes to align to
103
    """
104 1
    # pylint: disable=too-many-instance-attributes
105
    def __init__(self, field: list, mode: Optional[Mode]=Mode.Native, alignment: Optional[int]=1):
106
        # All of the type checks have already been performed by the class
107 1
        # factory
108
        self.name = field[0]
109
110 1
        # Callable elements use the normal struct packing method
111 1
        self.format = field[1]
112 1
113
        if isinstance(field[2], dict):
114 1
            self.ref = field[2]
115 1
116
            self._make_func = self.ref['make'][0]
117 1
            self._make_args = self.ref['make'][1:]
118
119 1
            self._pack_func = self.ref['pack'][0]
120
            self._pack_args = self.ref['pack'][1:]
121 1
122
            self._unpack_func = self.ref['unpack'][0]
123 1
            self._unpack_args = self.ref['unpack'][1:]
124
        elif isinstance(field[2], set):
125 1
            instruction = field[2].copy().pop()
126
            self.ref = {'all': instruction}
127 1
128
            self._make_func = self._pack_func = self._unpack_func = instruction[0]
129 1
            self._make_args = self._pack_args = self._unpack_args = instruction[1:]
130 1
131
        if len(field) == 4:
132
            self._error_on_bad_result = field[3]
133
        else:
134
            self._error_on_bad_result = True
135
136 1
        self._elements = []
137
138
        self.update(mode, alignment)
139
140
    @property
141
    def _struct(self):
142 1
        return struct.Struct(self._mode.value + self.format)
143
144
    @staticmethod
145
    def valid(field: tuple) -> bool:
146
        """
147
        See :py:func:`starstruct.element.Element.valid`
148
149 1
        :param field: The items to determine the structure of the element
150
        """
151
        required_keys = {'pack', 'unpack', 'make'}
152
153
        if len(field) >= 3 and isinstance(field[0], str) and isinstance(field[1], str):
154 1
            if isinstance(field[2], dict):
155
                if set(field[2].keys()) == required_keys and \
156 1
                        all(isinstance(val, tuple) for val in field[2].values()):
157 1
                    return True
158
            elif isinstance(field[2], set):
159 1
                return len(field[2]) == 1 and \
160 1
                    all(isinstance(val, tuple) for val in field[2])
161
162 1
        return False
163
164 1
165
    def validate(self, msg):
166 1
        """
167
        Ensure that the supplied message contains the required information for
168 1
        this element object to operate.
169 1
170
        All elements that are Variable must reference valid Length elements.
171
        """
172 1
        for action in self.ref.values():
173
            for arg in action[1:]:
174
                if isinstance(arg, str):
175 1
                    pass
176
                elif hasattr(arg, 'decode'):
177
                    arg = arg.decode('utf-8')
178
                elif hasattr(arg, 'to_bytes'):
179 1
                    arg = arg.to_bytes((arg.bit_length() + 7) // 8, self._mode.to_byteorder()).decode('utf-8')
180 1
181 1
                if arg not in msg:
182
                    raise ValueError('Need all keys to be in the message, {0} was not found\nAction: {1} -> {2}'.format(arg, action, action[1]))
183
184 1
    def update(self, mode=None, alignment=None):
185 1
        """change the mode of the struct format"""
186
        if mode:
187
            self._mode = mode
188
189
        if alignment:
190 1
            self._alignment = alignment
191
192 1
    def pack(self, msg):
193
        """Pack the provided values into the supplied buffer."""
194
        pack_values = self.call_func(msg, self._pack_func, self._pack_args)
195
196
        # Test if the object is iterable
197 1
        # If it isn't, then turn it into a list
198
        try:
199
            _ = (p for p in pack_values)
200 1
        except TypeError:
201
            pack_values = [pack_values]
202 1
203
        # Unpack the items for struct to allow for mutli-value
204 1
        # items to be passed in.
205 1
        return self._struct.pack(*pack_values)
0 ignored issues
show
Coding Style introduced by
Usage of * or ** arguments should usually be done with care.

Generally, there is nothing wrong with usage of * or ** arguments. For readability of the code base, we suggest to not over-use these language constructs though.

For more information, we can recommend this blog post from Ned Batchelder including its comments which also touches this aspect.

Loading history...
206 1
207 1
    def unpack(self, msg, buf):
208 1
        """Unpack data from the supplied buffer using the initialized format."""
209 1
        ret = self._struct.unpack_from(buf)
210 1
        if isinstance(ret, (list, tuple)) and len(ret) == 1:
211
            # We only change it not to a list if we expected one value.
212
            # Otherwise, we keep it as a list, because that's what we would
213
            # expect (like for a 16I type of struct
214 1
            ret = ret[0]
215
216
        # Only check for errors if they haven't told us not to
217
        if self._error_on_bad_result:
218 1
            # Pretend we're getting a dictionary to make our item,
219
            # but it has no reference to `self`. This is so we check
220 1
            # for errors correctly.
221 1
            temp_dict = copy.deepcopy(msg._asdict())
222 1
            temp_dict.pop(self.name)
223
            expected_value = self.call_func(msg, self._unpack_func, self._unpack_args)
224 1
225
            # Check for an error
226
            if expected_value != ret:
227
                raise ValueError('Expected value was: {0}, but got: {1}'.format(
228
                    expected_value,
229
                    ret,
230
                ))
231
232
        return (ret, buf[self._struct.size:])
233
234
    def make(self, msg):
235
        """Return the expected "made" value"""
236
        # If we aren't going to error on a bad result
237
        # and our name is in the message, just send the value
238
        # No need to do extra work.
239
        if not self._error_on_bad_result \
240
                and self.name in msg \
241
                and msg[self.name] is not None:
242
            return msg[self.name]
243
244
        ret = self.call_func(msg, self._make_func, self._make_args)
245
246
        if self.name in msg:
247
            if ret != msg[self.name]:
248
                raise ValueError('Excepted value: {0}, but got: {1}'.format(ret, msg[self.name]))
249
250
        return ret
251
252
    def call_func(self, msg, func, args):
253
        items = self.prepare_args(msg, args)
254
        return func(*items)
0 ignored issues
show
Coding Style introduced by
Usage of * or ** arguments should usually be done with care.

Generally, there is nothing wrong with usage of * or ** arguments. For readability of the code base, we suggest to not over-use these language constructs though.

For more information, we can recommend this blog post from Ned Batchelder including its comments which also touches this aspect.

Loading history...
255
256
    def prepare_args(self, msg, args):
257
        items = []
258
259
        if hasattr(msg, '_asdict'):
260
            msg = msg._asdict()
261
262
        for reference in args:
263
            if isinstance(reference, str):
264
                index = reference
265
                attr = 'make'
266
            elif isinstance(reference, bytes):
267
                index = reference.decode('utf-8')
268
                attr = 'pack'
269
            else:
270
                raise ValueError('Needed str or bytes for the reference')
271
272
            items.append(
273
                getattr(self._elements[index], attr)(msg)
274
            )
275
276
        return items
277