#12501

pending completion

Build # #12501

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/2d7cae808475b68099d321fbbf186e6a0fc86bc0">2d7cae808<a href="https://github.com/datajoint/datajoint-python/commit/2d7cae808475b68099d321fbbf186e6a0fc86bc0">">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/5a57b89fa19c68cfb2ed8cb8dbbfb3e6721377a4">5a57b89fa</a>">5a57b89fa</a><a href="https://github.com/datajoint/datajoint-python/commit/2d7cae808475b68099d321fbbf186e6a0fc86bc0"> into 07c5553e4">07c5553e4</a>

Run Details

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

3097 of 3429 relevant lines covered (90.32%)

0.9 hits per line

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

93.85

/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_int64_datetime,  # np.datetime64
                "Z": self.read_array_dt64,
                "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:
            self.set_dj0()
            return b"Z" + array.astype("datetime64[us]").tobytes()
        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_array_dt64(self):
        data = self.read_value(dtype="<M8[us]", count=-1)
        return data

    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	)
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_int64_datetime, # np.datetime64
145	"Z": self.read_array_dt64,
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	self.set_dj0()	×
249	return b"Z" + array.astype("datetime64[us]").tobytes()	×
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):	1✔
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):	1✔
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):	1✔
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):	1✔
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):	1✔
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):	1✔
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):	1✔
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():	1✔
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_array_dt64(self):	1✔
437	data = self.read_value(dtype="<M8[us]", count=-1)	×
438	return data	×
439
440	def read_struct(self):	1✔
441	"""deserialize matlab stuct"""
442	n_dims = self.read_value()	1✔
443	shape = self.read_value(count=n_dims)	1✔
444	n_elem = np.prod(shape, dtype=int)	1✔
445	n_fields = self.read_value("uint32")	1✔
446	if not n_fields:	1✔
447	return np.array(None) # empty array	×
448	field_names = [self.read_zero_terminated_string() for _ in range(n_fields)]	1✔
449	raw_data = [	1✔
450	tuple(
451	self.read_blob(n_bytes=int(self.read_value())) for _ in range(n_fields)
452	)
453	for __ in range(n_elem)
454	]
455	data = np.array(raw_data, dtype=list(zip(field_names, repeat(object))))	1✔
456	return self.squeeze(	1✔
457	data.reshape(shape, order="F"), convert_to_scalar=False
458	).view(MatStruct)
459
460	def pack_struct(self, array):	1✔
461	"""Serialize a Matlab struct array"""
462	return (	1✔
463	b"S"
464	+ np.array((array.ndim,) + array.shape, dtype=np.uint64).tobytes()
465	+ len_u32(array.dtype.names) # dimensionality
466	+ "\0".join(array.dtype.names).encode() # number of fields
467	+ b"\0"
468	+ b"".join( # field names
469	len_u64(it) + it
470	for it in (
471	self.pack_blob(e) for rec in array.flatten(order="F") for e in rec
472	)
473	)
474	) # values
475
476	def read_cell_array(self):	1✔
477	"""deserialize MATLAB cell array"""
478	n_dims = self.read_value()	1✔
479	shape = self.read_value(count=n_dims)	1✔
480	n_elem = int(np.prod(shape))	1✔
481	result = [self.read_blob(n_bytes=self.read_value()) for _ in range(n_elem)]	1✔
482	return (	1✔
483	self.squeeze(
484	np.array(result).reshape(shape, order="F"), convert_to_scalar=False
485	)
486	).view(MatCell)
487
488	def pack_cell_array(self, array):	1✔
489	return (	1✔
490	b"C"
491	+ np.array((array.ndim,) + array.shape, dtype=np.uint64).tobytes()
492	+ b"".join(
493	len_u64(it) + it
494	for it in (self.pack_blob(e) for e in array.flatten(order="F"))
495	)
496	)
497
498	def read_datetime(self):	1✔
499	"""deserialize datetime.date, .time, or .datetime"""
500	date, time = self.read_value("int32"), self.read_value("int64")	1✔
501	date = (	1✔
502	datetime.date(year=date // 10000, month=(date // 100) % 100, day=date % 100)
503	if date >= 0
504	else None
505	)
506	time = (	1✔
507	datetime.time(
508	hour=(time // 10000000000) % 100,
509	minute=(time // 100000000) % 100,
510	second=(time // 1000000) % 100,
511	microsecond=time % 1000000,
512	)
513	if time >= 0
514	else None
515	)
516	return time and date and datetime.datetime.combine(date, time) or time or date	1✔
517
518	def read_int64_datetime(self):	1✔
519	data = self.read_value()	1✔
520	return data.astype("datetime64[us]")	1✔
521
522	@staticmethod	1✔
523	def pack_datetime(d):	1✔
524	if isinstance(d, datetime.datetime):	1✔
525	date, time = d.date(), d.time()	1✔
526	elif isinstance(d, datetime.date):	1✔
527	date, time = d, None	1✔
528	elif isinstance(d, np.datetime64):	1✔
529	return b"T" + (d.astype("datetime64[us]")).tobytes()	1✔
530	else:
531	date, time = None, d	×
532	return b"t" + (	1✔
533	np.int32(
534	-1 if date is None else (date.year * 100 + date.month) * 100 + date.day
535	).tobytes()
536	+ np.int64(
537	-1
538	if time is None
539	else ((time.hour * 100 + time.minute) * 100 + time.second) * 1000000
540	+ time.microsecond
541	).tobytes()
542	)
543
544	def read_uuid(self):	1✔
545	q = self.read_binary(16)	1✔
546	return uuid.UUID(bytes=q)	1✔
547
548	@staticmethod	1✔
549	def pack_uuid(obj):	1✔
550	return b"u" + obj.bytes	1✔
551
552	def read_zero_terminated_string(self):	1✔
553	target = self._blob.find(b"\0", self._pos)	1✔
554	data = self._blob[self._pos : target].decode()	1✔
555	self._pos = target + 1	1✔
556	return data	1✔
557
558	def read_value(self, dtype=None, count=1):	1✔
559	if dtype is None:	1✔
560	dtype = "uint32" if use_32bit_dims else "uint64"	1✔
561	data = np.frombuffer(self._blob, dtype=dtype, count=count, offset=self._pos)	1✔
562	self._pos += data.dtype.itemsize * data.size	1✔
563	return data[0] if count == 1 else data	1✔
564
565	def read_binary(self, size):	1✔
566	self._pos += int(size)	1✔
567	return self._blob[self._pos - int(size) : self._pos]	1✔
568
569	def pack(self, obj, compress):	1✔
570	self.protocol = b"mYm\0" # will be replaced with dj0 if new features are used	1✔
571	blob = self.pack_blob(	1✔
572	obj
573	) # this may reset the protocol and must precede protocol evaluation
574	blob = self.protocol + blob	1✔
575	if compress and len(blob) > 1000:	1✔
576	compressed = b"ZL123\0" + len_u64(blob) + zlib.compress(blob)	1✔
577	if len(compressed) < len(blob):	1✔
578	blob = compressed	1✔
579	return blob	1✔
580
581
582	def pack(obj, compress=True):	1✔
583	if bypass_serialization:	1✔
584	# provide a way to move blobs quickly without de/serialization
585	assert isinstance(obj, bytes) and obj.startswith(	1✔
586	(b"ZL123\0", b"mYm\0", b"dj0\0")
587	)
588	return obj	1✔
589	return Blob().pack(obj, compress=compress)	1✔
590
591
592	def unpack(blob, squeeze=False):	1✔
593	if bypass_serialization:	1✔
594	# provide a way to move blobs quickly without de/serialization
595	assert isinstance(blob, bytes) and blob.startswith(	1✔
596	(b"ZL123\0", b"mYm\0", b"dj0\0")
597	)
598	return blob	1✔
599	if blob is not None:	1✔
600	return Blob(squeeze=squeeze).unpack(blob)	1✔

datajoint / datajoint-python / #12501

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