#12513

pending completion

Build # #12513

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/34da51cbc1fdbe332dc6cea1e6f55dadee5bd96c">34da51cbc<a href="https://github.com/datajoint/datajoint-python/commit/34da51cbc1fdbe332dc6cea1e6f55dadee5bd96c">">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/c2cc50bbcfd93b3e8df1c153c68674cb486c43c9">c2cc50bbc</a>">c2cc50bbc</a><a href="https://github.com/datajoint/datajoint-python/commit/34da51cbc1fdbe332dc6cea1e6f55dadee5bd96c"> into 07c5553e4">07c5553e4</a>

Run Details

17 of 17 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


scalar_id = dict(
    (
        # see http://www.mathworks.com/help/techdoc/apiref/mxclassid.html
        ("UNKNOWN", None),
        ("CELL", None),
        ("STRUCT", None),
        ("LOGICAL", np.dtype("bool")),
        ("CHAR", np.dtype("c")),
        ("VOID", np.dtype("O")),
        ("DOUBLE", np.dtype("float64")),
        ("SINGLE", np.dtype("float32")),
        ("INT8", np.dtype("int8")),
        ("UINT8", np.dtype("uint8")),
        ("INT16", np.dtype("int16")),
        ("UINT16", np.dtype("uint16")),
        ("INT32", np.dtype("int32")),
        ("UINT32", np.dtype("uint32")),
        ("INT64", np.dtype("int64")),
        ("UINT64", np.dtype("uint64")),
        ("FUNCTION", None),
        ("DATETIME64", np.dtype("<M8[us]")),
    )
)

# Matlab numeric codes
matlab_scalar_mapping = {
    np.dtype("bool"): 3,  # LOGICAL
    np.dtype("c"): 4,  # CHAR
    np.dtype("O"): 5,  # VOID
    np.dtype("float64"): 6,  # DOUBLE
    np.dtype("float32"): 7,  # SINGLE
    np.dtype("int8"): 8,  # INT8
    np.dtype("uint8"): 9,  # UINT8
    np.dtype("int16"): 10,  # INT16
    np.dtype("uint16"): 11,  # UINT16
    np.dtype("int32"): 12,  # INT32
    np.dtype("uint32"): 13,  # UINT32
    np.dtype("int64"): 14,  # INT64
    np.dtype("uint64"): 15,  # UINT64
}

dtype_list = list(scalar_id.values())
type_names = list(scalar_id)

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

datajoint / datajoint-python / #12513

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