Zmq pr

.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
@@ -102,6 +102,6 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Close parallel build
-        uses: coverallsapp/github-action@v1
+        uses: coverallsapp/github-action@v2
         with:
           parallel-finished: true

.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/1p_caiman/actors/1p_processor.py

@@ -58,13 +58,13 @@ def setup(self):
     def stop(self):
         super().stop()
 
-    def runStep(self):
+    def run_step(self):
         """Run process. Runs once per frame.
         Output is a location in the DS to continually
         place the Estimates results, with ref number that
         corresponds to the frame number
         """
-        super().runStep()
+        super().run_step()
 
     def putEstimates(self):
         """Put whatever estimates we currently have

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/actors/basic_processor.py

@@ -61,7 +61,7 @@ def stop(self):
         np.savetxt("output/timing/shape_time.txt", self.shape_time)
         np.savetxt("output/timing/detect_time.txt", self.detect_time)
 
-    def runStep(self):
+    def run_step(self):
         """Run process. Runs once per frame.
         Output is a location in the DS to continually
         place the Estimates results, with ref number that

demos/basic/actors/behavior.py

@@ -41,7 +41,7 @@ def setup(self):
         else:
             raise FileNotFoundError
 
-    def runStep(self):
+    def run_step(self):
         if self.done:
             pass
         elif self.frame_num < self.data.shape[2]:
@@ -95,7 +95,7 @@ def setup(self):
         else:
             raise FileNotFoundError
 
-    def runStep(self):
+    def run_step(self):
         if self.done:
             pass
         elif self.frame_num < len(self.data):

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/README.md

@@ -4,13 +4,13 @@ After installing _improv_, install additional dependencies (JAX on CPU by defaul
 
 - `pip install -r requirements.txt`
 
-To download the sample data from the paper, do:
+To download the sample data from the paper, run the following from the `improv` directory:
 
 - `python demos/bubblewrap/actors/utils.py`
 
-This may take a few minutes. After data is downloaded, run the GUI with:
+This may take a few minutes. After data is downloaded, run the demo with:
 
-- `python demos/bubblewrap/bubble_demo.py`
+- `improv run demos/bubblewrap/bubble_demo.yaml`
 
 A GUI will pop up with two buttons named "setup" and "run". First hit "setup" and wait ~5 seconds, then hit "run". Bubblewrap will perform dimensionality reduction of ~180 neurons to 2 dimensions, represented by grey dots popping up on the plot, and coarsely tile the space with red bubbles to represent transitions in the low-dimension space. All in real-time! 
 

demos/bubblewrap/actors/acquire.py

@@ -4,17 +4,42 @@
 import time
 import logging
 import traceback
-import time
+import sys, os
 
 
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)
 
+def load_data(filename):
+    """
+    Load filename, which may be a relative path. 
+    Return data.
+    """
+    if os.path.isfile(filename):
+        full_fname = filename
+    else:
+        for p in reversed(sys.path):
+        # traverse path in reverse order, on the theory that
+        # the yaml file comes near the end
+            full_fname = os.path.join(p,  filename)
+            if os.path.isfile(full_fname):
+                break
+
+    try:
+        data = mat73.loadmat(full_fname)
+    except FileNotFoundError:
+        logger.error("Bubblewrap data file not found!")
+
+    return data
+
+
+
 
 class Acquirer(Actor):
     def __init__(self, *args, filename=None, **kwargs):
         super().__init__(*args, **kwargs)
-        if not filename: logger.error('Error: Filename not specified')
+        if not filename: 
+            logger.error('Error: Filename not specified')
         self.file = filename
         self.frame_num = 0
         self.done = False
@@ -28,7 +53,7 @@ def setup(self):
         Note: A utility function that downloads the required data file can be found in utils.py
         """
         # get unsorted vs sorted units
-        data_dict = mat73.loadmat(self.file)
+        data_dict = load_data(self.file)
         units_unsorted = []
         units_sorted = []
         for ch_curr in data_dict['spikes']:
@@ -58,23 +83,23 @@ def setup(self):
         self.num_iters = np.floor((self.data.shape[0] - l1 - self.l)/self.l).astype('int')
 
         #send to dim reduction
-        init_id = self.client.put([self.data.shape[0], self.data[:l1, :]], "init_data")
+        init_id = self.client.put([self.data.shape[0], self.data[:l1, :]])
         logger.info("Putted init data")
         self.q_out.put(init_id)
 
     def stop(self):
         logger.info(f"Stopped running Acquire, avg time per frame: {np.mean(self.total_times)}")
         logger.info(f"Acquire got through {self.frame_num} frames")
 
-    def runStep(self):
+    def run_step(self):
         """Send data to dim reduction one frame at a time"""
         if self.done:
             pass
         elif self.frame_num < self.num_iters:
             start, end = self.t, self.t + 1
             frame = self.data[start:end, :]
             t = time.time()
-            id = self.client.put([self.t, frame], "acq_bubble" + str(self.frame_num))
+            id = self.client.put([self.t, frame])
             self.timestamp.append([time.time(), self.frame_num])
             try:
                 self.q_out.put([str(self.frame_num), id])

demos/bubblewrap/actors/bubble.py

@@ -4,7 +4,7 @@
 from bubblewrap import Bubblewrap
 from improv.actor import Actor
 import logging
-
+logging.getLogger("jax").setLevel(logging.WARNING)
 
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)
@@ -21,7 +21,10 @@ def setup(self):
         while shape_id is None:
             try:
                 shape_id = self.q_in.get(timeout=0.0005)
-            except Empty: pass
+            except Empty: 
+                pass
+            except TimeoutError:
+                pass
         dat_shape_0 = self.client.get(shape_id)
         # init bubblewrap
         M = 20
@@ -47,23 +50,24 @@ def setup(self):
         while id is None:
             try:
                 id = self.q_in.get(timeout = 0.0005)
-            except Empty: pass
-        init_data = self.client.getID(id)
+            except Empty: 
+                pass
+            except TimeoutError:
+                pass
+        init_data = self.client.get(id)
 
         for i in np.arange(0, M):
             self.bw.observe(init_data[i])
         self.bw.init_nodes()
         logger.info("Nodes initialized")
 
-        self._getStoreInterface()
-
-    def runStep(self):
+    def run_step(self):
         """Observe new data from dim reduction and update bubblewrap"""
         try:
             ids = self.q_in.get(timeout=0.0005)
 
             # expect ids of size 2 containing data location and frame number
-            new_data = self.client.getID(ids[1])
+            new_data = self.client.get(ids[1])
             self.frame_number = ids[0]
 
             self.bw.observe(new_data)
@@ -72,19 +76,17 @@ def runStep(self):
             self.putOutput()
         except Empty:
             pass
+        except TimeoutError:
+            pass
 
     def putOutput(self):
         """Function for putting updated results into the store"""
         ids = []
-        ids.append(self.client.put(np.array(self.bw.A), "A" + str(self.frame_number)))
-        ids.append(self.client.put(np.array(self.bw.L), "L" + str(self.frame_number)))
-        ids.append(self.client.put(np.array(self.bw.mu), "mu" + str(self.frame_number)))
-        ids.append(self.client.put(
-                    np.array(self.bw.n_obs), "n_obs" + str(self.frame_number)))
-        ids.append(self.client.put(
-                    np.array(self.bw.pred), "pred" + str(self.frame_number)))
-        ids.append(self.client.put(
-                    np.array(self.bw.entropy_list), "entropy" + str(self.frame_number)))
-        ids.append(self.client.put(
-                    np.array(self.bw.dead_nodes), "dead_nodes" + str(self.frame_number)))
+        ids.append(self.client.put(np.array(self.bw.A)))
+        ids.append(self.client.put(np.array(self.bw.L)))
+        ids.append(self.client.put(np.array(self.bw.mu)))
+        ids.append(self.client.put(np.array(self.bw.n_obs)))
+        ids.append(self.client.put(np.array(self.bw.pred)))
+        ids.append(self.client.put(np.array(self.bw.entropy_list)))
+        ids.append(self.client.put(np.array(self.bw.dead_nodes)))
         self.q_out.put([self.frame_number, ids])