Repository: intel/dffml

Issue:
df: memory: Dispatch operations without inputs as soon as DataFlow begins
We may have operations which need to be kicked off as soon as a dataflow starts, a good example would be something which is waiting for a message in a gitter channel, and the output it the message. This operation likely has no need for an input, it just returns the first message it sees. We'd want it to start waiting as soon as the dataflow started.

Discussion:
aghinsa: > Great job! :D I left a comment on a way to simplify this. Let's also make sure to add a test.

oh my bad,i thought i forgot to commit the test!

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index a6b219868..bb6d643b1 100644
--- a/CHANGELOG.md
+++ b/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
diff --git a/dffml/df/memory.py b/dffml/df/memory.py
index 738904798..6d016a61e 100644
--- a/dffml/df/memory.py
+++ b/dffml/df/memory.py
@@ -1026,6 +1026,23 @@ class MemoryOperationImplementationNetworkContext(
         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 @@ class MemoryOrchestratorContext(BaseOrchestratorContext):
         # 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:


Test patch:
diff --git a/tests/df/test_auto_start.py b/tests/df/test_auto_start.py
new file mode 100644
index 000000000..cf132a89f
--- /dev/null
+++ b/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"])