#12496

pending completion

Build # #12496

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/datajoint/datajoint-python/commit/<a class=hub.com/datajoint/datajoint-python/commit/1421ce9594e68f8be9df67f9c6aae83c0fc5b07a">1421ce959<a href="https://github.com/datajoint/datajoint-python/commit/1421ce9594e68f8be9df67f9c6aae83c0fc5b07a">">Merge </a><a class="double-link" href="https://github.com/datajoint/datajoint-python/commit/<a class="double-link" href="https://github.com/datajoint/datajoint-python/commit/b65338906ace14844deb4fed896853b3f9743eba">b65338906</a>">b65338906</a><a href="https://github.com/datajoint/datajoint-python/commit/1421ce9594e68f8be9df67f9c6aae83c0fc5b07a"> into 07c5553e4">07c5553e4</a>

Run Details

6 of 6 new or added lines in 1 file covered. (100.0%)

3095 of 3423 relevant lines covered (90.42%)

0.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.05

/datajoint/blob.py

"""
(De)serialization methods for basic datatypes and numpy.ndarrays with provisions for mutual
compatibility with Matlab-based serialization implemented by mYm.
"""

import zlib
from itertools import repeat
import collections
from decimal import Decimal
import datetime
import uuid
import numpy as np
from .errors import DataJointError
from .settings import config


mxClassID = dict(
    (
        # see http://www.mathworks.com/help/techdoc/apiref/mxclassid.html
        ("mxUNKNOWN_CLASS", None),
        ("mxCELL_CLASS", None),
        ("mxSTRUCT_CLASS", None),
        ("mxLOGICAL_CLASS", np.dtype("bool")),
        ("mxCHAR_CLASS", np.dtype("c")),
        ("mxVOID_CLASS", np.dtype("O")),
        ("mxDOUBLE_CLASS", np.dtype("float64")),
        ("mxSINGLE_CLASS", np.dtype("float32")),
        ("mxINT8_CLASS", np.dtype("int8")),
        ("mxUINT8_CLASS", np.dtype("uint8")),
        ("mxINT16_CLASS", np.dtype("int16")),
        ("mxUINT16_CLASS", np.dtype("uint16")),
        ("mxINT32_CLASS", np.dtype("int32")),
        ("mxUINT32_CLASS", np.dtype("uint32")),
        ("mxINT64_CLASS", np.dtype("int64")),
        ("mxUINT64_CLASS", np.dtype("uint64")),
        ("mxFUNCTION_CLASS", None),
    )
)

rev_class_id = {dtype: i for i, dtype in enumerate(mxClassID.values())}
dtype_list = list(mxClassID.values())
type_names = list(mxClassID)

compression = {b"ZL123\0": zlib.decompress}

bypass_serialization = False  # runtime setting to bypass blob (en|de)code

# runtime setting to read integers as 32-bit to read blobs created by the 32-bit
# version of the mYm library for MATLAB
use_32bit_dims = False


def len_u64(obj):
    return np.uint64(len(obj)).tobytes()


def len_u32(obj):
    return np.uint32(len(obj)).tobytes()


class MatCell(np.ndarray):
    """a numpy ndarray representing a Matlab cell array"""

    pass


class MatStruct(np.recarray):
    """numpy.recarray representing a Matlab struct array"""

    pass


class Blob:
    def __init__(self, squeeze=False):
        self._squeeze = squeeze
        self._blob = None
        self._pos = 0
        self.protocol = None

    def set_dj0(self):
        if not config.get("enable_python_native_blobs"):
            raise DataJointError(
                """v0.12+ python native blobs disabled.
                See also: https://github.com/datajoint/datajoint-python#python-native-blobs"""
            )

        self.protocol = b"dj0\0"  # when using new blob features

    def squeeze(self, array, convert_to_scalar=True):
        """
        Simplify the input array - squeeze out all singleton dimensions.
        If convert_to_scalar, then convert zero-dimensional arrays to scalars
        """
        if not self._squeeze:
            return array
        array = array.squeeze()
        return array.item() if array.ndim == 0 and convert_to_scalar else array

    def unpack(self, blob):
        self._blob = blob
        try:
            # decompress
            prefix = next(
                p for p in compression if self._blob[self._pos :].startswith(p)
            )
        except StopIteration:
            pass  # assume uncompressed but could be unrecognized compression
        else:
            self._pos += len(prefix)
            blob_size = self.read_value()
            blob = compression[prefix](self._blob[self._pos :])
            assert len(blob) == blob_size
            self._blob = blob
            self._pos = 0
        blob_format = self.read_zero_terminated_string()
        if blob_format in ("mYm", "dj0"):
            return self.read_blob(n_bytes=len(self._blob) - self._pos)

    def read_blob(self, n_bytes=None):
        start = self._pos
        data_structure_code = chr(self.read_value("uint8"))
        try:
            call = {
                # MATLAB-compatible, inherited from original mYm
                "A": self.read_array,  # matlab-compatible numeric arrays and scalars with ndim==0
                "P": self.read_sparse_array,  # matlab sparse array -- not supported yet
                "S": self.read_struct,  # matlab struct array
                "C": self.read_cell_array,  # matlab cell array
                # basic data types
                "\xFF": self.read_none,  # None
                "\x01": self.read_tuple,  # a Sequence (e.g. tuple)
                "\x02": self.read_list,  # a MutableSequence (e.g. list)
                "\x03": self.read_set,  # a Set
                "\x04": self.read_dict,  # a Mapping (e.g. dict)
                "\x05": self.read_string,  # a UTF8-encoded string
                "\x06": self.read_bytes,  # a ByteString
                "\x0a": self.read_int,  # unbounded scalar int
                "\x0b": self.read_bool,  # scalar boolean
                "\x0c": self.read_complex,  # scalar 128-bit complex number
                "\x0d": self.read_float,  # scalar 64-bit float
                "F": self.read_recarray,  # numpy array with fields, including recarrays
                "d": self.read_decimal,  # a decimal
                "t": self.read_datetime,  # date, time, or datetime
                "T": self.read_np_datetime,  # np.datetime64
                "u": self.read_uuid,  # UUID
            }[data_structure_code]
        except KeyError:
            raise DataJointError(
                'Unknown data structure code "%s". Upgrade datajoint.'
                % data_structure_code
            )
        v = call()
        if n_bytes is not None and self._pos - start != n_bytes:
            raise DataJointError("Blob length check failed! Invalid blob")
        return v

    def pack_blob(self, obj):
        # original mYm-based serialization from datajoint-matlab
        if isinstance(obj, MatCell):
            return self.pack_cell_array(obj)
        if isinstance(obj, MatStruct):
            return self.pack_struct(obj)
        if isinstance(obj, np.ndarray) and obj.dtype.fields is None:
            return self.pack_array(obj)

        # blob types in the expanded dj0 blob format
        self.set_dj0()
        if not isinstance(obj, (np.ndarray, np.number)):
            # python built-in data types
            if isinstance(obj, bool):
                return self.pack_bool(obj)
            if isinstance(obj, int):
                return self.pack_int(obj)
            if isinstance(obj, complex):
                return self.pack_complex(obj)
            if isinstance(obj, float):
                return self.pack_float(obj)
        if isinstance(obj, np.ndarray) and obj.dtype.fields:
            return self.pack_recarray(np.array(obj))
        if isinstance(obj, np.number):
            return self.pack_array(np.array(obj))
        if isinstance(obj, (bool, np.bool_)):
            return self.pack_array(np.array(obj))
        if isinstance(obj, (float, int, complex)):
            return self.pack_array(np.array(obj))
        if isinstance(
            obj, (datetime.datetime, datetime.date, datetime.time, np.datetime64)
        ):
            return self.pack_datetime(obj)
        if isinstance(obj, Decimal):
            return self.pack_decimal(obj)
        if isinstance(obj, uuid.UUID):
            return self.pack_uuid(obj)
        if isinstance(obj, collections.abc.Mapping):
            return self.pack_dict(obj)
        if isinstance(obj, str):
            return self.pack_string(obj)
        if isinstance(obj, collections.abc.ByteString):
            return self.pack_bytes(obj)
        if isinstance(obj, collections.abc.MutableSequence):
            return self.pack_list(obj)
        if isinstance(obj, collections.abc.Sequence):
            return self.pack_tuple(obj)
        if isinstance(obj, collections.abc.Set):
            return self.pack_set(obj)
        if obj is None:
            return self.pack_none()
        raise DataJointError(
            "Packing object of type %s currently not supported!" % type(obj)
        )

    def read_array(self):
        n_dims = int(self.read_value())
        shape = self.read_value(count=n_dims)
        n_elem = np.prod(shape, dtype=int)
        dtype_id, is_complex = self.read_value("uint32", 2)
        dtype = dtype_list[dtype_id]

        if type_names[dtype_id] == "mxVOID_CLASS":
            data = np.array(
                list(self.read_blob(self.read_value()) for _ in range(n_elem)),
                dtype=np.dtype("O"),
            )
        elif type_names[dtype_id] == "mxCHAR_CLASS":
            # compensate for MATLAB packing of char arrays
            data = self.read_value(dtype, count=2 * n_elem)
            data = data[::2].astype("U1")
            if n_dims == 2 and shape[0] == 1 or n_dims == 1:
                compact = data.squeeze()
                data = (
                    compact
                    if compact.shape == ()
                    else np.array("".join(data.squeeze()))
                )
                shape = (1,)
        else:
            data = self.read_value(dtype, count=n_elem)
            if is_complex:
                data = data + 1j * self.read_value(dtype, count=n_elem)
        return self.squeeze(data.reshape(shape, order="F"))

    def pack_array(self, array):
        """
        Serialize an np.ndarray into bytes.  Scalars are encoded with ndim=0.
        """
        blob = (
            b"A"
            + np.uint64(array.ndim).tobytes()
            + np.array(array.shape, dtype=np.uint64).tobytes()
        )
        is_complex = np.iscomplexobj(array)
        if is_complex:
            array, imaginary = np.real(array), np.imag(array)
        type_id = (
            rev_class_id[array.dtype]
            if array.dtype.char != "U"
            else rev_class_id[np.dtype("O")]
        )
        if dtype_list[type_id] is None:
            raise DataJointError("Type %s is ambiguous or unknown" % array.dtype)

        blob += np.array([type_id, is_complex], dtype=np.uint32).tobytes()
        if type_names[type_id] == "mxVOID_CLASS":  # array of dtype('O')
            blob += b"".join(
                len_u64(it) + it
                for it in (self.pack_blob(e) for e in array.flatten(order="F"))
            )
            self.set_dj0()  # not supported by original mym
        elif type_names[type_id] == "mxCHAR_CLASS":  # array of dtype('c')
            blob += (
                array.view(np.uint8).astype(np.uint16).tobytes()
            )  # convert to 16-bit chars for MATLAB
        else:  # numeric arrays
            if array.ndim == 0:  # not supported by original mym
                self.set_dj0()
            blob += array.tobytes(order="F")
            if is_complex:
                blob += imaginary.tobytes(order="F")
        return blob

    def read_recarray(self):
        """
        Serialize an np.ndarray with fields, including recarrays
        """
        n_fields = self.read_value("uint32")
        if not n_fields:
            return np.array(None)  # empty array
        field_names = [self.read_zero_terminated_string() for _ in range(n_fields)]
        arrays = [self.read_blob() for _ in range(n_fields)]
        rec = np.empty(
            arrays[0].shape,
            np.dtype([(f, t.dtype) for f, t in zip(field_names, arrays)]),
        )
        for f, t in zip(field_names, arrays):
            rec[f] = t
        return rec.view(np.recarray)

    def pack_recarray(self, array):
        """Serialize a Matlab struct array"""
        return (
            b"F"
            + len_u32(array.dtype)
            + "\0".join(array.dtype.names).encode()  # number of fields
            + b"\0"
            + b"".join(  # field names
                self.pack_recarray(array[f])
                if array[f].dtype.fields
                else self.pack_array(array[f])
                for f in array.dtype.names
            )
        )

    def read_sparse_array(self):
        raise DataJointError(
            "datajoint-python does not yet support sparse arrays. Issue (#590)"
        )

    def read_int(self):
        return int.from_bytes(
            self.read_binary(self.read_value("uint16")), byteorder="little", signed=True
        )

    @staticmethod
    def pack_int(v):
        n_bytes = v.bit_length() // 8 + 1
        assert 0 < n_bytes <= 0xFFFF, "Integers are limited to 65535 bytes"
        return (
            b"\x0a"
            + np.uint16(n_bytes).tobytes()
            + v.to_bytes(n_bytes, byteorder="little", signed=True)
        )

    def read_bool(self):
        return bool(self.read_value("bool"))

    @staticmethod
    def pack_bool(v):
        return b"\x0b" + np.array(v, dtype="bool").tobytes()

    def read_complex(self):
        return complex(self.read_value("complex128"))

    @staticmethod
    def pack_complex(v):
        return b"\x0c" + np.array(v, dtype="complex128").tobytes()

    def read_float(self):
        return float(self.read_value("float64"))

    @staticmethod
    def pack_float(v):
        return b"\x0d" + np.array(v, dtype="float64").tobytes()

    def read_decimal(self):
        return Decimal(self.read_string())

    @staticmethod
    def pack_decimal(d):
        s = str(d)
        return b"d" + len_u64(s) + s.encode()

    def read_string(self):
        return self.read_binary(self.read_value()).decode()

    @staticmethod
    def pack_string(s):
        blob = s.encode()
        return b"\5" + len_u64(blob) + blob

    def read_bytes(self):
        return self.read_binary(self.read_value())

    @staticmethod
    def pack_bytes(s):
        return b"\6" + len_u64(s) + s

    def read_none(self):
        pass

    @staticmethod
    def pack_none():
        return b"\xFF"

    def read_tuple(self):
        return tuple(
            self.read_blob(self.read_value()) for _ in range(self.read_value())
        )

    def pack_tuple(self, t):
        return (
            b"\1"
            + len_u64(t)
            + b"".join(len_u64(it) + it for it in (self.pack_blob(i) for i in t))
        )

    def read_list(self):
        return list(self.read_blob(self.read_value()) for _ in range(self.read_value()))

    def pack_list(self, t):
        return (
            b"\2"
            + len_u64(t)
            + b"".join(len_u64(it) + it for it in (self.pack_blob(i) for i in t))
        )

    def read_set(self):
        return set(self.read_blob(self.read_value()) for _ in range(self.read_value()))

    def pack_set(self, t):
        return (
            b"\3"
            + len_u64(t)
            + b"".join(len_u64(it) + it for it in (self.pack_blob(i) for i in t))
        )

    def read_dict(self):
        return dict(
            (self.read_blob(self.read_value()), self.read_blob(self.read_value()))
            for _ in range(self.read_value())
        )

    def pack_dict(self, d):
        return (
            b"\4"
            + len_u64(d)
            + b"".join(
                b"".join((len_u64(it) + it) for it in packed)
                for packed in (map(self.pack_blob, pair) for pair in d.items())
            )
        )

    def read_struct(self):
        """deserialize matlab stuct"""
        n_dims = self.read_value()
        shape = self.read_value(count=n_dims)
        n_elem = np.prod(shape, dtype=int)
        n_fields = self.read_value("uint32")
        if not n_fields:
            return np.array(None)  # empty array
        field_names = [self.read_zero_terminated_string() for _ in range(n_fields)]
        raw_data = [
            tuple(
                self.read_blob(n_bytes=int(self.read_value())) for _ in range(n_fields)
            )
            for __ in range(n_elem)
        ]
        data = np.array(raw_data, dtype=list(zip(field_names, repeat(object))))
        return self.squeeze(
            data.reshape(shape, order="F"), convert_to_scalar=False
        ).view(MatStruct)

    def pack_struct(self, array):
        """Serialize a Matlab struct array"""
        return (
            b"S"
            + np.array((array.ndim,) + array.shape, dtype=np.uint64).tobytes()
            + len_u32(array.dtype.names)  # dimensionality
            + "\0".join(array.dtype.names).encode()  # number of fields
            + b"\0"
            + b"".join(  # field names
                len_u64(it) + it
                for it in (
                    self.pack_blob(e) for rec in array.flatten(order="F") for e in rec
                )
            )
        )  # values

    def read_cell_array(self):
        """deserialize MATLAB cell array"""
        n_dims = self.read_value()
        shape = self.read_value(count=n_dims)
        n_elem = int(np.prod(shape))
        result = [self.read_blob(n_bytes=self.read_value()) for _ in range(n_elem)]
        return (
            self.squeeze(
                np.array(result).reshape(shape, order="F"), convert_to_scalar=False
            )
        ).view(MatCell)

    def pack_cell_array(self, array):
        return (
            b"C"
            + np.array((array.ndim,) + array.shape, dtype=np.uint64).tobytes()
            + b"".join(
                len_u64(it) + it
                for it in (self.pack_blob(e) for e in array.flatten(order="F"))
            )
        )

    def read_datetime(self):
        """deserialize datetime.date, .time, or .datetime"""
        date, time = self.read_value("int32"), self.read_value("int64")
        date = (
            datetime.date(year=date // 10000, month=(date // 100) % 100, day=date % 100)
            if date >= 0
            else None
        )
        time = (
            datetime.time(
                hour=(time // 10000000000) % 100,
                minute=(time // 100000000) % 100,
                second=(time // 1000000) % 100,
                microsecond=time % 1000000,
            )
            if time >= 0
            else None
        )
        return time and date and datetime.datetime.combine(date, time) or time or date

    def read_np_datetime(self):
        data = self.read_value()
        return data.astype("datetime64[us]")

    @staticmethod
    def pack_datetime(d):
        if isinstance(d, datetime.datetime):
            date, time = d.date(), d.time()
        elif isinstance(d, datetime.date):
            date, time = d, None
        elif isinstance(d, np.datetime64):
            return b"T" + (d.astype("datetime64[us]")).tobytes()
        else:
            date, time = None, d
        return b"t" + (
            np.int32(
                -1 if date is None else (date.year * 100 + date.month) * 100 + date.day
            ).tobytes()
            + np.int64(
                -1
                if time is None
                else ((time.hour * 100 + time.minute) * 100 + time.second) * 1000000
                + time.microsecond
            ).tobytes()
        )

    def read_uuid(self):
        q = self.read_binary(16)
        return uuid.UUID(bytes=q)

    @staticmethod
    def pack_uuid(obj):
        return b"u" + obj.bytes

    def read_zero_terminated_string(self):
        target = self._blob.find(b"\0", self._pos)
        data = self._blob[self._pos : target].decode()
        self._pos = target + 1
        return data

    def read_value(self, dtype=None, count=1):
        if dtype is None:
            dtype = "uint32" if use_32bit_dims else "uint64"
        data = np.frombuffer(self._blob, dtype=dtype, count=count, offset=self._pos)
        self._pos += data.dtype.itemsize * data.size
        return data[0] if count == 1 else data

    def read_binary(self, size):
        self._pos += int(size)
        return self._blob[self._pos - int(size) : self._pos]

    def pack(self, obj, compress):
        self.protocol = b"mYm\0"  # will be replaced with dj0 if new features are used
        blob = self.pack_blob(
            obj
        )  # this may reset the protocol and must precede protocol evaluation
        blob = self.protocol + blob
        if compress and len(blob) > 1000:
            compressed = b"ZL123\0" + len_u64(blob) + zlib.compress(blob)
            if len(compressed) < len(blob):
                blob = compressed
        return blob


def pack(obj, compress=True):
    if bypass_serialization:
        # provide a way to move blobs quickly without de/serialization
        assert isinstance(obj, bytes) and obj.startswith(
            (b"ZL123\0", b"mYm\0", b"dj0\0")
        )
        return obj
    return Blob().pack(obj, compress=compress)


def unpack(blob, squeeze=False):
    if bypass_serialization:
        # provide a way to move blobs quickly without de/serialization
        assert isinstance(blob, bytes) and blob.startswith(
            (b"ZL123\0", b"mYm\0", b"dj0\0")
        )
        return blob
    if blob is not None:
        return Blob(squeeze=squeeze).unpack(blob)

1	"""
2	(De)serialization methods for basic datatypes and numpy.ndarrays with provisions for mutual
3	compatibility with Matlab-based serialization implemented by mYm.
4	"""
5
6	import zlib	1✔
7	from itertools import repeat	1✔
8	import collections	1✔
9	from decimal import Decimal	1✔
10	import datetime	1✔
11	import uuid	1✔
12	import numpy as np	1✔
13	from .errors import DataJointError	1✔
14	from .settings import config	1✔
15
16
17	mxClassID = dict(	1✔
18	(
19	# see http://www.mathworks.com/help/techdoc/apiref/mxclassid.html
20	("mxUNKNOWN_CLASS", None),
21	("mxCELL_CLASS", None),
22	("mxSTRUCT_CLASS", None),
23	("mxLOGICAL_CLASS", np.dtype("bool")),
24	("mxCHAR_CLASS", np.dtype("c")),
25	("mxVOID_CLASS", np.dtype("O")),
26	("mxDOUBLE_CLASS", np.dtype("float64")),
27	("mxSINGLE_CLASS", np.dtype("float32")),
28	("mxINT8_CLASS", np.dtype("int8")),
29	("mxUINT8_CLASS", np.dtype("uint8")),
30	("mxINT16_CLASS", np.dtype("int16")),
31	("mxUINT16_CLASS", np.dtype("uint16")),
32	("mxINT32_CLASS", np.dtype("int32")),
33	("mxUINT32_CLASS", np.dtype("uint32")),
34	("mxINT64_CLASS", np.dtype("int64")),
35	("mxUINT64_CLASS", np.dtype("uint64")),
36	("mxFUNCTION_CLASS", None),
37	)
38	)
39
40	rev_class_id = {dtype: i for i, dtype in enumerate(mxClassID.values())}	1✔
41	dtype_list = list(mxClassID.values())	1✔
42	type_names = list(mxClassID)	1✔
43
44	compression = {b"ZL123\0": zlib.decompress}	1✔
45
46	bypass_serialization = False # runtime setting to bypass blob (en\|de)code	1✔
47
48	# runtime setting to read integers as 32-bit to read blobs created by the 32-bit
49	# version of the mYm library for MATLAB
50	use_32bit_dims = False	1✔
51
52
53	def len_u64(obj):	1✔
54	return np.uint64(len(obj)).tobytes()	1✔
55
56
57	def len_u32(obj):	1✔
58	return np.uint32(len(obj)).tobytes()	1✔
59
60
61	class MatCell(np.ndarray):	1✔
62	"""a numpy ndarray representing a Matlab cell array"""
63
64	pass	1✔
65
66
67	class MatStruct(np.recarray):	1✔
68	"""numpy.recarray representing a Matlab struct array"""
69
70	pass	1✔
71
72
73	class Blob:	1✔
74	def __init__(self, squeeze=False):	1✔
75	self._squeeze = squeeze	1✔
76	self._blob = None	1✔
77	self._pos = 0	1✔
78	self.protocol = None	1✔
79
80	def set_dj0(self):	1✔
81	if not config.get("enable_python_native_blobs"):	1✔
82	raise DataJointError(	×
83	"""v0.12+ python native blobs disabled.
84	See also: https://github.com/datajoint/datajoint-python#python-native-blobs"""
85	)
86
87	self.protocol = b"dj0\0" # when using new blob features	1✔
88
89	def squeeze(self, array, convert_to_scalar=True):	1✔
90	"""
91	Simplify the input array - squeeze out all singleton dimensions.
92	If convert_to_scalar, then convert zero-dimensional arrays to scalars
93	"""
94	if not self._squeeze:	1✔
95	return array	1✔
96	array = array.squeeze()	1✔
97	return array.item() if array.ndim == 0 and convert_to_scalar else array	1✔
98
99	def unpack(self, blob):	1✔
100	self._blob = blob	1✔
101	try:	1✔
102	# decompress
103	prefix = next(	1✔
104	p for p in compression if self._blob[self._pos :].startswith(p)
105	)
106	except StopIteration:	1✔
107	pass # assume uncompressed but could be unrecognized compression	1✔
108	else:
109	self._pos += len(prefix)	1✔
110	blob_size = self.read_value()	1✔
111	blob = compression[prefix](self._blob[self._pos :])	1✔
112	assert len(blob) == blob_size	1✔
113	self._blob = blob	1✔
114	self._pos = 0	1✔
115	blob_format = self.read_zero_terminated_string()	1✔
116	if blob_format in ("mYm", "dj0"):	1✔
117	return self.read_blob(n_bytes=len(self._blob) - self._pos)	1✔
118
119	def read_blob(self, n_bytes=None):	1✔
120	start = self._pos	1✔
121	data_structure_code = chr(self.read_value("uint8"))	1✔
122	try:	1✔
123	call = {	1✔
124	# MATLAB-compatible, inherited from original mYm
125	"A": self.read_array, # matlab-compatible numeric arrays and scalars with ndim==0
126	"P": self.read_sparse_array, # matlab sparse array -- not supported yet
127	"S": self.read_struct, # matlab struct array
128	"C": self.read_cell_array, # matlab cell array
129	# basic data types
130	"\xFF": self.read_none, # None
131	"\x01": self.read_tuple, # a Sequence (e.g. tuple)
132	"\x02": self.read_list, # a MutableSequence (e.g. list)
133	"\x03": self.read_set, # a Set
134	"\x04": self.read_dict, # a Mapping (e.g. dict)
135	"\x05": self.read_string, # a UTF8-encoded string
136	"\x06": self.read_bytes, # a ByteString
137	"\x0a": self.read_int, # unbounded scalar int
138	"\x0b": self.read_bool, # scalar boolean
139	"\x0c": self.read_complex, # scalar 128-bit complex number
140	"\x0d": self.read_float, # scalar 64-bit float
141	"F": self.read_recarray, # numpy array with fields, including recarrays
142	"d": self.read_decimal, # a decimal
143	"t": self.read_datetime, # date, time, or datetime
144	"T": self.read_np_datetime, # np.datetime64
145	"u": self.read_uuid, # UUID
146	}[data_structure_code]
147	except KeyError:	×
148	raise DataJointError(	×
149	'Unknown data structure code "%s". Upgrade datajoint.'
150	% data_structure_code
151	)
152	v = call()	1✔
153	if n_bytes is not None and self._pos - start != n_bytes:	1✔
154	raise DataJointError("Blob length check failed! Invalid blob")	×
155	return v	1✔
156
157	def pack_blob(self, obj):	1✔
158	# original mYm-based serialization from datajoint-matlab
159	if isinstance(obj, MatCell):	1✔
160	return self.pack_cell_array(obj)	1✔
161	if isinstance(obj, MatStruct):	1✔
162	return self.pack_struct(obj)	1✔
163	if isinstance(obj, np.ndarray) and obj.dtype.fields is None:	1✔
164	return self.pack_array(obj)	1✔
165
166	# blob types in the expanded dj0 blob format
167	self.set_dj0()	1✔
168	if not isinstance(obj, (np.ndarray, np.number)):	1✔
169	# python built-in data types
170	if isinstance(obj, bool):	1✔
171	return self.pack_bool(obj)	1✔
172	if isinstance(obj, int):	1✔
173	return self.pack_int(obj)	1✔
174	if isinstance(obj, complex):	1✔
175	return self.pack_complex(obj)	1✔
176	if isinstance(obj, float):	1✔
177	return self.pack_float(obj)	1✔
178	if isinstance(obj, np.ndarray) and obj.dtype.fields:	1✔
179	return self.pack_recarray(np.array(obj))	1✔
180	if isinstance(obj, np.number):	1✔
181	return self.pack_array(np.array(obj))	1✔
182	if isinstance(obj, (bool, np.bool_)):	1✔
183	return self.pack_array(np.array(obj))	1✔
184	if isinstance(obj, (float, int, complex)):	1✔
185	return self.pack_array(np.array(obj))	×
186	if isinstance(	1✔
187	obj, (datetime.datetime, datetime.date, datetime.time, np.datetime64)
188	):
189	return self.pack_datetime(obj)	1✔
190	if isinstance(obj, Decimal):	1✔
191	return self.pack_decimal(obj)	1✔
192	if isinstance(obj, uuid.UUID):	1✔
193	return self.pack_uuid(obj)	1✔
194	if isinstance(obj, collections.abc.Mapping):	1✔
195	return self.pack_dict(obj)	1✔
196	if isinstance(obj, str):	1✔
197	return self.pack_string(obj)	1✔
198	if isinstance(obj, collections.abc.ByteString):	1✔
199	return self.pack_bytes(obj)	×
200	if isinstance(obj, collections.abc.MutableSequence):	1✔
201	return self.pack_list(obj)	1✔
202	if isinstance(obj, collections.abc.Sequence):	1✔
203	return self.pack_tuple(obj)	1✔
204	if isinstance(obj, collections.abc.Set):	1✔
205	return self.pack_set(obj)	1✔
206	if obj is None:	1✔
207	return self.pack_none()	1✔
208	raise DataJointError(	×
209	"Packing object of type %s currently not supported!" % type(obj)
210	)
211
212	def read_array(self):	1✔
213	n_dims = int(self.read_value())	1✔
214	shape = self.read_value(count=n_dims)	1✔
215	n_elem = np.prod(shape, dtype=int)	1✔
216	dtype_id, is_complex = self.read_value("uint32", 2)	1✔
217	dtype = dtype_list[dtype_id]	1✔
218
219	if type_names[dtype_id] == "mxVOID_CLASS":	1✔
220	data = np.array(	1✔
221	list(self.read_blob(self.read_value()) for _ in range(n_elem)),
222	dtype=np.dtype("O"),
223	)
224	elif type_names[dtype_id] == "mxCHAR_CLASS":	1✔
225	# compensate for MATLAB packing of char arrays
226	data = self.read_value(dtype, count=2 * n_elem)	1✔
227	data = data[::2].astype("U1")	1✔
228	if n_dims == 2 and shape[0] == 1 or n_dims == 1:	1✔
229	compact = data.squeeze()	1✔
230	data = (	1✔
231	compact
232	if compact.shape == ()
233	else np.array("".join(data.squeeze()))
234	)
235	shape = (1,)	1✔
236	else:
237	data = self.read_value(dtype, count=n_elem)	1✔
238	if is_complex:	1✔
239	data = data + 1j * self.read_value(dtype, count=n_elem)	1✔
240	return self.squeeze(data.reshape(shape, order="F"))	1✔
241
242	def pack_array(self, array):	1✔
243	"""
244	Serialize an np.ndarray into bytes. Scalars are encoded with ndim=0.
245	"""
246	blob = (	1✔
247	b"A"
248	+ np.uint64(array.ndim).tobytes()
249	+ np.array(array.shape, dtype=np.uint64).tobytes()
250	)
251	is_complex = np.iscomplexobj(array)	1✔
252	if is_complex:	1✔
253	array, imaginary = np.real(array), np.imag(array)	1✔
254	type_id = (	1✔
255	rev_class_id[array.dtype]
256	if array.dtype.char != "U"
257	else rev_class_id[np.dtype("O")]
258	)
259	if dtype_list[type_id] is None:	1✔
260	raise DataJointError("Type %s is ambiguous or unknown" % array.dtype)	×
261
262	blob += np.array([type_id, is_complex], dtype=np.uint32).tobytes()	1✔
263	if type_names[type_id] == "mxVOID_CLASS": # array of dtype('O')	1✔
264	blob += b"".join(	1✔
265	len_u64(it) + it
266	for it in (self.pack_blob(e) for e in array.flatten(order="F"))
267	)
268	self.set_dj0() # not supported by original mym	1✔
269	elif type_names[type_id] == "mxCHAR_CLASS": # array of dtype('c')	1✔
270	blob += (	×
271	array.view(np.uint8).astype(np.uint16).tobytes()
272	) # convert to 16-bit chars for MATLAB
273	else: # numeric arrays
274	if array.ndim == 0: # not supported by original mym	1✔
275	self.set_dj0()	1✔
276	blob += array.tobytes(order="F")	1✔
277	if is_complex:	1✔
278	blob += imaginary.tobytes(order="F")	1✔
279	return blob	1✔
280
281	def read_recarray(self):	1✔
282	"""
283	Serialize an np.ndarray with fields, including recarrays
284	"""
285	n_fields = self.read_value("uint32")	1✔
286	if not n_fields:	1✔
287	return np.array(None) # empty array	×
288	field_names = [self.read_zero_terminated_string() for _ in range(n_fields)]	1✔
289	arrays = [self.read_blob() for _ in range(n_fields)]	1✔
290	rec = np.empty(	1✔
291	arrays[0].shape,
292	np.dtype([(f, t.dtype) for f, t in zip(field_names, arrays)]),
293	)
294	for f, t in zip(field_names, arrays):	1✔
295	rec[f] = t	1✔
296	return rec.view(np.recarray)	1✔
297
298	def pack_recarray(self, array):	1✔
299	"""Serialize a Matlab struct array"""
300	return (	1✔
301	b"F"
302	+ len_u32(array.dtype)
303	+ "\0".join(array.dtype.names).encode() # number of fields
304	+ b"\0"
305	+ b"".join( # field names
306	self.pack_recarray(array[f])
307	if array[f].dtype.fields
308	else self.pack_array(array[f])
309	for f in array.dtype.names
310	)
311	)
312
313	def read_sparse_array(self):	1✔
314	raise DataJointError(	×
315	"datajoint-python does not yet support sparse arrays. Issue (#590)"
316	)
317
318	def read_int(self):	1✔
319	return int.from_bytes(	1✔
320	self.read_binary(self.read_value("uint16")), byteorder="little", signed=True
321	)
322
323	@staticmethod	1✔
324	def pack_int(v):	1✔
325	n_bytes = v.bit_length() // 8 + 1	1✔
326	assert 0 < n_bytes <= 0xFFFF, "Integers are limited to 65535 bytes"	1✔
327	return (	1✔
328	b"\x0a"
329	+ np.uint16(n_bytes).tobytes()
330	+ v.to_bytes(n_bytes, byteorder="little", signed=True)
331	)
332
333	def read_bool(self):	1✔
334	return bool(self.read_value("bool"))	1✔
335
336	@staticmethod	1✔
337	def pack_bool(v):	1✔
338	return b"\x0b" + np.array(v, dtype="bool").tobytes()	1✔
339
340	def read_complex(self):	1✔
341	return complex(self.read_value("complex128"))	1✔
342
343	@staticmethod	1✔
344	def pack_complex(v):	1✔
345	return b"\x0c" + np.array(v, dtype="complex128").tobytes()	1✔
346
347	def read_float(self):	1✔
348	return float(self.read_value("float64"))	1✔
349
350	@staticmethod	1✔
351	def pack_float(v):	1✔
352	return b"\x0d" + np.array(v, dtype="float64").tobytes()	1✔
353
354	def read_decimal(self):	1✔
355	return Decimal(self.read_string())	1✔
356
357	@staticmethod	1✔
358	def pack_decimal(d):	1✔
359	s = str(d)	1✔
360	return b"d" + len_u64(s) + s.encode()	1✔
361
362	def read_string(self):	1✔
363	return self.read_binary(self.read_value()).decode()	1✔
364
365	@staticmethod	1✔
366	def pack_string(s):	1✔
367	blob = s.encode()	1✔
368	return b"\5" + len_u64(blob) + blob	1✔
369
370	def read_bytes(self):	1✔
371	return self.read_binary(self.read_value())	×
372
373	@staticmethod	1✔
374	def pack_bytes(s):	1✔
375	return b"\6" + len_u64(s) + s	×
376
377	def read_none(self):	1✔
378	pass	1✔
379
380	@staticmethod	1✔
381	def pack_none():	1✔
382	return b"\xFF"	1✔
383
384	def read_tuple(self):	1✔
385	return tuple(	1✔
386	self.read_blob(self.read_value()) for _ in range(self.read_value())
387	)
388
389	def pack_tuple(self, t):	1✔
390	return (	1✔
391	b"\1"
392	+ len_u64(t)
393	+ b"".join(len_u64(it) + it for it in (self.pack_blob(i) for i in t))
394	)
395
396	def read_list(self):	1✔
397	return list(self.read_blob(self.read_value()) for _ in range(self.read_value()))	1✔
398
399	def pack_list(self, t):	1✔
400	return (	1✔
401	b"\2"
402	+ len_u64(t)
403	+ b"".join(len_u64(it) + it for it in (self.pack_blob(i) for i in t))
404	)
405
406	def read_set(self):	1✔
407	return set(self.read_blob(self.read_value()) for _ in range(self.read_value()))	1✔
408
409	def pack_set(self, t):	1✔
410	return (	1✔
411	b"\3"
412	+ len_u64(t)
413	+ b"".join(len_u64(it) + it for it in (self.pack_blob(i) for i in t))
414	)
415
416	def read_dict(self):	1✔
417	return dict(	1✔
418	(self.read_blob(self.read_value()), self.read_blob(self.read_value()))
419	for _ in range(self.read_value())
420	)
421
422	def pack_dict(self, d):	1✔
423	return (	1✔
424	b"\4"
425	+ len_u64(d)
426	+ b"".join(
427	b"".join((len_u64(it) + it) for it in packed)
428	for packed in (map(self.pack_blob, pair) for pair in d.items())
429	)
430	)
431
432	def read_struct(self):	1✔
433	"""deserialize matlab stuct"""
434	n_dims = self.read_value()	1✔
435	shape = self.read_value(count=n_dims)	1✔
436	n_elem = np.prod(shape, dtype=int)	1✔
437	n_fields = self.read_value("uint32")	1✔
438	if not n_fields:	1✔
439	return np.array(None) # empty array	×
440	field_names = [self.read_zero_terminated_string() for _ in range(n_fields)]	1✔
441	raw_data = [	1✔
442	tuple(
443	self.read_blob(n_bytes=int(self.read_value())) for _ in range(n_fields)
444	)
445	for __ in range(n_elem)
446	]
447	data = np.array(raw_data, dtype=list(zip(field_names, repeat(object))))	1✔
448	return self.squeeze(	1✔
449	data.reshape(shape, order="F"), convert_to_scalar=False
450	).view(MatStruct)
451
452	def pack_struct(self, array):	1✔
453	"""Serialize a Matlab struct array"""
454	return (	1✔
455	b"S"
456	+ np.array((array.ndim,) + array.shape, dtype=np.uint64).tobytes()
457	+ len_u32(array.dtype.names) # dimensionality
458	+ "\0".join(array.dtype.names).encode() # number of fields
459	+ b"\0"
460	+ b"".join( # field names
461	len_u64(it) + it
462	for it in (
463	self.pack_blob(e) for rec in array.flatten(order="F") for e in rec
464	)
465	)
466	) # values
467
468	def read_cell_array(self):	1✔
469	"""deserialize MATLAB cell array"""
470	n_dims = self.read_value()	1✔
471	shape = self.read_value(count=n_dims)	1✔
472	n_elem = int(np.prod(shape))	1✔
473	result = [self.read_blob(n_bytes=self.read_value()) for _ in range(n_elem)]	1✔
474	return (	1✔
475	self.squeeze(
476	np.array(result).reshape(shape, order="F"), convert_to_scalar=False
477	)
478	).view(MatCell)
479
480	def pack_cell_array(self, array):	1✔
481	return (	1✔
482	b"C"
483	+ np.array((array.ndim,) + array.shape, dtype=np.uint64).tobytes()
484	+ b"".join(
485	len_u64(it) + it
486	for it in (self.pack_blob(e) for e in array.flatten(order="F"))
487	)
488	)
489
490	def read_datetime(self):	1✔
491	"""deserialize datetime.date, .time, or .datetime"""
492	date, time = self.read_value("int32"), self.read_value("int64")	1✔
493	date = (	1✔
494	datetime.date(year=date // 10000, month=(date // 100) % 100, day=date % 100)
495	if date >= 0
496	else None
497	)
498	time = (	1✔
499	datetime.time(
500	hour=(time // 10000000000) % 100,
501	minute=(time // 100000000) % 100,
502	second=(time // 1000000) % 100,
503	microsecond=time % 1000000,
504	)
505	if time >= 0
506	else None
507	)
508	return time and date and datetime.datetime.combine(date, time) or time or date	1✔
509
510	def read_np_datetime(self):	1✔
511	data = self.read_value()	1✔
512	return data.astype("datetime64[us]")	1✔
513
514	@staticmethod	1✔
515	def pack_datetime(d):	1✔
516	if isinstance(d, datetime.datetime):	1✔
517	date, time = d.date(), d.time()	1✔
518	elif isinstance(d, datetime.date):	1✔
519	date, time = d, None	1✔
520	elif isinstance(d, np.datetime64):	1✔
521	return b"T" + (d.astype("datetime64[us]")).tobytes()	1✔
522	else:
523	date, time = None, d	×
524	return b"t" + (	1✔
525	np.int32(
526	-1 if date is None else (date.year * 100 + date.month) * 100 + date.day
527	).tobytes()
528	+ np.int64(
529	-1
530	if time is None
531	else ((time.hour * 100 + time.minute) * 100 + time.second) * 1000000
532	+ time.microsecond
533	).tobytes()
534	)
535
536	def read_uuid(self):	1✔
537	q = self.read_binary(16)	1✔
538	return uuid.UUID(bytes=q)	1✔
539
540	@staticmethod	1✔
541	def pack_uuid(obj):	1✔
542	return b"u" + obj.bytes	1✔
543
544	def read_zero_terminated_string(self):	1✔
545	target = self._blob.find(b"\0", self._pos)	1✔
546	data = self._blob[self._pos : target].decode()	1✔
547	self._pos = target + 1	1✔
548	return data	1✔
549
550	def read_value(self, dtype=None, count=1):	1✔
551	if dtype is None:	1✔
552	dtype = "uint32" if use_32bit_dims else "uint64"	1✔
553	data = np.frombuffer(self._blob, dtype=dtype, count=count, offset=self._pos)	1✔
554	self._pos += data.dtype.itemsize * data.size	1✔
555	return data[0] if count == 1 else data	1✔
556
557	def read_binary(self, size):	1✔
558	self._pos += int(size)	1✔
559	return self._blob[self._pos - int(size) : self._pos]	1✔
560
561	def pack(self, obj, compress):	1✔
562	self.protocol = b"mYm\0" # will be replaced with dj0 if new features are used	1✔
563	blob = self.pack_blob(	1✔
564	obj
565	) # this may reset the protocol and must precede protocol evaluation
566	blob = self.protocol + blob	1✔
567	if compress and len(blob) > 1000:	1✔
568	compressed = b"ZL123\0" + len_u64(blob) + zlib.compress(blob)	1✔
569	if len(compressed) < len(blob):	1✔
570	blob = compressed	1✔
571	return blob	1✔
572
573
574	def pack(obj, compress=True):	1✔
575	if bypass_serialization:	1✔
576	# provide a way to move blobs quickly without de/serialization
577	assert isinstance(obj, bytes) and obj.startswith(	1✔
578	(b"ZL123\0", b"mYm\0", b"dj0\0")
579	)
580	return obj	1✔
581	return Blob().pack(obj, compress=compress)	1✔
582
583
584	def unpack(blob, squeeze=False):	1✔
585	if bypass_serialization:	1✔
586	# provide a way to move blobs quickly without de/serialization
587	assert isinstance(blob, bytes) and blob.startswith(	1✔
588	(b"ZL123\0", b"mYm\0", b"dj0\0")
589	)
590	return blob	1✔
591	if blob is not None:	1✔
592	return Blob(squeeze=squeeze).unpack(blob)	1✔

datajoint / datajoint-python / #12496

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous