<s> """<STR_LIT>""" <EOL> from rpyc . lib . compat import Struct , BytesIO , all , is_py3k , BYTES_LITERAL <EOL> TAG_NONE = BYTES_LITERAL ( "<STR_LIT:\x00>" ) <EOL> TAG_EMPTY_STR = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_EMPTY_TUPLE = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_TRUE = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_FALSE = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_NOT_IMPLEMENTED = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_ELLIPSIS = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_UNICODE = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_LONG = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_STR1 = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_STR2 = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_STR3 = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_STR4 = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_STR_L1 = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_STR_L4 = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_TUP1 = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_TUP2 = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_TUP3 = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_TUP4 = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_TUP_L1 = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_TUP_L4 = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_INT_L1 = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_INT_L4 = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_FLOAT = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_SLICE = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_FSET = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> TAG_COMPLEX = BYTES_LITERAL ( "<STR_LIT>" ) <EOL> if is_py3k : <EOL> IMM_INTS = dict ( ( i , bytes ( [ i + <NUM_LIT> ] ) ) for i in range ( - <NUM_LIT> , <NUM_LIT> ) ) <EOL> else : <EOL> IMM_INTS = dict ( ( i , chr ( i + <NUM_LIT> ) ) for i in range ( - <NUM_LIT> , <NUM_LIT> ) ) <EOL> I1 = Struct ( "<STR_LIT>" ) <EOL> I4 = Struct ( "<STR_LIT>" ) <EOL> F8 = Struct ( "<STR_LIT>" ) <EOL> C16 = Struct ( "<STR_LIT>" ) <EOL> _dump_registry = { } <EOL> _load_registry = { } <EOL> IMM_INTS_LOADER = dict ( ( v , k ) for k , v in IMM_INTS . items ( ) ) <EOL> def register ( coll , key ) : <EOL> def deco ( func ) : <EOL> coll [ key ] = func <EOL> return func <EOL> return deco <EOL> @ register ( _dump_registry , type ( None ) ) <EOL> def _dump_none ( obj , stream ) : <EOL> stream . append ( TAG_NONE ) <EOL> @ register ( _dump_registry , type ( NotImplemented ) ) <EOL> def _dump_notimplemeted ( obj , stream ) : <EOL> stream . append ( TAG_NOT_IMPLEMENTED ) <EOL> @ register ( _dump_registry , type ( Ellipsis ) ) <EOL> def _dump_ellipsis ( obj , stream ) : <EOL> stream . append ( TAG_ELLIPSIS ) <EOL> @ register ( _dump_registry , bool ) <EOL> def _dump_bool ( obj , stream ) : <EOL> if obj : <EOL> stream . append ( TAG_TRUE ) <EOL> else : <EOL> stream . append ( TAG_FALSE ) <EOL> @ register ( _dump_registry , slice ) <EOL> def _dump_slice ( obj , stream ) : <EOL> stream . append ( TAG_SLICE ) <EOL> _dump ( ( obj . start , obj . stop , obj . step ) , stream ) <EOL> @ register ( _dump_registry , frozenset ) <EOL> def _dump_frozenset ( obj , stream ) : <EOL> stream . append ( TAG_FSET ) <EOL> _dump ( tuple ( obj ) , stream ) <EOL> @ register ( _dump_registry , int ) <EOL> def _dump_int ( obj , stream ) : <EOL> if obj in IMM_INTS : <EOL> stream . append ( IMM_INTS [ obj ] ) <EOL> else : <EOL> obj = BYTES_LITERAL ( str ( obj ) ) <EOL> l = len ( obj ) <EOL> if l < <NUM_LIT> : <EOL> stream . append ( TAG_INT_L1 + I1 . pack ( l ) + obj ) <EOL> else : <EOL> stream . append ( TAG_INT_L4 + I4 . pack ( l ) + obj ) <EOL> @ register ( _dump_registry , float ) <EOL> def _dump_float ( obj , stream ) : <EOL> stream . append ( TAG_FLOAT + F8 . pack ( obj ) ) <EOL> @ register ( _dump_registry , complex ) <EOL> def _dump_complex ( obj , stream ) : <EOL> stream . append ( TAG_COMPLEX + C16 . pack ( obj . real , obj . imag ) ) <EOL> if is_py3k : <EOL> @ register ( _dump_registry , bytes ) <EOL> def _dump_bytes ( obj , stream ) : <EOL> l = len ( obj ) <EOL> if l == <NUM_LIT:0> : <EOL> stream . append ( TAG_EMPTY_STR ) <EOL> elif l == <NUM_LIT:1> : <EOL> stream . append ( TAG_STR1 + obj ) <EOL> elif l == <NUM_LIT:2> : <EOL> stream . append ( TAG_STR2 + obj ) <EOL> elif l == <NUM_LIT:3> : <EOL> stream . append ( TAG_STR3 + obj ) <EOL> elif l == <NUM_LIT:4> : <EOL> stream . append ( TAG_STR4 + obj ) <EOL> elif l < <NUM_LIT> : <EOL> stream . append ( TAG_STR_L1 + I1 . pack ( l ) + obj ) <EOL> else : <EOL> stream . append ( TAG_STR_L4 + I4 . pack ( l ) + obj ) <EOL> @ register ( _dump_registry , str ) <EOL> def _dump_str ( obj , stream ) : <EOL> stream . append ( TAG_UNICODE ) <EOL> _dump_bytes ( obj . encode ( "<STR_LIT:utf8>" ) , stream ) <EOL> else : <EOL> @ register ( _dump_registry , str ) <EOL> def _dump_str ( obj , stream ) : <EOL> l = len ( obj ) <EOL> if l == <NUM_LIT:0> : <EOL> stream . append ( TAG_EMPTY_STR ) <EOL> elif l == <NUM_LIT:1> : <EOL> stream . append ( TAG_STR1 + obj ) <EOL> elif l == <NUM_LIT:2> : <EOL> stream . append ( TAG_STR2 + obj ) <EOL> elif l == <NUM_LIT:3> : <EOL> stream . append ( TAG_STR3 + obj ) <EOL> elif l == <NUM_LIT:4> : <EOL> stream . append ( TAG_STR4 + obj ) <EOL> elif l < <NUM_LIT> : <EOL> stream . append ( TAG_STR_L1 + I1 . pack ( l ) + obj ) <EOL> else : <EOL> stream . append ( TAG_STR_L4 + I4 . pack ( l ) + obj ) <EOL> @ register ( _dump_registry , unicode ) <EOL> def _dump_unicode ( obj , stream ) : <EOL> stream . append ( TAG_UNICODE ) <EOL> _dump_str ( obj . encode ( "<STR_LIT:utf8>" ) , stream ) <EOL> @ register ( _dump_registry , long ) <EOL> def _dump_long ( obj , stream ) : <EOL> stream . append ( TAG_LONG ) <EOL> _dump_int ( obj , stream ) <EOL> @ register ( _dump_registry , tuple ) <EOL> def _dump_tuple ( obj , stream ) : <EOL> l = len ( obj ) <EOL> if l == <NUM_LIT:0> : <EOL> stream . append ( TAG_EMPTY_TUPLE ) <EOL> elif l == <NUM_LIT:1> : <EOL> stream . append ( TAG_TUP1 ) <EOL> elif l == <NUM_LIT:2> : <EOL> stream . append ( TAG_TUP2 ) <EOL> elif l == <NUM_LIT:3> : <EOL> stream . append ( TAG_TUP3 ) <EOL> elif l == <NUM_LIT:4> : <EOL> stream . append ( TAG_TUP4 ) <EOL> elif l < <NUM_LIT> : <EOL> stream . append ( TAG_TUP_L1 + I1 . pack ( l ) ) <EOL> else : <EOL> stream . append ( TAG_TUP_L4 + I4 . pack ( l ) ) <EOL> for item in obj : <EOL> _dump ( item , stream ) <EOL> def _undumpable ( obj , stream ) : <EOL> raise TypeError ( "<STR_LIT>" % ( obj , ) ) <EOL> def _dump ( obj , stream ) : <EOL> _dump_registry . get ( type ( obj ) , _undumpable ) ( obj , stream ) <EOL> @ register ( _load_registry , TAG_NONE ) <EOL> def _load_none ( stream ) : <EOL> return None <EOL> @ register ( _load_registry , TAG_NOT_IMPLEMENTED ) <EOL> def _load_nonimp ( stream ) : <EOL> return NotImplemented <EOL> @ register ( _load_registry , TAG_ELLIPSIS ) <EOL> def _load_elipsis ( stream ) : <EOL> return Ellipsis <EOL> @ register ( _load_registry , TAG_TRUE ) <EOL> def _load_true ( stream ) : <EOL> return True <EOL> @ register ( _load_registry , TAG_FALSE ) <EOL> def _load_false ( stream ) : <EOL> return False <EOL> @ register ( _load_registry , TAG_EMPTY_TUPLE ) <EOL> def _load_empty_tuple ( stream ) : <EOL> return ( ) <EOL> if is_py3k : <EOL> @ register ( _load_registry , TAG_EMPTY_STR ) <EOL> def _load_empty_str ( stream ) : <EOL> return BYTES_LITERAL ( "<STR_LIT>" ) <EOL> else : <EOL> @ register ( _load_registry , TAG_EMPTY_STR ) <EOL> def _load_empty_str ( stream ) : <EOL> return "<STR_LIT>" <EOL> if is_py3k : <EOL> @ register ( _load_registry , TAG_LONG ) <EOL> def _load_long ( stream ) : <EOL> obj = _load ( stream ) <EOL> return int ( obj ) <EOL> else : <EOL> @ register ( _load_registry , TAG_LONG ) <EOL> def _load_long ( stream ) : <EOL> obj = _load ( stream ) <EOL> return long ( obj ) <EOL> @ register ( _load_registry , TAG_FLOAT ) <EOL> def _load_float ( stream ) : <EOL> return F8 . unpack ( stream . read ( <NUM_LIT:8> ) ) [ <NUM_LIT:0> ] <EOL> @ register ( _load_registry , TAG_COMPLEX ) <EOL> def _load_complex ( stream ) : <EOL> real , imag = C16 . unpack ( stream . read ( <NUM_LIT:16> ) ) <EOL> return complex ( real , imag ) <EOL> @ register ( _load_registry , TAG_STR1 ) <EOL> def _load_str1 ( stream ) : <EOL> return stream . read ( <NUM_LIT:1> ) <EOL> @ register ( _load_registry , TAG_STR2 ) <EOL> def _load_str2 ( stream ) : <EOL> return stream . read ( <NUM_LIT:2> ) <EOL> @ register ( _load_registry , TAG_STR3 ) <EOL> def _load_str3 ( stream ) : <EOL> return stream . read ( <NUM_LIT:3> ) <EOL> @ register ( _load_registry , TAG_STR4 ) <EOL> def _load_str4 ( stream ) : <EOL> return stream . read ( <NUM_LIT:4> ) <EOL> @ register ( _load_registry , TAG_STR_L1 ) <EOL> def _load_str_l1 ( stream ) : <EOL> l , = I1 . unpack ( stream . read ( <NUM_LIT:1> ) ) <EOL> return stream . read ( l ) <EOL> @ register ( _load_registry , TAG_STR_L4 ) <EOL> def _load_str_l4 ( stream ) : <EOL> l , = I4 . unpack ( stream . read ( <NUM_LIT:4> ) ) <EOL> return stream . read ( l ) <EOL> @ register ( _load_registry , TAG_UNICODE ) <EOL> def _load_unicode ( stream ) : <EOL> obj = _load ( stream ) <EOL> return obj . decode ( "<STR_LIT:utf-8>" ) <EOL> @ register ( _load_registry , TAG_TUP1 ) <EOL> def _load_tup1 ( stream ) : <EOL> return ( _load ( stream ) , ) <EOL> @ register ( _load_registry , TAG_TUP2 ) <EOL> def _load_tup2 ( stream ) : <EOL> return ( _load ( stream ) , _load ( stream ) ) <EOL> @ register ( _load_registry , TAG_TUP3 ) <EOL> def _load_tup3 ( stream ) : <EOL> return ( _load ( stream ) , _load ( stream ) , _load ( stream ) ) <EOL> @ register ( _load_registry , TAG_TUP4 ) <EOL> def _load_tup4 ( stream ) : <EOL> return ( _load ( stream ) , _load ( stream ) , _load ( stream ) , _load ( stream ) ) <EOL> @ register ( _load_registry , TAG_TUP_L1 ) <EOL> def _load_tup_l1 ( stream ) : <EOL> l , = I1 . unpack ( stream . read ( <NUM_LIT:1> ) ) <EOL> return tuple ( _load ( stream ) for i in range ( l ) ) <EOL> if is_py3k : <EOL> @ register ( _load_registry , TAG_TUP_L4 ) <EOL> def _load_tup_l4 ( stream ) : <EOL> l , = I4 . unpack ( stream . read ( <NUM_LIT:4> ) ) <EOL> return tuple ( _load ( stream ) for i in range ( l ) ) <EOL> else : <EOL> @ register ( _load_registry , TAG_TUP_L4 ) <EOL> def _load_tup_l4 ( stream ) : <EOL> l , = I4 . unpack ( stream . read ( <NUM_LIT:4> ) ) <EOL> return tuple ( _load ( stream ) for i in xrange ( l ) ) <EOL> @ register ( _load_registry , TAG_SLICE ) <EOL> def _load_slice ( stream ) : <EOL> start , stop , step = _load ( stream ) <EOL> return slice ( start , stop , step ) <EOL> @ register ( _load_registry , TAG_FSET ) <EOL> def _load_frozenset ( stream ) : <EOL> return frozenset ( _load ( stream ) ) <EOL> @ register ( _load_registry , TAG_INT_L1 ) <EOL> def _load_int_l1 ( stream ) : <EOL> l , = I1 . unpack ( stream . read ( <NUM_LIT:1> ) ) <EOL> return int ( stream . read ( l ) ) <EOL> @ register ( _load_registry , TAG_INT_L4 ) <EOL> def _load_int_l4 ( stream ) : <EOL> l , = I4 . unpack ( stream . read ( <NUM_LIT:4> ) ) <EOL> return int ( stream . read ( l ) ) <EOL> def _load ( stream ) : <EOL> tag = stream . read ( <NUM_LIT:1> ) <EOL> if tag in IMM_INTS_LOADER : <EOL> return IMM_INTS_LOADER [ tag ] <EOL> return _load_registry . get ( tag ) ( stream ) <EOL> def dump ( obj ) : <EOL> """<STR_LIT>""" <EOL> stream = [ ] <EOL> _dump ( obj , stream ) <EOL> return BYTES_LITERAL ( "<STR_LIT>" ) . join ( stream ) <EOL> def load ( data ) : <EOL> """<STR_LIT>""" <EOL> stream = BytesIO ( data ) <EOL> return _load ( stream ) <EOL> if is_py3k : <EOL> simple_types = frozenset ( [ type ( None ) , int , bool , float , bytes , str , complex , <EOL> type ( NotImplemented ) , type ( Ellipsis ) ] ) <EOL> else : <EOL> simple_types = frozenset ( [ type ( None ) , int , long , bool , float , str , unicode , complex , <EOL> type ( NotImplemented ) , type ( Ellipsis ) ] ) <EOL> def dumpable ( obj ) : <EOL> """<STR_LIT>""" <EOL> if type ( obj ) in simple_types : <EOL> return True <EOL> if type ( obj ) in ( tuple , frozenset ) : <EOL> return all ( dumpable ( item ) for item in obj ) <EOL> if type ( obj ) is slice : <EOL> return dumpable ( obj . start ) and dumpable ( obj . stop ) and dumpable ( obj . step ) <EOL> return False <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> import doctest <EOL> doctest . testmod ( ) </s>
