#12539

pending completion

Build # #12539

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/eaa05b90a6bdd9e1aca0eb092af04f2b46761f3c">eaa05b90a">&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/eaa05b90a6bdd9e1aca0eb092af04f2b46761f3c">eaa05b90a</a><a href="https://github.com/datajoint/datajoint-python/commit/eaa05b90a6bdd9e1aca0eb092af04f2b46761f3c">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/datajoint/datajoint-python/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/datajoint/datajoint-python/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/datajoint/datajoint-python/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/datajoint/datajoint-python/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/datajoint/datajoint-python/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/datajoint/datajoint-python/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/datajoint/datajoint-python/commit/eaa05b90a6bdd9e1aca0eb092af04f2b46761f3c&quot;&gt;eaa05b90a&lt;/a&gt;&lt;a href=&quot;https://github.com/datajoint/datajoint-python/commit/eaa05b90a6bdd9e1aca0eb092af04f2b46761f3c&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;quo... (continued)

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[ns]"), "scalar_type": "DATETIME64[ns]"},
    138: {"dtype": np.dtype("<M8[ps]"), "scalar_type": "DATETIME64[ps]"},
    139: {"dtype": np.dtype("<M8[fs]"), "scalar_type": "DATETIME64[fs]"},
    140: {"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[ns]"), "scalar_type": "DATETIME64[ns]"},
45	138: {"dtype": np.dtype("<M8[ps]"), "scalar_type": "DATETIME64[ps]"},
46	139: {"dtype": np.dtype("<M8[fs]"), "scalar_type": "DATETIME64[fs]"},
47	140: {"dtype": np.dtype("<M8[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	if array.dtype.char == "U":	1✔
273	type_id = serialize_lookup[np.dtype("O")]["type_id"]	1✔
274	pass	1✔
275	else:
276	raise DataJointError("Type %s is ambiguous or unknown" % array.dtype)	×
277
278	blob += np.array([type_id, is_complex], dtype=np.uint32).tobytes()	1✔
279	# array of dtype('O'), U is for unicode string
280	if (	1✔
281	array.dtype.char == "U"
282	or serialize_lookup[array.dtype]["scalar_type"] == "VOID"
283	):
284	blob += b"".join(	1✔
285	len_u64(it) + it
286	for it in (self.pack_blob(e) for e in array.flatten(order="F"))
287	)
288	self.set_dj0() # not supported by original mym	1✔
289	# array of dtype('c')
290	elif serialize_lookup[array.dtype]["scalar_type"] == "CHAR":	1✔
291	blob += (	×
292	array.view(np.uint8).astype(np.uint16).tobytes()
293	) # convert to 16-bit chars for MATLAB
294	else: # numeric arrays
295	if array.ndim == 0: # not supported by original mym	1✔
296	self.set_dj0()	1✔
297	blob += array.tobytes(order="F")	1✔
298	if is_complex:	1✔
299	blob += imaginary.tobytes(order="F")	1✔
300	return blob	1✔
301
302	def read_recarray(self):	1✔
303	"""
304	Serialize an np.ndarray with fields, including recarrays
305	"""
306	n_fields = self.read_value("uint32")	1✔
307	if not n_fields:	1✔
308	return np.array(None) # empty array	×
309	field_names = [self.read_zero_terminated_string() for _ in range(n_fields)]	1✔
310	arrays = [self.read_blob() for _ in range(n_fields)]	1✔
311	rec = np.empty(	1✔
312	arrays[0].shape,
313	np.dtype([(f, t.dtype) for f, t in zip(field_names, arrays)]),
314	)
315	for f, t in zip(field_names, arrays):	1✔
316	rec[f] = t	1✔
317	return rec.view(np.recarray)	1✔
318
319	def pack_recarray(self, array):	1✔
320	"""Serialize a Matlab struct array"""
321	return (	1✔
322	b"F"
323	+ len_u32(array.dtype)
324	+ "\0".join(array.dtype.names).encode() # number of fields
325	+ b"\0"
326	+ b"".join( # field names
327	self.pack_recarray(array[f])
328	if array[f].dtype.fields
329	else self.pack_array(array[f])
330	for f in array.dtype.names
331	)
332	)
333
334	def read_sparse_array(self):	1✔
335	raise DataJointError(	×
336	"datajoint-python does not yet support sparse arrays. Issue (#590)"
337	)
338
339	def read_int(self):	1✔
340	return int.from_bytes(	1✔
341	self.read_binary(self.read_value("uint16")), byteorder="little", signed=True
342	)
343
344	@staticmethod	1✔
345	def pack_int(v):	1✔
346	n_bytes = v.bit_length() // 8 + 1	1✔
347	assert 0 < n_bytes <= 0xFFFF, "Integers are limited to 65535 bytes"	1✔
348	return (	1✔
349	b"\x0a"
350	+ np.uint16(n_bytes).tobytes()
351	+ v.to_bytes(n_bytes, byteorder="little", signed=True)
352	)
353
354	def read_bool(self):	1✔
355	return bool(self.read_value("bool"))	1✔
356
357	@staticmethod	1✔
358	def pack_bool(v):	1✔
359	return b"\x0b" + np.array(v, dtype="bool").tobytes()	1✔
360
361	def read_complex(self):	1✔
362	return complex(self.read_value("complex128"))	1✔
363
364	@staticmethod	1✔
365	def pack_complex(v):	1✔
366	return b"\x0c" + np.array(v, dtype="complex128").tobytes()	1✔
367
368	def read_float(self):	1✔
369	return float(self.read_value("float64"))	1✔
370
371	@staticmethod	1✔
372	def pack_float(v):	1✔
373	return b"\x0d" + np.array(v, dtype="float64").tobytes()	1✔
374
375	def read_decimal(self):	1✔
376	return Decimal(self.read_string())	1✔
377
378	@staticmethod	1✔
379	def pack_decimal(d):	1✔
380	s = str(d)	1✔
381	return b"d" + len_u64(s) + s.encode()	1✔
382
383	def read_string(self):	1✔
384	return self.read_binary(self.read_value()).decode()	1✔
385
386	@staticmethod	1✔
387	def pack_string(s):	1✔
388	blob = s.encode()	1✔
389	return b"\5" + len_u64(blob) + blob	1✔
390
391	def read_bytes(self):	1✔
392	return self.read_binary(self.read_value())	×
393
394	@staticmethod	1✔
395	def pack_bytes(s):	1✔
396	return b"\6" + len_u64(s) + s	×
397
398	def read_none(self):	1✔
399	pass	1✔
400
401	@staticmethod	1✔
402	def pack_none():	1✔
403	return b"\xFF"	1✔
404
405	def read_tuple(self):	1✔
406	return tuple(	1✔
407	self.read_blob(self.read_value()) for _ in range(self.read_value())
408	)
409
410	def pack_tuple(self, t):	1✔
411	return (	1✔
412	b"\1"
413	+ len_u64(t)
414	+ b"".join(len_u64(it) + it for it in (self.pack_blob(i) for i in t))
415	)
416
417	def read_list(self):	1✔
418	return list(self.read_blob(self.read_value()) for _ in range(self.read_value()))	1✔
419
420	def pack_list(self, t):	1✔
421	return (	1✔
422	b"\2"
423	+ len_u64(t)
424	+ b"".join(len_u64(it) + it for it in (self.pack_blob(i) for i in t))
425	)
426
427	def read_set(self):	1✔
428	return set(self.read_blob(self.read_value()) for _ in range(self.read_value()))	1✔
429
430	def pack_set(self, t):	1✔
431	return (	1✔
432	b"\3"
433	+ len_u64(t)
434	+ b"".join(len_u64(it) + it for it in (self.pack_blob(i) for i in t))
435	)
436
437	def read_dict(self):	1✔
438	return dict(	1✔
439	(self.read_blob(self.read_value()), self.read_blob(self.read_value()))
440	for _ in range(self.read_value())
441	)
442
443	def pack_dict(self, d):	1✔
444	return (	1✔
445	b"\4"
446	+ len_u64(d)
447	+ b"".join(
448	b"".join((len_u64(it) + it) for it in packed)
449	for packed in (map(self.pack_blob, pair) for pair in d.items())
450	)
451	)
452
453	def read_struct(self):	1✔
454	"""deserialize matlab stuct"""
455	n_dims = self.read_value()	1✔
456	shape = self.read_value(count=n_dims)	1✔
457	n_elem = np.prod(shape, dtype=int)	1✔
458	n_fields = self.read_value("uint32")	1✔
459	if not n_fields:	1✔
460	return np.array(None) # empty array	×
461	field_names = [self.read_zero_terminated_string() for _ in range(n_fields)]	1✔
462	raw_data = [	1✔
463	tuple(
464	self.read_blob(n_bytes=int(self.read_value())) for _ in range(n_fields)
465	)
466	for __ in range(n_elem)
467	]
468	data = np.array(raw_data, dtype=list(zip(field_names, repeat(object))))	1✔
469	return self.squeeze(	1✔
470	data.reshape(shape, order="F"), convert_to_scalar=False
471	).view(MatStruct)
472
473	def pack_struct(self, array):	1✔
474	"""Serialize a Matlab struct array"""
475	return (	1✔
476	b"S"
477	+ np.array((array.ndim,) + array.shape, dtype=np.uint64).tobytes()
478	+ len_u32(array.dtype.names) # dimensionality
479	+ "\0".join(array.dtype.names).encode() # number of fields
480	+ b"\0"
481	+ b"".join( # field names
482	len_u64(it) + it
483	for it in (
484	self.pack_blob(e) for rec in array.flatten(order="F") for e in rec
485	)
486	)
487	) # values
488
489	def read_cell_array(self):	1✔
490	"""deserialize MATLAB cell array"""
491	n_dims = self.read_value()	1✔
492	shape = self.read_value(count=n_dims)	1✔
493	n_elem = int(np.prod(shape))	1✔
494	result = [self.read_blob(n_bytes=self.read_value()) for _ in range(n_elem)]	1✔
495	return (	1✔
496	self.squeeze(
497	np.array(result).reshape(shape, order="F"), convert_to_scalar=False
498	)
499	).view(MatCell)
500
501	def pack_cell_array(self, array):	1✔
502	return (	1✔
503	b"C"
504	+ np.array((array.ndim,) + array.shape, dtype=np.uint64).tobytes()
505	+ b"".join(
506	len_u64(it) + it
507	for it in (self.pack_blob(e) for e in array.flatten(order="F"))
508	)
509	)
510
511	def read_datetime(self):	1✔
512	"""deserialize datetime.date, .time, or .datetime"""
513	date, time = self.read_value("int32"), self.read_value("int64")	1✔
514	date = (	1✔
515	datetime.date(year=date // 10000, month=(date // 100) % 100, day=date % 100)
516	if date >= 0
517	else None
518	)
519	time = (	1✔
520	datetime.time(
521	hour=(time // 10000000000) % 100,
522	minute=(time // 100000000) % 100,
523	second=(time // 1000000) % 100,
524	microsecond=time % 1000000,
525	)
526	if time >= 0
527	else None
528	)
529	return time and date and datetime.datetime.combine(date, time) or time or date	1✔
530
531	@staticmethod	1✔
532	def pack_datetime(d):	1✔
533	if isinstance(d, datetime.datetime):	1✔
534	date, time = d.date(), d.time()	1✔
535	elif isinstance(d, datetime.date):	1✔
536	date, time = d, None	1✔
537	else:
538	date, time = None, d	×
539	return b"t" + (	1✔
540	np.int32(
541	-1 if date is None else (date.year * 100 + date.month) * 100 + date.day
542	).tobytes()
543	+ np.int64(
544	-1
545	if time is None
546	else ((time.hour * 100 + time.minute) * 100 + time.second) * 1000000
547	+ time.microsecond
548	).tobytes()
549	)
550
551	def read_uuid(self):	1✔
552	q = self.read_binary(16)	1✔
553	return uuid.UUID(bytes=q)	1✔
554
555	@staticmethod	1✔
556	def pack_uuid(obj):	1✔
557	return b"u" + obj.bytes	1✔
558
559	def read_zero_terminated_string(self):	1✔
560	target = self._blob.find(b"\0", self._pos)	1✔
561	data = self._blob[self._pos : target].decode()	1✔
562	self._pos = target + 1	1✔
563	return data	1✔
564
565	def read_value(self, dtype=None, count=1):	1✔
566	if dtype is None:	1✔
567	dtype = "uint32" if use_32bit_dims else "uint64"	1✔
568	data = np.frombuffer(self._blob, dtype=dtype, count=count, offset=self._pos)	1✔
569	self._pos += data.dtype.itemsize * data.size	1✔
570	return data[0] if count == 1 else data	1✔
571
572	def read_binary(self, size):	1✔
573	self._pos += int(size)	1✔
574	return self._blob[self._pos - int(size) : self._pos]	1✔
575
576	def pack(self, obj, compress):	1✔
577	self.protocol = b"mYm\0" # will be replaced with dj0 if new features are used	1✔
578	blob = self.pack_blob(	1✔
579	obj
580	) # this may reset the protocol and must precede protocol evaluation
581	blob = self.protocol + blob	1✔
582	if compress and len(blob) > 1000:	1✔
583	compressed = b"ZL123\0" + len_u64(blob) + zlib.compress(blob)	1✔
584	if len(compressed) < len(blob):	1✔
585	blob = compressed	1✔
586	return blob	1✔
587
588
589	def pack(obj, compress=True):	1✔
590	if bypass_serialization:	1✔
591	# provide a way to move blobs quickly without de/serialization
592	assert isinstance(obj, bytes) and obj.startswith(	1✔
593	(b"ZL123\0", b"mYm\0", b"dj0\0")
594	)
595	return obj	1✔
596	return Blob().pack(obj, compress=compress)	1✔
597
598
599	def unpack(blob, squeeze=False):	1✔
600	if bypass_serialization:	1✔
601	# provide a way to move blobs quickly without de/serialization
602	assert isinstance(blob, bytes) and blob.startswith(	1✔
603	(b"ZL123\0", b"mYm\0", b"dj0\0")
604	)
605	return blob	1✔
606	if blob is not None:	1✔
607	return Blob(squeeze=squeeze).unpack(blob)	1✔

datajoint / datajoint-python / #12539

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