Add elevation nod sequence

Author: BrianJKoopman

src/sorunlib/seq.py

@@ -1,4 +1,5 @@
 import datetime as dt
+import time
 
 import sorunlib as run
 
@@ -9,16 +10,20 @@
 OP_TIMEOUT = 60
 
 
-def _stop_scan():
-    acu = run.CLIENTS['acu']
-
-    print("Stopping scan.")
+def _stop_smurfs():
     # Stop SMuRF streams
     try:
         run.smurf.stream('off')
     except RuntimeError as e:
         print(f"Caught error while shutting down SMuRF streams: {e}")
 
+
+def _stop_scan():
+    acu = run.CLIENTS['acu']
+
+    print("Stopping scan.")
+    _stop_smurfs()
+
     # Stop motion
     acu.generate_scan.stop()
     resp = acu.generate_scan.wait(timeout=OP_TIMEOUT)
@@ -85,3 +90,43 @@ def scan(description, stop_time, width, az_drift=0, tag=None, subtype=None,
         monitor_process(acu, 'generate_scan', stop_time)
     finally:
         _stop_scan()
+
+
+def el_nod(el1, el2, num=5, pause=5):
+    """Perform a set of elevation nods.
+
+    Elevation nods will be peformed at the current azimuth, and will start from
+    and return to the current elevation. The nod first moves to ``el1``,
+    pauses, then moves to ``el2``, pauses, and then repeats for the
+    specified number of iterations.
+
+    Args:
+        el1 (float): First elevation to move to during the nod.
+        el2 (float): Second elevation to move to during the nod.
+        num (int): Number of nods to peform. Defaults to 5.
+        pause (float): Length of pause, in seconds, at each elevation. Defaults
+            to 5 seconds.
+
+    """
+    acu = run.CLIENTS['acu']
+
+    # Enable SMuRF streams
+    run.smurf.stream('on', subtype='cal', tag='el_nods')
+
+    try:
+        # Grab current telescope position
+        resp = acu.monitor.status()
+        init_az = resp.session['data']['StatusDetailed']['Azimuth current position']
+        init_el = resp.session['data']['StatusDetailed']['Elevation current position']
+
+        # Perform nods
+        for x in range(num):
+            run.acu.move_to(az=init_az, el=el1)
+            time.sleep(pause)
+            run.acu.move_to(az=init_az, el=el2)
+            time.sleep(pause)
+        else:
+            # Return to initial position
+            run.acu.move_to(az=init_az, el=init_el)
+    finally:
+        _stop_smurfs()

tests/test_seq.py

@@ -5,7 +5,7 @@
 
 import ocs
 import pytest
-from unittest.mock import MagicMock, patch
+from unittest.mock import call, MagicMock, patch
 from ocs.ocs_client import OCSReply
 
 import sorunlib
@@ -92,3 +92,18 @@ def test_scan_failed_smurfs_on_shutdown(patch_clients):
 
     seq._stop_scan()
     seq.run.CLIENTS['acu'].generate_scan.wait.assert_called()
+
+
+@patch('sorunlib.seq.time.sleep', MagicMock())
+def test_el_nod(patch_clients):
+    sorunlib.acu.move_to(az=180, el=50)
+    seq.el_nod(el1=40, el2=60)
+
+    # Calls will be repeated, but these three are representative of the el nod
+    calls = [call(az=180, el=50),
+             call(az=180, el=40),
+             call(az=180, el=60)]
+    seq.run.CLIENTS['acu'].go_to.assert_has_calls(calls, any_order=True)
+
+    # Move back to initial position
+    seq.run.CLIENTS['acu'].go_to.assert_called_with(az=180, el=50)