#12529

pending completion

Build # #12529

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/f7bbdac5eaf9aab624e537ba53c6d0ad0b32acf1">f7bbdac5e<a href="https://github.com/datajoint/datajoint-python/commit/f7bbdac5eaf9aab624e537ba53c6d0ad0b32acf1">">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/fd24525a3b78697514ca547af9ea33b67f963970">fd24525a3</a>">fd24525a3</a><a href="https://github.com/datajoint/datajoint-python/commit/f7bbdac5eaf9aab624e537ba53c6d0ad0b32acf1"> into 07c5553e4">07c5553e4</a>

Run Details

26 of 26 new or added lines in 2 files covered. (100.0%)

3029 of 3355 relevant lines covered (90.28%)

0.9 hits per line

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

95.0

/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_type = 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]")),
    )
)

scalar_codes = {
    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
    np.dtype(
        "<M8[us]"
    ): 50,  # Datetime[us], skipped to 50 to accommodate more matlab types
}

# Lookup dict for quickly getting a scalar type from its code
scalar_code_lookup = dict((v, k) for k, v in scalar_codes.items())
# Lookup dict for quickly getting a scalar name from its type
scalar_name_lookup = dict((v, k) for k, v in scalar_type.items())


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)

        # Get dtype from type id
        dtype = scalar_code_lookup[dtype_id]

        # Check if name is void
        if scalar_name_lookup[dtype] == "VOID":
            data = np.array(
                list(self.read_blob(self.read_value()) for _ in range(n_elem)),
                dtype=np.dtype("O"),
            )
        # Check if name is char
        elif scalar_name_lookup[dtype] == "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)
        try:
            type_id = scalar_codes[array.dtype]
        except KeyError:
            if array.dtype.char == "U":
                type_id = scalar_codes[np.dtype("O")]
                pass
            else:
                raise DataJointError("Type %s is ambiguous or unknown" % array.dtype)

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

datajoint / datajoint-python / #12529

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