df: memory: Auto start operations without inputs

Fixes: #392

Author: aghinsa

CHANGELOG.md

@@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Add python code for tensorflow DNNEstimator
 - Ability to run a subflow as if it were an operation using the
   `dffml.dataflow.run` operation.
+- Support for operations without inputs.
 ### Fixed
 - New model tutorial mentions file paths that should be edited.
 - DataFlow is no longer a dataclass to prevent it from being exported

dffml/df/memory.py

@@ -1026,6 +1026,23 @@ async def dispatch(
         task.add_done_callback(ignore_args(self.completed_event.set))
         return task
 
+    async def dispatch_auto_starts(self, octx: BaseOrchestratorContext, ctx):
+        """
+        Schedule the running of all operations without inputs
+        """
+        empty_parameter_set = MemoryParameterSet(
+            MemoryParameterSetConfig(ctx=ctx, parameters=[])
+        )
+
+        for operation in octx.config.dataflow.operations.values():
+            if operation.inputs:
+                continue
+            task = asyncio.create_task(
+                self.run_dispatch(octx, operation, empty_parameter_set)
+            )
+            task.add_done_callback(ignore_args(self.completed_event.set))
+            yield task
+
 
 @entrypoint("memory")
 class MemoryOperationImplementationNetwork(
@@ -1416,6 +1433,10 @@ async def run_operations_for_ctx(
         # TODO(dfass) Make ictx.added(ctx) specific to dataflow
         input_set_enters_network = asyncio.create_task(self.ictx.added(ctx))
         tasks.add(input_set_enters_network)
+        # schedule running of operations with no inputs
+        async for task in self.nctx.dispatch_auto_starts(self, ctx):
+            tasks.add(task)
+
         try:
             # Return when outstanding operations reaches zero
             while tasks:

tests/df/test_auto_start.py

@@ -0,0 +1,39 @@
+from dffml.df.base import op
+from dffml.df.types import DataFlow, Input, Definition
+from dffml.operation.output import GetSingle
+from dffml.util.asynctestcase import AsyncTestCase
+from dffml.df.memory import MemoryOrchestrator
+
+STRING = Definition(name="string", primitive="str")
+
+
+@op(inputs={}, outputs={"string_out": STRING})
+async def announce():
+    return {"string_out": "EXISTS"}
+
+
+class TestAutoStart(AsyncTestCase):
+    async def setUp(self):
+        dataflow = DataFlow(
+            operations={
+                "announce": announce.op,
+                "get_single": GetSingle.imp.op,
+            },
+            seed=[
+                Input(
+                    value=[announce.op.outputs["string_out"].name],
+                    definition=GetSingle.op.inputs["spec"],
+                )
+            ],
+            implementations={announce.op.name: announce.imp,},
+        )
+
+        self.dataflow = dataflow
+
+    async def test_auto_start(self):
+        test_inputs = {"testStart": []}
+        async with MemoryOrchestrator.withconfig({}) as orchestrator:
+            async with orchestrator(self.dataflow) as octx:
+                async for ctx_str, results in octx.run(test_inputs):
+                    self.assertIn("string", results)
+                    self.assertEqual("EXISTS", results["string"])