<s> """<STR_LIT>""" <EOL> import bz2 <EOL> import gzip <EOL> import sys <EOL> from os . path import splitext <EOL> GZIP_MAX_READ_CHUNK = <NUM_LIT:100> * <NUM_LIT> * <NUM_LIT> <EOL> class BufferedGzipFile ( gzip . GzipFile ) : <EOL> """<STR_LIT>""" <EOL> if sys . version_info [ : <NUM_LIT:3> ] == ( <NUM_LIT:3> , <NUM_LIT:5> , <NUM_LIT:0> ) : <EOL> def __init__ ( self , fileish , mode = '<STR_LIT:rb>' , compresslevel = <NUM_LIT:9> , <EOL> buffer_size = <NUM_LIT:2> ** <NUM_LIT:32> - <NUM_LIT:1> ) : <EOL> super ( BufferedGzipFile , self ) . __init__ ( fileish , mode = mode , <EOL> compresslevel = compresslevel ) <EOL> self . buffer_size = buffer_size <EOL> def readinto ( self , buf ) : <EOL> """<STR_LIT>""" <EOL> n_bytes = len ( buf ) <EOL> if n_bytes < <NUM_LIT:2> ** <NUM_LIT:32> : <EOL> return super ( BufferedGzipFile , self ) . readinto ( buf ) <EOL> mv = memoryview ( buf ) <EOL> n_read = <NUM_LIT:0> <EOL> max_read = <NUM_LIT:2> ** <NUM_LIT:32> - <NUM_LIT:1> <EOL> while ( n_read < n_bytes ) : <EOL> n_wanted = min ( n_bytes - n_read , max_read ) <EOL> n_got = super ( BufferedGzipFile , self ) . readinto ( <EOL> mv [ n_read : n_read + n_wanted ] ) <EOL> n_read += n_got <EOL> if n_got != n_wanted : <EOL> break <EOL> return n_read <EOL> def _gzip_open ( fileish , * args , ** kwargs ) : <EOL> gzip_file = BufferedGzipFile ( fileish , * args , ** kwargs ) <EOL> if hasattr ( gzip_file , '<STR_LIT>' ) : <EOL> gzip_file . max_read_chunk = GZIP_MAX_READ_CHUNK <EOL> return gzip_file <EOL> class Opener ( object ) : <EOL> """<STR_LIT>""" <EOL> gz_def = ( _gzip_open , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> bz2_def = ( bz2 . BZ2File , ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> compress_ext_map = { <EOL> '<STR_LIT>' : gz_def , <EOL> '<STR_LIT>' : bz2_def , <EOL> None : ( open , ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> } <EOL> default_compresslevel = <NUM_LIT:1> <EOL> compress_ext_icase = True <EOL> def __init__ ( self , fileish , * args , ** kwargs ) : <EOL> if self . _is_fileobj ( fileish ) : <EOL> self . fobj = fileish <EOL> self . me_opened = False <EOL> self . _name = None <EOL> return <EOL> opener , arg_names = self . _get_opener_argnames ( fileish ) <EOL> full_kwargs = kwargs . copy ( ) <EOL> n_args = len ( args ) <EOL> full_kwargs . update ( dict ( zip ( arg_names [ : n_args ] , args ) ) ) <EOL> if '<STR_LIT>' not in full_kwargs : <EOL> kwargs [ '<STR_LIT>' ] = '<STR_LIT:rb>' <EOL> if '<STR_LIT>' in arg_names and '<STR_LIT>' not in kwargs : <EOL> kwargs [ '<STR_LIT>' ] = self . default_compresslevel <EOL> self . fobj = opener ( fileish , * args , ** kwargs ) <EOL> self . _name = fileish <EOL> self . me_opened = True <EOL> def _get_opener_argnames ( self , fileish ) : <EOL> _ , ext = splitext ( fileish ) <EOL> if self . compress_ext_icase : <EOL> ext = ext . lower ( ) <EOL> for key in self . compress_ext_map : <EOL> if key is None : <EOL> continue <EOL> if key . lower ( ) == ext : <EOL> return self . compress_ext_map [ key ] <EOL> elif ext in self . compress_ext_map : <EOL> return self . compress_ext_map [ ext ] <EOL> return self . compress_ext_map [ None ] <EOL> def _is_fileobj ( self , obj ) : <EOL> """<STR_LIT>""" <EOL> return hasattr ( obj , '<STR_LIT>' ) and hasattr ( obj , '<STR_LIT>' ) <EOL> @ property <EOL> def closed ( self ) : <EOL> return self . fobj . closed <EOL> @ property <EOL> def name ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self . fobj . name <EOL> except AttributeError : <EOL> return self . _name <EOL> @ property <EOL> def mode ( self ) : <EOL> return self . fobj . mode <EOL> def fileno ( self ) : <EOL> return self . fobj . fileno ( ) <EOL> def read ( self , * args , ** kwargs ) : <EOL> return self . fobj . read ( * args , ** kwargs ) <EOL> def write ( self , * args , ** kwargs ) : <EOL> return self . fobj . write ( * args , ** kwargs ) <EOL> def seek ( self , * args , ** kwargs ) : <EOL> return self . fobj . seek ( * args , ** kwargs ) <EOL> def tell ( self , * args , ** kwargs ) : <EOL> return self . fobj . tell ( * args , ** kwargs ) <EOL> def close ( self , * args , ** kwargs ) : <EOL> return self . fobj . close ( * args , ** kwargs ) <EOL> def __iter__ ( self ) : <EOL> return iter ( self . fobj ) <EOL> def close_if_mine ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . me_opened : <EOL> self . close ( ) <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , exc_type , exc_val , exc_tb ) : <EOL> self . close_if_mine ( ) <EOL> class ImageOpener ( Opener ) : <EOL> """<STR_LIT>""" <EOL> compress_ext_map = Opener . compress_ext_map . copy ( ) </s>
