#12534

pending completion

Build # #12534

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/12a82730b85cfc0b7c139b93009b47802cb72c7a">12a82730b<a href="https://github.com/datajoint/datajoint-python/commit/12a82730b85cfc0b7c139b93009b47802cb72c7a">">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/42b0537dbc350a6a73dc4365d7cf0f7716bd3267">42b0537db</a>">42b0537db</a><a href="https://github.com/datajoint/datajoint-python/commit/12a82730b85cfc0b7c139b93009b47802cb72c7a"> into 07c5553e4">07c5553e4</a>

Run Details

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

3094 of 3420 relevant lines covered (90.47%)

0.9 hits per line

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

95.03

/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


deserialize_lookup = {
    0: {"dtype": None, "scalar_type": "UNKNOWN"},
    1: {"dtype": None, "scalar_type": "CELL"},
    2: {"dtype": None, "scalar_type": "STRUCT"},
    3: {"dtype": np.dtype("bool"), "scalar_type": "LOGICAL"},
    4: {"dtype": np.dtype("c"), "scalar_type": "CHAR"},
    5: {"dtype": np.dtype("O"), "scalar_type": "VOID"},
    6: {"dtype": np.dtype("float64"), "scalar_type": "DOUBLE"},
    7: {"dtype": np.dtype("float32"), "scalar_type": "SINGLE"},
    8: {"dtype": np.dtype("int8"), "scalar_type": "INT8"},
    9: {"dtype": np.dtype("uint8"), "scalar_type": "UINT8"},
    10: {"dtype": np.dtype("int16"), "scalar_type": "INT16"},
    11: {"dtype": np.dtype("uint16"), "scalar_type": "UINT16"},
    12: {"dtype": np.dtype("int32"), "scalar_type": "INT32"},
    13: {"dtype": np.dtype("uint32"), "scalar_type": "UINT32"},
    14: {"dtype": np.dtype("int64"), "scalar_type": "INT64"},
    15: {"dtype": np.dtype("uint64"), "scalar_type": "UINT64"},
    16: {"dtype": None, "scalar_type": "FUNCTION"},
    128: {"dtype": np.dtype("<M8[Y]"), "scalar_type": "DATETIME64[Y]"},
    129: {"dtype": np.dtype("<M8[M]"), "scalar_type": "DATETIME64[M]"},
    130: {"dtype": np.dtype("<M8[W]"), "scalar_type": "DATETIME64[W]"},
    131: {"dtype": np.dtype("<M8[D]"), "scalar_type": "DATETIME64[D]"},
    132: {"dtype": np.dtype("<M8[h]"), "scalar_type": "DATETIME64[h]"},
    133: {"dtype": np.dtype("<M8[m]"), "scalar_type": "DATETIME64[m]"},
    134: {"dtype": np.dtype("<M8[s]"), "scalar_type": "DATETIME64[s]"},
    135: {"dtype": np.dtype("<M8[ms]"), "scalar_type": "DATETIME64[ms]"},
    136: {"dtype": np.dtype("<M8[us]"), "scalar_type": "DATETIME64[us]"},
    137: {"dtype": np.dtype("<M8[ps]"), "scalar_type": "DATETIME64[ps]"},
    138: {"dtype": np.dtype("<M8[fs]"), "scalar_type": "DATETIME64[fs]"},
    139: {"dtype": np.dtype("<M8[as]"), "scalar_type": "DATETIME64[as]"},
}
serialize_lookup = {
    v["dtype"]: {"type_id": k, "scalar_type": v["scalar_type"]}
    for k, v in deserialize_lookup.items()
    if v["dtype"] is not None
}


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
                "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, np.datetime64)):
            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)):
            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 = deserialize_lookup[dtype_id]["dtype"]

        # Check if name is void
        if deserialize_lookup[dtype_id]["scalar_type"] == "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 deserialize_lookup[dtype_id]["scalar_type"] == "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:
            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 = serialize_lookup[array.dtype]["type_id"]
        except KeyError:
            if array.dtype.char == "U":
                type_id = serialize_lookup[np.dtype("O")]["type_id"]
                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 serialize_lookup[array.dtype]["scalar_type"] == "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
        # array of dtype('c')
        elif serialize_lookup[array.dtype]["scalar_type"] == "CHAR":
            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

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

datajoint / datajoint-python / #12534

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