<s> import sys <EOL> import tempfile <EOL> from shutil import rmtree <EOL> from os import path as op <EOL> import numpy as np <EOL> try : <EOL> from scipy import sparse <EOL> except ImportError : <EOL> sparse = None <EOL> PY3 = sys . version_info [ <NUM_LIT:0> ] == <NUM_LIT:3> <EOL> text_type = str if PY3 else unicode <EOL> string_types = str if PY3 else basestring <EOL> def _check_h5py ( ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> import h5py <EOL> except ImportError : <EOL> raise ImportError ( '<STR_LIT>' ) <EOL> return h5py <EOL> def _create_titled_group ( root , key , title ) : <EOL> """<STR_LIT>""" <EOL> out = root . create_group ( key ) <EOL> out . attrs [ '<STR_LIT>' ] = title <EOL> return out <EOL> def _create_titled_dataset ( root , key , title , data , comp_kw = None ) : <EOL> """<STR_LIT>""" <EOL> comp_kw = { } if comp_kw is None else comp_kw <EOL> out = root . create_dataset ( key , data = data , ** comp_kw ) <EOL> out . attrs [ '<STR_LIT>' ] = title <EOL> return out <EOL> def write_hdf5 ( fname , data , overwrite = False , compression = <NUM_LIT:4> , <EOL> title = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> h5py = _check_h5py ( ) <EOL> if op . isfile ( fname ) and not overwrite : <EOL> raise IOError ( '<STR_LIT>' <EOL> % fname ) <EOL> if not isinstance ( title , string_types ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> comp_kw = dict ( ) <EOL> if compression > <NUM_LIT:0> : <EOL> comp_kw = dict ( compression = '<STR_LIT>' , compression_opts = compression ) <EOL> with h5py . File ( fname , mode = '<STR_LIT:w>' ) as fid : <EOL> _triage_write ( title , data , fid , comp_kw , str ( type ( data ) ) ) <EOL> def _triage_write ( key , value , root , comp_kw , where ) : <EOL> if isinstance ( value , dict ) : <EOL> sub_root = _create_titled_group ( root , key , '<STR_LIT>' ) <EOL> for key , sub_value in value . items ( ) : <EOL> if not isinstance ( key , string_types ) : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> _triage_write ( '<STR_LIT>' . format ( key ) , sub_value , sub_root , comp_kw , <EOL> where + '<STR_LIT>' % key ) <EOL> elif isinstance ( value , ( list , tuple ) ) : <EOL> title = '<STR_LIT:list>' if isinstance ( value , list ) else '<STR_LIT>' <EOL> sub_root = _create_titled_group ( root , key , title ) <EOL> for vi , sub_value in enumerate ( value ) : <EOL> _triage_write ( '<STR_LIT>' . format ( vi ) , sub_value , sub_root , comp_kw , <EOL> where + '<STR_LIT>' % vi ) <EOL> elif isinstance ( value , type ( None ) ) : <EOL> _create_titled_dataset ( root , key , '<STR_LIT:None>' , [ False ] ) <EOL> elif isinstance ( value , ( int , float ) ) : <EOL> if isinstance ( value , int ) : <EOL> title = '<STR_LIT:int>' <EOL> else : <EOL> title = '<STR_LIT:float>' <EOL> _create_titled_dataset ( root , key , title , np . atleast_1d ( value ) ) <EOL> elif isinstance ( value , string_types ) : <EOL> if isinstance ( value , text_type ) : <EOL> value = np . fromstring ( value . encode ( '<STR_LIT:utf-8>' ) , np . uint8 ) <EOL> title = '<STR_LIT>' <EOL> else : <EOL> value = np . fromstring ( value . encode ( '<STR_LIT>' ) , np . uint8 ) <EOL> title = '<STR_LIT:ascii>' <EOL> _create_titled_dataset ( root , key , title , value , comp_kw ) <EOL> elif isinstance ( value , np . ndarray ) : <EOL> _create_titled_dataset ( root , key , '<STR_LIT>' , value ) <EOL> elif sparse is not None and isinstance ( value , sparse . csc_matrix ) : <EOL> sub_root = _create_titled_group ( root , key , '<STR_LIT>' ) <EOL> _triage_write ( '<STR_LIT:data>' , value . data , sub_root , comp_kw , <EOL> where + '<STR_LIT>' ) <EOL> _triage_write ( '<STR_LIT>' , value . indices , sub_root , comp_kw , <EOL> where + '<STR_LIT>' ) <EOL> _triage_write ( '<STR_LIT>' , value . indptr , sub_root , comp_kw , <EOL> where + '<STR_LIT>' ) <EOL> else : <EOL> raise TypeError ( '<STR_LIT>' % ( type ( value ) , where ) ) <EOL> def read_hdf5 ( fname , title = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> h5py = _check_h5py ( ) <EOL> if not op . isfile ( fname ) : <EOL> raise IOError ( '<STR_LIT>' % fname ) <EOL> if not isinstance ( title , string_types ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> with h5py . File ( fname , mode = '<STR_LIT:r>' ) as fid : <EOL> if title not in fid . keys ( ) : <EOL> raise ValueError ( '<STR_LIT>' % title ) <EOL> data = _triage_read ( fid [ title ] ) <EOL> return data <EOL> def _triage_read ( node ) : <EOL> h5py = _check_h5py ( ) <EOL> type_str = node . attrs [ '<STR_LIT>' ] <EOL> if isinstance ( type_str , bytes ) : <EOL> type_str = type_str . decode ( ) <EOL> if isinstance ( node , h5py . Group ) : <EOL> if type_str == '<STR_LIT>' : <EOL> data = dict ( ) <EOL> for key , subnode in node . items ( ) : <EOL> data [ key [ <NUM_LIT:4> : ] ] = _triage_read ( subnode ) <EOL> elif type_str in [ '<STR_LIT:list>' , '<STR_LIT>' ] : <EOL> data = list ( ) <EOL> ii = <NUM_LIT:0> <EOL> while True : <EOL> subnode = node . get ( '<STR_LIT>' . format ( ii ) , None ) <EOL> if subnode is None : <EOL> break <EOL> data . append ( _triage_read ( subnode ) ) <EOL> ii += <NUM_LIT:1> <EOL> assert len ( data ) == ii <EOL> data = tuple ( data ) if type_str == '<STR_LIT>' else data <EOL> return data <EOL> elif type_str == '<STR_LIT>' : <EOL> if sparse is None : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> data = sparse . csc_matrix ( ( _triage_read ( node [ '<STR_LIT:data>' ] ) , <EOL> _triage_read ( node [ '<STR_LIT>' ] ) , <EOL> _triage_read ( node [ '<STR_LIT>' ] ) ) ) <EOL> else : <EOL> raise NotImplementedError ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( type_str ) ) <EOL> elif type_str == '<STR_LIT>' : <EOL> data = np . array ( node ) <EOL> elif type_str in ( '<STR_LIT:int>' , '<STR_LIT:float>' ) : <EOL> cast = int if type_str == '<STR_LIT:int>' else float <EOL> data = cast ( np . array ( node ) [ <NUM_LIT:0> ] ) <EOL> elif type_str in ( '<STR_LIT>' , '<STR_LIT:ascii>' , '<STR_LIT:str>' ) : <EOL> decoder = '<STR_LIT:utf-8>' if type_str == '<STR_LIT>' else '<STR_LIT>' <EOL> cast = text_type if type_str == '<STR_LIT>' else str <EOL> data = cast ( np . array ( node ) . tostring ( ) . decode ( decoder ) ) <EOL> elif type_str == '<STR_LIT:None>' : <EOL> data = None <EOL> else : <EOL> raise TypeError ( '<STR_LIT>' . format ( type_str ) ) <EOL> return data <EOL> def _sort_keys ( x ) : <EOL> """<STR_LIT>""" <EOL> keys = list ( x . keys ( ) ) <EOL> idx = np . argsort ( [ str ( k ) for k in keys ] ) <EOL> keys = [ keys [ ii ] for ii in idx ] <EOL> return keys <EOL> def object_diff ( a , b , pre = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> out = '<STR_LIT>' <EOL> if type ( a ) != type ( b ) : <EOL> out += pre + '<STR_LIT>' % ( type ( a ) , type ( b ) ) <EOL> elif isinstance ( a , dict ) : <EOL> k1s = _sort_keys ( a ) <EOL> k2s = _sort_keys ( b ) <EOL> m1 = set ( k2s ) - set ( k1s ) <EOL> if len ( m1 ) : <EOL> out += pre + '<STR_LIT>' % ( m1 ) <EOL> for key in k1s : <EOL> if key not in k2s : <EOL> out += pre + '<STR_LIT>' % key <EOL> else : <EOL> out += object_diff ( a [ key ] , b [ key ] , pre + '<STR_LIT>' % repr ( key ) ) <EOL> elif isinstance ( a , ( list , tuple ) ) : <EOL> if len ( a ) != len ( b ) : <EOL> out += pre + '<STR_LIT>' % ( len ( a ) , len ( b ) ) <EOL> else : <EOL> for xx1 , xx2 in zip ( a , b ) : <EOL> out += object_diff ( xx1 , xx2 , pre = '<STR_LIT>' ) <EOL> elif isinstance ( a , ( string_types , int , float , bytes ) ) : <EOL> if a != b : <EOL> out += pre + '<STR_LIT>' % ( a , b ) <EOL> elif a is None : <EOL> pass <EOL> elif isinstance ( a , np . ndarray ) : <EOL> if not np . array_equal ( a , b ) : <EOL> out += pre + '<STR_LIT>' <EOL> elif sparse is not None and sparse . isspmatrix ( a ) : <EOL> if b . shape != a . shape : <EOL> out += pre + ( '<STR_LIT>' <EOL> '<STR_LIT>' % ( a . shape , b . shape ) ) <EOL> else : <EOL> c = a - b <EOL> c . eliminate_zeros ( ) <EOL> if c . nnz > <NUM_LIT:0> : <EOL> out += pre + ( '<STR_LIT>' <EOL> '<STR_LIT>' % c . nnz ) <EOL> else : <EOL> raise RuntimeError ( pre + '<STR_LIT>' % ( type ( a ) , a ) ) <EOL> return out <EOL> class _TempDir ( str ) : <EOL> """<STR_LIT>""" <EOL> def __new__ ( self ) : <EOL> new = str . __new__ ( self , tempfile . mkdtemp ( ) ) <EOL> return new <EOL> def __init__ ( self ) : <EOL> self . _path = self . __str__ ( ) <EOL> def __del__ ( self ) : <EOL> rmtree ( self . _path , ignore_errors = True ) </s>
