Restructure the relationship between PipelineComponent and AlgorithmBase

Rather than each specific PipelineComponent have a link to the AlgorithmBase that describes the contract that the algorithm must implement, each AlgorithmBase implementation points to the PipelineComponent that it serves.

The current arrangement forces all the algorithms to be defined and evaluated in the interpreter before the pipeline component is defined and evaluated.  If we want to create an algorithm that's only used in tests, this is not possible.

The new arrangement ensures that any implementation of a specific AlgorithmBase is registered with its corresponding PipelineComponent.

This adds explicit abstract classes to the AlgorithmBase class hierarchy to make it easier to detect what's an actual algorithm implementation.

Depends on #1094

Test plan: `make -j fast`

starfish/image/_filter/__init__.py

@@ -1,42 +1,3 @@
-from typing import Type
-
-from starfish.imagestack.imagestack import ImageStack
-from starfish.pipeline import AlgorithmBase, import_all_submodules, PipelineComponent
-from starfish.util import click
-from . import _base
+from starfish.pipeline import import_all_submodules
+from ._base import Filter
 import_all_submodules(__file__, __package__)
-
-
-COMPONENT_NAME = "filter"
-
-
-class Filter(PipelineComponent):
-    @classmethod
-    def pipeline_component_type_name(cls) -> str:
-        return COMPONENT_NAME
-
-    @classmethod
-    def _get_algorithm_base_class(cls) -> Type[AlgorithmBase]:
-        return _base.FilterAlgorithmBase
-
-    @classmethod
-    def _cli_run(cls, ctx, instance):
-        output = ctx.obj["output"]
-        stack = ctx.obj["stack"]
-        filtered = instance.run(stack)
-        filtered.export(output)
-
-    @staticmethod
-    @click.group(COMPONENT_NAME)
-    @click.option("-i", "--input", type=click.Path(exists=True))
-    @click.option("-o", "--output", required=True)
-    @click.pass_context
-    def _cli(ctx, input, output):
-        """smooth, sharpen, denoise, etc"""
-        print("Filtering images...")
-        ctx.obj = dict(
-            component=Filter,
-            input=input,
-            output=output,
-            stack=ImageStack.from_path_or_url(input),
-        )

starfish/image/_filter/_base.py

@@ -1,8 +1,49 @@
+from abc import abstractmethod
+from typing import Type
+
 from starfish.imagestack.imagestack import ImageStack
 from starfish.pipeline.algorithmbase import AlgorithmBase
+from starfish.pipeline.pipelinecomponent import PipelineComponent
+from starfish.util import click
+
+
+COMPONENT_NAME = "filter"
+
+
+class Filter(PipelineComponent):
+    @classmethod
+    def pipeline_component_type_name(cls) -> str:
+        return COMPONENT_NAME
+
+    @classmethod
+    def _cli_run(cls, ctx, instance):
+        output = ctx.obj["output"]
+        stack = ctx.obj["stack"]
+        filtered = instance.run(stack)
+        filtered.export(output)
+
+    @staticmethod
+    @click.group(COMPONENT_NAME)
+    @click.option("-i", "--input", type=click.Path(exists=True))
+    @click.option("-o", "--output", required=True)
+    @click.pass_context
+    def _cli(ctx, input, output):
+        """smooth, sharpen, denoise, etc"""
+        print("Filtering images...")
+        ctx.obj = dict(
+            component=Filter,
+            input=input,
+            output=output,
+            stack=ImageStack.from_path_or_url(input),
+        )
 
 
 class FilterAlgorithmBase(AlgorithmBase):
+    @classmethod
+    def get_pipeline_component_class(cls) -> Type[PipelineComponent]:
+        return Filter
+
+    @abstractmethod
     def run(self, stack: ImageStack) -> ImageStack:
         """Performs filtering on an ImageStack."""
         raise NotImplementedError()

starfish/image/_registration/__init__.py

@@ -1,43 +1,3 @@
-from typing import Type
-
-from starfish.imagestack.imagestack import ImageStack
-from starfish.pipeline import AlgorithmBase, import_all_submodules, PipelineComponent
-from starfish.util import click
-from . import _base
+from starfish.pipeline import import_all_submodules
+from ._base import Registration
 import_all_submodules(__file__, __package__)
-
-
-COMPONENT_NAME = "registration"
-
-
-class Registration(PipelineComponent):
-
-    @classmethod
-    def pipeline_component_type_name(cls) -> str:
-        return COMPONENT_NAME
-
-    @classmethod
-    def _get_algorithm_base_class(cls) -> Type[AlgorithmBase]:
-        return _base.RegistrationAlgorithmBase
-
-    @classmethod
-    def _cli_run(cls, ctx, instance):
-        output = ctx.obj["output"]
-        stack = ctx.obj["stack"]
-        instance.run(stack)
-        stack.export(output)
-
-    @staticmethod
-    @click.group(COMPONENT_NAME)
-    @click.option("-i", "--input", type=click.Path(exists=True))
-    @click.option("-o", "--output", required=True)
-    @click.pass_context
-    def _cli(ctx, input, output):
-        """translation correction of image stacks"""
-        print("Registering...")
-        ctx.obj = dict(
-            component=Registration,
-            input=input,
-            output=output,
-            stack=ImageStack.from_path_or_url(input),
-        )

starfish/image/_registration/_base.py

@@ -1,10 +1,51 @@
-from typing import Optional
+from abc import abstractmethod
+from typing import Optional, Type
+
+import click
 
 from starfish.imagestack.imagestack import ImageStack
+from starfish.pipeline import PipelineComponent
 from starfish.pipeline.algorithmbase import AlgorithmBase
 
 
+COMPONENT_NAME = "registration"
+
+
+class Registration(PipelineComponent):
+
+    @classmethod
+    def pipeline_component_type_name(cls) -> str:
+        return COMPONENT_NAME
+
+    @classmethod
+    def _cli_run(cls, ctx, instance):
+        output = ctx.obj["output"]
+        stack = ctx.obj["stack"]
+        instance.run(stack)
+        stack.export(output)
+
+    @staticmethod
+    @click.group(COMPONENT_NAME)
+    @click.option("-i", "--input", type=click.Path(exists=True))
+    @click.option("-o", "--output", required=True)
+    @click.pass_context
+    def _cli(ctx, input, output):
+        """translation correction of image stacks"""
+        print("Registering...")
+        ctx.obj = dict(
+            component=Registration,
+            input=input,
+            output=output,
+            stack=ImageStack.from_path_or_url(input),
+        )
+
+
 class RegistrationAlgorithmBase(AlgorithmBase):
+    @classmethod
+    def get_pipeline_component_class(cls) -> Type[PipelineComponent]:
+        return Registration
+
+    @abstractmethod
     def run(self, stack) -> Optional[ImageStack]:
         """Performs registration on the stack provided."""
         raise NotImplementedError()

starfish/image/_segmentation/__init__.py

@@ -1,50 +1,3 @@
-from typing import Type
-
-from skimage.io import imsave
-
-from starfish.imagestack.imagestack import ImageStack
-from starfish.pipeline import AlgorithmBase, import_all_submodules, PipelineComponent
-from starfish.util import click
-from . import _base
+from starfish.pipeline import import_all_submodules
+from ._base import Segmentation
 import_all_submodules(__file__, __package__)
-
-
-COMPONENT_NAME = "segment"
-
-
-class Segmentation(PipelineComponent):
-
-    @classmethod
-    def pipeline_component_type_name(cls) -> str:
-        return COMPONENT_NAME
-
-    @classmethod
-    def _get_algorithm_base_class(cls) -> Type[AlgorithmBase]:
-        return _base.SegmentationAlgorithmBase
-
-    @classmethod
-    def _cli_run(cls, ctx, instance):
-        output = ctx.obj["output"]
-        pri_stack = ctx.obj["primary_images"]
-        nuc_stack = ctx.obj["nuclei"]
-
-        label_image = instance.run(pri_stack, nuc_stack)
-
-        print(f"Writing label image to {output}")
-        imsave(output, label_image)
-
-    @staticmethod
-    @click.group(COMPONENT_NAME)
-    @click.option("--primary-images", required=True, type=click.Path(exists=True))
-    @click.option("--nuclei", required=True, type=click.Path(exists=True))
-    @click.option("-o", "--output", required=True)
-    @click.pass_context
-    def _cli(ctx, primary_images, nuclei, output):
-        """define polygons for cell boundaries and assign spots"""
-        print('Segmenting ...')
-        ctx.obj = dict(
-            component=Segmentation,
-            output=output,
-            primary_images=ImageStack.from_path_or_url(primary_images),
-            nuclei=ImageStack.from_path_or_url(nuclei),
-        )

starfish/image/_segmentation/_base.py

@@ -1,8 +1,57 @@
+from abc import abstractmethod
+from typing import Type
+
+import click
+from skimage.io import imsave
+
 from starfish.imagestack.imagestack import ImageStack
+from starfish.pipeline import PipelineComponent
 from starfish.pipeline.algorithmbase import AlgorithmBase
 
 
+COMPONENT_NAME = "segment"
+
+
+class Segmentation(PipelineComponent):
+
+    @classmethod
+    def pipeline_component_type_name(cls) -> str:
+        return COMPONENT_NAME
+
+    @classmethod
+    def _cli_run(cls, ctx, instance):
+        output = ctx.obj["output"]
+        pri_stack = ctx.obj["primary_images"]
+        nuc_stack = ctx.obj["nuclei"]
+
+        label_image = instance.run(pri_stack, nuc_stack)
+
+        print(f"Writing label image to {output}")
+        imsave(output, label_image)
+
+    @staticmethod
+    @click.group(COMPONENT_NAME)
+    @click.option("--primary-images", required=True, type=click.Path(exists=True))
+    @click.option("--nuclei", required=True, type=click.Path(exists=True))
+    @click.option("-o", "--output", required=True)
+    @click.pass_context
+    def _cli(ctx, primary_images, nuclei, output):
+        """define polygons for cell boundaries and assign spots"""
+        print('Segmenting ...')
+        ctx.obj = dict(
+            component=Segmentation,
+            output=output,
+            primary_images=ImageStack.from_path_or_url(primary_images),
+            nuclei=ImageStack.from_path_or_url(nuclei),
+        )
+
+
 class SegmentationAlgorithmBase(AlgorithmBase):
+    @classmethod
+    def get_pipeline_component_class(cls) -> Type[PipelineComponent]:
+        return Segmentation
+
+    @abstractmethod
     def run(self, primary_image_stack: ImageStack, nuclei_stack: ImageStack):
         """Performs registration on the stack provided."""
         raise NotImplementedError()

starfish/pipeline/algorithmbase.py

@@ -1,29 +1,43 @@
+import inspect
+from abc import ABCMeta, abstractmethod
+from typing import Type
+
 from starfish.imagestack.imagestack import ImageStack
 from starfish.intensity_table.intensity_table import IntensityTable
 from starfish.types import LOG
 from starfish.types._constants import STARFISH_EXTRAS_KEY
 from starfish.util.logging import LogEncoder
+from .pipelinecomponent import PipelineComponent
 
 
-class AlgorithmBaseType(type):
-
+class AlgorithmBaseType(ABCMeta):
     def __init__(cls, name, bases, namespace):
         super().__init__(name, bases, namespace)
-        if len(bases) != 0:
-            # this is _not_ AlgorithmBase.  Instead, it's a subclass of AlgorithmBase.
+        if not inspect.isabstract(cls):
+            AlgorithmBaseType.register_with_pipeline_component(cls)
             cls.run = AlgorithmBaseType.run_with_logging(cls.run)
 
+    @staticmethod
+    def register_with_pipeline_component(algorithm_cls):
+        pipeline_component_cls = algorithm_cls.get_pipeline_component_class()
+        if pipeline_component_cls._algorithm_to_class_map_int is None:
+            pipeline_component_cls._algorithm_to_class_map_int = {}
+        pipeline_component_cls._algorithm_to_class_map_int[algorithm_cls.__name__] = algorithm_cls
+        setattr(pipeline_component_cls, algorithm_cls._get_algorithm_name(), algorithm_cls)
+
+        pipeline_component_cls._cli.add_command(algorithm_cls._cli)
+
     @staticmethod
     def run_with_logging(func):
         """
         This method extends each pipeline component.run() method to also log itself and
-        runtime parameters to the IntensityTable and Imagestack objects. There are two
+        runtime parameters to the IntensityTable and ImageStack objects. There are two
         scenarios for this method:
             1.) Filtering:
-                    Imagestack -> Imagestack
+                    ImageStack -> ImageStack
             2.) Spot Detection:
-                    Imagestack -> IntensityTable
-                    Imagestack -> [IntenistyTable, ConnectedComponentDecodingResult]
+                    ImageStack -> IntensityTable
+                    ImageStack -> [IntensityTable, ConnectedComponentDecodingResult]
             TODO segmentation and decoding
         """
         def helper(*args, **kwargs):
@@ -91,3 +105,11 @@ def _get_algorithm_name(cls):
         https://docs.python.org/3/reference/lexical_analysis.html#identifiers
         """
         return cls.__name__
+
+    @classmethod
+    @abstractmethod
+    def get_pipeline_component_class(cls) -> Type[PipelineComponent]:
+        """
+        Returns the class of PipelineComponent this algorithm implements.
+        """
+        raise NotImplementedError()