Indirect File click types

starfish/util/click/indirectparams.py

@@ -0,0 +1,63 @@
+import abc
+from typing import Generic, Iterable, TypeVar
+
+from starfish.codebook.codebook import Codebook
+from starfish.imagestack.imagestack import ImageStack
+from starfish.util.indirectfile import (
+    ConversionRecipe,
+    convert,
+    GetCodebook,
+    GetCodebookFromExperiment,
+    GetImageStack,
+    GetImageStackFromExperiment,
+    NoApplicableConversionRecipeError,
+    NoSuccessfulConversionRecipeError,
+)
+from . import ParamType
+
+
+IndirectResultType = TypeVar("IndirectResultType")
+
+
+class IndirectFile(ParamType, Generic[IndirectResultType]):
+    def convert(self, value: str, param, ctx):
+        conversion_recipes = self.get_conversion_recipes()
+        try:
+            return convert(value, conversion_recipes)
+        except (NoApplicableConversionRecipeError, NoSuccessfulConversionRecipeError) as ex:
+            self.fail(ex.args[0])
+
+    @abc.abstractmethod
+    def get_conversion_recipes(self) -> Iterable[ConversionRecipe[IndirectResultType]]:
+        """Return one or more conversion recipes to get from an input string to the type of object
+        we want.
+        """
+        raise NotImplementedError()
+
+
+class CodebookParam(IndirectFile[Codebook]):
+    def __init__(self):
+        self.name = "codebook"
+
+    def get_conversion_recipes(self) -> Iterable[ConversionRecipe[Codebook]]:
+        return [
+            GetCodebookFromExperiment(),
+            GetCodebook(),
+        ]
+
+
+CodebookParamType = CodebookParam()
+
+
+class ImageStackParam(IndirectFile[ImageStack]):
+    def __init__(self):
+        self.name = "imagestack"
+
+    def get_conversion_recipes(self) -> Iterable[ConversionRecipe[ImageStack]]:
+        return [
+            GetImageStackFromExperiment(),
+            GetImageStack(),
+        ]
+
+
+ImageStackParamType = ImageStackParam()

starfish/util/indirectfile/__init__.py

@@ -0,0 +1,8 @@
+from ._base import (
+    ConversionRecipe,
+    convert,
+    NoApplicableConversionRecipeError,
+    NoSuccessfulConversionRecipeError,
+)
+from ._codebook import GetCodebook, GetCodebookFromExperiment
+from ._imagestack import GetImageStack, GetImageStackFromExperiment

starfish/util/indirectfile/_base.py

@@ -0,0 +1,45 @@
+import abc
+from typing import Generic, Iterable, TypeVar
+
+
+RecipeResultType = TypeVar("RecipeResultType")
+
+
+class ConversionRecipe(Generic[RecipeResultType]):
+    @abc.abstractmethod
+    def applicable(self, input_parameter: str) -> bool:
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def load(self, input_parameter: str) -> RecipeResultType:
+        """Attempt to run this conversion recipe against this input."""
+        raise NotImplementedError()
+
+
+class NoApplicableConversionRecipeError(Exception):
+    """Raised when no conversion recipe declared itself applicable to this input string."""
+    pass
+
+
+class NoSuccessfulConversionRecipeError(Exception):
+    """Raised when all the conversion recipes that declared itself applicable failed to execute
+    successfully."""
+    pass
+
+
+def convert(value: str, conversion_recipes: Iterable[ConversionRecipe]):
+    none_applied = True
+
+    for conversion_recipe in conversion_recipes:
+        if conversion_recipe.applicable(value):
+            none_applied = False
+            try:
+                return conversion_recipe.load(value)
+            except Exception:
+                pass
+
+    if none_applied:
+        raise NoApplicableConversionRecipeError(
+            f"Could not find applicable gonversion recipe for {value}")
+    raise NoSuccessfulConversionRecipeError(
+        f"All applicable conversion recipes failed to run successfully for {value}.")

starfish/util/indirectfile/_codebook.py

@@ -0,0 +1,21 @@
+from starfish.codebook.codebook import Codebook
+from starfish.experiment.experiment import Experiment
+from starfish.util.indirectfile._base import ConversionRecipe
+
+
+class GetCodebookFromExperiment(ConversionRecipe):
+    def applicable(self, input_parameter: str) -> bool:
+        return input_parameter.startswith("@")
+
+    def load(self, input_parameter: str) -> Codebook:
+        path = input_parameter[1:]
+        experiment = Experiment.from_json(path)
+        return experiment.codebook
+
+
+class GetCodebook(ConversionRecipe):
+    def applicable(self, input_parameter: str) -> bool:
+        return not input_parameter.startswith("@")
+
+    def load(self, input_parameter: str) -> Codebook:
+        return Codebook.from_json(input_parameter)

starfish/util/indirectfile/_imagestack.py

@@ -0,0 +1,28 @@
+import re
+
+from starfish.experiment.experiment import Experiment
+from starfish.imagestack.imagestack import ImageStack
+from starfish.util.indirectfile._base import ConversionRecipe
+
+
+CRE = re.compile("@(?P<path>.+)\[(?P<fov>[^\[\]]+)\]\[(?P<image_type>[^\[\]]+)\]")  # noqa: W605
+
+
+class GetImageStackFromExperiment(ConversionRecipe[ImageStack]):
+    def applicable(self, input_parameter: str) -> bool:
+        return CRE.match(input_parameter) is not None
+
+    def load(self, input_parameter: str) -> ImageStack:
+        mo = CRE.match(input_parameter)
+        assert mo is not None
+        experiment = Experiment.from_json(mo.group("path"))
+        fov = experiment[mo.group("fov")]
+        return fov.get_image(mo.group("image_type"))
+
+
+class GetImageStack(ConversionRecipe[ImageStack]):
+    def applicable(self, input_parameter: str) -> bool:
+        return not CRE.match(input_parameter)
+
+    def load(self, input_parameter: str) -> ImageStack:
+        return ImageStack.from_path_or_url(input_parameter)