Merge pull request #77 from simonsobs/koopman/boresight-commands

Add boresight control

src/sorunlib/acu.py

@@ -10,5 +10,26 @@ def move_to(az, el):
         el (float): destination angle for the elevation axis
 
     """
+    # Az/El motion
     resp = run.CLIENTS['acu'].go_to(az=az, el=el)
     check_response(resp)
+
+
+def set_boresight(target):
+    """Move the third axis to a specific target angle.
+
+    Args:
+        target (float): destination angle for boresight rotation
+
+    Raises:
+        RuntimeError: If boresight is passed to a non-satp platform.
+
+    """
+    # Check platform type
+    resp = run.CLIENTS['acu'].monitor.status()
+    platform = resp.session['data']['PlatformType']
+    if platform == "satp":
+        resp = run.CLIENTS['acu'].set_boresight(target=target)
+        check_response(resp)
+    else:
+        raise RuntimeError(f"Platform type {platform} does not support boresight motion")

tests/test_acu.py

@@ -5,9 +5,28 @@
 
 from unittest.mock import MagicMock, patch
 
+import ocs
 from ocs.ocs_client import OCSReply
 from sorunlib import acu
 
+from util import create_session
+
+
+def create_acu_client(platform_type):
+    """Create an ACU client with mock monitor Process session.data.
+
+    Args:
+        platform_type (str): Either 'satp' or 'ccat'.
+
+    """
+    acu_client = MagicMock()
+    session = create_session('monitor')
+    session.data = {'PlatformType': platform_type}
+    reply = OCSReply(ocs.OK, 'msg', session.encoded())
+    acu_client.monitor.status = MagicMock(return_value=reply)
+
+    return acu_client
+
 
 def mocked_clients(test_mode):
     clients = {'acu': MagicMock(),
@@ -23,6 +42,22 @@ def test_move_to():
     acu.run.CLIENTS['acu'].go_to.assert_called_with(az=180, el=60)
 
 
+@patch('sorunlib.create_clients', mocked_clients)
+def test_set_boresight():
+    acu.run.initialize(test_mode=True)
+    acu.run.CLIENTS['acu'] = create_acu_client('satp')
+    acu.set_boresight(20)
+    acu.run.CLIENTS['acu'].set_boresight.assert_called_with(target=20)
+
+
+@patch('sorunlib.create_clients', mocked_clients)
+def test_set_boresight_lat():
+    acu.run.initialize(test_mode=True)
+    acu.run.CLIENTS['acu'] = create_acu_client('ccat')
+    with pytest.raises(RuntimeError):
+        acu.set_boresight(20)
+
+
 @patch('sorunlib.create_clients', mocked_clients)
 def test_move_to_failed():
     acu.run.initialize()

tests/util.py

@@ -0,0 +1,11 @@
+from unittest import mock
+
+from ocs.ocs_agent import OpSession
+
+
+def create_session(op_name):
+    """Create an OpSession with a mocked app for testing."""
+    mock_app = mock.MagicMock()
+    session = OpSession(1, op_name, app=mock_app)
+
+    return session