Create ACTiCameraAgent (#591)

* initial

* create latest image file

* update docstrings

* add docs

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

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

* move datetime info

* agent name change

* address comments

* restructure naming and session.data

* fix data structure

* address comments

* fix flake8

* docs: Fix build errors

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Brian Koopman <brian.koopman@yale.edu>

docs/agents/acti_camera.rst

@@ -0,0 +1,83 @@
+.. highlight:: rst
+
+.. _acti_camera:
+
+====================
+ACTi Camera Agent
+====================
+
+The ACTi Camera Agent is an OCS Agent which grabs screenshots from ACTi cameras
+and saves files to a directory.
+
+.. argparse::
+    :filename: ../socs/agents/acti_camera/agent.py
+    :func: add_agent_args
+    :prog: python3 agent.py
+
+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 ACTi Camera Agent we need to add a ACTiCameraAgent
+block to our ocs configuration file. Here is an example configuration block
+using all of the available arguments::
+
+      {'agent-class': 'ACTiCameraAgent',
+       'instance-id': 'cameras',
+       'arguments': [['--mode', 'acq'],
+                     ['--camera-addresses', ['10.10.10.41', '10.10.10.42', '10.10.10.43']],
+                     ['--locations', ['location1', 'location2', 'location3']],
+                     ['--user', 'admin'],
+                     ['--password', 'password']]},
+
+.. note::
+    The ``--camera-addresses`` argument should be a list of the IP addresses
+    of the cameras on the network.
+    The ``--locations`` argument should be a list of names for camera locations.
+    This should be in the same order as the list of IP addresses.
+
+Docker Compose
+``````````````
+
+The iBootbar Agent should be configured to run in a Docker container. An
+example docker-compose service configuration is shown here::
+
+  ocs-cameras:
+    image: simonsobs/socs:latest
+    hostname: ocs-docker
+    environment:
+      - INSTANCE_ID=cameras
+      - SITE_HUB=ws://127.0.0.1:8001/ws
+      - SITE_HTTP=http://127.0.0.1:8001/call
+      - LOGLEVEL=info
+    volumes:
+      - ${OCS_CONFIG_DIR}:/config:ro
+      - /mnt/nfs/data/cameras:/screenshots
+    user: 9000:9000
+
+The ``LOGLEVEL`` environment variable can be used to set the log level for
+debugging. The default level is "info".
+The volume must mount to ``/screenshots``. The user must have permissions to write
+to the mounted local directory.
+
+Description
+-----------
+
+The ACTi cameras will be used to monitor conditions at the SO site.
+The ACTi Camera Agent periodically (1 minute) grabs screenshots from each
+camera on the network. The images are saved to a location on disk. A webserver
+should then be configured to serve this directory to some URL. Then we can use
+HTML to access the webserver and display ``latest.jpg`` for an up-to-date
+view of the camera. For example, this can be done directly in Grafana
+using the Text panel in HTML mode.
+
+Agent API
+---------
+
+.. autoclass:: socs.agents.acti_camera.agent.ACTiCameraAgent
+    :members:

docs/index.rst

@@ -37,6 +37,7 @@ API Reference        Full API documentation for core parts of the SOCS library.
     :caption: Agent Reference
     :maxdepth: 2
 
+    agents/acti_camera
     agents/acu_agent
     agents/bluefors_agent
     agents/cryomech_cpa

socs/agents/acti_camera/agent.py

@@ -0,0 +1,219 @@
+import argparse
+import os
+import shutil
+import time
+from pathlib import Path
+
+import requests
+import txaio
+from ocs import ocs_agent, site_config
+from ocs.ocs_twisted import Pacemaker, TimeoutLock
+
+# For logging
+txaio.use_twisted()
+
+
+class ACTiCameraAgent:
+    """Grab screenshots from ACTi cameras.
+
+    Parameters
+    ----------
+    agent : OCSAgent
+        OCSAgent object which forms this Agent
+    camera_addresses : list
+        List of IP addresses (as strings) of cameras.
+    user : str
+        Username of cameras.
+    password : str
+        Password of cameras.
+
+    Attributes
+    ----------
+    agent : OCSAgent
+        OCSAgent object which forms this Agent
+    is_streaming : bool
+        Tracks whether or not the agent is actively issuing requests to grab
+        screenshots from cameras. Setting to false stops sending commands.
+    log : txaio.tx.Logger
+        txaio logger object, created by the OCSAgent
+    """
+
+    def __init__(self, agent, camera_addresses, locations, user, password):
+        self.agent = agent
+        self.is_streaming = False
+        self.log = self.agent.log
+        self.lock = TimeoutLock()
+
+        self.cameras = []
+        for (location, address) in (zip(locations, camera_addresses)):
+            self.cameras.append({'location': location,
+                                 'address': address,
+                                 'connected': True})
+        self.user = user
+        self.password = password
+
+        agg_params = {
+            'frame_length': 10 * 60  # [sec]
+        }
+        self.agent.register_feed('cameras',
+                                 record=True,
+                                 agg_params=agg_params,
+                                 buffer_time=0)
+
+    @ocs_agent.param('test_mode', default=False, type=bool)
+    def acq(self, session, params=None):
+        """acq(test_mode=False)
+
+        **Process** - Grab screenshots from ACTi cameras.
+
+        Parameters
+        ----------
+        test_mode : bool, optional
+            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
+        structure::
+
+            >>> response.session['data']
+            # for each camera
+            {'location1': {'location': 'location1',
+                         'last_attempt': 1701983575.032506,
+                         'connected': True,
+                         'address': '10.10.10.41'},
+             'location2': ...
+            }
+        """
+        pm = Pacemaker(1 / 60, quantize=False)
+
+        session.set_status('running')
+        self.is_streaming = True
+        while self.is_streaming:
+            # Use UTC
+            timestamp = time.time()
+            data = {}
+
+            for camera in self.cameras:
+                data[camera['location']] = {'location': camera['location']}
+                self.log.info(f"Grabbing screenshot from {camera['location']}")
+                payload = {'USER': self.user,
+                           'PWD': self.password,
+                           'SNAPSHOT': 'N640x480,100'}
+                url = f"http://{camera['address']}/cgi-bin/encoder"
+
+                # Format directory and filename
+                ctime = int(timestamp)
+                ctime_dir = int(str(timestamp)[:5])
+                Path(f"screenshots/{camera['location']}/{ctime_dir}").mkdir(parents=True, exist_ok=True)
+                filename = f"screenshots/{camera['location']}/{ctime_dir}/{ctime}.jpg"
+                latest_filename = f"screenshots/{camera['location']}/latest.jpg"
+
+                # If no response from camera, update connection status and continue
+                try:
+                    response = requests.get(url, params=payload, stream=True, timeout=5)
+                except requests.exceptions.RequestException as e:
+                    self.log.error(f'{e}')
+                    self.log.info("Unable to get response from camera.")
+                    camera['connected'] = False
+                    data[camera['location']]['last_attempt'] = time.time()
+                    data[camera['location']]['connected'] = camera['connected']
+                    continue
+                camera['connected'] = True
+                self.log.debug("Received screenshot from camera.")
+
+                # Write screenshot to file and update latest file
+                with open(filename, 'wb') as out_file:
+                    shutil.copyfileobj(response.raw, out_file)
+                self.log.debug(f"Wrote {ctime}.jpg to /{camera['location']}/{ctime_dir}.")
+                shutil.copy2(filename, latest_filename)
+                self.log.debug(f"Updated latest.jpg in /{camera['location']}.")
+                del response
+
+                data[camera['location']]['last_attempt'] = time.time()
+                data[camera['location']]['connected'] = camera['connected']
+
+            # Update session.data and publish to feed
+            for camera in self.cameras:
+                data[camera['location']]['address'] = camera['address']
+            session.data = data
+            self.log.debug("{data}", data=session.data)
+
+            message = {
+                'block_name': 'cameras',
+                'timestamp': timestamp,
+                'data': {}
+            }
+            for camera in self.cameras:
+                message['data'][camera['location'] + "_connected"] = int(camera['connected'])
+            session.app.publish_to_feed('cameras', message)
+            self.log.debug("{msg}", msg=message)
+
+            if params['test_mode']:
+                break
+            pm.sleep()
+
+        return True, "Finished Recording"
+
+    def _stop_acq(self, session, params=None):
+        """_stop_acq()
+        **Task** - Stop task associated with acq process.
+        """
+        if self.is_streaming:
+            session.set_status('stopping')
+            self.is_streaming = False
+            return True, "Stopping Recording"
+        else:
+            return False, "Acq is not currently running"
+
+
+def add_agent_args(parser=None):
+    """
+    Build the argument parser for the Agent. Allows sphinx to automatically
+    build documentation based on this function.
+    """
+    if parser is None:
+        parser = argparse.ArgumentParser()
+
+    pgroup = parser.add_argument_group("Agent Options")
+    pgroup.add_argument("--camera-addresses", nargs='+', type=str, help="List of camera IP addresses.")
+    pgroup.add_argument("--locations", nargs='+', type=str, help="List of camera locations.")
+    pgroup.add_argument("--user", help="Username of camera.")
+    pgroup.add_argument("--password", help="Password of camera.")
+    pgroup.add_argument("--mode", choices=['acq', 'test'])
+
+    return parser
+
+
+def main(args=None):
+    # Start logging
+    txaio.start_logging(level=os.environ.get("LOGLEVEL", "info"))
+
+    parser = add_agent_args()
+    args = site_config.parse_args(agent_class='ACTiCameraAgent',
+                                  parser=parser,
+                                  args=args)
+
+    if args.mode == 'acq':
+        init_params = True
+    elif args.mode == 'test':
+        init_params = False
+
+    agent, runner = ocs_agent.init_site_agent(args)
+    p = ACTiCameraAgent(agent,
+                        camera_addresses=args.camera_addresses,
+                        locations=args.locations,
+                        user=args.user,
+                        password=args.password)
+
+    agent.register_process("acq",
+                           p.acq,
+                           p._stop_acq,
+                           startup=init_params)
+
+    runner.run(agent, auto_reconnect=True)
+
+
+if __name__ == "__main__":
+    main()

socs/agents/ocs_plugin_so.py

@@ -11,6 +11,7 @@
 root = os.path.abspath(os.path.split(__file__)[0])
 
 for n, f in [
+        ('ACTiCameraAgent', 'acti_camera/agent.py'),
         ('ACUAgent', 'acu/agent.py'),
         ('BlueforsAgent', 'bluefors/agent.py'),
         ('CrateAgent', 'smurf_crate_monitor/agent.py'),

socs/plugin.py

@@ -1,5 +1,6 @@
 package_name = 'socs'
 agents = {
+    'ACTiCameraAgent': {'module': 'socs.agents.acti_camera.agent', 'entry_point': 'main'},
     'ACUAgent': {'module': 'socs.agents.acu.agent', 'entry_point': 'main'},
     'BlueforsAgent': {'module': 'socs.agents.bluefors.agent', 'entry_point': 'main'},
     'CrateAgent': {'module': 'socs.agents.smurf_crate_monitor.agent', 'entry_point': 'main'},