adds pysmurf-controller tag params (#424)

* adds pysmurf-controller tag params

* Update test

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Fix docs and small changes

* Fix tests

* Adds choices to stream_type

* bump sodetlib

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

docker/pysmurf_controller/Dockerfile

@@ -1,4 +1,4 @@
-FROM simonsobs/sodetlib:v0.5.0-2-g0f91202
+FROM simonsobs/sodetlib:v0.5.0-35-ga937bff
 
 ENV OCS_CONFIG_DIR /config
 

socs/agents/pysmurf_controller/agent.py

@@ -331,6 +331,9 @@ def _stop_check_state(self, session, params):
     @ocs_agent.param("duration", default=None, type=float)
     @ocs_agent.param('kwargs', default=None)
     @ocs_agent.param('load_tune', default=True, type=bool)
+    @ocs_agent.param('stream_type', default='obs', choices=['obs', 'oper'])
+    @ocs_agent.param('subtype', default=None)
+    @ocs_agent.param('tag', default=None)
     @ocs_agent.param('test_mode', default=False, type=bool)
     def stream(self, session, params):
         """stream(duration=None)
@@ -350,6 +353,15 @@ def stream(self, session, params):
         load_tune : bool
             If true, will load a tune-file to the pysmurf object on
             instantiation.
+        stream_type : string, optional
+            Stream type. This can be either 'obs' or 'oper', and will be 'obs'
+            by default. The tags attached to the stream will be
+            ``<stream_type>,<subtype>,<tag>``.
+        subtype : string, optional
+            Operation subtype used tot tag the stream.
+        tag : string, optional
+            Tag (or comma-separated list of tags) to attach to the G3 stream.
+            This has precedence over the `tag` key in the kwargs dict.
 
         Notes
         ------
@@ -364,6 +376,13 @@ def stream(self, session, params):
         if params['kwargs'] is None:
             params['kwargs'] = {}
 
+        if params['stream_type']:
+            params['kwargs']['tag'] = params['stream_type']
+        if params['subtype'] is not None:
+            params['kwargs']['subtype'] = params['subtype']
+        if params['tag'] is not None:
+            params['kwargs']['tag'] = params['tag']
+
         with self.lock.acquire_timeout(0, job='stream') as acquired:
             if not acquired:
                 return False, f"Operation failed: {self.lock.job} is running."
@@ -533,6 +552,7 @@ def uxm_relock(self, session, params):
 
     @ocs_agent.param('duration', default=30., type=float)
     @ocs_agent.param('kwargs', default=None)
+    @ocs_agent.param('tag', default=None)
     def take_noise(self, session, params):
         """take_noise(duration=30., kwargs=None)
 
@@ -549,6 +569,10 @@ def take_noise(self, session, params):
         kwargs : dict
             Dict containing additional keyword args to pass to the take_noise
             function.
+        tag : string, optional
+            Tag (or comma-separated list of tags) to attach to the G3 stream.
+            This has precedence over the `tag` key in the kwargs dict.
+
 
         Notes
         -------
@@ -565,6 +589,9 @@ def take_noise(self, session, params):
         if params['kwargs'] is None:
             params['kwargs'] = {}
 
+        if params['tag'] is not None:
+            params['kwargs']['g3_tag'] = params['tag']
+
         with self.lock.acquire_timeout(0, job='take_noise') as acquired:
             if not acquired:
                 return False, f"Operation failed: {self.lock.job} is running."
@@ -575,6 +602,7 @@ def take_noise(self, session, params):
             return True, "Finished taking noise"
 
     @ocs_agent.param('kwargs', default=None)
+    @ocs_agent.param('tag', default=None)
     def take_bgmap(self, session, params):
         """take_bgmap(kwargs=None)
 
@@ -589,6 +617,9 @@ def take_bgmap(self, session, params):
         ----
         kwargs : dict
             Additional kwargs to pass to take_bgmap function.
+        tag : Optional[str]
+            String containing a tag or comma-separated list of tags to attach
+            to the g3 stream.
 
         Notes
         ------
@@ -606,6 +637,9 @@ def take_bgmap(self, session, params):
         if params['kwargs'] is None:
             params['kwargs'] = {}
 
+        if params['tag'] is not None:
+            params['kwargs']['g3_tag'] = params['tag']
+
         with self.lock.acquire_timeout(0, job='take_bgmap') as acquired:
             if not acquired:
                 return False, f"Operation failed: {self.lock.job} is running."
@@ -630,6 +664,7 @@ def take_bgmap(self, session, params):
             return True, "Finished taking bgmap"
 
     @ocs_agent.param('kwargs', default=None)
+    @ocs_agent.param('tag', default=None)
     def take_iv(self, session, params):
         """take_iv(kwargs=None)
 
@@ -644,6 +679,9 @@ def take_iv(self, session, params):
         ----
         kwargs : dict
             Additional kwargs to pass to the ``take_iv`` function.
+        tag : Optional[str]
+            String containing a tag or comma-separated list of tags to attach
+            to the g3 stream.
 
         Notes
         ------
@@ -661,6 +699,9 @@ def take_iv(self, session, params):
         if params['kwargs'] is None:
             params['kwargs'] = {}
 
+        if params['tag'] is not None:
+            params['kwargs']['g3_tag'] = params['tag']
+
         with self.lock.acquire_timeout(0, job='take_iv') as acquired:
             if not acquired:
                 return False, f"Operation failed: {self.lock.job} is running."
@@ -679,6 +720,7 @@ def take_iv(self, session, params):
 
     @ocs_agent.param('kwargs', default=None)
     @ocs_agent.param('rfrac_range', default=(0.2, 0.9))
+    @ocs_agent.param('tag', default=None)
     def take_bias_steps(self, session, params):
         """take_bias_steps(kwargs=None, rfrac_range=(0.2, 0.9))
 
@@ -694,6 +736,9 @@ def take_bias_steps(self, session, params):
         rfrac_range : tuple
             Range of valid rfracs to check against when determining the number
             of good detectors.
+        tag : Optional[str]
+            String containing a tag or comma-separated list of tags to attach
+            to the g3 stream.
 
         Notes
         ------
@@ -718,6 +763,9 @@ def take_bias_steps(self, session, params):
         if params['kwargs'] is None:
             params['kwargs'] = {}
 
+        if params['tag'] is not None:
+            params['kwargs']['g3_tag'] = params['tag']
+
         with self.lock.acquire_timeout(0, job='bias_steps') as acquired:
             if not acquired:
                 return False, f"Operation failed: {self.lock.job} is running."

tests/agents/test_pysmurf_controller_agent.py

@@ -283,7 +283,7 @@ def test_take_bgmap(agent):
     **Test** - Tests take_bgmap task.
     """
     session = create_session('take_bgmap')
-    res = agent.take_bgmap(session, {'kwargs': {'high_current_mode': False}})
+    res = agent.take_bgmap(session, {'kwargs': {'high_current_mode': False}, 'tag': None})
     assert res[0] is True
     assert session.data['nchans_per_bg'] == [NCHANS, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
     assert session.data['filepath'] == 'bias_step_analysis.npy'
@@ -307,7 +307,7 @@ def test_take_iv(agent):
     **Test** - Tests take_iv task.
     """
     session = create_session('take_iv')
-    res = agent.take_iv(session, {'kwargs': {'run_analysis': False}})
+    res = agent.take_iv(session, {'kwargs': {'run_analysis': False}, 'tag': None})
     assert res[0] is True
     assert session.data['bands'] == np.zeros(NCHANS).tolist()
     assert session.data['channels'] == np.arange(NCHANS).tolist()
@@ -330,7 +330,7 @@ def test_take_bias_steps(agent):
     **Test** - Tests take_bias_steps task.
     """
     session = create_session('take_bias_steps')
-    res = agent.take_bias_steps(session, {'kwargs': None, 'rfrac_range': (0.3, 0.9)})
+    res = agent.take_bias_steps(session, {'kwargs': None, 'rfrac_range': (0.3, 0.9), 'tag': None})
     assert res[0] is True
     assert session.data['filepath'] == 'bias_step_analysis.npy'
 
@@ -344,7 +344,7 @@ def test_take_noise(agent):
     **Test** - Tests take_noise task.
     """
     session = create_session('take_noise')
-    res = agent.take_noise(session, {'duration': 30, 'kwargs': None})
+    res = agent.take_noise(session, {'duration': 30, 'kwargs': None, 'tag': None})
     assert res[0] is True
 
 
@@ -377,7 +377,9 @@ def test_stream(agent):
     **Test** - Tests stream process.
     """
     session = create_session('stream')
-    res = agent.stream(session, {'duration': None, 'load_tune': False, 'kwargs': None, 'test_mode': True})
+    res = agent.stream(session, {'duration': None, 'load_tune': False,
+                                 'kwargs': None, 'test_mode': True, 'tag': None,
+                                 'stream_type': 'obs', 'subtype': None})
     assert res[0] is True