Merge pull request #1 from marqh/validation

Validation

Author: marqh

.gitignore

@@ -60,3 +60,6 @@ target/
 
 #Ipython Notebook
 .ipynb_checkpoints
+
+*~
+

lib/bald/__init__.py

@@ -0,0 +1,112 @@
+import contextlib
+import re
+
+import h5py
+import requests
+
+from bald.validation import ContainerValidation, DatasetValidation
+
+
+class HttpCache(object):
+    """
+    Requests cache.
+    """
+    def __init__(self):
+        self.cache = {}
+
+    def __getitem__(self, item):
+
+        if not item.startswith('http://') or item.startswith('https://'):
+            raise ValueError('{} is not a HTTP URI.'.format(item))
+        if item not in self.cache:
+            headers={'Accept':'text/turtle'}
+            self.cache[item] = requests.get(item, headers=headers)
+
+        return self.cache[item]
+        
+
+    def check_uri(self, uri):
+        result = False
+        if self[uri].status_code == 200:
+            result = True
+        return result
+
+class Subject(object):
+    def __init__(self, attrs=None):
+        """
+        A subject of metadata statements.
+
+        attrs: an dictionary of key value pair attributes
+        """
+        if attrs is None:
+            attrs = []
+        self.attrs = attrs
+        self._prefix_suffix = re.compile('(^(?:(?!__).)*)__((?!.*__).*$)')
+        _http_p = 'http[s]?://.*'
+        self._http_uri = re.compile('{}'.format(_http_p))
+        self._http_uri_prefix = re.compile('{}/|#'.format(_http_p))
+
+
+    def prefixes(self):
+        prefixes = {}
+        for key, value in self.attrs.iteritems():
+            if key.endswith('__') and self._http_uri_prefix.match(value):
+                pref = key.rstrip('__')
+                if prefixes.has_key(pref):
+                    raise ValueError('This container has conflicting prefix definitions')
+                prefixes[pref] = value
+        return prefixes
+
+
+    def unpack_uri(self, astring):
+        result = astring
+        if self._prefix_suffix.match(astring):
+            prefix, suffix = self._prefix_suffix.match(astring).groups()
+
+            if self.prefixes().has_key(prefix):
+                if self._http_uri.match(self.prefixes()[prefix]):
+                    result = astring.replace('{}__'.format(prefix), self.prefixes()[prefix])
+        return result
+
+@contextlib.contextmanager
+def load(afilepath):
+    if afilepath.endswith('.hdf'):
+        loader = h5py.File
+    else:
+        raise ValueError('filepath suffix not supported')
+    try:
+        f = loader(afilepath, "r")
+        yield f
+    finally:
+        f.close()
+
+
+def validate_hdf5(afilepath):
+    """
+    Validate a file with respect ot binarry-array-linked-data.
+    Returns a :class:`bald.validation.Validation` 
+    """
+    
+    with load(afilepath) as fhandle:
+        valid = True
+        cache = {}
+        root_container = Subject(fhandle.attrs)
+        root_val = ContainerValidation(subject=root_container, fhandle=fhandle)
+        if not root_val.is_valid():
+            valid = False
+        # iterate through the datasets
+        for name, dataset in fhandle.items():
+            # a dataset's attribute collection inherits from and specialises it's
+            # container's attrbiute collection
+            sattrs = dict(fhandle.attrs).copy()
+            sattrs.update(dataset.attrs)
+            dset = Subject(sattrs)
+            dset_val = DatasetValidation(name, dataset, fhandle=fhandle, subject=dset)
+            if not dset_val.is_valid():
+                valid = False
+        
+        return valid
+        
+        
+
+    

lib/bald/tests/__init__.py

@@ -0,0 +1,16 @@
+import contextlib
+import os
+import tempfile
+import unittest
+
+
+class BaldTestCase(unittest.TestCase):
+    @contextlib.contextmanager
+    def temp_filename(self, suffix=''):
+        temp_file = tempfile.mkstemp(suffix)
+        os.close(temp_file[0])
+        filename = temp_file[1]
+        try:
+            yield filename
+        finally:
+            os.remove(filename)

lib/bald/tests/__init__.pyc

@@ -0,0 +1,119 @@
+import unittest
+
+import h5py
+import numpy as np
+
+import bald
+from bald.tests import BaldTestCase
+
+def _fattrs(f):
+    f.attrs['bald__'] = 'http://binary-array-ld.net/experimental/'
+    f.attrs['rdf__'] = 'http://www.w3.org/1999/02/22-rdf-syntax-ns#'
+    f.attrs['rdf__type'] = 'bald__Container'
+    return f
+
+def _create_parent_child(f, pshape, cshape):
+    dsetp = f.create_dataset("parent_dataset", pshape, dtype='i')
+    dsetc = f.create_dataset("child_dataset", cshape, dtype='i')
+    dsetp.attrs['rdf__type'] = 'bald__Dataset'
+    dsetp.attrs['bald__references'] = dsetc.ref
+    dsetc.attrs['rdf__type'] = 'bald__Dataset'
+    dsetc.attrs['rdf__type'] = 'bald__Reference'
+    dsetc.attrs['bald__dataset'] = dsetc.ref
+    return f
+
+
+class Test(BaldTestCase):
+        
+    def test_valid_uri(self):
+        with self.temp_filename('.hdf') as tfile:
+            f = h5py.File(tfile, "w")
+            f = _fattrs(f)
+            f = _create_parent_child(f, (11, 17), (11, 17))
+            f.close()
+            self.assertTrue(bald.validate_hdf5(tfile))
+
+    def test_invalid_uri(self):
+        with self.temp_filename('.hdf') as tfile:
+            f = h5py.File(tfile, "w")
+            f = _fattrs(f)
+            f = _create_parent_child(f, (11, 17), (11, 17))
+            f.attrs['bald__turtle'] = 'bald__walnut'
+            f.close()
+            self.assertFalse(bald.validate_hdf5(tfile))
+
+class TestArrayReference(BaldTestCase):
+    def test_match(self):
+        with self.temp_filename('.hdf') as tfile:
+            f = h5py.File(tfile, "w")
+            f = _fattrs(f)
+            f = _create_parent_child(f, (11, 17), (11, 17))
+            f.close()
+            self.assertTrue(bald.validate_hdf5(tfile))
+
+    def test_mismatch_zeroth(self):
+        with self.temp_filename('.hdf') as tfile:
+            f = h5py.File(tfile, "w")
+            f = _fattrs(f)
+            f = _create_parent_child(f, (11, 17), (11, 13))
+            f.close()
+            self.assertFalse(bald.validate_hdf5(tfile))
+
+    def test_mismatch_oneth(self):
+        with self.temp_filename('.hdf') as tfile:
+            f = h5py.File(tfile, "w")
+            f = _fattrs(f)
+            f = _create_parent_child(f, (11, 17), (13, 17))
+            f.close()
+            self.assertFalse(bald.validate_hdf5(tfile))
+
+    def test_match_plead_dim(self):
+        with self.temp_filename('.hdf') as tfile:
+            f = h5py.File(tfile, "w")
+            f = _fattrs(f)
+            # parent has leading dimension wrt child
+            f = _create_parent_child(f, (4, 13, 17), (13, 17))
+            f.close()
+            self.assertTrue(bald.validate_hdf5(tfile))
+
+    def test_match_clead_dim(self):
+        with self.temp_filename('.hdf') as tfile:
+            f = h5py.File(tfile, "w")
+            f = _fattrs(f)
+            # child has leading dimension wrt parent
+            f = _create_parent_child(f, (13, 17), (7, 13, 17))
+            f.close()
+            self.assertTrue(bald.validate_hdf5(tfile))
+
+    def test_mismatch_pdisjc_lead_dim(self):
+        with self.temp_filename('.hdf') as tfile:
+            f = h5py.File(tfile, "w")
+            f = _fattrs(f)
+            # child and parent have disjoint leading dimensions
+            f = _create_parent_child(f, (4, 13, 17), (7, 13, 17))
+            
+            f.close()
+            self.assertFalse(bald.validate_hdf5(tfile))
+
+    def test_mismatch_pdisjc_trail_dim(self):
+        with self.temp_filename('.hdf') as tfile:
+            f = h5py.File(tfile, "w")
+            f = _fattrs(f)
+            # child and parent have disjoint trailing dimensions
+            f = _create_parent_child(f, (13, 17, 2), (13, 17, 9))
+            f.close()
+            self.assertFalse(bald.validate_hdf5(tfile))
+
+
+
+    # def test_match_(self):
+    #      with self.temp_filename('.hdf') as tfile:
+    #         f = h5py.File(tfile, "w")
+    #         f = _fattrs(f)
+    #         # 
+    #         f = _create_parent_child(f, (), ())
+    #         f.close()
+    #         self.assert(bald.validate_hdf5(tfile))
+
+if __name__ == '__main__':
+    unittest.main()

lib/bald/tests/integration/__init__.py

@@ -0,0 +1,32 @@
+import unittest
+
+import h5py
+import numpy as np
+
+from bald.tests import BaldTestCase
+
+def _fattrs(f):
+    f.attrs['bald__'] = 'http://binary_array_ld.net/experimental'
+    f.attrs['bald__type'] = 'bald__Container'
+    return f
+
+def _create_parent_child(f, pshape, cshape):
+    dsetp = f.create_dataset("parent_dataset", pshape, dtype='i')
+    dsetc = f.create_dataset("child_dataset", cshape, dtype='i')
+    dsetp.attrs['bald__type'] = 'bald__Dataset'
+    dsetp.attrs['bald__reference'] = dsetc.ref
+    dsetc.attrs['bald__type'] = 'bald__Dataset'
+    return f
+
+
+class TestArrayReference(BaldTestCase):
+    def test_match_array_reference(self):
+        with self.temp_filename('.hdf') as tfile:
+            f = h5py.File(tfile, "w")
+            f = _fattrs(f)
+            f = _create_parent_child(f, (11, 17), (11, 17))
+            f.close()
+
+
+if __name__ == '__main__':
+    unittest.main()

lib/bald/tests/integration/test_validation.py

@@ -0,0 +1,36 @@
+import unittest
+
+import h5py
+import numpy as np
+
+from bald.tests import BaldTestCase
+
+
+class Test(BaldTestCase):
+    def setUp(self):
+        self.this = 'this'
+
+    def test_this(self):
+        self.assertTrue(self.this, 'this')
+
+    def test_make_file(self):
+        with self.temp_filename('.hdf') as tfile:
+            f = h5py.File(tfile, "w")
+            dset = f.create_dataset("mydataset", (100,), dtype='i')
+
+    def test_make_load_file(self):
+        with self.temp_filename('.hdf') as tfile:
+            f = h5py.File(tfile, "w")
+            dset = f.create_dataset("mydataset", (100,), dtype='i')
+            dset.attrs['bald__'] = 'http://binary_array_ld.net/experimental'
+            f.close()
+            newf = h5py.File(tfile, "r")
+            nset = newf.get('mydataset')
+            self.assertEqual(nset.shape, (100,))
+            self.assertEqual(nset.dtype, 'int32')
+            self.assertEqual(nset.attrs.get('bald__'),
+                             'http://binary_array_ld.net/experimental')
+
+
+if __name__ == '__main__':
+    unittest.main()

lib/bald/tests/test_array_reference.py

@@ -0,0 +1,20 @@
+import unittest
+
+import h5py
+import numpy as np
+
+from bald.tests import BaldTestCase
+from bald import validation
+
+class Test(unittest.TestCase):
+    def test_check_uri_200(self):
+        auri = 'http://binary-array-ld.net/experimental'
+        self.assertTrue(validation.check_uri(auri))
+
+    def test_check_uri_404(self):
+        notauri = 'http://binary-array-ld.net/experimentalish'
+        self.assertFalse(validation.check_uri(notauri))
+
+
+if __name__ == '__main__':
+    unittest.main()