<s> import functools <EOL> from . . import Variable <EOL> from . . conventions import cf_encoder <EOL> from . . core import indexing <EOL> from . . core . utils import FrozenOrderedDict , close_on_error , Frozen <EOL> from . . core . pycompat import iteritems , bytes_type , unicode_type , OrderedDict <EOL> from . common import WritableCFDataStore <EOL> from . netCDF4_ import _nc4_group , _nc4_values_and_dtype , _extract_nc4_encoding <EOL> def maybe_decode_bytes ( txt ) : <EOL> if isinstance ( txt , bytes_type ) : <EOL> return txt . decode ( '<STR_LIT:utf-8>' ) <EOL> else : <EOL> return txt <EOL> def _read_attributes ( h5netcdf_var ) : <EOL> attrs = OrderedDict ( ) <EOL> for k in h5netcdf_var . ncattrs ( ) : <EOL> v = h5netcdf_var . getncattr ( k ) <EOL> if k not in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> v = maybe_decode_bytes ( v ) <EOL> attrs [ k ] = v <EOL> return attrs <EOL> _extract_h5nc_encoding = functools . partial ( _extract_nc4_encoding , <EOL> lsd_okay = False , backend = '<STR_LIT>' ) <EOL> class H5NetCDFStore ( WritableCFDataStore ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , filename , mode = '<STR_LIT:r>' , format = None , group = None , <EOL> writer = None ) : <EOL> import h5netcdf . legacyapi <EOL> if format not in [ None , '<STR_LIT>' ] : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> ds = h5netcdf . legacyapi . Dataset ( filename , mode = mode ) <EOL> with close_on_error ( ds ) : <EOL> self . ds = _nc4_group ( ds , group , mode ) <EOL> self . format = format <EOL> self . _filename = filename <EOL> super ( H5NetCDFStore , self ) . __init__ ( writer ) <EOL> def open_store_variable ( self , var ) : <EOL> dimensions = var . dimensions <EOL> data = indexing . LazilyIndexedArray ( var ) <EOL> attrs = _read_attributes ( var ) <EOL> encoding = dict ( var . filters ( ) ) <EOL> chunking = var . chunking ( ) <EOL> encoding [ '<STR_LIT>' ] = chunking if chunking != '<STR_LIT>' else None <EOL> encoding [ '<STR_LIT:source>' ] = self . _filename <EOL> encoding [ '<STR_LIT>' ] = var . shape <EOL> return Variable ( dimensions , data , attrs , encoding ) <EOL> def get_variables ( self ) : <EOL> return FrozenOrderedDict ( ( k , self . open_store_variable ( v ) ) <EOL> for k , v in iteritems ( self . ds . variables ) ) <EOL> def get_attrs ( self ) : <EOL> return Frozen ( _read_attributes ( self . ds ) ) <EOL> def get_dimensions ( self ) : <EOL> return self . ds . dimensions <EOL> def set_dimension ( self , name , length ) : <EOL> self . ds . createDimension ( name , size = length ) <EOL> def set_attribute ( self , key , value ) : <EOL> self . ds . setncattr ( key , value ) <EOL> def prepare_variable ( self , name , variable , check_encoding = False ) : <EOL> import h5py <EOL> attrs = variable . attrs . copy ( ) <EOL> variable , dtype = _nc4_values_and_dtype ( variable ) <EOL> if dtype is str : <EOL> dtype = h5py . special_dtype ( vlen = unicode_type ) <EOL> self . set_necessary_dimensions ( variable ) <EOL> fill_value = attrs . pop ( '<STR_LIT>' , None ) <EOL> if fill_value in [ '<STR_LIT:\x00>' ] : <EOL> fill_value = None <EOL> encoding = _extract_h5nc_encoding ( variable , <EOL> raise_on_invalid = check_encoding ) <EOL> kwargs = { } <EOL> for key in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] : <EOL> if key in encoding : <EOL> kwargs [ key ] = encoding [ key ] <EOL> nc4_var = self . ds . createVariable ( name , dtype , variable . dims , <EOL> fill_value = fill_value , ** kwargs ) <EOL> for k , v in iteritems ( attrs ) : <EOL> nc4_var . setncattr ( k , v ) <EOL> return nc4_var , variable . data <EOL> def sync ( self ) : <EOL> super ( H5NetCDFStore , self ) . sync ( ) <EOL> self . ds . sync ( ) <EOL> def close ( self ) : <EOL> ds = self . ds <EOL> while ds . parent is not None : <EOL> ds = ds . parent <EOL> ds . close ( ) </s>
