#2425

pending completion

Build # #2425

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/ccdd330fc12bc4a55e6c41263f382c753d303df1">ccdd330fc<a href="https://github.com/datajoint/datajoint-python/commit/ccdd330fc12bc4a55e6c41263f382c753d303df1">">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/eac63ab758ed4156b78fe1864d5770bcaf8cb6c5">eac63ab75</a>">eac63ab75</a><a href="https://github.com/datajoint/datajoint-python/commit/ccdd330fc12bc4a55e6c41263f382c753d303df1"> into ed7cf6a78">ed7cf6a78</a>

Run Details

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

2638 of 2893 relevant lines covered (91.19%)

0.91 hits per line

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

94.64

/datajoint/blob.py

"""
(De)serialization methods for python 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 .utils import OrderedDict


DJ0_ENABLE = True  # temporary hook to allow dj0 linting


mxClassID = OrderedDict((
    # 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


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 DJ0_ENABLE:
            raise RuntimeError('dj0 encoding disabled by user')

        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('uint64')
            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
                # Python-native
                "\xFF": self.read_none,      # None
                "\1": self.read_tuple,     # a Sequence
                "\2": self.read_list,      # a MutableSequence
                "\3": self.read_set,       # a Set
                "\4": self.read_dict,      # a Mapping
                "\5": self.read_string,    # a UTF8-encoded string
                "\6": self.read_bytes,     # a ByteString
                "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"' % 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 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, np.bool_)):
            return self.pack_array(np.array(obj))
        if isinstance(obj, float):
            return self.pack_array(np.array(obj, dtype=np.float64))
        if isinstance(obj, int):
            return self.pack_array(np.array(obj, dtype=np.int64))
        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('uint64'))
        shape = self.read_value('uint64', 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) +  # number of fields
                '\0'.join(array.dtype.names).encode() + b"\0" +  # field names
                b"".join(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_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 OrderedDict((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('uint64'))) for _ in range(n_fields))
            for __ in range(n_elem)]
        data = np.array(raw_data, dtype=list(zip(field_names, repeat(np.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() +  # dimensionality
                len_u32(array.dtype.names) +  # number of fields
                "\0".join(array.dtype.names).encode() + b"\0" +  # field names
                b"".join(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='uint64', count=1):
        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 python 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 .utils import OrderedDict	1✔
15
16
17	DJ0_ENABLE = True # temporary hook to allow dj0 linting	1✔
18
19
20	mxClassID = OrderedDict((	1✔
21	# see http://www.mathworks.com/help/techdoc/apiref/mxclassid.html
22	('mxUNKNOWN_CLASS', None),
23	('mxCELL_CLASS', None),
24	('mxSTRUCT_CLASS', None),
25	('mxLOGICAL_CLASS', np.dtype('bool')),
26	('mxCHAR_CLASS', np.dtype('c')),
27	('mxVOID_CLASS', np.dtype('O')),
28	('mxDOUBLE_CLASS', np.dtype('float64')),
29	('mxSINGLE_CLASS', np.dtype('float32')),
30	('mxINT8_CLASS', np.dtype('int8')),
31	('mxUINT8_CLASS', np.dtype('uint8')),
32	('mxINT16_CLASS', np.dtype('int16')),
33	('mxUINT16_CLASS', np.dtype('uint16')),
34	('mxINT32_CLASS', np.dtype('int32')),
35	('mxUINT32_CLASS', np.dtype('uint32')),
36	('mxINT64_CLASS', np.dtype('int64')),
37	('mxUINT64_CLASS', np.dtype('uint64')),
38	('mxFUNCTION_CLASS', None)))
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 = {	1✔
45	b'ZL123\0': zlib.decompress
46	}
47
48	bypass_serialization = False # runtime setting to bypass blob (en\|de)code	1✔
49
50
51	def len_u64(obj):	1✔
52	return np.uint64(len(obj)).tobytes()	1✔
53
54
55	def len_u32(obj):	1✔
56	return np.uint32(len(obj)).tobytes()	1✔
57
58
59	class MatCell(np.ndarray):	1✔
60	""" a numpy ndarray representing a Matlab cell array """
61	pass	1✔
62
63
64	class MatStruct(np.recarray):	1✔
65	""" numpy.recarray representing a Matlab struct array """
66	pass	1✔
67
68
69	class Blob:	1✔
70	def __init__(self, squeeze=False):	1✔
71	self._squeeze = squeeze	1✔
72	self._blob = None	1✔
73	self._pos = 0	1✔
74	self.protocol = None	1✔
75
76	def set_dj0(self):	1✔
77	if not DJ0_ENABLE:	1✔
78	raise RuntimeError('dj0 encoding disabled by user')	×
79
80	self.protocol = b"dj0\0" # when using new blob features	1✔
81
82	def squeeze(self, array, convert_to_scalar=True):	1✔
83	"""
84	Simplify the input array - squeeze out all singleton dimensions.
85	If convert_to_scalar, then convert zero-dimensional arrays to scalars
86	"""
87	if not self._squeeze:	1✔
88	return array	1✔
89	array = array.squeeze()	1✔
90	return array.item() if array.ndim == 0 and convert_to_scalar else array	1✔
91
92	def unpack(self, blob):	1✔
93	self._blob = blob	1✔
94	try:	1✔
95	# decompress
96	prefix = next(p for p in compression if self._blob[self._pos:].startswith(p))	1✔
97	except StopIteration:	1✔
98	pass # assume uncompressed but could be unrecognized compression	1✔
99	else:
100	self._pos += len(prefix)	1✔
101	blob_size = self.read_value('uint64')	1✔
102	blob = compression[prefix](self._blob[self._pos:])	1✔
103	assert len(blob) == blob_size	1✔
104	self._blob = blob	1✔
105	self._pos = 0	1✔
106	blob_format = self.read_zero_terminated_string()	1✔
107	if blob_format in ('mYm', 'dj0'):	1✔
108	return self.read_blob(n_bytes=len(self._blob) - self._pos)	1✔
109
110	def read_blob(self, n_bytes=None):	1✔
111	start = self._pos	1✔
112	data_structure_code = chr(self.read_value('uint8'))	1✔
113	try:	1✔
114	call = {	1✔
115	# MATLAB-compatible, inherited from original mYm
116	"A": self.read_array, # matlab-compatible numeric arrays and scalars with ndim==0
117	"P": self.read_sparse_array, # matlab sparse array -- not supported yet
118	"S": self.read_struct, # matlab struct array
119	"C": self.read_cell_array, # matlab cell array
120	# Python-native
121	"\xFF": self.read_none, # None
122	"\1": self.read_tuple, # a Sequence
123	"\2": self.read_list, # a MutableSequence
124	"\3": self.read_set, # a Set
125	"\4": self.read_dict, # a Mapping
126	"\5": self.read_string, # a UTF8-encoded string
127	"\6": self.read_bytes, # a ByteString
128	"F": self.read_recarray, # numpy array with fields, including recarrays
129	"d": self.read_decimal, # a decimal
130	"t": self.read_datetime, # date, time, or datetime
131	"u": self.read_uuid, # UUID
132	}[data_structure_code]
133	except KeyError:	×
134	raise DataJointError('Unknown data structure code "%s"' % data_structure_code)	×
135	v = call()	1✔
136	if n_bytes is not None and self._pos - start != n_bytes:	1✔
137	raise DataJointError('Blob length check failed! Invalid blob')	×
138	return v	1✔
139
140	def pack_blob(self, obj):	1✔
141	# original mYm-based serialization from datajoint-matlab
142	if isinstance(obj, MatCell):	1✔
143	return self.pack_cell_array(obj)	1✔
144	if isinstance(obj, MatStruct):	1✔
145	return self.pack_struct(obj)	1✔
146	if isinstance(obj, np.ndarray) and obj.dtype.fields is None:	1✔
147	return self.pack_array(obj)	1✔
148
149	# blob types in the expanded dj0 blob format
150	self.set_dj0()	1✔
151	if isinstance(obj, np.ndarray) and obj.dtype.fields:	1✔
152	return self.pack_recarray(np.array(obj))	1✔
153	if isinstance(obj, np.number):	1✔
154	return self.pack_array(np.array(obj))	1✔
155	if isinstance(obj, (bool, np.bool, np.bool_)):	1✔
156	return self.pack_array(np.array(obj))	1✔
157	if isinstance(obj, float):	1✔
158	return self.pack_array(np.array(obj, dtype=np.float64))	1✔
159	if isinstance(obj, int):	1✔
160	return self.pack_array(np.array(obj, dtype=np.int64))	1✔
161	if isinstance(obj, (datetime.datetime, datetime.date, datetime.time)):	1✔
162	return self.pack_datetime(obj)	1✔
163	if isinstance(obj, Decimal):	1✔
164	return self.pack_decimal(obj)	1✔
165	if isinstance(obj, uuid.UUID):	1✔
166	return self.pack_uuid(obj)	1✔
167	if isinstance(obj, collections.Mapping):	1✔
168	return self.pack_dict(obj)	1✔
169	if isinstance(obj, str):	1✔
170	return self.pack_string(obj)	1✔
171	if isinstance(obj, collections.ByteString):	1✔
172	return self.pack_bytes(obj)	×
173	if isinstance(obj, collections.MutableSequence):	1✔
174	return self.pack_list(obj)	1✔
175	if isinstance(obj, collections.Sequence):	1✔
176	return self.pack_tuple(obj)	1✔
177	if isinstance(obj, collections.Set):	1✔
178	return self.pack_set(obj)	1✔
179	if obj is None:	1✔
180	return self.pack_none()	1✔
181	raise DataJointError("Packing object of type %s currently not supported!" % type(obj))	×
182
183	def read_array(self):	1✔
184	n_dims = int(self.read_value('uint64'))	1✔
185	shape = self.read_value('uint64', count=n_dims)	1✔
186	n_elem = np.prod(shape, dtype=int)	1✔
187	dtype_id, is_complex = self.read_value('uint32', 2)	1✔
188	dtype = dtype_list[dtype_id]	1✔
189
190	if type_names[dtype_id] == 'mxVOID_CLASS':	1✔
191	data = np.array(	1✔
192	list(self.read_blob(self.read_value()) for _ in range(n_elem)), dtype=np.dtype('O'))
193	elif type_names[dtype_id] == 'mxCHAR_CLASS':	1✔
194	# compensate for MATLAB packing of char arrays
195	data = self.read_value(dtype, count=2 * n_elem)	1✔
196	data = data[::2].astype('U1')	1✔
197	if n_dims == 2 and shape[0] == 1 or n_dims == 1:	1✔
198	compact = data.squeeze()	1✔
199	data = compact if compact.shape == () else np.array(''.join(data.squeeze()))	1✔
200	shape = (1,)	1✔
201	else:
202	data = self.read_value(dtype, count=n_elem)	1✔
203	if is_complex:	1✔
204	data = data + 1j * self.read_value(dtype, count=n_elem)	1✔
205	return self.squeeze(data.reshape(shape, order="F"))	1✔
206
207	def pack_array(self, array):	1✔
208	"""
209	Serialize an np.ndarray into bytes. Scalars are encoded with ndim=0.
210	"""
211	blob = b"A" + np.uint64(array.ndim).tobytes() + np.array(array.shape, dtype=np.uint64).tobytes()	1✔
212	is_complex = np.iscomplexobj(array)	1✔
213	if is_complex:	1✔
214	array, imaginary = np.real(array), np.imag(array)	1✔
215	type_id = (rev_class_id[array.dtype] if array.dtype.char != 'U'	1✔
216	else rev_class_id[np.dtype('O')])
217	if dtype_list[type_id] is None:	1✔
218	raise DataJointError("Type %s is ambiguous or unknown" % array.dtype)	×
219
220	blob += np.array([type_id, is_complex], dtype=np.uint32).tobytes()	1✔
221	if type_names[type_id] == 'mxVOID_CLASS': # array of dtype('O')	1✔
222	blob += b"".join(len_u64(it) + it for it in (self.pack_blob(e) for e in array.flatten(order="F")))	1✔
223	self.set_dj0() # not supported by original mym	1✔
224	elif type_names[type_id] == 'mxCHAR_CLASS': # array of dtype('c')	1✔
225	blob += array.view(np.uint8).astype(np.uint16).tobytes() # convert to 16-bit chars for MATLAB	×
226	else: # numeric arrays
227	if array.ndim == 0: # not supported by original mym	1✔
228	self.set_dj0()	1✔
229	blob += array.tobytes(order="F")	1✔
230	if is_complex:	1✔
231	blob += imaginary.tobytes(order="F")	1✔
232	return blob	1✔
233
234	def read_recarray(self):	1✔
235	"""
236	Serialize an np.ndarray with fields, including recarrays
237	"""
238	n_fields = self.read_value('uint32')	1✔
239	if not n_fields:	1✔
240	return np.array(None) # empty array	×
241	field_names = [self.read_zero_terminated_string() for _ in range(n_fields)]	1✔
242	arrays = [self.read_blob() for _ in range(n_fields)]	1✔
243	rec = np.empty(arrays[0].shape, np.dtype([(f, t.dtype) for f, t in zip(field_names, arrays)]))	1✔
244	for f, t in zip(field_names, arrays):	1✔
245	rec[f] = t	1✔
246	return rec.view(np.recarray)	1✔
247
248	def pack_recarray(self, array):	1✔
249	""" Serialize a Matlab struct array """
250	return (b"F" + len_u32(array.dtype) + # number of fields	1✔
251	'\0'.join(array.dtype.names).encode() + b"\0" + # field names
252	b"".join(self.pack_recarray(array[f]) if array[f].dtype.fields else self.pack_array(array[f])
253	for f in array.dtype.names))
254
255	def read_sparse_array(self):	1✔
256	raise DataJointError('datajoint-python does not yet support sparse arrays. Issue (#590)')	×
257
258	def read_decimal(self):	1✔
259	return Decimal(self.read_string())	1✔
260
261	@staticmethod	1✔
262	def pack_decimal(d):
263	s = str(d)	1✔
264	return b"d" + len_u64(s) + s.encode()	1✔
265
266	def read_string(self):	1✔
267	return self.read_binary(self.read_value()).decode()	1✔
268
269	@staticmethod	1✔
270	def pack_string(s):
271	blob = s.encode()	1✔
272	return b"\5" + len_u64(blob) + blob	1✔
273
274	def read_bytes(self):	1✔
275	return self.read_binary(self.read_value())	×
276
277	@staticmethod	1✔
278	def pack_bytes(s):
279	return b"\6" + len_u64(s) + s	×
280
281	def read_none(self):	1✔
282	pass	1✔
283
284	@staticmethod	1✔
285	def pack_none():
286	return b"\xFF"	1✔
287
288	def read_tuple(self):	1✔
289	return tuple(self.read_blob(self.read_value()) for _ in range(self.read_value()))	1✔
290
291	def pack_tuple(self, t):	1✔
292	return b"\1" + len_u64(t) + b"".join(	1✔
293	len_u64(it) + it for it in (self.pack_blob(i) for i in t))
294
295	def read_list(self):	1✔
296	return list(self.read_blob(self.read_value()) for _ in range(self.read_value()))	1✔
297
298	def pack_list(self, t):	1✔
299	return b"\2" + len_u64(t) + b"".join(	1✔
300	len_u64(it) + it for it in (self.pack_blob(i) for i in t))
301
302	def read_set(self):	1✔
303	return set(self.read_blob(self.read_value()) for _ in range(self.read_value()))	1✔
304
305	def pack_set(self, t):	1✔
306	return b"\3" + len_u64(t) + b"".join(	1✔
307	len_u64(it) + it for it in (self.pack_blob(i) for i in t))
308
309	def read_dict(self):	1✔
310	return OrderedDict((self.read_blob(self.read_value()), self.read_blob(self.read_value()))	1✔
311	for _ in range(self.read_value()))
312
313	def pack_dict(self, d):	1✔
314	return b"\4" + len_u64(d) + b"".join(	1✔
315	b"".join((len_u64(it) + it) for it in packed)
316	for packed in (map(self.pack_blob, pair) for pair in d.items()))
317
318	def read_struct(self):	1✔
319	"""deserialize matlab stuct"""
320	n_dims = self.read_value()	1✔
321	shape = self.read_value(count=n_dims)	1✔
322	n_elem = np.prod(shape, dtype=int)	1✔
323	n_fields = self.read_value('uint32')	1✔
324	if not n_fields:	1✔
325	return np.array(None) # empty array	×
326	field_names = [self.read_zero_terminated_string() for _ in range(n_fields)]	1✔
327	raw_data = [	1✔
328	tuple(self.read_blob(n_bytes=int(self.read_value('uint64'))) for _ in range(n_fields))
329	for __ in range(n_elem)]
330	data = np.array(raw_data, dtype=list(zip(field_names, repeat(np.object))))	1✔
331	return self.squeeze(data.reshape(shape, order="F"), convert_to_scalar=False).view(MatStruct)	1✔
332
333	def pack_struct(self, array):	1✔
334	""" Serialize a Matlab struct array """
335	return (b"S" + np.array((array.ndim,) + array.shape, dtype=np.uint64).tobytes() + # dimensionality	1✔
336	len_u32(array.dtype.names) + # number of fields
337	"\0".join(array.dtype.names).encode() + b"\0" + # field names
338	b"".join(len_u64(it) + it for it in (
339	self.pack_blob(e) for rec in array.flatten(order="F") for e in rec))) # values
340
341	def read_cell_array(self):	1✔
342	""" deserialize MATLAB cell array """
343	n_dims = self.read_value()	1✔
344	shape = self.read_value(count=n_dims)	1✔
345	n_elem = int(np.prod(shape))	1✔
346	result = [self.read_blob(n_bytes=self.read_value()) for _ in range(n_elem)]	1✔
347	return (self.squeeze(np.array(result).reshape(shape, order="F"), convert_to_scalar=False)).view(MatCell)	1✔
348
349	def pack_cell_array(self, array):	1✔
350	return (b"C" + np.array((array.ndim,) + array.shape, dtype=np.uint64).tobytes() +	1✔
351	b"".join(len_u64(it) + it for it in (self.pack_blob(e) for e in array.flatten(order="F"))))
352
353	def read_datetime(self):	1✔
354	""" deserialize datetime.date, .time, or .datetime """
355	date, time = self.read_value('int32'), self.read_value('int64')	1✔
356	date = datetime.date(	1✔
357	year=date // 10000,
358	month=(date // 100) % 100,
359	day=date % 100) if date >= 0 else None
360	time = datetime.time(	1✔
361	hour=(time // 10000000000) % 100,
362	minute=(time // 100000000) % 100,
363	second=(time // 1000000) % 100,
364	microsecond=time % 1000000) if time >= 0 else None
365	return time and date and datetime.datetime.combine(date, time) or time or date	1✔
366
367	@staticmethod	1✔
368	def pack_datetime(d):
369	if isinstance(d, datetime.datetime):	1✔
370	date, time = d.date(), d.time()	1✔
371	elif isinstance(d, datetime.date):	1✔
372	date, time = d, None	1✔
373	else:
374	date, time = None, d	×
375	return b"t" + (	1✔
376	np.int32(-1 if date is None else (date.year100 + date.month)100 + date.day).tobytes() +
377	np.int64(-1 if time is None else
378	((time.hour100 + time.minute)100 + time.second)*1000000 + time.microsecond).tobytes())
379
380	def read_uuid(self):	1✔
381	q = self.read_binary(16)	1✔
382	return uuid.UUID(bytes=q)	1✔
383
384	@staticmethod	1✔
385	def pack_uuid(obj):
386	return b"u" + obj.bytes	1✔
387
388	def read_zero_terminated_string(self):	1✔
389	target = self._blob.find(b'\0', self._pos)	1✔
390	data = self._blob[self._pos:target].decode()	1✔
391	self._pos = target + 1	1✔
392	return data	1✔
393
394	def read_value(self, dtype='uint64', count=1):	1✔
395	data = np.frombuffer(self._blob, dtype=dtype, count=count, offset=self._pos)	1✔
396	self._pos += data.dtype.itemsize * data.size	1✔
397	return data[0] if count == 1 else data	1✔
398
399	def read_binary(self, size):	1✔
400	self._pos += int(size)	1✔
401	return self._blob[self._pos-int(size):self._pos]	1✔
402
403	def pack(self, obj, compress):	1✔
404	self.protocol = b"mYm\0" # will be replaced with dj0 if new features are used	1✔
405	blob = self.pack_blob(obj) # this may reset the protocol and must precede protocol evaluation	1✔
406	blob = self.protocol + blob	1✔
407	if compress and len(blob) > 1000:	1✔
408	compressed = b'ZL123\0' + len_u64(blob) + zlib.compress(blob)	1✔
409	if len(compressed) < len(blob):	1✔
410	blob = compressed	1✔
411	return blob	1✔
412
413
414	def pack(obj, compress=True):	1✔
415	if bypass_serialization:	1✔
416	# provide a way to move blobs quickly without de/serialization
417	assert isinstance(obj, bytes) and obj.startswith((b'ZL123\0', b'mYm\0', b'dj0\0'))	1✔
418	return obj	1✔
419	return Blob().pack(obj, compress=compress)	1✔
420
421
422	def unpack(blob, squeeze=False):	1✔
423	if bypass_serialization:	1✔
424	# provide a way to move blobs quickly without de/serialization
425	assert isinstance(blob, bytes) and blob.startswith((b'ZL123\0', b'mYm\0', b'dj0\0'))	1✔
426	return blob	1✔
427	if blob is not None:	1✔
428	return Blob(squeeze=squeeze).unpack(blob)	1✔

datajoint / datajoint-python / #2425

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