Phase 1: Session Metadata Foundation

examples/ezmsg_configs.py

@@ -82,8 +82,8 @@ async def listen(self, msg: int) -> None:
 
 
 class PassthroughCollection(ez.Collection):
-    INPUT = ez.InputStream(int)
-    OUTPUT = ez.OutputStream(int)
+    INPUT = ez.InputTopic(int)
+    OUTPUT = ez.OutputTopic(int)
 
     def network(self) -> ez.NetworkDefinition:
         return ((self.INPUT, self.OUTPUT),)
@@ -136,7 +136,7 @@ def configure(self) -> None:
 
 
 class PubNoSubCollection(ez.Collection):
-    OUTPUT = ez.OutputStream(int)
+    OUTPUT = ez.OutputTopic(int)
     GENERATE = Generator()
     LOG = DebugLog()
 
@@ -148,7 +148,7 @@ def network(self) -> ez.NetworkDefinition:
 
 
 class SubNoPubCollection(ez.Collection):
-    INPUT = ez.InputStream(int)
+    INPUT = ez.InputTopic(int)
     LISTEN = Listener()
 
     def network(self) -> ez.NetworkDefinition:
@@ -175,7 +175,7 @@ class PubNoSubPassthroughCollection(ez.Collection):
     COLLECTION = PubNoSubCollection()
     PASSTHROUGH = PassthroughCollection()
 
-    OUTPUT = ez.OutputStream(int)
+    OUTPUT = ez.OutputTopic(int)
 
     def network(self) -> ez.NetworkDefinition:
         return (
@@ -188,7 +188,7 @@ class SubNoPubPassthroughCollection(ez.Collection):
     COLLECTION = SubNoPubCollection()
     PASSTHROUGH = PassthroughCollection()
 
-    INPUT = ez.InputStream(int)
+    INPUT = ez.InputTopic(int)
 
     def network(self) -> ez.NetworkDefinition:
         return (

examples/ezmsg_toy.py

@@ -24,38 +24,64 @@ class LFOSettings(ez.Settings):
     update_rate: float = 2.0  # Hz, update rate
 
 
+class LFOState(ez.State):
+    start_time: float
+    cur_settings: LFOSettings
+
+
 class LFO(ez.Unit):
     SETTINGS = LFOSettings
+    STATE = LFOState
 
     OUTPUT = ez.OutputStream(float)
 
+    INPUT_SETTINGS = ez.InputStream(LFOSettings)
+
     async def initialize(self) -> None:
-        self.start_time = time.time()
+        self.STATE.cur_settings = self.SETTINGS
+        self.STATE.start_time = time.time()
 
+    @ez.subscriber(INPUT_SETTINGS)
+    async def on_settings(self, msg: LFOSettings) -> None:
+        self.STATE.cur_settings = msg
+
     @ez.publisher(OUTPUT)
     async def generate(self) -> AsyncGenerator:
         while True:
-            t = time.time() - self.start_time
-            yield self.OUTPUT, math.sin(2.0 * math.pi * self.SETTINGS.freq * t)
-            await asyncio.sleep(1.0 / self.SETTINGS.update_rate)
+            t = time.time() - self.STATE.start_time
+            yield self.OUTPUT, math.sin(2.0 * math.pi * self.STATE.cur_settings.freq * t)
+            await asyncio.sleep(1.0 / self.STATE.cur_settings.update_rate)
 
 
 # MESSAGE GENERATOR
 class MessageGeneratorSettings(ez.Settings):
     message: str
 
 
+class MessageGeneratorState(ez.State):
+    cur_settings: MessageGeneratorSettings
+
+
 class MessageGenerator(ez.Unit):
     SETTINGS = MessageGeneratorSettings
+    STATE = MessageGeneratorState
 
     OUTPUT = ez.OutputStream(str)
+    INPUT_SETTINGS = ez.InputStream(MessageGeneratorSettings)
+
+    async def initialize(self) -> None:
+        self.STATE.cur_settings = self.SETTINGS
+
+    @ez.subscriber(INPUT_SETTINGS)
+    async def on_settings(self, msg: MessageGeneratorSettings) -> None:
+        self.STATE.cur_settings = msg
 
     @ez.publisher(OUTPUT)
     async def spawn_message(self) -> AsyncGenerator:
         while True:
             await asyncio.sleep(1.0)
-            ez.logger.info(f"Spawning {self.SETTINGS.message}")
-            yield self.OUTPUT, self.SETTINGS.message
+            ez.logger.info(f"Spawning {self.STATE.cur_settings.message}")
+            yield self.OUTPUT, self.STATE.cur_settings.message
 
     @ez.publisher(OUTPUT)
     async def spawn_once(self) -> AsyncGenerator:
@@ -123,8 +149,8 @@ class ModifierCollection(ez.Collection):
     """This collection will subscribe to messages
     and append the most recent LFO output"""
 
-    INPUT = ez.InputStream(str)
-    OUTPUT = ez.OutputStream(str)
+    INPUT = ez.InputTopic(str)
+    OUTPUT = ez.OutputTopic(str)
 
     SIN = LFO()
     # SIN2 = LFO()
@@ -152,6 +178,8 @@ class TestSystemSettings(ez.Settings):
 class TestSystem(ez.Collection):
     SETTINGS = TestSystemSettings
 
+    OUTPUT_PING = ez.OutputTopic(str)
+
     # Publishers
     PING = MessageGenerator()
     FOO = MessageGenerator()
@@ -173,6 +201,7 @@ def configure(self) -> None:
     # Define Connections
     def network(self) -> ez.NetworkDefinition:
         return (
+            (self.PING.OUTPUT, self.OUTPUT_PING),
             (self.PING.OUTPUT, self.PINGSUB1.INPUT),
             (self.PING.OUTPUT, self.MODIFIER_COLLECTION.INPUT),
             (self.MODIFIER_COLLECTION.OUTPUT, self.PINGSUB2.INPUT),
@@ -193,7 +222,7 @@ def process_components(self):
     ez.run(
         SYSTEM=system,
         connections=[
-            # Make PING.OUTPUT available on a topic ezmsg_attach.py
-            (system.PING.OUTPUT, "GLOBAL_PING_TOPIC"),
+            # Make a system output available on a topic ezmsg_attach.py
+            (system.OUTPUT_PING, "GLOBAL_PING_TOPIC"),
         ],
     )