<s> """<STR_LIT>""" <EOL> from pypy . annotation import model as annmodel <EOL> from pypy . annotation . signature import annotation <EOL> from pypy . annotation . listdef import ListDef , TooLateForChange <EOL> from pypy . tool . pairtype import pair , pairtype <EOL> from pypy . rlib . rarithmetic import formatd , r_longlong , intmask <EOL> from pypy . rlib . rarithmetic import break_up_float , parts_to_float <EOL> from pypy . rlib . unroll import unrolling_iterable <EOL> class CannotMarshal ( Exception ) : <EOL> pass <EOL> class CannotUnmarshall ( Exception ) : <EOL> pass <EOL> def get_marshaller ( type ) : <EOL> """<STR_LIT>""" <EOL> s_obj = annotation ( type , None ) <EOL> try : <EOL> return find_dumper ( s_obj ) <EOL> except CannotMarshal : <EOL> pair ( _tag , s_obj ) . install_marshaller ( ) <EOL> return find_dumper ( s_obj ) <EOL> get_marshaller . _annspecialcase_ = '<STR_LIT>' <EOL> def get_loader ( type ) : <EOL> s_obj = annotation ( type , None ) <EOL> try : <EOL> return find_loader ( s_obj ) <EOL> except CannotUnmarshall : <EOL> pair ( _tag , s_obj ) . install_unmarshaller ( ) <EOL> return find_loader ( s_obj ) <EOL> def get_unmarshaller ( type ) : <EOL> """<STR_LIT>""" <EOL> loaditem = get_loader ( type ) <EOL> try : <EOL> return _unmarshaller_cache [ loaditem ] <EOL> except KeyError : <EOL> def unmarshaller ( buf ) : <EOL> loader = Loader ( buf ) <EOL> result = loaditem ( loader ) <EOL> loader . check_finished ( ) <EOL> return result <EOL> _unmarshaller_cache [ loaditem ] = unmarshaller <EOL> return unmarshaller <EOL> get_unmarshaller . _annspecialcase_ = '<STR_LIT>' <EOL> _unmarshaller_cache = { } <EOL> TYPE_NONE = '<STR_LIT:N>' <EOL> TYPE_FALSE = '<STR_LIT:F>' <EOL> TYPE_TRUE = '<STR_LIT:T>' <EOL> TYPE_INT = '<STR_LIT:i>' <EOL> TYPE_INT64 = '<STR_LIT:I>' <EOL> TYPE_FLOAT = '<STR_LIT:f>' <EOL> TYPE_STRING = '<STR_LIT:s>' <EOL> TYPE_TUPLE = '<STR_LIT:(>' <EOL> TYPE_LIST = '<STR_LIT:[>' <EOL> dumpers = [ ] <EOL> loaders = [ ] <EOL> s_list_of_chars = annmodel . SomeList ( ListDef ( None , annmodel . SomeChar ( ) , <EOL> mutated = True , resized = True ) ) <EOL> def add_dumper ( s_obj , dumper ) : <EOL> dumpers . append ( ( s_obj , dumper ) ) <EOL> dumper . s_obj = s_obj <EOL> dumper . _annenforceargs_ = [ s_list_of_chars , s_obj ] <EOL> def add_loader ( s_obj , loader ) : <EOL> loaders . append ( ( s_obj , loader ) ) <EOL> def get_dumper_annotation ( dumper ) : <EOL> return dumper . s_obj <EOL> def find_dumper ( s_obj ) : <EOL> for s_cond , dumper in dumpers : <EOL> if weakly_contains ( s_cond , s_obj ) : <EOL> return dumper <EOL> raise CannotMarshal ( s_obj ) <EOL> def find_loader ( s_obj ) : <EOL> for s_cond , loader in loaders [ : : - <NUM_LIT:1> ] : <EOL> if s_obj . contains ( s_cond ) : <EOL> return loader <EOL> if s_obj == annmodel . s_None : <EOL> return load_none <EOL> raise CannotUnmarshall ( s_obj ) <EOL> def w_long ( buf , x ) : <EOL> buf . append ( chr ( x & <NUM_LIT> ) ) <EOL> x >>= <NUM_LIT:8> <EOL> buf . append ( chr ( x & <NUM_LIT> ) ) <EOL> x >>= <NUM_LIT:8> <EOL> buf . append ( chr ( x & <NUM_LIT> ) ) <EOL> x >>= <NUM_LIT:8> <EOL> buf . append ( chr ( x & <NUM_LIT> ) ) <EOL> w_long . _annenforceargs_ = [ None , int ] <EOL> def dump_none ( buf , x ) : <EOL> buf . append ( TYPE_NONE ) <EOL> add_dumper ( annmodel . s_None , dump_none ) <EOL> def load_none ( loader ) : <EOL> if readchr ( loader ) != TYPE_NONE : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return None <EOL> def dump_bool ( buf , x ) : <EOL> if x : <EOL> buf . append ( TYPE_TRUE ) <EOL> else : <EOL> buf . append ( TYPE_FALSE ) <EOL> add_dumper ( annmodel . s_Bool , dump_bool ) <EOL> def load_bool ( loader ) : <EOL> t = readchr ( loader ) <EOL> if t == TYPE_TRUE : <EOL> return True <EOL> elif t == TYPE_FALSE : <EOL> return False <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> add_loader ( annmodel . s_Bool , load_bool ) <EOL> def dump_int ( buf , x ) : <EOL> buf . append ( TYPE_INT ) <EOL> w_long ( buf , x ) <EOL> add_dumper ( annmodel . SomeInteger ( ) , dump_int ) <EOL> def load_int_nonneg ( loader ) : <EOL> x = load_int ( loader ) <EOL> if x < <NUM_LIT:0> : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return x <EOL> add_loader ( annmodel . SomeInteger ( nonneg = True ) , load_int_nonneg ) <EOL> def load_int ( loader ) : <EOL> if readchr ( loader ) != TYPE_INT : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return readlong ( loader ) <EOL> add_loader ( annmodel . SomeInteger ( ) , load_int ) <EOL> def dump_longlong ( buf , x ) : <EOL> buf . append ( TYPE_INT64 ) <EOL> w_long ( buf , intmask ( x ) ) <EOL> w_long ( buf , intmask ( x >> <NUM_LIT:32> ) ) <EOL> add_dumper ( annotation ( r_longlong ) , dump_longlong ) <EOL> r_32bits_mask = r_longlong ( <NUM_LIT> ) <EOL> def load_longlong ( loader ) : <EOL> if readchr ( loader ) != TYPE_INT64 : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> x = r_longlong ( readlong ( loader ) ) & r_32bits_mask <EOL> x |= ( r_longlong ( readlong ( loader ) ) << <NUM_LIT:32> ) <EOL> return x <EOL> add_loader ( annotation ( r_longlong ) , load_longlong ) <EOL> def dump_float ( buf , x ) : <EOL> buf . append ( TYPE_FLOAT ) <EOL> s = formatd ( "<STR_LIT>" , x ) <EOL> buf . append ( chr ( len ( s ) ) ) <EOL> buf += s <EOL> add_dumper ( annmodel . SomeFloat ( ) , dump_float ) <EOL> def load_float ( loader ) : <EOL> if readchr ( loader ) != TYPE_FLOAT : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> length = ord ( readchr ( loader ) ) <EOL> s = readstr ( loader , length ) <EOL> return parts_to_float ( * break_up_float ( s ) ) <EOL> add_loader ( annmodel . SomeFloat ( ) , load_float ) <EOL> def dump_string_or_none ( buf , x ) : <EOL> if x is None : <EOL> dump_none ( buf , x ) <EOL> else : <EOL> buf . append ( TYPE_STRING ) <EOL> w_long ( buf , len ( x ) ) <EOL> buf += x <EOL> add_dumper ( annmodel . SomeString ( can_be_None = True ) , dump_string_or_none ) <EOL> def load_single_char ( loader ) : <EOL> if readchr ( loader ) != TYPE_STRING or readlong ( loader ) != <NUM_LIT:1> : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return readchr ( loader ) <EOL> add_loader ( annmodel . SomeChar ( ) , load_single_char ) <EOL> def load_string ( loader ) : <EOL> if readchr ( loader ) != TYPE_STRING : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> length = readlong ( loader ) <EOL> return readstr ( loader , length ) <EOL> add_loader ( annmodel . SomeString ( can_be_None = False ) , load_string ) <EOL> def load_string_or_none ( loader ) : <EOL> t = readchr ( loader ) <EOL> if t == TYPE_STRING : <EOL> length = readlong ( loader ) <EOL> return readstr ( loader , length ) <EOL> elif t == TYPE_NONE : <EOL> return None <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> add_loader ( annmodel . SomeString ( can_be_None = True ) , load_string_or_none ) <EOL> class Loader ( object ) : <EOL> def __init__ ( self , buf ) : <EOL> self . buf = buf <EOL> self . pos = <NUM_LIT:0> <EOL> def check_finished ( self ) : <EOL> if self . pos != len ( self . buf ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> def need_more_data ( self ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> def readstr ( loader , count ) : <EOL> if count < <NUM_LIT:0> : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> pos = loader . pos <EOL> end = pos + count <EOL> while end > len ( loader . buf ) : <EOL> loader . need_more_data ( ) <EOL> loader . pos = end <EOL> return loader . buf [ pos : end ] <EOL> def readchr ( loader ) : <EOL> pos = loader . pos <EOL> while pos >= len ( loader . buf ) : <EOL> loader . need_more_data ( ) <EOL> loader . pos = pos + <NUM_LIT:1> <EOL> return loader . buf [ pos ] <EOL> def readlong ( loader ) : <EOL> a = ord ( readchr ( loader ) ) <EOL> b = ord ( readchr ( loader ) ) <EOL> c = ord ( readchr ( loader ) ) <EOL> d = ord ( readchr ( loader ) ) <EOL> if d >= <NUM_LIT> : <EOL> d -= <NUM_LIT> <EOL> return a | ( b << <NUM_LIT:8> ) | ( c << <NUM_LIT:16> ) | ( d << <NUM_LIT> ) <EOL> class MTag ( object ) : <EOL> """<STR_LIT>""" <EOL> _tag = MTag ( ) <EOL> def weakly_contains ( s_bigger , s_smaller ) : <EOL> try : <EOL> s_union = annmodel . unionof ( s_bigger , s_smaller ) <EOL> return s_bigger . contains ( s_union ) <EOL> except ( annmodel . UnionError , TooLateForChange ) : <EOL> return False <EOL> class __extend__ ( pairtype ( MTag , annmodel . SomeObject ) ) : <EOL> def install_marshaller ( ( tag , s_obj ) ) : <EOL> if not hasattr ( s_obj , '<STR_LIT>' ) : <EOL> raise CannotMarshal ( s_obj ) <EOL> def dump_with_custom_reduce ( buf , x ) : <EOL> reduced_obj = fn_reduce ( x ) <EOL> reduceddumper ( buf , reduced_obj ) <EOL> s_reduced_obj , fn_reduce , fn_recreate = s_obj . _get_rmarshall_support_ ( ) <EOL> reduceddumper = get_marshaller ( s_reduced_obj ) <EOL> add_dumper ( s_obj , dump_with_custom_reduce ) <EOL> def install_unmarshaller ( ( tag , s_obj ) ) : <EOL> if not hasattr ( s_obj , '<STR_LIT>' ) : <EOL> raise CannotUnmarshall ( s_obj ) <EOL> def load_with_custom_recreate ( loader ) : <EOL> reduced_obj = reducedloader ( loader ) <EOL> return fn_recreate ( reduced_obj ) <EOL> s_reduced_obj , fn_reduce , fn_recreate = s_obj . _get_rmarshall_support_ ( ) <EOL> reducedloader = get_loader ( s_reduced_obj ) <EOL> add_loader ( s_obj , load_with_custom_recreate ) <EOL> class __extend__ ( pairtype ( MTag , annmodel . SomeList ) ) : <EOL> def install_marshaller ( ( tag , s_list ) ) : <EOL> def dump_list_or_none ( buf , x ) : <EOL> if x is None : <EOL> dump_none ( buf , x ) <EOL> else : <EOL> buf . append ( TYPE_LIST ) <EOL> w_long ( buf , len ( x ) ) <EOL> for item in x : <EOL> itemdumper ( buf , item ) <EOL> itemdumper = get_marshaller ( s_list . listdef . listitem . s_value ) <EOL> if s_list . listdef . listitem . dont_change_any_more : <EOL> s_general_list = s_list <EOL> else : <EOL> s_item = get_dumper_annotation ( itemdumper ) <EOL> s_general_list = annotation ( [ s_item ] ) <EOL> add_dumper ( s_general_list , dump_list_or_none ) <EOL> def install_unmarshaller ( ( tag , s_list ) ) : <EOL> def load_list_or_none ( loader ) : <EOL> t = readchr ( loader ) <EOL> if t == TYPE_LIST : <EOL> length = readlong ( loader ) <EOL> result = [ ] <EOL> for i in range ( length ) : <EOL> result . append ( itemloader ( loader ) ) <EOL> return result <EOL> elif t == TYPE_NONE : <EOL> return None <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> itemloader = get_loader ( s_list . listdef . listitem . s_value ) <EOL> add_loader ( s_list , load_list_or_none ) <EOL> class __extend__ ( pairtype ( MTag , annmodel . SomeTuple ) ) : <EOL> def install_marshaller ( ( tag , s_tuple ) ) : <EOL> def dump_tuple ( buf , x ) : <EOL> buf . append ( TYPE_TUPLE ) <EOL> w_long ( buf , len ( x ) ) <EOL> for i , itemdumper in unroll_item_dumpers : <EOL> itemdumper ( buf , x [ i ] ) <EOL> itemdumpers = [ get_marshaller ( s_item ) for s_item in s_tuple . items ] <EOL> unroll_item_dumpers = unrolling_iterable ( enumerate ( itemdumpers ) ) <EOL> dumper_annotations = [ get_dumper_annotation ( itemdumper ) <EOL> for itemdumper in itemdumpers ] <EOL> s_general_tuple = annmodel . SomeTuple ( dumper_annotations ) <EOL> add_dumper ( s_general_tuple , dump_tuple ) <EOL> def install_unmarshaller ( ( tag , s_tuple ) ) : <EOL> def load_tuple ( loader ) : <EOL> if readchr ( loader ) != TYPE_TUPLE : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if readlong ( loader ) != expected_length : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> result = ( ) <EOL> for i , itemloader in unroll_item_loaders : <EOL> result += ( itemloader ( loader ) , ) <EOL> return result <EOL> itemloaders = [ get_loader ( s_item ) for s_item in s_tuple . items ] <EOL> expected_length = len ( itemloaders ) <EOL> unroll_item_loaders = unrolling_iterable ( enumerate ( itemloaders ) ) <EOL> add_loader ( s_tuple , load_tuple ) </s>
