Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def decode(bits, descriptors, operators, descriptor_overlay):
"""
:param bits: Bit stream to decode from
:param descriptors: Descriptor iterator
:param dict operators: Operators in effect, indexed by opcode
:param dict descriptor_overlay: Overlay descriptors affected by CHANGE_REFERENCE_VALUES operator
"""
values = []
for descriptor in descriptors:
descriptor = descriptor_overlay.get(descriptor.code, descriptor)
if isinstance(descriptor, ElementDescriptor):
op_crf = operators.get(OpCode.CHANGE_REFERENCE_VALUES, None)
if op_crf is not None:
ref_value = Bits._readuint(bits, op_crf.bits(), bits.pos)
bits.pos += op_crf.bits()
top_bit_mask = (1 << op_crf.bits()-1)
if ref_value & top_bit_mask:
ref_value = -(ref_value & ~top_bit_mask)
overlay_descriptor = ElementDescriptor(descriptor.code, descriptor.length, descriptor.scale, ref_value, descriptor.significance, descriptor.unit)
descriptor_overlay[descriptor.code] = overlay_descriptor
continue
op_aaf = operators.get(OpCode.ADD_ASSOCIATED_FIELD, None)
if op_aaf is not None and descriptor.code != fxy2int("031021"):
# Don't apply to ASSOCIATED FIELD SIGNIFICANCE
associated_value = Bits._readuint(bits, op_aaf.bits(), bits.pos)
bits.pos += op_aaf.bits()
# Use dummy descriptor 999999 for associated field, like Geo::BUFR and libbufr
dummy_descriptor = ElementDescriptor(fxy2int("999999"), op_aaf.bits(), 0, 0, "ASSOCIATED FIELD", "NUMERIC")
values.append(BufrValue(associated_value, associated_value, dummy_descriptor))
if isinstance(descriptor, ElementDescriptor):
op_crf = operators.get(OpCode.CHANGE_REFERENCE_VALUES, None)
if op_crf is not None:
ref_value = Bits._readuint(bits, op_crf.bits(), bits.pos)
bits.pos += op_crf.bits()
top_bit_mask = (1 << op_crf.bits()-1)
if ref_value & top_bit_mask:
ref_value = -(ref_value & ~top_bit_mask)
overlay_descriptor = ElementDescriptor(descriptor.code, descriptor.length, descriptor.scale, ref_value, descriptor.significance, descriptor.unit)
descriptor_overlay[descriptor.code] = overlay_descriptor
continue
op_aaf = operators.get(OpCode.ADD_ASSOCIATED_FIELD, None)
if op_aaf is not None and descriptor.code != fxy2int("031021"):
# Don't apply to ASSOCIATED FIELD SIGNIFICANCE
associated_value = Bits._readuint(bits, op_aaf.bits(), bits.pos)
bits.pos += op_aaf.bits()
# Use dummy descriptor 999999 for associated field, like Geo::BUFR and libbufr
dummy_descriptor = ElementDescriptor(fxy2int("999999"), op_aaf.bits(), 0, 0, "ASSOCIATED FIELD", "NUMERIC")
values.append(BufrValue(associated_value, associated_value, dummy_descriptor))
read_length = _calculate_read_length(descriptor, operators)
if descriptor.unit == 'CCITTIA5':
raw_value = Bits._readhex(bits, read_length, bits.pos)
else:
raw_value = Bits._readuint(bits, read_length, bits.pos)
bits.pos += read_length
values.append(_decode_raw_value(raw_value, descriptor, operators))
elif isinstance(descriptor, ReplicationDescriptor):
aggregation = []
if descriptor.count:
bval = None
continue
op_aaf = operators.get(OpCode.ADD_ASSOCIATED_FIELD, None)
if op_aaf is not None and descriptor.code != fxy2int("031021"):
# Don't apply to ASSOCIATED FIELD SIGNIFICANCE
associated_value = Bits._readuint(bits, op_aaf.bits(), bits.pos)
bits.pos += op_aaf.bits()
# Use dummy descriptor 999999 for associated field, like Geo::BUFR and libbufr
dummy_descriptor = ElementDescriptor(fxy2int("999999"), op_aaf.bits(), 0, 0, "ASSOCIATED FIELD", "NUMERIC")
values.append(BufrValue(associated_value, associated_value, dummy_descriptor))
read_length = _calculate_read_length(descriptor, operators)
if descriptor.unit == 'CCITTIA5':
raw_value = Bits._readhex(bits, read_length, bits.pos)
else:
raw_value = Bits._readuint(bits, read_length, bits.pos)
bits.pos += read_length
values.append(_decode_raw_value(raw_value, descriptor, operators))
elif isinstance(descriptor, ReplicationDescriptor):
aggregation = []
if descriptor.count:
bval = None
count = descriptor.count
else:
bval = decode(bits, itertools.islice(descriptors, 1), {}, {})[0]
count = bval.value
n_fields = descriptor.fields
field_descriptors = list(itertools.islice(descriptors, n_fields))
if bval is None or bval.descriptor.code in REPLICATION_DESCRIPTORS:
# Regular replication, X elements repeated Y or <element value=""> times in the file
for _ in range(count):
aggregation.append(decode(bits, iter(field_descriptors), operators, descriptor_overlay))</element>
continue
op_aaf = operators.get(OpCode.ADD_ASSOCIATED_FIELD, None)
if op_aaf is not None and descriptor.code != fxy2int("031021"):
# Don't apply to ASSOCIATED FIELD SIGNIFICANCE
# Use dummy descriptor 999999 for associated field, like Geo::BUFR and libbufr
dummy_descriptors = iter([ElementDescriptor(fxy2int("999999"), op_aaf.bits(), 0, 0, "ASSOCIATED FIELD", "NUMERIC")])
vals = decode_compressed(bits, dummy_descriptors, n_subsets, {}, {})
for i,ss in enumerate(vals):
subsets[i].extend(ss)
read_length = _calculate_read_length(descriptor, operators)
if descriptor.unit == 'CCITTIA5':
ref_value = Bits._readhex(bits, read_length, bits.pos)
else:
ref_value = Bits._readuint(bits, read_length, bits.pos)
bits.pos += read_length
n_bits = Bits._readuint(bits, 6, bits.pos)
bits.pos += 6
for i in range(n_subsets):
if descriptor.unit == 'CCITTIA5':
n_chars = n_bits
if n_chars:
raw_value = Bits._readhex(bits, n_chars*8, bits.pos)
bits.pos += n_chars*8
value = _decode_raw_value(raw_value, descriptor, operators)
else:
value = _decode_raw_value(ref_value, descriptor, operators)
else:
if n_bits:
n_bits = Bits._readuint(bits, 6, bits.pos)
bits.pos += 6
for i in range(n_subsets):
if descriptor.unit == 'CCITTIA5':
n_chars = n_bits
if n_chars:
raw_value = Bits._readhex(bits, n_chars*8, bits.pos)
bits.pos += n_chars*8
value = _decode_raw_value(raw_value, descriptor, operators)
else:
value = _decode_raw_value(ref_value, descriptor, operators)
else:
if n_bits:
increment = Bits._readuint(bits, n_bits, bits.pos)
bits.pos += n_bits
if increment ^ ((1 << n_bits)-1) == 0: # Missing value, all-ones
value = _decode_raw_value((1 << descriptor.length)-1, descriptor, operators)
else:
value = _decode_raw_value(ref_value + increment, descriptor, operators)
else:
value = _decode_raw_value(ref_value, descriptor, operators)
subsets[i].append(value)
elif isinstance(descriptor, ReplicationDescriptor):
aggregations = [[] for x in range(n_subsets)]
if descriptor.count:
bval = None
count = descriptor.count
else:
bval = decode_compressed(bits, itertools.islice(descriptors, 1), n_subsets, {}, {})[0][0]
count = bval.value
if op_aaf is not None and descriptor.code != fxy2int("031021"):
# Don't apply to ASSOCIATED FIELD SIGNIFICANCE
# Use dummy descriptor 999999 for associated field, like Geo::BUFR and libbufr
dummy_descriptors = iter([ElementDescriptor(fxy2int("999999"), op_aaf.bits(), 0, 0, "ASSOCIATED FIELD", "NUMERIC")])
vals = decode_compressed(bits, dummy_descriptors, n_subsets, {}, {})
for i,ss in enumerate(vals):
subsets[i].extend(ss)
read_length = _calculate_read_length(descriptor, operators)
if descriptor.unit == 'CCITTIA5':
ref_value = Bits._readhex(bits, read_length, bits.pos)
else:
ref_value = Bits._readuint(bits, read_length, bits.pos)
bits.pos += read_length
n_bits = Bits._readuint(bits, 6, bits.pos)
bits.pos += 6
for i in range(n_subsets):
if descriptor.unit == 'CCITTIA5':
n_chars = n_bits
if n_chars:
raw_value = Bits._readhex(bits, n_chars*8, bits.pos)
bits.pos += n_chars*8
value = _decode_raw_value(raw_value, descriptor, operators)
else:
value = _decode_raw_value(ref_value, descriptor, operators)
else:
if n_bits:
increment = Bits._readuint(bits, n_bits, bits.pos)
bits.pos += n_bits
if increment ^ ((1 << n_bits)-1) == 0: # Missing value, all-ones