<s> import sys <EOL> from pypy . interpreter . gateway import ObjSpace , W_Root , interp2app <EOL> from pypy . interpreter . baseobjspace import Wrappable <EOL> from pypy . interpreter . typedef import TypeDef , interp_attrproperty <EOL> from pypy . interpreter . error import OperationError <EOL> from pypy . rlib . rarithmetic import intmask , r_uint <EOL> from pypy . rlib . objectmodel import keepalive_until_here <EOL> from pypy . rlib import rzlib <EOL> if intmask ( <NUM_LIT:2> ** <NUM_LIT> ) == - <NUM_LIT:2> ** <NUM_LIT> : <EOL> unsigned_to_signed_32bit = intmask <EOL> else : <EOL> def unsigned_to_signed_32bit ( x ) : <EOL> SIGN_EXTEND2 = <NUM_LIT:1> << <NUM_LIT> <EOL> return intmask ( ( x ^ SIGN_EXTEND2 ) - SIGN_EXTEND2 ) <EOL> def crc32 ( space , string , w_start = rzlib . CRC32_DEFAULT_START ) : <EOL> """<STR_LIT>""" <EOL> if space . is_true ( space . isinstance ( w_start , space . w_long ) ) : <EOL> num = space . bigint_w ( w_start ) <EOL> ustart = num . uintmask ( ) <EOL> elif space . is_true ( space . isinstance ( w_start , space . w_int ) ) : <EOL> start = space . int_w ( w_start ) <EOL> ustart = r_uint ( start ) <EOL> else : <EOL> raise OperationError ( space . w_TypeError , <EOL> space . wrap ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> checksum = rzlib . crc32 ( string , ustart ) <EOL> checksum = unsigned_to_signed_32bit ( checksum ) <EOL> return space . wrap ( checksum ) <EOL> crc32 . unwrap_spec = [ ObjSpace , '<STR_LIT>' , W_Root ] <EOL> def adler32 ( space , string , start = rzlib . ADLER32_DEFAULT_START ) : <EOL> """<STR_LIT>""" <EOL> checksum = rzlib . adler32 ( string , start ) <EOL> checksum = unsigned_to_signed_32bit ( checksum ) <EOL> return space . wrap ( checksum ) <EOL> adler32 . unwrap_spec = [ ObjSpace , '<STR_LIT>' , r_uint ] <EOL> def zlib_error ( space , msg ) : <EOL> w_module = space . getbuiltinmodule ( '<STR_LIT>' ) <EOL> w_error = space . getattr ( w_module , space . wrap ( '<STR_LIT:error>' ) ) <EOL> return OperationError ( w_error , space . wrap ( msg ) ) <EOL> def compress ( space , string , level = rzlib . Z_DEFAULT_COMPRESSION ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> try : <EOL> stream = rzlib . deflateInit ( level ) <EOL> except ValueError : <EOL> raise zlib_error ( space , "<STR_LIT>" ) <EOL> try : <EOL> result = rzlib . compress ( stream , string , rzlib . Z_FINISH ) <EOL> finally : <EOL> rzlib . deflateEnd ( stream ) <EOL> except rzlib . RZlibError , e : <EOL> raise zlib_error ( space , e . msg ) <EOL> return space . wrap ( result ) <EOL> compress . unwrap_spec = [ ObjSpace , '<STR_LIT>' , int ] <EOL> def decompress ( space , string , wbits = rzlib . MAX_WBITS , bufsize = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> try : <EOL> stream = rzlib . inflateInit ( wbits ) <EOL> except ValueError : <EOL> raise zlib_error ( space , "<STR_LIT>" ) <EOL> try : <EOL> result , _ , _ = rzlib . decompress ( stream , string , rzlib . Z_FINISH ) <EOL> finally : <EOL> rzlib . inflateEnd ( stream ) <EOL> except rzlib . RZlibError , e : <EOL> raise zlib_error ( space , e . msg ) <EOL> return space . wrap ( result ) <EOL> decompress . unwrap_spec = [ ObjSpace , '<STR_LIT>' , int , int ] <EOL> class ZLibObject ( Wrappable ) : <EOL> """<STR_LIT>""" <EOL> stream = rzlib . null_stream <EOL> def __init__ ( self , space ) : <EOL> self . space = space <EOL> self . _lock = space . allocate_lock ( ) <EOL> def lock ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _lock . acquire ( True ) <EOL> def unlock ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _lock . release ( ) <EOL> keepalive_until_here ( self ) <EOL> class Compress ( ZLibObject ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , space , level = rzlib . Z_DEFAULT_COMPRESSION , <EOL> method = rzlib . Z_DEFLATED , <EOL> wbits = rzlib . MAX_WBITS , <EOL> memLevel = rzlib . DEF_MEM_LEVEL , <EOL> strategy = rzlib . Z_DEFAULT_STRATEGY ) : <EOL> ZLibObject . __init__ ( self , space ) <EOL> try : <EOL> self . stream = rzlib . deflateInit ( level , method , wbits , <EOL> memLevel , strategy ) <EOL> except rzlib . RZlibError , e : <EOL> raise zlib_error ( self . space , e . msg ) <EOL> except ValueError : <EOL> raise OperationError ( space . w_ValueError , <EOL> space . wrap ( "<STR_LIT>" ) ) <EOL> def __del__ ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . stream : <EOL> rzlib . deflateEnd ( self . stream ) <EOL> self . stream = rzlib . null_stream <EOL> def compress ( self , data ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . lock ( ) <EOL> try : <EOL> if not self . stream : <EOL> raise zlib_error ( self . space , <EOL> "<STR_LIT>" ) <EOL> result = rzlib . compress ( self . stream , data ) <EOL> finally : <EOL> self . unlock ( ) <EOL> except rzlib . RZlibError , e : <EOL> raise zlib_error ( self . space , e . msg ) <EOL> return self . space . wrap ( result ) <EOL> compress . unwrap_spec = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> def flush ( self , mode = rzlib . Z_FINISH ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . lock ( ) <EOL> try : <EOL> if not self . stream : <EOL> raise zlib_error ( self . space , <EOL> "<STR_LIT>" ) <EOL> result = rzlib . compress ( self . stream , '<STR_LIT>' , mode ) <EOL> if mode == rzlib . Z_FINISH : <EOL> rzlib . deflateEnd ( self . stream ) <EOL> self . stream = rzlib . null_stream <EOL> finally : <EOL> self . unlock ( ) <EOL> except rzlib . RZlibError , e : <EOL> raise zlib_error ( self . space , e . msg ) <EOL> return self . space . wrap ( result ) <EOL> flush . unwrap_spec = [ '<STR_LIT>' , int ] <EOL> def Compress___new__ ( space , w_subtype , level = rzlib . Z_DEFAULT_COMPRESSION , <EOL> method = rzlib . Z_DEFLATED , <EOL> wbits = rzlib . MAX_WBITS , <EOL> memLevel = rzlib . DEF_MEM_LEVEL , <EOL> strategy = rzlib . Z_DEFAULT_STRATEGY ) : <EOL> """<STR_LIT>""" <EOL> stream = space . allocate_instance ( Compress , w_subtype ) <EOL> stream = space . interp_w ( Compress , stream ) <EOL> Compress . __init__ ( stream , space , level , <EOL> method , wbits , memLevel , strategy ) <EOL> return space . wrap ( stream ) <EOL> Compress___new__ . unwrap_spec = [ ObjSpace , W_Root , int , int , int , int , int ] <EOL> Compress . typedef = TypeDef ( <EOL> '<STR_LIT>' , <EOL> __new__ = interp2app ( Compress___new__ ) , <EOL> compress = interp2app ( Compress . compress ) , <EOL> flush = interp2app ( Compress . flush ) , <EOL> __doc__ = """<STR_LIT>""" ) <EOL> class Decompress ( ZLibObject ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , space , wbits = rzlib . MAX_WBITS ) : <EOL> """<STR_LIT>""" <EOL> ZLibObject . __init__ ( self , space ) <EOL> self . unused_data = '<STR_LIT>' <EOL> self . unconsumed_tail = '<STR_LIT>' <EOL> try : <EOL> self . stream = rzlib . inflateInit ( wbits ) <EOL> except rzlib . RZlibError , e : <EOL> raise zlib_error ( self . space , e . msg ) <EOL> except ValueError : <EOL> raise OperationError ( space . w_ValueError , <EOL> space . wrap ( "<STR_LIT>" ) ) <EOL> def __del__ ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . stream : <EOL> rzlib . inflateEnd ( self . stream ) <EOL> self . stream = rzlib . null_stream <EOL> def decompress ( self , data , max_length = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> if max_length == <NUM_LIT:0> : <EOL> max_length = sys . maxint <EOL> elif max_length < <NUM_LIT:0> : <EOL> raise OperationError ( self . space . w_ValueError , <EOL> self . space . wrap ( "<STR_LIT>" <EOL> "<STR_LIT>" ) ) <EOL> try : <EOL> self . lock ( ) <EOL> try : <EOL> result = rzlib . decompress ( self . stream , data , <EOL> max_length = max_length ) <EOL> finally : <EOL> self . unlock ( ) <EOL> except rzlib . RZlibError , e : <EOL> raise zlib_error ( self . space , e . msg ) <EOL> string , finished , unused_len = result <EOL> unused_start = len ( data ) - unused_len <EOL> assert unused_start >= <NUM_LIT:0> <EOL> tail = data [ unused_start : ] <EOL> if finished : <EOL> self . unconsumed_tail = '<STR_LIT>' <EOL> self . unused_data = tail <EOL> else : <EOL> self . unconsumed_tail = tail <EOL> return self . space . wrap ( string ) <EOL> decompress . unwrap_spec = [ '<STR_LIT>' , '<STR_LIT>' , int ] <EOL> def flush ( self , length = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> return self . space . wrap ( "<STR_LIT>" ) <EOL> flush . unwrap_spec = [ '<STR_LIT>' , int ] <EOL> def Decompress___new__ ( space , w_subtype , wbits = rzlib . MAX_WBITS ) : <EOL> """<STR_LIT>""" <EOL> stream = space . allocate_instance ( Decompress , w_subtype ) <EOL> stream = space . interp_w ( Decompress , stream ) <EOL> Decompress . __init__ ( stream , space , wbits ) <EOL> return space . wrap ( stream ) <EOL> Decompress___new__ . unwrap_spec = [ ObjSpace , W_Root , int ] <EOL> Decompress . typedef = TypeDef ( <EOL> '<STR_LIT>' , <EOL> __new__ = interp2app ( Decompress___new__ ) , <EOL> decompress = interp2app ( Decompress . decompress ) , <EOL> flush = interp2app ( Decompress . flush ) , <EOL> unused_data = interp_attrproperty ( '<STR_LIT>' , Decompress ) , <EOL> unconsumed_tail = interp_attrproperty ( '<STR_LIT>' , Decompress ) , <EOL> __doc__ = """<STR_LIT>""" ) </s>
