Create Water Vapor Radiometer Agent (#429)

* commit PWV agent

* commit pwv_agent.py

* commit changes to pwv agent

* found and fixed errors to functions outside of class

* keeping variable names consistent

* Setup demo Flask server for fetching latest pwv value

* Automatically fetch latest pwv file in api

* Add docstring, mark private methods as such, and data dir outside of helper function

* Setup to run with gunicorn and write README

* rename pwv agent script to match latest socs version

* move pwv_agent dir to correct up-to-date socs path

* update pwv agent for socs pluginification and flask server changes

* clean off unnecessary functions and add docstrings

* implement flake8 comments

* change directory/class name of agent to vendor/hardware name

* change acq name process

* clean up acq, _stop_acq, and __init__, and add test_mode

* add radiometer agent docs

* update radiometer agent description

* fix typo to actually compare timestamps in acq()

* add pacemaker

* commit flake8 rules

* commit tests for radioimeter agent

* fix pwv agent in plugin.py to updated dir and agent class name

* update ocs_plugin_so.py to match PWV agent's updated name/class

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

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

* Fix flake8 warning

* remove year arg for agent

* change flask server working to radiometer web server

* add session.data for pwv vals

* remove argparse args

* add session.data output to docstring

* remove year in example

* remove supporting APIs section in docs

* add blank line for docs consistency

* change phrasing to be specific about PWV agent mechanics

* change the port back to 5000

* skip pacemaker step if in test mode

* remove api from socs repo

* change port on http mock test for consistency

* update test to work pytest-docker plugin instead of pytest-docker-compose

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

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

* change port for radiometer in config file

* fix data key to avoid nested structure

* move pacemaker outside of loop

* add test_mode to startup params

* bye bye 'hi'

* remove acq.stop() in test now that test_params exist

* fix test_mode arg

* fix test mode arg string and replace type with action

* add dependencies to the docs

* fix test_params dict for agent startup

* clean up test_mode conditional and clarify test-mode argument

* add test-mode to tests config file for agents

* Flatten args list for radiometer agent

* Remove install instructions from readme

I expect the installation process to change, just rely on the linked repo for
instructions to reduce maintenance burden.

---------

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

docs/agents/ucsc_radiometer.rst

@@ -0,0 +1,72 @@
+.. highlight:: rst
+
+.. _ucsc_radiometer:
+
+=====================
+UCSC Radiometer Agent
+=====================
+
+The UCSC Radiometer Agent monitors the PWV through the UCSC Radiometer web server.
+
+.. argparse::
+    :filename: ../socs/agents/ucsc_radiometer/agent.py
+    :func: add_agent_args
+    :prog: python3 agent.py
+
+Dependencies
+------------
+
+The UCSC Radiometer Agent requires the `UCSC Radiometer Server
+<https://github.com/simonsobs/ucscradiometer-server>`_. This server runs where
+the radiometer data files are written and makes readings available to the Agent
+over HTTP.
+
+Configuration File Examples
+---------------------------
+
+Below are configuration examples for the ocs config file and for running the
+Agent in a docker container
+
+OCS Site Config
+```````````````
+
+To configure the UCSC Radiometer Agent we need to a UCSCRadiometerAgent
+block to our ocs configuration file. Here is an example configuration block
+using all of the available arguments::
+
+        {'agent-class': 'UCSCRadiometerAgent',
+         'instance-id': 'pwvs',
+         'arguments':[['--url', 'http://127.0.0.1:5000']]},
+
+.. note::
+   The ``--url`` argument should be the address of the Flask server on the
+   Web which is publishing pwv data from a server connected to the
+   radiometer on-site.
+
+Docker Compose
+``````````````
+
+The UCSC Radiometer Agent should be configured to run in a Docker container. An
+example docker-compose service configuration is shown here::
+
+  ocs-ucsc-radiometer:
+       image: simonsobs/socs:latest
+       hostname: ocs-docker
+       network_mode: host
+       volumes:
+           - ${OCS_CONFIG_DIR}:/config
+       environment:
+           - INSTANCE_ID=pwvs
+
+Description
+-----------
+
+The UCSC radiometer measures precipitable water vapor (pwv) of the atmosphere,
+and outputs the values to disk on a computer at the site where OCS
+is not setup. A web server makes the PWV values available over HTTP. The Agent requests the PWV data and then publishes it to OCS.
+
+Agent API
+---------
+
+.. autoclass:: socs.agents.ucsc_radiometer.agent.UCSCRadiometerAgent
+    :members:

docs/index.rst

@@ -71,6 +71,7 @@ API Reference        Full API documentation for core parts of the SOCS library.
     agents/synacc
     agents/tektronix3021c
     agents/thorlabs_mc2000b
+    agents/ucsc_radiometer
     agents/ups
     agents/vantage_pro2
     agents/wiregrid_actuator

socs/agents/ocs_plugin_so.py

@@ -45,6 +45,7 @@
         ('SynthAgent', 'holo_synth/agent.py'),
         ('TektronixAWGAgent', 'tektronix3021c/agent.py'),
         ('ThorlabsMC2000BAgent', 'thorlabs_mc2000b/agent.py'),
+        ('UCSCRadiometerAgent', 'ucsc_radiometer/agent.py'),
         ('UPSAgent', 'ups/agent.py'),
         ('VantagePro2Agent', 'vantagepro2/agent.py'),
         ('WiregridActuatorAgent', 'wiregrid_actuator/agent.py'),

socs/agents/ucsc_radiometer/agent.py

@@ -0,0 +1,142 @@
+import argparse
+from os import environ
+
+import requests
+import txaio
+from ocs import ocs_agent, site_config
+from ocs.ocs_twisted import Pacemaker, TimeoutLock
+
+
+class UCSCRadiometerAgent:
+    """Monitor the PWV Flask Server.
+
+    Parameters
+    ----------
+    agent : OCS Agent
+        OCSAgent object which forms this Agent
+    url : str
+        url of the radiometer web server on the internet
+
+    """
+
+    def __init__(self, agent, url):
+        self.agent = agent
+        self.log = agent.log
+        self.lock = TimeoutLock()
+
+        self.url = url
+
+        self.take_data = False
+
+        agg_params = {'frame_length': 60,
+                      'exclude_influx': False}
+
+        # register the feed
+        self.agent.register_feed('pwvs',
+                                 record=True,
+                                 agg_params=agg_params,
+                                 buffer_time=1
+                                 )
+
+        self.last_published_reading = None
+
+    @ocs_agent.param('test_mode', default=False, type=bool)
+    def acq(self, session, params=None):
+        """acq()
+
+        **Process** - Fetch values from PWV Flask Server
+
+        Parameters
+        ----------
+        test_mode : bool, option
+            Run the Process loop only once. Meant only for testing.
+            Default is False.
+
+        Notes
+        -----
+        The most recent data collected is stored in session data in the
+        following structure::
+
+            >>> response.session['data']
+            {'timestamp': 1678820419.0,
+             'pwv': 0.49253026985972237}
+
+        """
+        pm = Pacemaker(1 / 60, quantize=False)
+
+        self.take_data = True
+        while self.take_data:
+            r = requests.get(self.url)
+            data = r.json()
+            last_pwv = data['pwv']
+            last_timestamp = data['timestamp']
+
+            pwvs = {'block_name': 'pwvs',
+                    'timestamp': last_timestamp,
+                    'data': {'pwv': last_pwv}
+                    }
+
+            if self.last_published_reading is not None:
+                if last_timestamp > self.last_published_reading[1]:
+                    self.agent.publish_to_feed('pwvs', pwvs)
+                    self.last_published_reading = (last_pwv, last_timestamp)
+            else:
+                self.agent.publish_to_feed('pwvs', pwvs)
+                self.last_published_reading = (last_pwv, last_timestamp)
+
+            session.data = {"timestamp": last_timestamp,
+                            "pwv": {}}
+
+            session.data['pwv'] = last_pwv
+
+            if params['test_mode']:
+                break
+            else:
+                pm.sleep()
+
+        return True, 'Acquisition exited cleanly.'
+
+    def _stop_acq(self, session, params=None):
+        """
+        Stops acq process.
+        """
+        self.take_data = False
+        return True, 'Stopping acq process'
+
+
+def add_agent_args(parser=None):
+    if parser is None:
+        parser = argparse.ArgumentParser()
+    pgroup = parser.add_argument_group('Agent Options')
+    pgroup.add_argument("--url", type=str, help="url for radiometer web server")
+    pgroup.add_argument("--test-mode", action='store_true',
+                        help="Determines whether agent runs in test mode."
+                        "Default is False.")
+    return parser
+
+
+def main(args=None):
+    # For logging
+    txaio.use_twisted()
+    txaio.make_logger()
+
+    txaio.start_logging(level=environ.get("LOGLEVEL", "info"))
+
+    parser = add_agent_args()
+    args = site_config.parse_args(agent_class='UCSCRadiometerAgent', parser=parser, args=args)
+
+    # test params
+    test_params = {'test_mode': False}
+    if args.test_mode:
+        test_params = {'test_mode': True}
+
+    agent, runner = ocs_agent.init_site_agent(args)
+    pwv_agent = UCSCRadiometerAgent(agent, args.url)
+
+    agent.register_process('acq', pwv_agent.acq, pwv_agent._stop_acq, startup=test_params)
+
+    runner.run(agent, auto_reconnect=True)
+
+
+if __name__ == "__main__":
+    main()

socs/plugin.py

@@ -36,6 +36,7 @@
     'SynthAgent': {'module': 'socs.agents.holo_synth.agent', 'entry_point': 'main'},
     'TektronixAWGAgent': {'module': 'socs.agents.tektronix3021c.agent', 'entry_point': 'main'},
     'ThorlabsMC2000BAgent': {'module': 'socs.agents.thorlabs_mc2000b.agent', 'entry_point': 'main'},
+    'UCSCRadiometerAgent': {'module': 'socs.agents.ucsc_radiometer.agent', 'entry_point': 'main'},
     'UPSAgent': {'module': 'socs.agents.ups.agent', 'entry_point': 'main'},
     'VantagePro2Agent': {'module': 'socs.agents.vantagepro2.agent', 'entry_point': 'main'},
     'WiregridActuatorAgent': {'module': 'socs.agents.wiregrid_actuator.agent', 'entry_point': 'main'},

tests/default.yaml

@@ -108,5 +108,10 @@ hosts:
                      ['--test-mode']
                     ]
       },
+      {'agent-class': 'UCSCRadiometerAgent',
+      'instance-id': 'pwvs',
+      'arguments': ['--url', 'http://127.0.0.1:5000',
+                    '--test-mode']
+      },
     ]
   }

tests/integration/test_ucsc_radiometer_agent_integration.py

@@ -0,0 +1,47 @@
+import time
+from datetime import datetime
+
+import ocs
+import pytest
+from flask import jsonify, request
+from http_server_mock import HttpServerMock
+from integration.util import docker_compose_file  # noqa: F401
+from integration.util import create_crossbar_fixture
+from ocs.base import OpCode
+from ocs.testing import create_agent_runner_fixture, create_client_fixture
+
+wait_for_crossbar = create_crossbar_fixture()
+run_agent = create_agent_runner_fixture(
+    "../socs/agents/ucsc_radiometer/agent.py",
+    "radiometer",
+    args=["--log-dir", "./logs/"],
+)
+client = create_client_fixture("pwvs")
+
+
+@pytest.fixture
+def http_mock():
+    app = HttpServerMock(__name__)
+
+    @app.route("/", methods=["GET"])
+    def route_fn():
+        if request.method == "GET":
+            time_now = datetime.now()
+            timestamp = time.mktime(time_now.timetuple())
+            data = {'pwv': 1.2, 'timestamp': timestamp}
+            return jsonify(data)
+        else:
+            assert False, "Bad query"
+
+    return app.run("127.0.0.1", 5000)
+
+
+@pytest.mark.integtest
+def test_ucsc_radiometer_acq(wait_for_crossbar, http_mock, run_agent, client):
+    with http_mock:
+        resp = client.acq.start(test_mode=True)
+        resp = client.acq.wait()
+        print(resp)
+        assert resp.status == ocs.OK
+        print(resp.session)
+        assert resp.session['op_code'] == OpCode.SUCCEEDED.value