Pass failures through to next layer (#36)

Author: j6k4m8

bossphorus/__init__.py

@@ -28,7 +28,11 @@
 import numpy as np
 
 from . import storagemanager
-from .storagemanager import FilesystemStorageManager, RelayStorageManager
+from .storagemanager import (
+    FilesystemStorageManager,
+    RelayStorageManager,
+    ChunkedFilesystemStorageManager,
+)
 from . import version
 
 __version__ = version.__version__
@@ -43,12 +47,12 @@ def create_app(mgr: storagemanager.StorageManager = None):
     if mgr:
         manager = mgr
     else:
-        manager = FilesystemStorageManager(
+        manager = ChunkedFilesystemStorageManager(
             "./uploads",
             (256, 256, 256),
             next_layer=(
                 RelayStorageManager(
-                    upstream_uri="api.bossdb.org", protocol="https", token="PUBLIC"
+                    upstream_uri="api.bossdb.io", protocol="https", token="public"
                 )
             ),
         )

bossphorus/storagemanager/_ChunkedFilesystemStorageManager.py

@@ -91,6 +91,16 @@ def store(
         )
         return np.save(fname, data)
 
+    def hasfile(self, col: str, exp: str, chan: str, res: int, b: Tuple[int, int, int]) -> bool:
+        if not (
+            os.path.isdir("{}/{}".format(self.storage_path, col))
+            and os.path.isdir("{}/{}/{}".format(self.storage_path, col, exp))
+            and os.path.isdir("{}/{}/{}/{}".format(self.storage_path, col, exp, chan))
+        ):
+            return False
+        return True
+
+
     def retrieve(
         self, col: str, exp: str, chan: str, res: int, b: Tuple[int, int, int]
     ):
@@ -163,6 +173,10 @@ def hasdata(
     ):
         # TODO: Should know when it has data and return false even if it's
         # in terminal mode
+        try:
+            self.getdata(col, exp, chan, res, xs, ys, zs)
+        except:
+            return False
         return self.is_terminal
 
     def setdata(
@@ -223,20 +237,33 @@ def getdata(
         files = file_compute(
             xs[0], xs[1], ys[0], ys[1], zs[0], zs[1], block_size=self.block_size
         )
+
         indices = blockfile_indices(xs, ys, zs, block_size=self.block_size)
 
         payload = np.zeros(
             ((xs[1] - xs[0]), (ys[1] - ys[0]), (zs[1] - zs[0])), dtype="uint8"
         )
+
         for f, i in zip(files, indices):
-            try:
+            print(f)
+            if self.fs.hasfile(col, exp, chan, res, f):
                 data_partial = self.fs.retrieve(col, exp, chan, res, f)[
                     i[0][0] : i[0][1], i[1][0] : i[1][1], i[2][0] : i[2][1]
                 ]
-            except:
-                data_partial = np.zeros(self.block_size, dtype="uint8")[
-                    i[0][0] : i[0][1], i[1][0] : i[1][1], i[2][0] : i[2][1]
-                ]
+            else:
+                # what to do if the file doesn't exist
+                if self.is_terminal:
+                    # this is a terminal; must return something, so return 0s
+                    data_partial = np.zeros(self.block_size, dtype="uint8")[
+                        i[0][0] : i[0][1], i[1][0] : i[1][1], i[2][0] : i[2][1]
+                    ]
+                else:
+                    # we can cascade to a downstream:
+                    # TODO: Should just fetch the data-partial instead of
+                    # aborting the whole download and getting it from a
+                    # remote.
+                    return self._next.getdata(col, exp, chan, res, xs, ys, zs)
+
             payload[
                 (f[0] + i[0][0]) - xs[0] : (f[0] + i[0][1]) - xs[0],
                 (f[1] + i[1][0]) - ys[0] : (f[1] + i[1][1]) - ys[0],

bossphorus/storagemanager/_FilesystemStorageManager.py

@@ -135,9 +135,11 @@ def hasdata(
         zs: Tuple[int, int],
     ):
         fname = f"{self.storage_path}/{col}/{exp}/{chan}/{res}.h5"
+        if not os.path.isfile(fname):
+            return False
         return h5py.File(fname, "r")["mask"][
             xs[0] : xs[1], ys[0] : ys[1], zs[0] : zs[1]
-        ]
+        ].all()
 
 
 class FilesystemStorageManager(StorageManager):
@@ -166,6 +168,7 @@ def __init__(
             self.is_terminal = True
         self.storage_path = storage_path
         self.block_size = block_size
+        self._cache = kwargs.get("cache", True)
 
         self.fs = ({"h5": H5FileInterface}.get(kwargs.get("preferred_format", "h5")))(
             self.storage_path
@@ -184,7 +187,7 @@ def hasdata(
         if self.is_terminal:
             return True
 
-        return self.fs.hasdata(col, exp, chan, res, xs, ys, zs).all()
+        return self.fs.hasdata(col, exp, chan, res, xs, ys, zs)
 
     def setdata(
         self,
@@ -226,7 +229,10 @@ def getdata(
             return self.fs.retrieve(col, exp, chan, res, xs, ys, zs)
         if self.is_terminal:
             raise IndexError("Cannot find data at: ", (col, exp, chan, res, xs, ys, zs))
-        return self._next.getdata(col, exp, chan, res, xs, ys, zs)
+        data = self._next.getdata(col, exp, chan, res, xs, ys, zs)
+        if self._cache:
+            self.setdata(data, col, exp, chan, res, xs, ys, zs)
+        return data
 
     def __str__(self):
         return f"<FilesystemStorageManager [{str(self.fs)}]>"

bossphorus/storagemanager/_RelayStorageManager.py

@@ -46,11 +46,12 @@ def __init__(self, **kwargs):
         if "boss_remote" in kwargs:
             self.boss_remote = kwargs["boss_remote"]
         elif "upstream_uri" in kwargs:
+            self.token = kwargs.get("token", "public")
             self.boss_remote = BossRemote(
                 {
                     "host": kwargs["upstream_uri"],
                     "protocol": kwargs.get("protocol", "http"),
-                    "token": kwargs.get("token", "no-token"),
+                    "token": kwargs.get("token", "public"),
                 }
             )