#12544

pending completion

Build # #12544

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/e6ced59b92cd3b1b7d237e6aa66de928b23e60b4">e6ced59b9<a href="https://github.com/datajoint/datajoint-python/commit/e6ced59b92cd3b1b7d237e6aa66de928b23e60b4">">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/ef243e6afc04725b22fb6a1ec1b1a4b4374dafe1">ef243e6af</a>">ef243e6af</a><a href="https://github.com/datajoint/datajoint-python/commit/e6ced59b92cd3b1b7d237e6aa66de928b23e60b4"> into 07c5553e4">07c5553e4</a>

Run Details

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

3093 of 3419 relevant lines covered (90.47%)

0.9 hits per line

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

95.02

/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"},
    65_536: {"dtype": np.dtype("datetime64[Y]"), "scalar_type": "DATETIME64[Y]"},
    65_537: {"dtype": np.dtype("datetime64[M]"), "scalar_type": "DATETIME64[M]"},
    65_538: {"dtype": np.dtype("datetime64[W]"), "scalar_type": "DATETIME64[W]"},
    65_539: {"dtype": np.dtype("datetime64[D]"), "scalar_type": "DATETIME64[D]"},
    65_540: {"dtype": np.dtype("datetime64[h]"), "scalar_type": "DATETIME64[h]"},
    65_541: {"dtype": np.dtype("datetime64[m]"), "scalar_type": "DATETIME64[m]"},
    65_542: {"dtype": np.dtype("datetime64[s]"), "scalar_type": "DATETIME64[s]"},
    65_543: {"dtype": np.dtype("datetime64[ms]"), "scalar_type": "DATETIME64[ms]"},
    65_544: {"dtype": np.dtype("datetime64[us]"), "scalar_type": "DATETIME64[us]"},
    65_545: {"dtype": np.dtype("datetime64[ns]"), "scalar_type": "DATETIME64[ns]"},
    65_546: {"dtype": np.dtype("datetime64[ps]"), "scalar_type": "DATETIME64[ps]"},
    65_547: {"dtype": np.dtype("datetime64[fs]"), "scalar_type": "DATETIME64[fs]"},
    65_548: {"dtype": np.dtype("datetime64[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:
            # U is for unicode string
            if array.dtype.char == "U":
                type_id = serialize_lookup[np.dtype("O")]["type_id"]
            else:
                raise DataJointError(f"Type {array.dtype} is ambiguous or unknown")

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

datajoint / datajoint-python / #12544

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