#12508

pending completion

Build # #12508

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/4bbedaabe9dc0ab0f1d0df6032e26275788aa824">4bbedaabe<a href="https://github.com/datajoint/datajoint-python/commit/4bbedaabe9dc0ab0f1d0df6032e26275788aa824">">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/ae7b93cba495f76fd3f567d7c436565013f1141d">ae7b93cba</a>">ae7b93cba</a><a href="https://github.com/datajoint/datajoint-python/commit/4bbedaabe9dc0ab0f1d0df6032e26275788aa824"> into 07c5553e4">07c5553e4</a>

Run Details

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

3025 of 3353 relevant lines covered (90.22%)

0.9 hits per line

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

94.93

/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),
        ("NON_mx_DATETIME64", np.dtype("<M8[us]")),
    )
)

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

datajoint / datajoint-python / #12508

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