<s> r"""<STR_LIT>""" <EOL> import io <EOL> from contextlib import contextmanager , ExitStack <EOL> from skbio . io import IOSourceError <EOL> from skbio . io . _iosources import get_io_sources , get_compression_handler <EOL> from skbio . io . _fileobject import ( <EOL> is_binary_file , SaneTextIOWrapper , CompressedBufferedReader , <EOL> CompressedBufferedWriter ) <EOL> from skbio . util . _decorator import stable <EOL> _d = dict ( mode = '<STR_LIT:r>' , encoding = None , errors = None , newline = None , <EOL> compression = '<STR_LIT>' , compresslevel = <NUM_LIT:9> ) <EOL> def _resolve ( file , mode = _d [ '<STR_LIT>' ] , encoding = _d [ '<STR_LIT>' ] , <EOL> errors = _d [ '<STR_LIT>' ] , newline = _d [ '<STR_LIT>' ] , <EOL> compression = _d [ '<STR_LIT>' ] , compresslevel = _d [ '<STR_LIT>' ] ) : <EOL> arguments = locals ( ) . copy ( ) <EOL> if mode not in { '<STR_LIT:r>' , '<STR_LIT:w>' } : <EOL> raise ValueError ( "<STR_LIT>" % mode ) <EOL> newfile = None <EOL> source = None <EOL> for source_handler in get_io_sources ( ) : <EOL> source = source_handler ( file , arguments ) <EOL> if mode == '<STR_LIT:r>' and source . can_read ( ) : <EOL> newfile = source . get_reader ( ) <EOL> break <EOL> elif mode == '<STR_LIT:w>' and source . can_write ( ) : <EOL> newfile = source . get_writer ( ) <EOL> break <EOL> if newfile is None : <EOL> raise IOSourceError ( <EOL> "<STR_LIT>" % ( file , mode ) ) <EOL> return newfile , source , is_binary_file ( newfile ) <EOL> @ stable ( as_of = "<STR_LIT>" ) <EOL> def open ( file , mode = _d [ '<STR_LIT>' ] , encoding = _d [ '<STR_LIT>' ] , errors = _d [ '<STR_LIT>' ] , <EOL> newline = _d [ '<STR_LIT>' ] , compression = _d [ '<STR_LIT>' ] , <EOL> compresslevel = _d [ '<STR_LIT>' ] ) : <EOL> r"""<STR_LIT>""" <EOL> arguments = locals ( ) . copy ( ) <EOL> del arguments [ '<STR_LIT:file>' ] <EOL> file , _ , is_binary_file = _resolve ( file , ** arguments ) <EOL> return _munge_file ( file , is_binary_file , arguments ) <EOL> def _munge_file ( file , is_binary_file , arguments ) : <EOL> mode = arguments . get ( '<STR_LIT>' , _d [ '<STR_LIT>' ] ) <EOL> encoding = arguments . get ( '<STR_LIT>' , _d [ '<STR_LIT>' ] ) <EOL> errors = arguments . get ( '<STR_LIT>' , _d [ '<STR_LIT>' ] ) <EOL> newline = arguments . get ( '<STR_LIT>' , _d [ '<STR_LIT>' ] ) <EOL> compression = arguments . get ( '<STR_LIT>' , _d [ '<STR_LIT>' ] ) <EOL> is_output_binary = encoding == '<STR_LIT>' <EOL> newfile = file <EOL> compression_handler = get_compression_handler ( compression ) <EOL> if is_output_binary and newline is not _d [ '<STR_LIT>' ] : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if compression is not None and not compression_handler : <EOL> raise ValueError ( "<STR_LIT>" % compression ) <EOL> if is_binary_file : <EOL> if compression : <EOL> c = compression_handler ( newfile , arguments ) <EOL> if mode == '<STR_LIT:w>' : <EOL> newfile = CompressedBufferedWriter ( file , c . get_writer ( ) , <EOL> streamable = c . streamable ) <EOL> else : <EOL> newfile = CompressedBufferedReader ( file , c . get_reader ( ) ) <EOL> if not is_output_binary : <EOL> newfile = SaneTextIOWrapper ( newfile , encoding = encoding , <EOL> errors = errors , newline = newline ) <EOL> else : <EOL> if compression is not None and compression != '<STR_LIT>' : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if is_output_binary : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return newfile <EOL> @ contextmanager <EOL> def _resolve_file ( file , ** kwargs ) : <EOL> file , source , is_binary_file = _resolve ( file , ** kwargs ) <EOL> try : <EOL> yield file , source , is_binary_file <EOL> finally : <EOL> if source . closeable : <EOL> file . close ( ) <EOL> @ contextmanager <EOL> @ stable ( as_of = "<STR_LIT>" ) <EOL> def open_file ( file , ** kwargs ) : <EOL> r"""<STR_LIT>""" <EOL> with _resolve_file ( file , ** kwargs ) as ( file , source , is_binary_file ) : <EOL> newfile = _munge_file ( file , is_binary_file , source . options ) <EOL> try : <EOL> yield newfile <EOL> finally : <EOL> if not newfile . closed : <EOL> newfile . flush ( ) <EOL> _flush_compressor ( newfile ) <EOL> def _flush_compressor ( file ) : <EOL> if isinstance ( file , io . TextIOBase ) and hasattr ( file , '<STR_LIT>' ) : <EOL> file = file . buffer <EOL> if isinstance ( file , CompressedBufferedWriter ) and not file . streamable : <EOL> file . raw . close ( ) <EOL> @ contextmanager <EOL> @ stable ( as_of = "<STR_LIT>" ) <EOL> def open_files ( files , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> with ExitStack ( ) as stack : <EOL> yield [ stack . enter_context ( open_file ( f , ** kwargs ) ) for f in files ] </s>
