Completed
Pull Request — master (#42)
by TJ
08:30
created

ElementCallable.__init__()   B

Complexity

Conditions 4

Size

Total Lines 35

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 15
CRAP Score 4

Importance

Changes 3
Bugs 0 Features 0
Metric Value
cc 4
c 3
b 0
f 0
dl 0
loc 35
ccs 15
cts 15
cp 1
crap 4
rs 8.5806
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
0 ignored issues
show
best-practice introduced by
Too many instance attributes (13/7)
Loading history...
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
    accepted_mesages = (True, False)
105
106
    # pylint: disable=too-many-instance-attributes
107 1
    def __init__(self, field: list, mode: Optional[Mode]=Mode.Native, alignment: Optional[int]=1):
108
        # All of the type checks have already been performed by the class
109
        # factory
110 1
        self.name = field[0]
111 1
112 1
        # Callable elements use the normal struct packing method
113
        self.format = field[1]
114 1
115 1
        if isinstance(field[2], dict):
116
            self.ref = field[2]
117 1
118
            default_list = [None, None]
119 1
            self._make_func = self.ref.get('make', default_list)[0]
120
            self._make_args = self.ref.get('make', default_list)[1:]
121 1
122
            self._pack_func = self.ref.get('pack', default_list)[0]
123 1
            self._pack_args = self.ref.get('pack', default_list)[1:]
124
125 1
            self._unpack_func = self.ref.get('unpack', default_list)[0]
126
            self._unpack_args = self.ref.get('unpack', default_list)[1:]
127 1
        elif isinstance(field[2], set):
128
            instruction = field[2].copy().pop()
129 1
            self.ref = {'all': instruction}
130 1
131
            self._make_func = self._pack_func = self._unpack_func = instruction[0]
132
            self._make_args = self._pack_args = self._unpack_args = instruction[1:]
133
134
        if len(field) == 4:
135
            self._error_on_bad_result = field[3]
136 1
        else:
137
            self._error_on_bad_result = True
138
139
        self._elements = []
140
141
        self.update(mode, alignment)
142 1
143
    @property
144
    def _struct(self):
145
        return struct.Struct(self._mode.value + self.format)
146
147
    @staticmethod
148
    def valid(field: tuple) -> bool:
149 1
        """
150
        See :py:func:`starstruct.element.Element.valid`
151
152
        :param field: The items to determine the structure of the element
153
        """
154 1
        required_keys = {'pack', 'unpack', 'make'}
155
156 1
        if len(field) >= 3 and isinstance(field[0], str) and isinstance(field[1], str):
157 1
            if isinstance(field[2], dict):
158
                if set(field[2].keys()) <= required_keys and \
159 1
                        all(isinstance(val, tuple) for val in field[2].values()):
160 1
                    return True
161
            elif isinstance(field[2], set):
162 1
                return len(field[2]) == 1 and \
163
                    all(isinstance(val, tuple) for val in field[2])
164 1
165
        return False
166 1
167
168 1
    def validate(self, msg):
169 1
        """
170
        Ensure that the supplied message contains the required information for
171
        this element object to operate.
172 1
173
        All elements that are Variable must reference valid Length elements.
174
        """
175 1
        for action in self.ref.values():
176
            for arg in action[1:]:
177
                if arg in ElementCallable.accepted_mesages:
178
                    continue
179 1
                elif isinstance(arg, str):
180 1
                    pass
181 1
                elif hasattr(arg, 'decode'):
182
                    arg = arg.decode('utf-8')
183
                elif hasattr(arg, 'to_bytes'):
184 1
                    arg = arg.to_bytes((arg.bit_length() + 7) // 8, self._mode.to_byteorder()).decode('utf-8')
185 1
186
                if arg not in msg:
187
                    raise ValueError('Need all keys to be in the message, {0} was not found\nAction: {1} -> {2}'.format(arg, action, action[1:]))
188
189
    def update(self, mode=None, alignment=None):
190 1
        """change the mode of the struct format"""
191
        if mode:
192 1
            self._mode = mode
193
194
        if alignment:
195
            self._alignment = alignment
196
197 1
    def pack(self, msg):
198
        """Pack the provided values into the supplied buffer."""
199
        pack_values = self.call_func(msg, self._pack_func, self._pack_args)
200 1
201
        # Test if the object is iterable
202 1
        # If it isn't, then turn it into a list
203
        try:
204 1
            _ = (p for p in pack_values)
205 1
        except TypeError:
206 1
            pack_values = [pack_values]
207 1
208 1
        # Unpack the items for struct to allow for mutli-value
209 1
        # items to be passed in.
210 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...
211
212
    def unpack(self, msg, buf):
213
        """Unpack data from the supplied buffer using the initialized format."""
214 1
        ret = self._struct.unpack_from(buf)
215
        if isinstance(ret, (list, tuple)) and len(ret) == 1:
216
            # We only change it not to a list if we expected one value.
217
            # Otherwise, we keep it as a list, because that's what we would
218 1
            # expect (like for a 16I type of struct
219
            ret = ret[0]
220 1
221 1
        # Only check for errors if they haven't told us not to
222 1
        if self._error_on_bad_result:
223
            # Pretend we're getting a dictionary to make our item,
224 1
            # but it has no reference to `self`. This is so we check
225
            # for errors correctly.
226
            temp_dict = copy.deepcopy(msg._asdict())
227
            temp_dict.pop(self.name)
228
            expected_value = self.call_func(msg, self._unpack_func, self._unpack_args)
229
230
            # Check for an error
231
            if expected_value != ret:
232
                raise ValueError('Expected value was: {0}, but got: {1}'.format(
233
                    expected_value,
234
                    ret,
235
                ))
236
237
        if self.name in self._unpack_args:
238
            msg = msg._replace(**{self.name: ret})
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...
239
240
        ret = self.call_func(msg, self._unpack_func, self._unpack_args, original=ret)
241
242
        return (ret, buf[self._struct.size:])
243
244
    def make(self, msg):
245
        """Return the expected "made" value"""
246
        # If we aren't going to error on a bad result
247
        # and our name is in the message, just send the value
248
        # No need to do extra work.
249
        if not self._error_on_bad_result \
250
                and self.name in msg \
251
                and msg[self.name] is not None:
252
            return msg[self.name]
253
254
        ret = self.call_func(msg, self._make_func, self._make_args)
255
256
        if self.name in msg:
257
            if ret != msg[self.name]:
258
                raise ValueError('Excepted value: {0}, but got: {1}'.format(ret, msg[self.name]))
259
260
        return ret
261
262
    def call_func(self, msg, func, args, original=None):
263
        if func is None:
264
            return original
265
266
        items = self.prepare_args(msg, args)
267
        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...
268
269
    def prepare_args(self, msg, args):
270
        items = []
271
272
        if hasattr(msg, '_asdict'):
273
            msg = msg._asdict()
274
275
        for reference in args:
276
            if isinstance(reference, str):
277
                index = reference
278
                attr = 'make'
279
            elif isinstance(reference, bytes):
280
                index = reference.decode('utf-8')
281
                attr = 'pack'
282
            else:
283
                raise ValueError('Needed str or bytes for the reference')
284
285
            items.append(
286
                getattr(self._elements[index], attr)(msg)
287
            )
288
289
        return items
290