WIP: Implement IPC with ZMQ

.github/workflows/CI.yaml

@@ -20,7 +20,7 @@ jobs:
       max-parallel: 1
       fail-fast: false
       matrix:
-        python-version: ["3.8", "3.9", "3.10"]
+        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
         os: [ubuntu-latest, macos-latest] # [ubuntu-latest, macos-latest, windows-latest]
 
     steps:
@@ -65,7 +65,7 @@ jobs:
 
       - name: Test with pytest
         run: |
-          python -m pytest --cov=improv 
+          python -m pytest -x -s -l --cov=improv 
 
       - name: Coveralls
         uses: coverallsapp/github-action@v2

.gitignore

@@ -128,6 +128,9 @@ dmypy.json
 arrow
 
 .vscode/
+.idea/
 
 *.code-workspace
 improv/_version.py
+
+*venv

demos/1p_caiman/1p_demo.yaml

@@ -34,6 +34,3 @@ connections:
   Processor.q_out: [Analysis.q_in]
   Analysis.q_out: [Visual.q_in]
   InputStim.q_out: [Analysis.input_stim_queue]
-
-# settings:
-#   use_watcher: [Acquirer, Processor, Visual, Analysis]

demos/basic/Behavior_demo.py

@@ -7,7 +7,7 @@
 loadFile = "./Behavior_demo.yaml"
 
 nexus = Nexus("Nexus")
-nexus.createNexus(file=loadFile)
+nexus.create_nexus(file=loadFile)
 
 # All modules needed have been imported
 # so we can change the level of logging here
@@ -20,4 +20,4 @@
 # logger = logging.getLogger("improv")
 # logger.setLevel(logging.INFO)
 
-nexus.startNexus()
+nexus.start_nexus()

demos/basic/Behavior_demo.yaml

@@ -1,6 +1,3 @@
-settings:
-  use_watcher: [Acquirer, Processor, Visual, Analysis, Behavior, Motion]
-
 actors:
   GUI:
     package: actors.visual

demos/basic/basic_demo.py

@@ -7,7 +7,7 @@
 loadFile = "./basic_demo.yaml"
 
 nexus = Nexus("Nexus")
-nexus.createNexus(file=loadFile)
+nexus.create_nexus(file=loadFile)
 
 # All modules needed have been imported
 # so we can change the level of logging here
@@ -20,4 +20,4 @@
 # logger = logging.getLogger("improv")
 # logger.setLevel(logging.INFO)
 
-nexus.startNexus()
+nexus.start_nexus()

demos/basic/basic_demo.yaml

@@ -33,6 +33,3 @@ connections:
   Processor.q_out: [Analysis.q_in]
   Analysis.q_out: [Visual.q_in]
   InputStim.q_out: [Analysis.input_stim_queue]
-
-# settings:
-#   use_watcher: [Acquirer, Processor, Visual, Analysis]

demos/bubblewrap/actors/bubble.py

@@ -55,7 +55,7 @@ def setup(self):
         self.bw.init_nodes()
         logger.info("Nodes initialized")
 
-        self._getStoreInterface()
+        self._get_store_interface()
 
     def runStep(self):
         """Observe new data from dim reduction and update bubblewrap"""

demos/live/live_demo.py

@@ -7,7 +7,7 @@
 loadFile = "./live_demo.yaml"
 
 nexus = Nexus("Nexus")
-nexus.createNexus(file=loadFile)
+nexus.create_nexus(file=loadFile)
 
 # All modules needed have been imported
 # so we can change the level of logging here
@@ -20,4 +20,4 @@
 # logger = logging.getLogger("improv")
 # logger.setLevel(logging.INFO)
 
-nexus.startNexus()
+nexus.start_nexus()

demos/zmq/actors/zmq_ps_sample_generator.py → demos/minimal/actors/sample_generator_zmq.py

@@ -1,15 +1,14 @@
+from improv.actor import ZmqActor
+from datetime import date  # used for saving
 import numpy as np
 import logging
 
-from demos.sample_actors.zmqActor import ZmqActor
-
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)
 
 
 class Generator(ZmqActor):
-    """Sample actor to generate data to pass into a sample processor
-    using sync ZMQ to communicate.
+    """Sample actor to generate data to pass into a sample processor.
 
     Intended for use along with sample_processor.py.
     """
@@ -25,39 +24,53 @@ def __str__(self):
 
     def setup(self):
         """Generates an array that serves as an initial source of data.
-        Sets up a ZmqPSActor to send data to the processor.
 
         Initial array is a 100 row, 5 column numpy matrix that contains
         integers from 1-99, inclusive.
         """
 
-        logger.info("Beginning setup for Generator")
         self.data = np.asmatrix(np.random.randint(100, size=(100, 5)))
-        logger.info("Completed setup for Generator")
+        self.improv_logger.info("Completed setup for Generator")
+
+    # def run(self):
+    #     """ Send array into the store.
+    #     """
+    #     self.fcns = {}
+    #     self.fcns['setup'] = self.setup
+    #     self.fcns['run'] = self.runStep
+    #     self.fcns['stop'] = self.stop
+
+    #     with RunManager(self.name, self.fcns, self.links) as rm:
+    #         logger.info(rm)
 
     def stop(self):
         """Save current randint vector to a file."""
 
-        logger.info("Generator stopping")
-        np.save("sample_generator_data.npy", self.data)
+        self.improv_logger.info("Generator stopping")
+        np.save(f"sample_generator_data", self.data)
+        # This is not the best example of a save function,
+        # will overwrite previous files with the same name.
         return 0
 
-    def runStep(self):
+    def run_step(self):
         """Generates additional data after initial setup data is exhausted.
-        Sends data to the processor using a ZmqPSActor.
 
         Data is of a different form as the setup data in that although it is
         the same size (5x1 vector), it is uniformly distributed in [1, 10]
         instead of in [1, 100]. Therefore, the average over time should
         converge to 5.5.
         """
+
         if self.frame_num < np.shape(self.data)[0]:
-            data_id = self.client.put(self.data[self.frame_num], str(f"Gen_raw: {self.frame_num}"))
+            data_id = self.client.put(self.data[self.frame_num])
             try:
-                self.put(data_id) #[data_id, str(self.frame_num)])
+                self.q_out.put(data_id)
+                # self.improv_logger.info(f"Sent {self.data[self.frame_num]} with key {data_id}")
                 self.frame_num += 1
+
             except Exception as e:
-                logger.error(f"---------Generator Exception: {e}")
+                self.improv_logger.error(f"Generator Exception: {e}")
         else:
-            new_data = np.asmatrix(np.random.randint(10, size=(1, 5)))
-            self.data = np.concatenate((self.data, new_data), axis=0)
+            self.data = np.concatenate(
+                (self.data, np.asmatrix(np.random.randint(10, size=(1, 5)))), axis=0
+            )

demos/minimal/actors/sample_persistence_generator.py

@@ -0,0 +1,78 @@
+import random
+import time
+
+from improv.actor import ZmqActor
+from datetime import date  # used for saving
+import numpy as np
+import logging
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+
+class Generator(ZmqActor):
+    """Sample actor to generate data to pass into a sample processor.
+
+    Intended for use along with sample_processor.py.
+    """
+
+    def __init__(self, output_filename, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.data = None
+        self.name = "Generator"
+        self.frame_num = 0
+        self.output_filename = output_filename
+
+    def __str__(self):
+        return f"Name: {self.name}, Data: {self.data}"
+
+    def setup(self):
+        """Generates an array that serves as an initial source of data.
+
+        Initial array is a 100 row, 5 column numpy matrix that contains
+        integers from 1-99, inclusive.
+        """
+
+        self.data = np.asmatrix(np.random.randint(100, size=(100, 5)))
+        self.improv_logger.info("Completed setup for Generator")
+
+    def stop(self):
+        """Save current randint vector to a file."""
+
+        self.improv_logger.info("Generator stopping")
+        return 0
+
+    def run_step(self):
+        """Generates additional data after initial setup data is exhausted.
+
+        Data is of a different form as the setup data in that although it is
+        the same size (5x1 vector), it is uniformly distributed in [1, 10]
+        instead of in [1, 100]. Therefore, the average over time should
+        converge to 5.5.
+        """
+
+
+        device_time = time.time_ns()
+        time.sleep(0.0003)
+        acquired_time = time.time_ns()  # mock the time the generator "actually received" the data
+
+        if self.frame_num < np.shape(self.data)[0]:
+
+            with open(self.output_filename, "a+") as f:
+                device_data = self.data[self.frame_num]
+                packaged_data = (acquired_time, (device_time, device_data))
+                # save the data to the flat file before sending it downstream
+                f.write(f"{packaged_data[0]}, {packaged_data[1][0]}, {packaged_data[1][1]}\n")
+
+                data_id = self.client.put(packaged_data)
+                try:
+                    self.q_out.put(data_id)
+                    # self.improv_logger.info(f"Sent {self.data[self.frame_num]} with key {data_id}")
+                    self.frame_num += 1
+
+                except Exception as e:
+                    self.improv_logger.error(f"Generator Exception: {e}")
+        else:
+            self.data = np.concatenate(
+                (self.data, np.asmatrix(np.random.randint(10, size=(1, 5)))), axis=0
+            )

demos/zmq/actors/zmq_ps_sample_processor.py → demos/minimal/actors/sample_persistence_processor.py

@@ -1,44 +1,46 @@
+from improv.actor import ZmqActor
 import numpy as np
 import logging
 
-from demos.sample_actors.zmqActor import ZmqActor
-
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)
 
 
 class Processor(ZmqActor):
-    """Sample processor used to calculate the average of an array of integers
-    using sync ZMQ to communicate.
+    """Sample processor used to calculate the average of an array of integers.
 
     Intended for use with sample_generator.py.
     """
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
+        if "name" in kwargs:
+            self.name = kwargs["name"]
 
     def setup(self):
         """Initializes all class variables.
-        Sets up a ZmqRRActor to receive data from the generator.
 
         self.name (string): name of the actor.
-        self.frame (ObjectID): Store object id referencing data from the store.
+        self.frame (ObjectID): StoreInterface object id referencing data from the store.
         self.avg_list (list): list that contains averages of individual vectors.
         self.frame_num (int): index of current frame.
         """
-        self.name = "Processor"
+
+        if not hasattr(self, "name"):
+            self.name = "Processor"
         self.frame = None
         self.avg_list = []
         self.frame_num = 1
-        logger.info("Completed setup for Processor")
+        self.improv_logger.info("Completed setup for Processor")
 
     def stop(self):
         """Trivial stop function for testing purposes."""
-        logger.info("Processor stopping; have received {} frames so far".format(self.frame_num))
 
-    def runStep(self):
+        self.improv_logger.info("Processor stopping")
+        return 0
+
+    def run_step(self):
         """Gets from the input queue and calculates the average.
-        Receives data from the generator using a ZmqRRActor.
 
         Receives an ObjectID, references data in the store using that
         ObjectID, calculates the average of that data, and finally prints
@@ -47,19 +49,18 @@ def runStep(self):
 
         frame = None
         try:
-            frame = self.get()
-
-        except:
-            logger.error("Could not get frame!")
+            frame = self.q_in.get(timeout=0.05)
+        except Exception as e:
+            # logger.error(f"{self.name} could not get frame! At {self.frame_num}: {e}")
             pass
 
-        if frame is not None:
+        if frame is not None and self.frame_num is not None:
             self.done = False
-            self.frame = self.client.getID(frame)
-            avg = np.mean(self.frame[0])
-
-            # logger.info(f"Average: {avg}")
+            self.frame = self.client.get(frame)
+            device_data = self.frame[1][1]
+            avg = np.mean(device_data)
+            # self.improv_logger.info(f"Average: {avg}")
             self.avg_list.append(avg)
-            logger.info(f"Overall Average: {np.mean(self.avg_list)}")
-            # logger.info(f"Frame number: {self.frame_num}")
+            # self.improv_logger.info(f"Overall Average: {np.mean(self.avg_list)}")
+            # self.improv_logger.info(f"Frame number: {self.frame_num}")
             self.frame_num += 1