Class PyArraySerializer

python/pyfory/serializer.py:540–582

class PyArraySerializer(Serializer):
    typecode_dict = typecode_dict
    typecodearray_type = {typecode: ftype for typecode, (_itemsize, ftype, _type_id) in typecode_dict.items()}

    def __init__(self, type_resolver, ftype, type_id: str):
        super().__init__(type_resolver, ftype)
        self.typecode = typeid_code[type_id]
        self.itemsize, ftype, self.type_id = typecode_dict[self.typecode]

    def _array_type_id(self, value):
        entry = typecode_dict.get(value.typecode)
        if entry is None:
            raise TypeError(f"Unsupported array.array typecode {value.typecode!r}")
        itemsize, _ftype, type_id = entry
        if value.itemsize != itemsize:
            raise TypeError(f"array.array typecode {value.typecode!r} has itemsize {value.itemsize}, expected {itemsize}")
        return type_id

    def write(self, buffer, value):
        actual_type_id = self._array_type_id(value)
        if actual_type_id != self.type_id:
            raise TypeError(f"array.array typecode {value.typecode!r} maps to type id {actual_type_id}, expected {self.type_id}")
        view = memoryview(value)
        assert view.itemsize == self.itemsize
        assert view.c_contiguous  # TODO handle contiguous
        nbytes = len(value) * self.itemsize
        buffer.write_var_uint32(nbytes)
        if is_little_endian or self.itemsize == 1:
            buffer.write_buffer(value)
        else:
            # Swap bytes on big-endian machines for multi-byte types
            swapped = array.array(self.typecode, value)
            swapped.byteswap()
            buffer.write_buffer(swapped)

    def read(self, buffer):
        data = buffer.read_bytes_and_size()
        arr = array.array(self.typecode, [])
        arr.frombytes(data)
        if not is_little_endian and self.itemsize > 1:
            # Swap bytes on big-endian machines for multi-byte types
            arr.byteswap()
        return arr


FORY_ARRAY_WRAPPERS = {