refactorization
- separate fs specific code to hdf5 specific codes
- user better names
- use generators instead of list
- move functions to class method as they should be overwriten for future implementations
--- a/h5fs.py Tue Apr 03 20:34:00 2012 +0200
+++ b/h5fs.py Tue Apr 02 12:21:13 2013 +0200
@@ -7,13 +7,15 @@
#
'''
-HDF% filesystem implementation.
+HDF5 filesystem implementation.
'''
import os
import os.path as osp
import stat
import errno
+from itertools import imap, chain
+from functools import partial
import h5py
@@ -59,7 +61,7 @@
h5file = None
- def _get_entry(self, path):
+ def get_entry(self, path):
'''Return the entry recorded at ``path`` or raise H5IOError'''
try:
h5entry = self.h5file[path]
@@ -68,34 +70,34 @@
return h5entry
@staticmethod
- def _readdir(h5entry, offset):
- '''A fonction generator that yields entries from the h5 entry'''
- for name in ('.', '..'):
- yield fuse.Direntry(name)
- for content in h5entry.itervalues():
- yield fuse.Direntry(get_name(content), offset=offset)
+ def get_stat(h5entry):
+ '''Return the entry stat information (an instance of Stat)'''
+ if isinstance(h5entry, h5py.highlevel.Group):
+ return GroupStat()
+ if isinstance(h5entry, h5py.highlevel.Dataset):
+ return DatasetStat()
+ raise H5IOError(errno.ENOENT, 'Unknown entry type', h5entry.name)
+
+ @staticmethod
+ def get_name(h5entry):
+ '''Return the entry file name as string'''
+ # XXX unicode seems not to be allowed
+ return str(osp.basename(h5entry.name))
+
+ def list_group(self, group, specs=('.', '..')):
+ '''A fonction generator that yields name of the entries in the group.
+ The list is in arbritary order with specials entries ``specs``.'''
+ return chain(iter(specs), imap(self.get_name, group.itervalues()))
def readdir(self, path, offset):
'''Return a generator that yields entries from the given Group path'''
- return self._readdir(self._get_entry(path), offset)
+ convert = partial(fuse.Direntry, offset=offset)
+ return imap(convert, self.list_group(self.get_entry(path)))
def getattr(self, path):
'''Return the stat attributes of the given ``path``'''
- return get_stat(self._get_entry(path))
-
+ return self.get_stat(self.get_entry(path))
-def get_stat(h5entry):
- '''Return the entry stat information (an instance of Stat)'''
- if isinstance(h5entry, h5py.highlevel.Group):
- return GroupStat()
- if isinstance(h5entry, h5py.highlevel.Dataset):
- return DatasetStat()
- raise H5IOError(errno.ENOENT, 'Unknown entry type', h5entry.name)
-
-def get_name(h5entry):
- '''Return the entry file name as string'''
- # XXX unicode seems not to be allowed
- return str(osp.basename(h5entry.name))
def __main__():
'''Main entry point that start the HDF5 filesystem server'''
--- a/test/test_H5FS.py Tue Apr 03 20:34:00 2012 +0200
+++ b/test/test_H5FS.py Tue Apr 02 12:21:13 2013 +0200
@@ -7,60 +7,69 @@
from logilab.common.testlib import within_tempdir
import h5fs
+FSM = h5fs.FsMixin()
__folder__=osp.dirname(osp.abspath(__file__))
def getdata(path):
return osp.join(__folder__, 'data', path)
-class GetStat_TC(TC):
-
+def with_h5file(func):
@within_tempdir
- def test_group(self):
+ def wrapped(self):
h5file = h5py.File('file.hdf5')
- group = h5file.create_group(u'sam')
- self.assertTrue(isinstance(h5fs.get_stat(group), h5fs.GroupStat))
+ out = func(self, h5file)
h5file.close()
+ return out
+ return wrapped
- @within_tempdir
- def test_dataset(self):
- h5file = h5py.File('file.hdf5')
- group = h5file.create_dataset(u'sam', [])
- self.assertTrue(isinstance(h5fs.get_stat(group), h5fs.DatasetStat))
- h5file.close()
+def with_fsmixin(func):
+ @with_h5file
+ def wrapped(self, h5file):
+ h5file.create_group(u'sam')
+ h5file.create_dataset(u'melu', data=xrange(12), shape=(4, 3), dtype='u8')
+ fsmx = h5fs.FsMixin()
+ fsmx.h5file = h5file
+ return func(self, fsmx)
+ return wrapped
+
class FsMixin_TC(TC):
- def init(self):
- h5file = h5py.File('file.hdf5')
- h5file.create_group(u'sam')
- h5file.create_dataset(u'melu', (0,))
- h5mx = h5fs.FsMixin()
- h5mx.h5file = h5file
- return h5file, h5mx
+ @with_fsmixin
+ def test_get_stat_with_group(self, fsmx):
+ group = fsmx.h5file['sam']
+ self.assertTrue(isinstance(fsmx.get_stat(group), h5fs.GroupStat))
+
+ @with_fsmixin
+ def test_get_stat_with_dataset(self, fsmx):
+ dataset = fsmx.h5file['melu']
+ self.assertTrue(isinstance(fsmx.get_stat(dataset), h5fs.DatasetStat))
- @within_tempdir
- def test_readdir(self):
- h5file, h5mx = self.init()
- result = list(h5mx.readdir(u'/', 0))
- expected = list(h5file[u'/'].keys()) + ['.', '..']
+ @with_fsmixin
+ def test_get_name(self, fsmx):
+ dataset = fsmx.h5file['melu']
+ self.assertEqual(FSM.get_name(dataset), 'melu')
+
+ @with_fsmixin
+ def test_readdir(self, fsmx):
+ result = list(fsmx.readdir(u'/', 0))
+ expected = list(fsmx.h5file[u'/'].keys()) + ['.', '..']
self.assertEqual(len(expected), len(result))
- @within_tempdir
- def test_readdir_wrong(self):
- h5file, h5mx = self.init()
- self.assertRaises(h5fs.H5IOError, h5mx.readdir, u'/wrong data path', 0)
+ @with_fsmixin
+ def test_readdir_wrong(self, fsmx):
+ self.assertRaises(h5fs.H5IOError, fsmx.readdir, u'/wrong data path', 0)
- @within_tempdir
- def test_getattr(self):
- h5file, h5mx = self.init()
- self.assertTrue(isinstance(h5mx.getattr('/'), h5fs.GroupStat))
- self.assertTrue(isinstance(h5mx.getattr('/sam'), h5fs.GroupStat))
- self.assertTrue(isinstance(h5mx.getattr('/melu'), h5fs.DatasetStat))
+ @with_fsmixin
+ def test_getattr(self, fsmx):
+ self.assertTrue(isinstance(fsmx.getattr('/'), h5fs.GroupStat))
+ self.assertTrue(isinstance(fsmx.getattr('/sam'), h5fs.GroupStat))
+ self.assertTrue(isinstance(fsmx.getattr('/melu'), h5fs.DatasetStat))
- @within_tempdir
- def test_getattr_wrong(self):
- h5file, h5mx = self.init()
- self.assertRaises(h5fs.H5IOError, h5mx.getattr, u'/wrong data path')
+ @with_fsmixin
+ def test_getattr_wrong(self, fsmx):
+ self.assertRaises(h5fs.H5IOError, fsmx.getattr, u'/wrong data path')
+
if __name__ == '__main__':
main()