"""
The module implements abstraction for arrays used by Zserio python extension.
"""
import typing
from zserio.bitposition import alignto
from zserio.bitsizeof import (
bitsizeof_varuint16,
bitsizeof_varuint32,
bitsizeof_varuint64,
bitsizeof_varuint,
bitsizeof_varint16,
bitsizeof_varint32,
bitsizeof_varint64,
bitsizeof_varint,
bitsizeof_varsize,
bitsizeof_bytes,
bitsizeof_string,
bitsizeof_bitbuffer,
)
from zserio.bitreader import BitStreamReader
from zserio.bitwriter import BitStreamWriter
from zserio.bitbuffer import BitBuffer
from zserio.hashcode import (
HASH_SEED,
calc_hashcode_bool_array,
calc_hashcode_int_array,
calc_hashcode_float32_array,
calc_hashcode_float64_array,
calc_hashcode_bytes_array,
calc_hashcode_string_array,
calc_hashcode_object_array,
)
from zserio.exception import PythonRuntimeException
[docs]class Array:
"""
Abstraction for arrays to which Zserio arrays are mapped in python.
"""
def __init__(
self,
array_traits: typing.Any,
raw_array: typing.Optional[typing.List] = None,
*,
is_auto: bool = False,
is_implicit: bool = False,
set_offset_method: typing.Optional[typing.Callable[[int, int], None]] = None,
check_offset_method: typing.Optional[typing.Callable[[int, int], None]] = None
) -> None:
"""
Constructor.
:param array_traits: Array traits which specify the array type.
:param raw_array: Native python list which will be hold by this abstraction.
:param is_auto: True if mapped Zserio array is auto array.
:param is_implicit: True if mapped Zserio array is implicit array.
:param set_offset_method: Set offset method if mapped Zserio array is indexed offset array.
:param check_offset_method: Check offset method if mapped Zserio array is indexed offset array.
"""
self._raw_array: typing.List = [] if raw_array is None else raw_array
self._array_traits: typing.Any = array_traits
self._packed_array_traits: typing.Any = array_traits.packed_traits
self._is_auto: bool = is_auto
self._is_implicit: bool = is_implicit
self._set_offset_method: typing.Optional[typing.Callable[[int, int], None]] = set_offset_method
self._check_offset_method: typing.Optional[typing.Callable[[int, int], None]] = check_offset_method
[docs] @classmethod
def from_reader(
cls: typing.Type["Array"],
array_traits: typing.Any,
reader: BitStreamReader,
size: int = 0,
*,
is_auto: bool = False,
is_implicit: bool = False,
set_offset_method: typing.Optional[typing.Callable[[int, int], None]] = None,
check_offset_method: typing.Optional[typing.Callable[[int, int], None]] = None
) -> "Array":
"""
Constructs array and reads elements from the given bit stream reader.
:param array_traits: Array traits which specify the array type.
:param reader: Bit stream from which to read.
:param size: Number of elements to read or None in case of implicit or auto arrays.
:param raw_array: Native python list which will be hold by this abstraction.
:param is_auto: True if mapped Zserio array is auto array.
:param is_implicit: True if mapped Zserio array is implicit array.
:param set_offset_method: Set offset method if mapped Zserio array is indexed offset array.
:param check_offset_method: Check offset method if mapped Zserio array is indexed offset array.
:returns: Array instance filled using given bit stream reader.
"""
instance = cls(
array_traits,
is_auto=is_auto,
is_implicit=is_implicit,
set_offset_method=set_offset_method,
check_offset_method=check_offset_method,
)
instance.read(reader, size)
return instance
[docs] @classmethod
def from_reader_packed(
cls: typing.Type["Array"],
array_traits: typing.Any,
reader: BitStreamReader,
size: int = 0,
*,
is_auto: bool = False,
set_offset_method: typing.Optional[typing.Callable[[int, int], None]] = None,
check_offset_method: typing.Optional[typing.Callable[[int, int], None]] = None
) -> "Array":
"""
Constructs packed array and reads elements from the given bit stream reader.
:param array_traits: Array traits which specify the array type.
:param reader: Bit stream from which to read.
:param size: Number of elements to read or None in case of implicit or auto arrays.
:param raw_array: Native python list which will be hold by this abstraction.
:param is_auto: True if mapped Zserio array is auto array.
:param set_offset_method: Set offset method if mapped Zserio array is indexed offset array.
:param check_offset_method: Check offset method if mapped Zserio array is indexed offset array.
:returns: Array instance filled using given bit stream reader.
"""
instance = cls(
array_traits,
is_auto=is_auto,
is_implicit=False,
set_offset_method=set_offset_method,
check_offset_method=check_offset_method,
)
instance.read_packed(reader, size)
return instance
def __eq__(self, other: object) -> bool:
# it's enough to check only raw_array because compound types which call this are always the same type
if isinstance(other, Array):
return self._raw_array == other._raw_array
return False
def __hash__(self) -> int:
return self._array_traits.CALC_HASHCODE_FUNC(HASH_SEED, self._raw_array)
def __len__(self) -> int:
return len(self._raw_array)
def __getitem__(self, key: int) -> typing.Any:
return self._raw_array[key]
def __setitem__(self, key: int, value: typing.Any) -> None:
self._raw_array[key] = value
@property
def raw_array(self) -> typing.List:
"""
Gets raw array.
:returns: Native python list which is hold by the array.
"""
return self._raw_array
[docs] def bitsizeof(self, bitposition: int) -> int:
"""
Returns length of array stored in the bit stream in bits.
:param bitposition: Current bit stream position.
:returns: Length of the array stored in the bit stream in bits.
"""
end_bitposition = bitposition
size = len(self._raw_array)
if self._is_auto:
end_bitposition += bitsizeof_varsize(size)
if self._array_traits.HAS_BITSIZEOF_CONSTANT and size > 0:
element_size = self._array_traits.bitsizeof()
if self._set_offset_method is None:
end_bitposition += size * element_size
else:
end_bitposition = alignto(8, end_bitposition)
end_bitposition += element_size + (size - 1) * alignto(8, element_size)
else:
array_traits_bitsizeof = self._array_traits.bitsizeof
if self._set_offset_method is not None:
if self._array_traits.NEEDS_BITSIZEOF_POSITION:
for element in self._raw_array:
end_bitposition = alignto(8, end_bitposition)
end_bitposition += array_traits_bitsizeof(end_bitposition, element)
else:
for element in self._raw_array:
end_bitposition = alignto(8, end_bitposition)
end_bitposition += array_traits_bitsizeof(element)
else:
if self._array_traits.NEEDS_BITSIZEOF_POSITION:
for element in self._raw_array:
end_bitposition += array_traits_bitsizeof(end_bitposition, element)
else:
for element in self._raw_array:
end_bitposition += array_traits_bitsizeof(element)
return end_bitposition - bitposition
[docs] def bitsizeof_packed(self, bitposition: int) -> int:
"""
Returns length of the packed array stored in the bit stream in bits.
:param bitposition: Current bit stream position.
:returns: Length of the array stored in the bit stream in bits.
"""
end_bitposition = bitposition
size = len(self._raw_array)
if self._is_auto:
end_bitposition += bitsizeof_varsize(size)
if size > 0:
context = self._packed_array_traits.create_context()
packed_array_traits_init_context = self._packed_array_traits.init_context
packed_array_traits_bitsizeof = self._packed_array_traits.bitsizeof
for element in self._raw_array:
packed_array_traits_init_context(context, element)
if self._set_offset_method is not None:
for element in self._raw_array:
end_bitposition = alignto(8, end_bitposition)
end_bitposition += packed_array_traits_bitsizeof(context, end_bitposition, element)
else:
for element in self._raw_array:
end_bitposition += packed_array_traits_bitsizeof(context, end_bitposition, element)
return end_bitposition - bitposition
[docs] def initialize_offsets(self, bitposition: int) -> int:
"""
Initializes indexed offsets for the array.
:param bitposition: Current bit stream position.
:returns: Updated bit stream position which points to the first bit after the array.
"""
end_bitposition = bitposition
size = len(self._raw_array)
if self._is_auto:
end_bitposition += bitsizeof_varsize(size)
array_traits_initialize_offsets = self._array_traits.initialize_offsets
if self._set_offset_method is not None:
set_offset = self._set_offset_method
for index in range(size):
end_bitposition = alignto(8, end_bitposition)
set_offset(index, end_bitposition)
end_bitposition = array_traits_initialize_offsets(end_bitposition, self._raw_array[index])
else:
for element in self._raw_array:
end_bitposition = array_traits_initialize_offsets(end_bitposition, element)
return end_bitposition
[docs] def initialize_offsets_packed(self, bitposition: int) -> int:
"""
Initializes indexed offsets for the packed array.
:param bitposition: Current bit stream position.
:returns: Updated bit stream position which points to the first bit after the array.
"""
end_bitposition = bitposition
size = len(self._raw_array)
if self._is_auto:
end_bitposition += bitsizeof_varsize(size)
if size > 0:
context = self._packed_array_traits.create_context()
packed_array_traits_init_context = self._packed_array_traits.init_context
packed_array_traits_initialize_offsets = self._packed_array_traits.initialize_offsets
for element in self._raw_array:
packed_array_traits_init_context(context, element)
if self._set_offset_method is not None:
set_offset = self._set_offset_method
for index in range(size):
end_bitposition = alignto(8, end_bitposition)
set_offset(index, end_bitposition)
end_bitposition = packed_array_traits_initialize_offsets(
context, end_bitposition, self._raw_array[index]
)
else:
for element in self._raw_array:
end_bitposition = packed_array_traits_initialize_offsets(context, end_bitposition, element)
return end_bitposition
[docs] def read(self, reader: BitStreamReader, size: int = 0) -> None:
"""
Reads array from the bit stream.
:param reader: Bit stream from which to read.
:param size: Number of elements to read or None in case of implicit or auto arrays.
:raises PythonRuntimeException: If the implicit array does not have elements with constant bit size.
"""
self._raw_array.clear()
append = self._raw_array.append
array_traits_read = self._array_traits.read
if self._is_implicit:
if not self._array_traits.HAS_BITSIZEOF_CONSTANT:
raise PythonRuntimeException("Array: Implicit array elements must have constant bit size!")
element_size = self._array_traits.bitsizeof()
remaining_bits = reader.buffer_bitsize - reader.bitposition
read_size = remaining_bits // element_size
for _ in range(read_size):
# we know that no traits NEEDS_READ_INDEX here
append(array_traits_read(reader))
else:
if self._is_auto:
read_size = reader.read_varsize()
else:
read_size = size
if self._check_offset_method is not None:
check_offset = self._check_offset_method
reader_alignto = reader.alignto
if self._array_traits.NEEDS_READ_INDEX:
for index in range(read_size):
reader_alignto(8)
check_offset(index, reader.bitposition)
append(array_traits_read(reader, index))
else:
for index in range(read_size):
reader_alignto(8)
check_offset(index, reader.bitposition)
append(array_traits_read(reader))
else:
if self._array_traits.NEEDS_READ_INDEX:
for index in range(read_size):
append(array_traits_read(reader, index))
else:
for _ in range(read_size):
append(array_traits_read(reader))
[docs] def read_packed(self, reader: BitStreamReader, size: int = 0) -> None:
"""
Reads packed array from the bit stream.
:param reader: Bit stream from which to read.
:param size: Number of elements to read or 0 in case of auto arrays.
"""
self._raw_array.clear()
if self._is_implicit:
raise PythonRuntimeException("Array: Implicit array cannot be packed!")
if self._is_auto:
read_size = reader.read_varsize()
else:
read_size = size
if read_size > 0:
context = self._packed_array_traits.create_context()
append = self._raw_array.append
packed_array_traits_read = self._packed_array_traits.read
if self._check_offset_method is not None:
check_offset = self._check_offset_method
reader_alignto = reader.alignto
for index in range(read_size):
reader_alignto(8)
check_offset(index, reader.bitposition)
append(packed_array_traits_read(context, reader, index))
else:
for index in range(read_size):
append(packed_array_traits_read(context, reader, index))
[docs] def write(self, writer: BitStreamWriter) -> None:
"""
Writes array to the bit stream.
:param writer: Bit stream where to write.
"""
size = len(self._raw_array)
if self._is_auto:
writer.write_varsize(size)
array_traits_write = self._array_traits.write
if self._check_offset_method is not None:
check_offset = self._check_offset_method
writer_alignto = writer.alignto
for index in range(size):
writer_alignto(8)
check_offset(index, writer.bitposition)
array_traits_write(writer, self._raw_array[index])
else:
for element in self._raw_array:
array_traits_write(writer, element)
[docs] def write_packed(self, writer: BitStreamWriter) -> None:
"""
Writes packed array to the bit stream.
:param writer: Bit stream where to write.
"""
size = len(self._raw_array)
if self._is_auto:
writer.write_varsize(size)
if size > 0:
context = self._packed_array_traits.create_context()
packed_array_traits_init_context = self._packed_array_traits.init_context
packed_array_traits_write = self._packed_array_traits.write
for element in self._raw_array:
packed_array_traits_init_context(context, element)
if self._check_offset_method is not None:
check_offset = self._check_offset_method
writer_alignto = writer.alignto
for index in range(size):
writer_alignto(8)
check_offset(index, writer.bitposition)
packed_array_traits_write(context, writer, self._raw_array[index])
else:
for element in self._raw_array:
packed_array_traits_write(context, writer, element)
[docs]class DeltaContext:
"""
Context for delta packing created for each packable field.
Contexts are always newly created for each array operation (bitsizeof, initialize_offsets, read, write).
They must be initialized at first via calling the init method for each packable element present in the
array. After the full initialization, only a single method (bitsizeof, read, write) can be repeatedly
called for exactly the same sequence of packable elements.
Example::
context = DeltaContext(array_traits)
for element in array:
context.init(element) # initialization step, needed for max_bit_number calculation
context.write_descriptor(writer) # finishes the initialization
for element in array:
context.write(writer, element)
"""
def __init__(self) -> None:
"""Constructor."""
self._is_packed = False
self._max_bit_number = 0
self._previous_element: typing.Optional[int] = None
self._processing_started = False
self._unpacked_bitsize = 0
self._first_element_bitsize = 0
self._num_elements = 0
[docs] def init(self, array_traits: typing.Any, element: int) -> None:
"""
Makes initialization step for the provided array element.
:param array_traits: Standard array traits.
:param element: Current element of the array.
"""
self._num_elements += 1
self._unpacked_bitsize += DeltaContext._bitsizeof_unpacked(array_traits, element)
if self._previous_element is None:
self._previous_element = element
self._first_element_bitsize = self._unpacked_bitsize
else:
if self._max_bit_number <= self._MAX_BIT_NUMBER_LIMIT:
self._is_packed = True
delta = element - self._previous_element
max_bit_number = delta.bit_length()
# if delta is negative, we need one bit more because of sign
# if delta is positive, we need one bit more because delta are treated as signed number
# if delta is zero, we need one bit more because bit_length() returned zero
if max_bit_number > self._max_bit_number:
self._max_bit_number = max_bit_number
if max_bit_number > self._MAX_BIT_NUMBER_LIMIT:
self._is_packed = False
self._previous_element = element
[docs] def bitsizeof(self, array_traits: typing.Any, element: int) -> int:
"""
Returns length of the element representation stored in the bit stream in bits.
:param array_traits: Standard integral array traits.
:param element: Current element.
:returns: Length of the element representation stored in the bit stream in bits.
"""
if not self._processing_started:
self._processing_started = True
self._finish_init()
return self._bitsizeof_descriptor() + DeltaContext._bitsizeof_unpacked(array_traits, element)
elif not self._is_packed:
return DeltaContext._bitsizeof_unpacked(array_traits, element)
else:
return self._max_bit_number + 1 if self._max_bit_number > 0 else 0
[docs] def read(self, array_traits: typing.Any, reader: BitStreamReader) -> int:
"""
Reads the packed element from the bit stream.
:param array_traits: Standard array traits.
:param reader: Bit stream reader.
"""
if not self._processing_started:
self._processing_started = True
self._read_descriptor(reader)
return self._read_unpacked(array_traits, reader)
elif not self._is_packed:
return self._read_unpacked(array_traits, reader)
else:
assert self._previous_element is not None
if self._max_bit_number > 0:
delta = reader.read_signed_bits_unchecked(self._max_bit_number + 1)
self._previous_element += delta
return self._previous_element
[docs] def write(self, array_traits: typing.Any, writer: BitStreamWriter, element: int) -> None:
"""
Writes the packed element representation to the bit stream.
:param array_traits: Standard array traits.
:param writer: Bit stream writer.
:param element: Element to write.
"""
if not self._processing_started:
self._processing_started = True
self._finish_init()
self._write_descriptor(writer)
self._write_unpacked(array_traits, writer, element)
elif not self._is_packed:
self._write_unpacked(array_traits, writer, element)
else: # packed and not first
assert self._previous_element is not None
if self._max_bit_number > 0:
delta = element - self._previous_element
writer.write_signed_bits_unchecked(delta, self._max_bit_number + 1)
self._previous_element = element
def _finish_init(self) -> None:
if self._is_packed:
delta_bitsize = self._max_bit_number + 1 if self._max_bit_number > 0 else 0
packed_bitsize_with_descriptor = (
1
+ self._MAX_BIT_NUMBER_BITS # descriptor
+ self._first_element_bitsize
+ (self._num_elements - 1) * delta_bitsize
)
unpacked_bitsize_with_descriptor = 1 + self._unpacked_bitsize
if packed_bitsize_with_descriptor >= unpacked_bitsize_with_descriptor:
self._is_packed = False
def _bitsizeof_descriptor(self) -> int:
if self._is_packed:
return 1 + self._MAX_BIT_NUMBER_BITS
else:
return 1
@staticmethod
def _bitsizeof_unpacked(array_traits: typing.Any, element: int) -> int:
if array_traits.HAS_BITSIZEOF_CONSTANT:
return array_traits.bitsizeof()
else: # we know that NEEDS_BITSIZEOF_POSITION is False here
return array_traits.bitsizeof(element)
def _read_descriptor(self, reader: BitStreamReader) -> None:
self._is_packed = reader.read_bool()
if self._is_packed:
self._max_bit_number = reader.read_bits_unchecked(self._MAX_BIT_NUMBER_BITS)
def _read_unpacked(self, array_traits: typing.Any, reader: BitStreamReader) -> int:
element = array_traits.read(reader)
self._previous_element = element
return element
def _write_descriptor(self, writer: BitStreamWriter) -> None:
writer.write_bool(self._is_packed)
if self._is_packed:
writer.write_bits_unchecked(self._max_bit_number, self._MAX_BIT_NUMBER_BITS)
def _write_unpacked(self, array_traits: typing.Any, writer: BitStreamWriter, element: int) -> None:
self._previous_element = element
array_traits.write(writer, element)
_MAX_BIT_NUMBER_BITS = 6
_MAX_BIT_NUMBER_LIMIT = 62
[docs]class PackedArrayTraits:
"""
Packed array traits.
Packed array traits are used for all built-in types.
"""
def __init__(self, array_traits: typing.Any) -> None:
"""
Constructor.
:param array_traits: Standard array traits.
"""
self._array_traits = array_traits
[docs] @staticmethod
def create_context() -> DeltaContext:
"""
Creates new packing context - DeltaContext is used for non-object arrays.
:returns: New packing context.
"""
return DeltaContext()
[docs] def init_context(self, delta_context: DeltaContext, element: int) -> None:
"""
Calls context initialization step for the current element.
:param delta_context: Delta context.
:param element: Current element.
"""
delta_context.init(self._array_traits, element)
[docs] def bitsizeof(self, delta_context: DeltaContext, _bitposition: int, element: int) -> int:
"""
Returns length of the array element stored in the bit stream in bits.
:param delta_context: Delta context.
:param _bitposition: Current bit stream position (not used).
:param elemnet: Current element.
:returns: Length of the array element stored in the bit stream in bits.
"""
return delta_context.bitsizeof(self._array_traits, element)
[docs] def initialize_offsets(self, delta_context: DeltaContext, bitposition: int, element: int) -> int:
"""
Calls indexed offsets initialization for the current element.
:param delta_context: Delta context.
:param _bitposition: Current bit stream position.
:param element: Current element.
:returns: Updated bit stream position which points to the first bit after this element.
"""
return bitposition + delta_context.bitsizeof(self._array_traits, element)
[docs] def write(self, delta_context: DeltaContext, writer: BitStreamWriter, element: int) -> None:
"""
Writes the element to the bit stream.
:param delta_context: Delta context.
:param writer: Bit stream writer.
:param element: Element to write.
"""
delta_context.write(self._array_traits, writer, element)
[docs] def read(self, delta_context: DeltaContext, reader: BitStreamReader, _index: int) -> int:
"""
Read an element from the bit stream.
:param delta_context: Delta context.
:param reader: Bit stream reader.
:param _index: Not used.
:returns: Read element value.
"""
return delta_context.read(self._array_traits, reader)
[docs]class ObjectPackedArrayTraits:
"""
Packed array traits for Zserio objects.
This traits are used for Zserio objects which must implement special *_packed* methods to allow itself
to be used in a packed array.
"""
def __init__(self, element_factory: typing.Any):
"""
Constructor.
:param element_factory: Element factory which creates packed object from the element index.
"""
self._element_factory = element_factory
[docs] def create_context(self) -> typing.Any:
"""
Creates new packing context - generated ZserioPackingContext is used for object arrays.
:returns: New packing context.
"""
return self._element_factory.create_packing_context()
[docs] @staticmethod
def init_context(context: typing.Any, element: typing.Any) -> None:
"""
Calls context initialization step for the current element.
:param context: Packing context.
:param element: Current element.
"""
element.init_packing_context(context)
[docs] @staticmethod
def bitsizeof(context: typing.Any, bitposition: int, element: typing.Any) -> int:
"""
Returns length of the array element stored in the bit stream in bits.
:param context: Packing context.
:param bitposition: Current bit stream position.
:param elemnet: Current element.
:returns: Length of the array element stored in the bit stream in bits.
"""
return element.bitsizeof_packed(context, bitposition)
[docs] @staticmethod
def initialize_offsets(context: typing.Any, bitposition: int, element: typing.Any) -> int:
"""
Calls indexed offsets initialization for the current element.
:param context: Packing context.
:param bitposition: Current bit stream position.
:param element: Current element.
:returns: Updated bit stream position which points to the first bit after this element.
"""
return element.initialize_offsets_packed(context, bitposition)
[docs] @staticmethod
def write(context: typing.Any, writer: BitStreamWriter, element: typing.Any) -> None:
"""
Writes the element to the bit stream.
:param context: Packing context.
:param writer: Bit stream writer.
:param element: Element to write.
"""
element.write_packed(context, writer)
[docs] def read(self, context: typing.Any, reader: BitStreamReader, index: int) -> typing.Any:
"""
Read an element from the bit stream.
:param context: Packing context.
:param reader: Bit stream reader.
:param index: Index of the current element.
:returns: Read element value.
"""
return self._element_factory.create_packed(context, reader, index)
[docs]class BitFieldArrayTraits:
"""
Array traits for unsigned fixed integer Zserio types (uint16, uint32, uint64, bit:5, etc...).
"""
HAS_BITSIZEOF_CONSTANT = True
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_int_array)
def __init__(self, numbits: int) -> None:
"""
Constructor.
:param numbits: Number of bits for unsigned fixed integer Zserio type.
"""
self._numbits = numbits
@property
def packed_traits(self) -> PackedArrayTraits:
"""
Gets packed array traits.
:returns: PackedArrayTraits instance.
"""
return PackedArrayTraits(self)
[docs] def bitsizeof(self) -> int:
"""
Returns length of unsigned fixed integer Zserio type stored in the bit stream in bits.
:returns: Length of unsigned fixed integer Zserio type in bits.
"""
return self._numbits
[docs] def initialize_offsets(self, bitposition: int, _value: int) -> int:
"""
Initializes indexed offsets for unsigned fixed integer Zserio type.
:param bitposition: Current bit stream position.
:param _value: Not used.
:returns: Updated bit stream position which points to the first bit after unsigned fixed integer type.
"""
return bitposition + self.bitsizeof()
[docs] def read(self, reader: BitStreamReader) -> int:
"""
Reads unsigned fixed integer Zserio type from the bit stream.
:param reader: Bit stream from which to read.
:returns: Read unsigned int value.
"""
return reader.read_bits(self._numbits)
[docs] def write(self, writer: BitStreamWriter, value: int) -> None:
"""
Writes unsigned fixed integer Zserio type to the bit stream.
:param writer: Bit stream where to write.
:param value: Unsigned fixed integer Zserio type to write.
"""
writer.write_bits(value, self._numbits)
[docs]class SignedBitFieldArrayTraits:
"""
Array traits for signed fixed integer Zserio types (int16, int32, int64, int:5, etc...).
"""
HAS_BITSIZEOF_CONSTANT = True
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_int_array)
def __init__(self, numbits: int) -> None:
"""
Constructor.
:param numbits: Number of bits for signed fixed integer Zserio type.
"""
self._numbits = numbits
@property
def packed_traits(self) -> PackedArrayTraits:
"""
Gets packed array traits.
:returns: PackedArrayTraits instance.
"""
return PackedArrayTraits(self)
[docs] def bitsizeof(self) -> int:
"""
Returns length of signed fixed integer Zserio type stored in the bit stream in bits.
:returns: Length of signed fixed integer Zserio type in bits.
"""
return self._numbits
[docs] def initialize_offsets(self, bitposition: int, _value: int) -> int:
"""
Initializes indexed offsets for signed fixed integer Zserio type.
:param bitposition: Current bit stream position.
:param _value: Not used.
:returns: Updated bit stream position which points to the first bit after signed fixed integer type.
"""
return bitposition + self.bitsizeof()
[docs] def read(self, reader: BitStreamReader) -> int:
"""
Reads signed fixed integer Zserio type from the bit stream.
:param reader: Bit stream from which to read.
:returns: Read signed int value.
"""
return reader.read_signed_bits(self._numbits)
[docs] def write(self, writer: BitStreamWriter, value: int) -> None:
"""
Writes signed fixed integer Zserio type to the bit stream.
:param writer: Bit stream where to write.
:param value: Signed fixed integer Zserio type to write.
"""
writer.write_signed_bits(value, self._numbits)
[docs]class VarUInt16ArrayTraits:
"""
Array traits for Zserio varuint16 type.
"""
HAS_BITSIZEOF_CONSTANT = False
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_int_array)
@property
def packed_traits(self) -> PackedArrayTraits:
"""
Gets packed array traits.
:returns: PackedArrayTraits instance.
"""
return PackedArrayTraits(self)
[docs] @staticmethod
def bitsizeof(value: int) -> int:
"""
Returns length of Zserio varuint16 type stored in the bit stream in bits.
:param value: Zserio varuint16 type value.
:returns: Length of given Zserio varuint16 type in bits.
"""
return bitsizeof_varuint16(value)
[docs] @staticmethod
def initialize_offsets(bitposition: int, value: int) -> int:
"""
Initializes indexed offsets for Zserio varuint16 type.
:param bitposition: Current bit stream position.
:param value: Zserio varuint16 type value.
:returns: Updated bit stream position which points to the first bit after Zserio varuint16 type.
"""
return bitposition + VarUInt16ArrayTraits.bitsizeof(value)
[docs] @staticmethod
def read(reader: BitStreamReader) -> int:
"""
Reads Zserio varuint16 type from the bit stream.
:param reader: Bit stream from which to read.
"""
return reader.read_varuint16()
[docs] @staticmethod
def write(writer: BitStreamWriter, value: int) -> None:
"""
Writes Zserio varuint16 type to the bit stream.
:param writer: Bit stream where to write.
:param value: Zserio varuint16 type to write.
:returns: Read varuint16 value.
"""
writer.write_varuint16(value)
[docs]class VarUInt32ArrayTraits:
"""
Array traits for Zserio varuint32 type.
"""
HAS_BITSIZEOF_CONSTANT = False
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_int_array)
@property
def packed_traits(self) -> PackedArrayTraits:
"""
Gets packed array traits.
:returns: PackedArrayTraits instance.
"""
return PackedArrayTraits(self)
[docs] @staticmethod
def bitsizeof(value: int) -> int:
"""
Returns length of Zserio varuint32 type stored in the bit stream in bits.
:param value: Zserio varuint32 type value.
:returns: Length of given Zserio varuint32 type in bits.
"""
return bitsizeof_varuint32(value)
[docs] @staticmethod
def initialize_offsets(bitposition: int, value: int) -> int:
"""
Initializes indexed offsets for Zserio varuint32 type.
:param bitposition: Current bit stream position.
:param value: Zserio varuint32 type value.
:returns: Updated bit stream position which points to the first bit after Zserio varuint32 type.
"""
return bitposition + VarUInt32ArrayTraits.bitsizeof(value)
[docs] @staticmethod
def read(reader: BitStreamReader) -> int:
"""
Reads Zserio varuint32 type from the bit stream.
:param reader: Bit stream from which to read.
:returns: Read varuint32 value.
"""
return reader.read_varuint32()
[docs] @staticmethod
def write(writer: BitStreamWriter, value: int) -> None:
"""
Writes Zserio varuint32 type to the bit stream.
:param writer: Bit stream where to write.
:param value: Zserio varuint32 type to write.
"""
writer.write_varuint32(value)
[docs]class VarUInt64ArrayTraits:
"""
Array traits for Zserio varuint64 type.
"""
HAS_BITSIZEOF_CONSTANT = False
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_int_array)
@property
def packed_traits(self) -> PackedArrayTraits:
"""
Gets packed array traits.
:returns: PackedArrayTraits instance.
"""
return PackedArrayTraits(self)
[docs] @staticmethod
def bitsizeof(value: int) -> int:
"""
Returns length of Zserio varuint64 type stored in the bit stream in bits.
:param value: Zserio varuint64 type value.
:returns: Length of given Zserio varuint64 type in bits.
"""
return bitsizeof_varuint64(value)
[docs] @staticmethod
def initialize_offsets(bitposition: int, value: int) -> int:
"""
Initializes indexed offsets for Zserio varuint64 type.
:param value: Zserio varuint64 type value.
:returns: Updated bit stream position which points to the first bit after Zserio varuint64 type.
"""
return bitposition + VarUInt64ArrayTraits.bitsizeof(value)
[docs] @staticmethod
def read(reader: BitStreamReader) -> int:
"""
Reads Zserio varuint64 type from the bit stream.
:param reader: Bit stream from which to read.
:returns: Read varuint64 value.
"""
return reader.read_varuint64()
[docs] @staticmethod
def write(writer: BitStreamWriter, value: int) -> None:
"""
Writes Zserio varuint64 type to the bit stream.
:param writer: Bit stream where to write.
:param value: Zserio varuint64 type to write.
"""
writer.write_varuint64(value)
[docs]class VarUIntArrayTraits:
"""
Array traits for Zserio varuint type.
"""
HAS_BITSIZEOF_CONSTANT = False
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_int_array)
@property
def packed_traits(self) -> PackedArrayTraits:
"""
Gets packed array traits.
:returns: PackedArrayTraits instance.
"""
return PackedArrayTraits(self)
[docs] @staticmethod
def bitsizeof(value: int) -> int:
"""
Returns length of Zserio varuint type stored in the bit stream in bits.
:param value: Zserio varuint type value.
:returns: Length of given Zserio varuint type in bits.
"""
return bitsizeof_varuint(value)
[docs] @staticmethod
def initialize_offsets(bitposition: int, value: int) -> int:
"""
Initializes indexed offsets for Zserio varuint type.
:param bitposition: Current bit stream position.
:param value: Zserio varuint type value.
:returns: Updated bit stream position which points to the first bit after Zserio varuint type.
"""
return bitposition + VarUIntArrayTraits.bitsizeof(value)
[docs] @staticmethod
def read(reader: BitStreamReader) -> int:
"""
Reads Zserio varuint type from the bit stream.
:param reader: Bit stream from which to read.
:returns: Read varuint value.
"""
return reader.read_varuint()
[docs] @staticmethod
def write(writer: BitStreamWriter, value: int) -> None:
"""
Writes Zserio varuint type to the bit stream.
:param writer: Bit stream where to write.
:param value: Zserio varuint type to write.
"""
writer.write_varuint(value)
[docs]class VarSizeArrayTraits:
"""
Array traits for Zserio varsize type.
"""
HAS_BITSIZEOF_CONSTANT = False
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_int_array)
@property
def packed_traits(self) -> PackedArrayTraits:
"""
Gets packed array traits.
:returns: PackedArrayTraits instance.
"""
return PackedArrayTraits(self)
[docs] @staticmethod
def bitsizeof(value: int) -> int:
"""
Returns length of Zserio varsize type stored in the bit stream in bits.
:param value: Zserio varsize type value.
:returns: Length of given Zserio varsize type in bits.
"""
return bitsizeof_varsize(value)
[docs] @staticmethod
def initialize_offsets(bitposition: int, value: int) -> int:
"""
Initializes indexed offsets for Zserio varsize type.
:param bitposition: Current bit stream position.
:param value: Zserio varsize type value.
:returns: Updated bit stream position which points to the first bit after Zserio varsize type.
"""
return bitposition + VarSizeArrayTraits.bitsizeof(value)
[docs] @staticmethod
def read(reader: BitStreamReader) -> int:
"""
Reads Zserio varsize type from the bit stream.
:param reader: Bit stream from which to read.
:returns: Read varsize value.
"""
return reader.read_varsize()
[docs] @staticmethod
def write(writer: BitStreamWriter, value: int) -> None:
"""
Writes Zserio varsize type to the bit stream.
:param writer: Bit stream where to write.
:param value: Zserio varsize type to write.
"""
writer.write_varsize(value)
[docs]class VarInt16ArrayTraits:
"""
Array traits for Zserio varint16 type.
"""
HAS_BITSIZEOF_CONSTANT = False
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_int_array)
@property
def packed_traits(self) -> PackedArrayTraits:
"""
Gets packed array traits.
:returns: PackedArrayTraits instance.
"""
return PackedArrayTraits(self)
[docs] @staticmethod
def bitsizeof(value: int) -> int:
"""
Returns length of Zserio varint16 type stored in the bit stream in bits.
:param value: Zserio varint16 type value.
:returns: Length of given Zserio varint16 type in bits.
"""
return bitsizeof_varint16(value)
[docs] @staticmethod
def initialize_offsets(bitposition: int, value: int) -> int:
"""
Initializes indexed offsets for Zserio varint16 type.
:param bitposition: Current bit stream position.
:param value: Zserio varint16 type value.
:returns: Updated bit stream position which points to the first bit after Zserio varint16 type.
"""
return bitposition + VarInt16ArrayTraits.bitsizeof(value)
[docs] @staticmethod
def read(reader: BitStreamReader) -> int:
"""
Reads Zserio varint16 type from the bit stream.
:param reader: Bit stream from which to read.
:returns: Read varint16 value.
"""
return reader.read_varint16()
[docs] @staticmethod
def write(writer: BitStreamWriter, value: int) -> None:
"""
Writes Zserio varint16 type to the bit stream.
:param writer: Bit stream where to write.
:param value: Zserio varint16 type to write.
"""
writer.write_varint16(value)
[docs]class VarInt32ArrayTraits:
"""
Array traits for Zserio varint32 type.
"""
HAS_BITSIZEOF_CONSTANT = False
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_int_array)
@property
def packed_traits(self) -> PackedArrayTraits:
"""
Gets packed array traits.
:returns: PackedArrayTraits instance.
"""
return PackedArrayTraits(self)
[docs] @staticmethod
def bitsizeof(value: int) -> int:
"""
Returns length of Zserio varint32 type stored in the bit stream in bits.
:param value: Zserio varint32 type value.
:returns: Length of given Zserio varint32 type in bits.
"""
return bitsizeof_varint32(value)
[docs] @staticmethod
def initialize_offsets(bitposition: int, value: int) -> int:
"""
Initializes indexed offsets for Zserio varint32 type.
:param bitposition: Current bit stream position.
:param value: Zserio varint32 type value.
:returns: Updated bit stream position which points to the first bit after Zserio varint32 type.
"""
return bitposition + VarInt32ArrayTraits.bitsizeof(value)
[docs] @staticmethod
def read(reader: BitStreamReader) -> int:
"""
Reads Zserio varint32 type from the bit stream.
:param reader: Bit stream from which to read.
:returns: Read varint32 value.
"""
return reader.read_varint32()
[docs] @staticmethod
def write(writer: BitStreamWriter, value: int) -> None:
"""
Writes Zserio varint32 type to the bit stream.
:param writer: Bit stream where to write.
:param value: Zserio varint32 type to write.
"""
writer.write_varint32(value)
[docs]class VarInt64ArrayTraits:
"""
Array traits for Zserio varint64 type.
"""
HAS_BITSIZEOF_CONSTANT = False
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_int_array)
@property
def packed_traits(self) -> PackedArrayTraits:
"""
Gets packed array traits.
:returns: PackedArrayTraits instance.
"""
return PackedArrayTraits(self)
[docs] @staticmethod
def bitsizeof(value: int) -> int:
"""
Returns length of Zserio varint64 type stored in the bit stream in bits.
:param value: Zserio varint64 type value.
:returns: Length of given Zserio varint64 type in bits.
"""
return bitsizeof_varint64(value)
[docs] @staticmethod
def initialize_offsets(bitposition: int, value: int) -> int:
"""
Initializes indexed offsets for Zserio varint64 type.
:param bitposition: Current bit stream position.
:param value: Zserio varint64 type value.
:returns: Updated bit stream position which points to the first bit after Zserio varint64 type.
"""
return bitposition + VarInt64ArrayTraits.bitsizeof(value)
[docs] @staticmethod
def read(reader: BitStreamReader) -> int:
"""
Reads Zserio varint64 type from the bit stream.
:param reader: Bit stream from which to read.
:returns: Read varint64 value.
"""
return reader.read_varint64()
[docs] @staticmethod
def write(writer: BitStreamWriter, value: int) -> None:
"""
Writes Zserio varint64 type to the bit stream.
:param writer: Bit stream where to write.
:param value: Zserio varint64 type to write.
"""
writer.write_varint64(value)
[docs]class VarIntArrayTraits:
"""
Array traits for Zserio varint type.
"""
HAS_BITSIZEOF_CONSTANT = False
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_int_array)
@property
def packed_traits(self) -> PackedArrayTraits:
"""
Gets packed array traits.
:returns: PackedArrayTraits instance.
"""
return PackedArrayTraits(self)
[docs] @staticmethod
def bitsizeof(value: int) -> int:
"""
Returns length of Zserio varint type stored in the bit stream in bits.
:param value: Zserio varint type value.
:returns: Length of given Zserio varint type in bits.
"""
return bitsizeof_varint(value)
[docs] @staticmethod
def initialize_offsets(bitposition: int, value: int) -> int:
"""
Initializes indexed offsets for Zserio varint type.
:param bitposition: Current bit stream position.
:param value: Zserio varint type value.
:returns: Updated bit stream position which points to the first bit after Zserio varint type.
"""
return bitposition + VarIntArrayTraits.bitsizeof(value)
[docs] @staticmethod
def read(reader: BitStreamReader) -> int:
"""
Reads Zserio varint type from the bit stream.
:param reader: Bit stream from which to read.
:returns: Read varint value.
"""
return reader.read_varint()
[docs] @staticmethod
def write(writer: BitStreamWriter, value: int) -> None:
"""
Writes Zserio varint type to the bit stream.
:param writer: Bit stream where to write.
:param value: Zserio varint type to write.
"""
writer.write_varint(value)
[docs]class Float16ArrayTraits:
"""
Array traits for Zserio float16 type.
"""
HAS_BITSIZEOF_CONSTANT = True
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_float32_array)
@property
def packed_traits(self) -> None:
"""
Returns None since float type is not packable.
:returns: None.
"""
return None
[docs] @staticmethod
def bitsizeof() -> int:
"""
Returns length of Zserio float16 type stored in the bit stream in bits.
:returns: Length of Zserio float16 type in bits.
"""
return 16
[docs] @staticmethod
def initialize_offsets(bitposition: int, _value: float) -> int:
"""
Initializes indexed offsets for Zserio float16 type.
:param bitposition: Current bit stream position.
:param _value: Not used.
:returns: Updated bit stream position which points to the first bit after Zserio float16 type.
"""
return bitposition + Float16ArrayTraits.bitsizeof()
[docs] @staticmethod
def read(reader: BitStreamReader) -> float:
"""
Reads Zserio float16 type from the bit stream.
:param reader: Bit stream from which to read.
"""
return reader.read_float16()
[docs] @staticmethod
def write(writer: BitStreamWriter, value: float) -> None:
"""
Writes Zserio float16 type to the bit stream.
:param writer: Bit stream where to write.
:param value: Zserio float16 type to write.
:returns: Read float16 value.
"""
writer.write_float16(value)
[docs]class Float32ArrayTraits:
"""
Array traits for Zserio float32 type.
"""
HAS_BITSIZEOF_CONSTANT = True
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_float32_array)
@property
def packed_traits(self) -> None:
"""
Returns None since float type is not packable.
:returns: None.
"""
return None
[docs] @staticmethod
def bitsizeof() -> int:
"""
Returns length of Zserio float32 type stored in the bit stream in bits.
:returns: Length of Zserio float32 type in bits.
"""
return 32
[docs] @staticmethod
def initialize_offsets(bitposition: int, _value: float) -> int:
"""
Initializes indexed offsets for Zserio float32 type.
:param bitposition: Current bit stream position.
:param _value: Not used.
:returns: Updated bit stream position which points to the first bit after Zserio float32 type.
"""
return bitposition + Float32ArrayTraits.bitsizeof()
[docs] @staticmethod
def read(reader: BitStreamReader) -> float:
"""
Reads Zserio float32 type from the bit stream.
:param reader: Bit stream from which to read.
:returns: Read float32 value.
"""
return reader.read_float32()
[docs] @staticmethod
def write(writer: BitStreamWriter, value: float) -> None:
"""
Writes Zserio float32 type to the bit stream.
:param writer: Bit stream where to write.
:param value: Zserio float32 type to write.
"""
writer.write_float32(value)
[docs]class Float64ArrayTraits:
"""
Array traits for Zserio float64 type.
"""
HAS_BITSIZEOF_CONSTANT = True
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_float64_array)
@property
def packed_traits(self) -> None:
"""
Returns None since float type is not packable.
:returns: None.
"""
return None
[docs] @staticmethod
def bitsizeof() -> int:
"""
Returns length of Zserio float64 type stored in the bit stream in bits.
:returns: Length of Zserio float64 type in bits.
"""
return 64
[docs] @staticmethod
def initialize_offsets(bitposition: int, _value: float) -> int:
"""
Initializes indexed offsets for Zserio float64 type.
:param bitposition: Current bit stream position.
:param _value: Not used.
:returns: Updated bit stream position which points to the first bit after Zserio float64 type.
"""
return bitposition + Float64ArrayTraits.bitsizeof()
[docs] @staticmethod
def read(reader: BitStreamReader) -> float:
"""
Reads Zserio float64 type from the bit stream.
:param reader: Bit stream from which to read.
:returns: Read float64 value.
"""
return reader.read_float64()
[docs] @staticmethod
def write(writer: BitStreamWriter, value: float) -> None:
"""
Writes Zserio float64 type to the bit stream.
:param writer: Bit stream where to write.
:param value: Zserio float64 type to write.
"""
writer.write_float64(value)
[docs]class BytesArrayTraits:
"""
Array traits for Zserio bytes type.
"""
HAS_BITSIZEOF_CONSTANT = False
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_bytes_array)
@property
def packed_traits(self) -> None:
"""
Returns None since Bytes type is not packable.
:returns: None.
"""
return None
[docs] @staticmethod
def bitsizeof(value: bytearray) -> int:
"""
Returns length of Zserio bytes type stored in the bit stream in bits.
:param value: Zserio bytes type value.
:returns: Length of given Zserio bytes type in bits.
"""
return bitsizeof_bytes(value)
[docs] @staticmethod
def initialize_offsets(bitposition: int, value: bytearray) -> int:
"""
Initializes indexed offsets for Zserio bytes type.
:param bitposition: Current bit stream position.
:param value: Zserio bytes type value.
:returns: Updated bit stream position which points to the first bit after Zserio bytes type.
"""
return bitposition + BytesArrayTraits.bitsizeof(value)
[docs] @staticmethod
def read(reader: BitStreamReader) -> bytearray:
"""
Reads Zserio bytes type from the bit stream.
:param reader: Bit stream from which to read.
:returns: Read bytes value.
"""
return reader.read_bytes()
[docs] @staticmethod
def write(writer: BitStreamWriter, value: bytearray) -> None:
"""
Writes Zserio bytes type to the bit stream.
:param writer: Bit stream where to write.
:param value: Zserio bytes type to write.
"""
writer.write_bytes(value)
[docs]class StringArrayTraits:
"""
Array traits for Zserio string type.
"""
HAS_BITSIZEOF_CONSTANT = False
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_string_array)
@property
def packed_traits(self) -> None:
"""
Returns None since String type is not packable.
:returns: None.
"""
return None
[docs] @staticmethod
def bitsizeof(value: str) -> int:
"""
Returns length of Zserio string type stored in the bit stream in bits.
:param value: Zserio string type value.
:returns: Length of given Zserio string type in bits.
"""
return bitsizeof_string(value)
[docs] @staticmethod
def initialize_offsets(bitposition: int, value: str) -> int:
"""
Initializes indexed offsets for Zserio string type.
:param bitposition: Current bit stream position.
:param value: Zserio string type value.
:returns: Updated bit stream position which points to the first bit after Zserio string type.
"""
return bitposition + StringArrayTraits.bitsizeof(value)
[docs] @staticmethod
def read(reader: BitStreamReader) -> str:
"""
Reads Zserio string type from the bit stream.
:param reader: Bit stream from which to read.
:returns: Read string value.
"""
return reader.read_string()
[docs] @staticmethod
def write(writer: BitStreamWriter, value: str) -> None:
"""
Writes Zserio string type to the bit stream.
:param writer: Bit stream where to write.
:param value: Zserio string type to write.
"""
writer.write_string(value)
[docs]class BoolArrayTraits:
"""
Array traits for Zserio bool type.
"""
HAS_BITSIZEOF_CONSTANT = True
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_bool_array)
@property
def packed_traits(self) -> None:
"""
Returns None since Bool type is not packable.
:returns: None.
"""
return None
[docs] @staticmethod
def bitsizeof() -> int:
"""
Returns length of Zserio bool type stored in the bit stream in bits.
:returns: Length of Zserio bool type in bits.
"""
return 1
[docs] @staticmethod
def initialize_offsets(bitposition: int, _value: bool) -> int:
"""
Initializes indexed offsets for Zserio bool type.
:param bitposition: Current bit stream position.
:param _value: Not used.
:returns: Updated bit stream position which points to the first bit after Zserio bool type.
"""
return bitposition + BoolArrayTraits.bitsizeof()
[docs] @staticmethod
def read(reader: BitStreamReader) -> bool:
"""
Reads Zserio bool type from the bit stream.
:param reader: Bit stream from which to read.
:returns: Read bool value.
"""
return reader.read_bool()
[docs] @staticmethod
def write(writer: BitStreamWriter, value: bool) -> None:
"""
Writes Zserio bool type to the bit stream.
:param writer: Bit stream where to write.
:param value: Zserio bool type to write.
"""
writer.write_bool(value)
[docs]class BitBufferArrayTraits:
"""
Array traits for Zserio extern bit buffer type.
"""
HAS_BITSIZEOF_CONSTANT = False
NEEDS_BITSIZEOF_POSITION = False
NEEDS_READ_INDEX = False
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_object_array)
@property
def packed_traits(self) -> None:
"""
Returns None since BitBuffer type is not packable.
:returns: None.
"""
return None
[docs] @staticmethod
def bitsizeof(value: BitBuffer) -> int:
"""
Returns length of Zserio extern bit buffer type stored in the bit stream in bits.
:param value: Zserio extern bit buffer type value.
:returns: Length of given Zserio string type in bits.
"""
return bitsizeof_bitbuffer(value)
[docs] @staticmethod
def initialize_offsets(bitposition: int, value: BitBuffer) -> int:
"""
Initializes indexed offsets for Zserio extern bit buffer type.
:param bitposition: Current bit stream position.
:param value: Zserio extern bit buffer type value.
:returns: Updated bit stream position which points to the first bit after Zserio extern bit buffer type.
"""
return bitposition + BitBufferArrayTraits.bitsizeof(value)
[docs] @staticmethod
def read(reader: BitStreamReader) -> BitBuffer:
"""
Reads Zserio extern bit buffer type from the bit stream.
:param reader: Bit stream from which to read.
:returns: Read bit buffer value.
"""
return reader.read_bitbuffer()
[docs] @staticmethod
def write(writer: BitStreamWriter, value: BitBuffer) -> None:
"""
Writes Zserio extern bit buffer type to the bit stream.
:param writer: Bit stream where to write.
:param value: Zserio extern bit buffer type to write.
"""
writer.write_bitbuffer(value)
[docs]class ObjectArrayTraits:
"""
Array traits for Zserio structure, choice, union and enum types.
"""
HAS_BITSIZEOF_CONSTANT = False
NEEDS_BITSIZEOF_POSITION = True
NEEDS_READ_INDEX = True
CALC_HASHCODE_FUNC = staticmethod(calc_hashcode_object_array)
def __init__(self, element_factory: typing.Any) -> None:
"""
Constructor.
:param element_factory: Element factory which creates object from the element index.
"""
self._element_factory = element_factory
self._packed_traits = (
ObjectPackedArrayTraits(element_factory) if element_factory.IS_OBJECT_PACKABLE else None
)
@property
def packed_traits(self) -> typing.Optional[ObjectPackedArrayTraits]:
"""
Gets packed array traits.
:returns: ObjectPackedArrayTraits instance.
"""
return self._packed_traits
[docs] @staticmethod
def bitsizeof(bitposition: int, value: typing.Any) -> int:
"""
Returns length of Zserio object type stored in the bit stream in bits.
:param bitposition: Current bit position in bit stream.
:param value: Zserio object type value.
:returns: Length of given Zserio object type in bits.
"""
return value.bitsizeof(bitposition)
[docs] @staticmethod
def initialize_offsets(bitposition: int, value: typing.Any) -> int:
"""
Initializes indexed offsets for the Zserio object type.
:param bitposition: Current bit stream position.
:param value: Zserio object type value.
:returns: Updated bit stream position which points to the first bit after the Zserio object type.
"""
return value.initialize_offsets(bitposition)
[docs] def read(self, reader: BitStreamReader, index: int) -> typing.Any:
"""
Reads Zserio object type from the bit stream.
:param reader: Bit stream from which to read.
:param index: Element index in the array.
:returns: Read object.
"""
return self._element_factory.create(reader, index)
[docs] @staticmethod
def write(writer: BitStreamWriter, value: typing.Any) -> None:
"""
Writes Zserio object type to the bit stream.
:param writer: Bit stream where to write.
:param value: Zserio object type to write.
"""
value.write(writer)
