/usr/lib/python3/dist-packages/xarray/backends/pynio_.py is in python3-xarray 0.10.2-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | from __future__ import absolute_import, division, print_function
import functools
import numpy as np
from .. import Variable
from ..core import indexing
from ..core.utils import Frozen, FrozenOrderedDict
from .common import AbstractDataStore, BackendArray, DataStorePickleMixin
class NioArrayWrapper(BackendArray):
def __init__(self, variable_name, datastore):
self.datastore = datastore
self.variable_name = variable_name
array = self.get_array()
self.shape = array.shape
self.dtype = np.dtype(array.typecode())
def get_array(self):
self.datastore.assert_open()
return self.datastore.ds.variables[self.variable_name]
def __getitem__(self, key):
key, np_inds = indexing.decompose_indexer(
key, self.shape, indexing.IndexingSupport.BASIC)
with self.datastore.ensure_open(autoclose=True):
array = self.get_array()
if key.tuple == () and self.ndim == 0:
return array.get_value()
array = array[key.tuple]
if len(np_inds.tuple) > 0:
array = indexing.NumpyIndexingAdapter(array)[np_inds]
return array
class NioDataStore(AbstractDataStore, DataStorePickleMixin):
"""Store for accessing datasets via PyNIO
"""
def __init__(self, filename, mode='r', autoclose=False):
import Nio
opener = functools.partial(Nio.open_file, filename, mode=mode)
self._ds = opener()
self._autoclose = autoclose
self._isopen = True
self._opener = opener
self._mode = mode
# xarray provides its own support for FillValue,
# so turn off PyNIO's support for the same.
self.ds.set_option('MaskedArrayMode', 'MaskedNever')
def open_store_variable(self, name, var):
data = indexing.LazilyOuterIndexedArray(NioArrayWrapper(name, self))
return Variable(var.dimensions, data, var.attributes)
def get_variables(self):
with self.ensure_open(autoclose=False):
return FrozenOrderedDict((k, self.open_store_variable(k, v))
for k, v in self.ds.variables.items())
def get_attrs(self):
with self.ensure_open(autoclose=True):
return Frozen(self.ds.attributes)
def get_dimensions(self):
with self.ensure_open(autoclose=True):
return Frozen(self.ds.dimensions)
def get_encoding(self):
encoding = {}
encoding['unlimited_dims'] = set(
[k for k in self.ds.dimensions if self.ds.unlimited(k)])
return encoding
def close(self):
if self._isopen:
self.ds.close()
self._isopen = False
|