df: memory: Cleanup and move methods

- Move {operation_parameter,validator_target}set_pair to
  MemoryOrchestratorContext
- Make removing inputs from input sets async instead of
   non-async

Fixes: #477
Fixes: #452

Author: aghinsa

CHANGELOG.md

@@ -19,6 +19,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - New model tutorial mentions file paths that should be edited.
 - DataFlow is no longer a dataclass to prevent it from being exported
   incorrectly.
+- `operations_parameter_set_pairs` moved to `MemoryOrchestratorContext`
 
 ## [0.3.5] - 2020-03-10
 ### Added

dffml/df/base.py

@@ -482,6 +482,19 @@ async def _asdict(self) -> Dict[str, Any]:
             item.definition.name: item.value async for item in self.inputs()
         }
 
+    @abc.abstractmethod
+    async def remove_input(self, item: Input) -> None:
+        """
+        Removes item from input set
+        """
+        pass
+
+    @abc.abstractmethod
+    async def remove_unvalidated_inputs(self) -> "BaseInputSet":
+        """
+        Removes `unvalidated` inputs from internal list and returns the same.
+        """
+
 
 class BaseParameterSetConfig(NamedTuple):
     ctx: BaseInputSetContext
@@ -737,24 +750,6 @@ async def dispatch(
         Schedule the running of an operation
         """
 
-    @abc.abstractmethod
-    async def operations_parameter_set_pairs(
-        self,
-        ictx: BaseInputNetworkContext,
-        octx: BaseOperationNetworkContext,
-        rctx: BaseRedundancyCheckerContext,
-        ctx: BaseInputSetContext,
-        *,
-        new_input_set: BaseInputSet = None,
-        stage: Stage = Stage.PROCESSING,
-    ) -> AsyncIterator[Tuple[Operation, BaseParameterSet]]:
-        """
-        Use new_input_set to determine which operations in the network might be
-        up for running. Cross check using existing inputs to generate per
-        input set context novel input pairings. Yield novel input pairings
-        along with their operations as they are generated.
-        """
-
 
 # TODO We should be able to specify multiple operation implementation  networks.
 # This would enable operations to live in different place, accessed via the
@@ -790,6 +785,21 @@ async def run_operations(
         Run all the operations then run cleanup and output operations
         """
 
+    @abc.abstractmethod
+    async def operations_parameter_set_pairs(
+        self,
+        ctx: BaseInputSetContext,
+        *,
+        new_input_set: BaseInputSet = None,
+        stage: Stage = Stage.PROCESSING,
+    ) -> AsyncIterator[Tuple[Operation, BaseParameterSet]]:
+        """
+        Use new_input_set to determine which operations in the network might be
+        up for running. Cross check using existing inputs to generate per
+        input set context novel input pairings. Yield novel input pairings
+        along with their operations as they are generated.
+        """
+
 
 @base_entry_point("dffml.orchestrator", "orchestrator")
 class BaseOrchestrator(BaseDataFlowObject):

dffml/df/memory.py

@@ -127,13 +127,13 @@ async def inputs(self) -> AsyncIterator[Input]:
         for item in self.__inputs:
             yield item
 
-    def remove_input(self, item: Input):
+    async def remove_input(self, item: Input):
         for x in self.__inputs[:]:
             if x.uid == item.uid:
                 self.__inputs.remove(x)
                 break
 
-    def remove_unvalidated_inputs(self) -> "MemoryInputSet":
+    async def remove_unvalidated_inputs(self) -> "MemoryInputSet":
         """
         Removes `unvalidated` inputs from internal list and returns the same.
         """
@@ -278,7 +278,7 @@ async def add(self, input_set: BaseInputSet):
         # self.ctxhd
 
         # remove unvalidated inputs
-        unvalidated_input_set = input_set.remove_unvalidated_inputs()
+        unvalidated_input_set = await input_set.remove_unvalidated_inputs()
 
         # If the context for this input set does not exist create a
         # NotificationSet for it to notify the orchestrator
@@ -1026,65 +1026,6 @@ async def dispatch(
         task.add_done_callback(ignore_args(self.completed_event.set))
         return task
 
-    async def operations_parameter_set_pairs(
-        self,
-        ictx: BaseInputNetworkContext,
-        octx: BaseOperationNetworkContext,
-        rctx: BaseRedundancyCheckerContext,
-        ctx: BaseInputSetContext,
-        dataflow: DataFlow,
-        *,
-        new_input_set: Optional[BaseInputSet] = None,
-        stage: Stage = Stage.PROCESSING,
-    ) -> AsyncIterator[Tuple[Operation, BaseInputSet]]:
-        """
-        Use new_input_set to determine which operations in the network might be
-        up for running. Cross check using existing inputs to generate per
-        input set context novel input pairings. Yield novel input pairings
-        along with their operations as they are generated.
-        """
-        # Get operations which may possibly run as a result of these new inputs
-        async for operation in octx.operations(
-            dataflow, input_set=new_input_set, stage=stage
-        ):
-            # Generate all pairs of un-run input combinations
-            async for parameter_set in ictx.gather_inputs(
-                rctx, operation, dataflow, ctx=ctx
-            ):
-                yield operation, parameter_set
-
-    async def validator_target_set_pairs(
-        self,
-        octx: BaseOperationNetworkContext,
-        rctx: BaseRedundancyCheckerContext,
-        ctx: BaseInputSetContext,
-        dataflow: DataFlow,
-        unvalidated_input_set: BaseInputSet,
-    ):
-        async for unvalidated_input in unvalidated_input_set.inputs():
-            validator_instance_name = unvalidated_input.definition.validate
-            validator = dataflow.validators.get(validator_instance_name, None)
-            if validator is None:
-                raise ValidatorMissing(
-                    "Validator with instance_name {validator_instance_name} not found"
-                )
-            # There is only one `input` in `validators`
-            input_name, input_definition = list(validator.inputs.items())[0]
-            parameter = Parameter(
-                key=input_name,
-                value=unvalidated_input.value,
-                origin=unvalidated_input,
-                definition=input_definition,
-            )
-            parameter_set = MemoryParameterSet(
-                MemoryParameterSetConfig(ctx=ctx, parameters=[parameter])
-            )
-            async for parameter_set, exists in rctx.exists(
-                validator, parameter_set
-            ):
-                if not exists:
-                    yield validator, parameter_set
-
 
 @entrypoint("memory")
 class MemoryOperationImplementationNetwork(
@@ -1408,6 +1349,60 @@ async def run(
                 else:
                     task.exception()
 
+    async def operations_parameter_set_pairs(
+        self,
+        ctx: BaseInputSetContext,
+        dataflow: DataFlow,
+        *,
+        new_input_set: Optional[BaseInputSet] = None,
+        stage: Stage = Stage.PROCESSING,
+    ) -> AsyncIterator[Tuple[Operation, BaseInputSet]]:
+        """
+        Use new_input_set to determine which operations in the network might be
+        up for running. Cross check using existing inputs to generate per
+        input set context novel input pairings. Yield novel input pairings
+        along with their operations as they are generated.
+        """
+        # Get operations which may possibly run as a result of these new inputs
+        async for operation in self.octx.operations(
+            dataflow, input_set=new_input_set, stage=stage
+        ):
+            # Generate all pairs of un-run input combinations
+            async for parameter_set in self.ictx.gather_inputs(
+                self.rctx, operation, dataflow, ctx=ctx
+            ):
+                yield operation, parameter_set
+
+    async def validator_target_set_pairs(
+        self,
+        ctx: BaseInputSetContext,
+        dataflow: DataFlow,
+        unvalidated_input_set: BaseInputSet,
+    ):
+        async for unvalidated_input in unvalidated_input_set.inputs():
+            validator_instance_name = unvalidated_input.definition.validate
+            validator = dataflow.validators.get(validator_instance_name, None)
+            if validator is None:
+                raise ValidatorMissing(
+                    "Validator with instance_name {validator_instance_name} not found"
+                )
+            # There is only one `input` in `validators`
+            input_name, input_definition = list(validator.inputs.items())[0]
+            parameter = Parameter(
+                key=input_name,
+                value=unvalidated_input.value,
+                origin=unvalidated_input,
+                definition=input_definition,
+            )
+            parameter_set = MemoryParameterSet(
+                MemoryParameterSetConfig(ctx=ctx, parameters=[parameter])
+            )
+            async for parameter_set, exists in self.rctx.exists(
+                validator, parameter_set
+            ):
+                if not exists:
+                    yield validator, parameter_set
+
     async def run_operations_for_ctx(
         self, ctx: BaseContextHandle, *, strict: bool = True
     ) -> AsyncIterator[Tuple[BaseContextHandle, Dict[str, Any]]]:
@@ -1468,9 +1463,7 @@ async def run_operations_for_ctx(
                             unvalidated_input_set,
                             new_input_set,
                         ) in new_input_sets:
-                            async for operation, parameter_set in self.nctx.validator_target_set_pairs(
-                                self.octx,
-                                self.rctx,
+                            async for operation, parameter_set in self.validator_target_set_pairs(
                                 ctx,
                                 self.config.dataflow,
                                 unvalidated_input_set,
@@ -1497,10 +1490,7 @@ async def run_operations_for_ctx(
                             )
                             # Identify which operations have completed contextually
                             # appropriate input sets which haven't been run yet
-                            async for operation, parameter_set in self.nctx.operations_parameter_set_pairs(
-                                self.ictx,
-                                self.octx,
-                                self.rctx,
+                            async for operation, parameter_set in self.operations_parameter_set_pairs(
                                 ctx,
                                 self.config.dataflow,
                                 new_input_set=new_input_set,
@@ -1568,13 +1558,8 @@ async def run_operations_for_ctx(
     async def run_stage(self, ctx: BaseInputSetContext, stage: Stage):
         # Identify which operations have complete contextually appropriate
         # input sets which haven't been run yet and are stage operations
-        async for operation, parameter_set in self.nctx.operations_parameter_set_pairs(
-            self.ictx,
-            self.octx,
-            self.rctx,
-            ctx,
-            self.config.dataflow,
-            stage=stage,
+        async for operation, parameter_set in self.operations_parameter_set_pairs(
+            ctx, self.config.dataflow, stage=stage
         ):
             # Add inputs and operation to redundancy checker before dispatch
             await self.rctx.add(operation, parameter_set)

dffml/df/types.py

@@ -556,7 +556,7 @@ def export(self, *, linked: bool = False):
             "operations": {
                 instance_name: operation.export()
                 for instance_name, operation in self.operations.items()
-            },
+            }
         }
         if self.seed:
             exported["seed"] = self.seed.copy()

model/tensorflow/examples/tfdnnr/tfdnnr.py

@@ -4,7 +4,7 @@
 
 model = DNNRegressionModel(
     features=Features(
-        DefFeature("Feature1", float, 1), DefFeature("Feature2", float, 1),
+        DefFeature("Feature1", float, 1), DefFeature("Feature2", float, 1)
     ),
     predict=DefFeature("TARGET", float, 1),
     epochs=300,
@@ -20,7 +20,7 @@
 
 # Make prediction
 for i, features, prediction in predict(
-    model, {"Feature1": 0.21, "Feature2": 0.18, "TARGET": 0.84},
+    model, {"Feature1": 0.21, "Feature2": 0.18, "TARGET": 0.84}
 ):
     features["TARGET"] = prediction["TARGET"]["value"]
     print(features)