#12106

pending completion

Build # #12106

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/<a class="double-link" href="https://git"><a class=hub.com/datajoint/datajoint-python/commit/<a class="double-link" href="https://git"><a class=hub.com/datajoint/datajoint-python/commit/<a class="double-link" href="https://git"><a class=hub.com/datajoint/datajoint-python/commit/<a class="double-link" href="https://git"><a class=hub.com/datajoint/datajoint-python/commit/b61b69485dc48b5d17489728a600b42b11e1d272">b61b69485">&lt;a href=&quot;https://github.com/datajoint/datajoint-python/commit/</a><a class="double-link" href="https://github.com/datajoint/datajoint-python/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/datajoint/datajoint-python/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/datajoint/datajoint-python/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/datajoint/datajoint-python/commit/b61b69485dc48b5d17489728a600b42b11e1d272">b61b69485</a><a href="https://github.com/datajoint/datajoint-python/commit/b61b69485dc48b5d17489728a600b42b11e1d272">&lt;a href=&quot;https://github.com/datajoint/datajoint-python/commit/b61b69485dc48b5d17489728a600b42b11e1d272&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/datajoint/datajoint-python/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/datajoint/datajoint-python/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;https://git&amp;quot;&amp;gt;&amp;amp;lt;a class=&amp;lt;/a&amp;gt;hub.com/datajoint/datajoint-python/commit/b61b69485dc48b5d17489728a600b42b11e1d272&amp;quot;&amp;gt;b61b69485&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/datajoint/datajoint-python/commit/b61b69485dc48b5d17489728a600b42b11e1d272&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/datajoint/datajoint-python/commit/b61b69485dc48b5d17489728a600b42b&amp;lt;/a&amp;gt;11e1d272&amp;quot;</a>&gt;&amp;quot;&amp;gt;Merge &amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/datajoint/datajoint-python/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/datajoint/datajoint-python/commit/&lt;a cla... (continued)

Run Details

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

3063 of 3388 relevant lines covered (90.41%)

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


mxClassID = dict(
    (
        # see http://www.mathworks.com/help/techdoc/apiref/mxclassid.html
        ("mxUNKNOWN_CLASS", None),
        ("mxCELL_CLASS", None),
        ("mxSTRUCT_CLASS", None),
        ("mxLOGICAL_CLASS", np.dtype("bool")),
        ("mxCHAR_CLASS", np.dtype("c")),
        ("mxVOID_CLASS", np.dtype("O")),
        ("mxDOUBLE_CLASS", np.dtype("float64")),
        ("mxSINGLE_CLASS", np.dtype("float32")),
        ("mxINT8_CLASS", np.dtype("int8")),
        ("mxUINT8_CLASS", np.dtype("uint8")),
        ("mxINT16_CLASS", np.dtype("int16")),
        ("mxUINT16_CLASS", np.dtype("uint16")),
        ("mxINT32_CLASS", np.dtype("int32")),
        ("mxUINT32_CLASS", np.dtype("uint32")),
        ("mxINT64_CLASS", np.dtype("int64")),
        ("mxUINT64_CLASS", np.dtype("uint64")),
        ("mxFUNCTION_CLASS", None),
    )
)

rev_class_id = {dtype: i for i, dtype in enumerate(mxClassID.values())}
dtype_list = list(mxClassID.values())
type_names = list(mxClassID)

compression = {b"ZL123\0": zlib.decompress}

bypass_serialization = False  # runtime setting to bypass blob (en|de)code

# runtime setting to read integers as 32-bit to read blobs created by the 32-bit
# version of the mYm library for MATLAB
use_32bit_dims = False


def len_u64(obj):
    return np.uint64(len(obj)).tobytes()


def len_u32(obj):
    return np.uint32(len(obj)).tobytes()


class MatCell(np.ndarray):
    """a numpy ndarray representing a Matlab cell array"""

    pass


class MatStruct(np.recarray):
    """numpy.recarray representing a Matlab struct array"""

    pass


class Blob:
    def __init__(self, squeeze=False):
        self._squeeze = squeeze
        self._blob = None
        self._pos = 0
        self.protocol = None

    def set_dj0(self):
        if not config.get("enable_python_native_blobs"):
            raise DataJointError(
                """v0.12+ python native blobs disabled.
                See also: https://github.com/datajoint/datajoint-python#python-native-blobs"""
            )

        self.protocol = b"dj0\0"  # when using new blob features

    def squeeze(self, array, convert_to_scalar=True):
        """
        Simplify the input array - squeeze out all singleton dimensions.
        If convert_to_scalar, then convert zero-dimensional arrays to scalars
        """
        if not self._squeeze:
            return array
        array = array.squeeze()
        return array.item() if array.ndim == 0 and convert_to_scalar else array

    def unpack(self, blob):
        self._blob = blob
        try:
            # decompress
            prefix = next(
                p for p in compression if self._blob[self._pos :].startswith(p)
            )
        except StopIteration:
            pass  # assume uncompressed but could be unrecognized compression
        else:
            self._pos += len(prefix)
            blob_size = self.read_value()
            blob = compression[prefix](self._blob[self._pos :])
            assert len(blob) == blob_size
            self._blob = blob
            self._pos = 0
        blob_format = self.read_zero_terminated_string()
        if blob_format in ("mYm", "dj0"):
            return self.read_blob(n_bytes=len(self._blob) - self._pos)

    def read_blob(self, n_bytes=None):
        start = self._pos
        data_structure_code = chr(self.read_value("uint8"))
        try:
            call = {
                # MATLAB-compatible, inherited from original mYm
                "A": self.read_array,  # matlab-compatible numeric arrays and scalars with ndim==0
                "P": self.read_sparse_array,  # matlab sparse array -- not supported yet
                "S": self.read_struct,  # matlab struct array
                "C": self.read_cell_array,  # matlab cell array
                # basic data types
                "\xFF": self.read_none,  # None
                "\x01": self.read_tuple,  # a Sequence (e.g. tuple)
                "\x02": self.read_list,  # a MutableSequence (e.g. list)
                "\x03": self.read_set,  # a Set
                "\x04": self.read_dict,  # a Mapping (e.g. dict)
                "\x05": self.read_string,  # a UTF8-encoded string
                "\x06": self.read_bytes,  # a ByteString
                "\x0a": self.read_int,  # unbounded scalar int
                "\x0b": self.read_bool,  # scalar boolean
                "\x0c": self.read_complex,  # scalar 128-bit complex number
                "\x0d": self.read_float,  # scalar 64-bit float
                "F": self.read_recarray,  # numpy array with fields, including recarrays
                "d": self.read_decimal,  # a decimal
                "t": self.read_datetime,  # date, time, or datetime
                "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)):
            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.Mapping):
            return self.pack_dict(obj)
        if isinstance(obj, str):
            return self.pack_string(obj)
        if isinstance(obj, collections.ByteString):
            return self.pack_bytes(obj)
        if isinstance(obj, collections.MutableSequence):
            return self.pack_list(obj)
        if isinstance(obj, collections.Sequence):
            return self.pack_tuple(obj)
        if isinstance(obj, collections.Set):
            return self.pack_set(obj)
        if obj is None:
            return self.pack_none()
        raise DataJointError(
            "Packing object of type %s currently not supported!" % type(obj)
        )

    def read_array(self):
        n_dims = int(self.read_value())
        shape = self.read_value(count=n_dims)
        n_elem = np.prod(shape, dtype=int)
        dtype_id, is_complex = self.read_value("uint32", 2)
        dtype = dtype_list[dtype_id]

        if type_names[dtype_id] == "mxVOID_CLASS":
            data = np.array(
                list(self.read_blob(self.read_value()) for _ in range(n_elem)),
                dtype=np.dtype("O"),
            )
        elif type_names[dtype_id] == "mxCHAR_CLASS":
            # compensate for MATLAB packing of char arrays
            data = self.read_value(dtype, count=2 * n_elem)
            data = data[::2].astype("U1")
            if n_dims == 2 and shape[0] == 1 or n_dims == 1:
                compact = data.squeeze()
                data = (
                    compact
                    if compact.shape == ()
                    else np.array("".join(data.squeeze()))
                )
                shape = (1,)
        else:
            data = self.read_value(dtype, count=n_elem)
            if is_complex:
                data = data + 1j * self.read_value(dtype, count=n_elem)
        return self.squeeze(data.reshape(shape, order="F"))

    def pack_array(self, array):
        """
        Serialize an np.ndarray into bytes.  Scalars are encoded with ndim=0.
        """
        blob = (
            b"A"
            + np.uint64(array.ndim).tobytes()
            + np.array(array.shape, dtype=np.uint64).tobytes()
        )
        is_complex = np.iscomplexobj(array)
        if is_complex:
            array, imaginary = np.real(array), np.imag(array)
        type_id = (
            rev_class_id[array.dtype]
            if array.dtype.char != "U"
            else rev_class_id[np.dtype("O")]
        )
        if dtype_list[type_id] is None:
            raise DataJointError("Type %s is ambiguous or unknown" % array.dtype)

        blob += np.array([type_id, is_complex], dtype=np.uint32).tobytes()
        if type_names[type_id] == "mxVOID_CLASS":  # array of dtype('O')
            blob += b"".join(
                len_u64(it) + it
                for it in (self.pack_blob(e) for e in array.flatten(order="F"))
            )
            self.set_dj0()  # not supported by original mym
        elif type_names[type_id] == "mxCHAR_CLASS":  # array of dtype('c')
            blob += (
                array.view(np.uint8).astype(np.uint16).tobytes()
            )  # convert to 16-bit chars for MATLAB
        else:  # numeric arrays
            if array.ndim == 0:  # not supported by original mym
                self.set_dj0()
            blob += array.tobytes(order="F")
            if is_complex:
                blob += imaginary.tobytes(order="F")
        return blob

    def read_recarray(self):
        """
        Serialize an np.ndarray with fields, including recarrays
        """
        n_fields = self.read_value("uint32")
        if not n_fields:
            return np.array(None)  # empty array
        field_names = [self.read_zero_terminated_string() for _ in range(n_fields)]
        arrays = [self.read_blob() for _ in range(n_fields)]
        rec = np.empty(
            arrays[0].shape,
            np.dtype([(f, t.dtype) for f, t in zip(field_names, arrays)]),
        )
        for f, t in zip(field_names, arrays):
            rec[f] = t
        return rec.view(np.recarray)

    def pack_recarray(self, array):
        """Serialize a Matlab struct array"""
        return (
            b"F"
            + len_u32(array.dtype)
            + "\0".join(array.dtype.names).encode()  # number of fields
            + b"\0"
            + b"".join(  # field names
                self.pack_recarray(array[f])
                if array[f].dtype.fields
                else self.pack_array(array[f])
                for f in array.dtype.names
            )
        )

    def read_sparse_array(self):
        raise DataJointError(
            "datajoint-python does not yet support sparse arrays. Issue (#590)"
        )

    def read_int(self):
        return int.from_bytes(
            self.read_binary(self.read_value("uint16")), byteorder="little", signed=True
        )

    @staticmethod
    def pack_int(v):
        n_bytes = v.bit_length() // 8 + 1
        assert 0 < n_bytes <= 0xFFFF, "Integers are limited to 65535 bytes"
        return (
            b"\x0a"
            + np.uint16(n_bytes).tobytes()
            + v.to_bytes(n_bytes, byteorder="little", signed=True)
        )

    def read_bool(self):
        return bool(self.read_value("bool"))

    @staticmethod
    def pack_bool(v):
        return b"\x0b" + np.array(v, dtype="bool").tobytes()

    def read_complex(self):
        return complex(self.read_value("complex128"))

    @staticmethod
    def pack_complex(v):
        return b"\x0c" + np.array(v, dtype="complex128").tobytes()

    def read_float(self):
        return float(self.read_value("float64"))

    @staticmethod
    def pack_float(v):
        return b"\x0d" + np.array(v, dtype="float64").tobytes()

    def read_decimal(self):
        return Decimal(self.read_string())

    @staticmethod
    def pack_decimal(d):
        s = str(d)
        return b"d" + len_u64(s) + s.encode()

    def read_string(self):
        return self.read_binary(self.read_value()).decode()

    @staticmethod
    def pack_string(s):
        blob = s.encode()
        return b"\5" + len_u64(blob) + blob

    def read_bytes(self):
        return self.read_binary(self.read_value())

    @staticmethod
    def pack_bytes(s):
        return b"\6" + len_u64(s) + s

    def read_none(self):
        pass

    @staticmethod
    def pack_none():
        return b"\xFF"

    def read_tuple(self):
        return tuple(
            self.read_blob(self.read_value()) for _ in range(self.read_value())
        )

    def pack_tuple(self, t):
        return (
            b"\1"
            + len_u64(t)
            + b"".join(len_u64(it) + it for it in (self.pack_blob(i) for i in t))
        )

    def read_list(self):
        return list(self.read_blob(self.read_value()) for _ in range(self.read_value()))

    def pack_list(self, t):
        return (
            b"\2"
            + len_u64(t)
            + b"".join(len_u64(it) + it for it in (self.pack_blob(i) for i in t))
        )

    def read_set(self):
        return set(self.read_blob(self.read_value()) for _ in range(self.read_value()))

    def pack_set(self, t):
        return (
            b"\3"
            + len_u64(t)
            + b"".join(len_u64(it) + it for it in (self.pack_blob(i) for i in t))
        )

    def read_dict(self):
        return dict(
            (self.read_blob(self.read_value()), self.read_blob(self.read_value()))
            for _ in range(self.read_value())
        )

    def pack_dict(self, d):
        return (
            b"\4"
            + len_u64(d)
            + b"".join(
                b"".join((len_u64(it) + it) for it in packed)
                for packed in (map(self.pack_blob, pair) for pair in d.items())
            )
        )

    def read_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_str = str(date)
        time_str = str(time)

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

datajoint / datajoint-python / #12106

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