Implement alerts monitoring

python/jaeger/__init__.py

@@ -1,16 +1,21 @@
 # encoding: utf-8
 # isort: skip_file
 
+from __future__ import annotations
+
 import logging
 import os
-from typing import Union
+from typing import TYPE_CHECKING
 import warnings
 
 from sdsstools import get_config, get_logger, get_package_version
 from sdsstools.configuration import __ENVVARS__
 
 from .exceptions import JaegerUserWarning
 
+if TYPE_CHECKING:
+    from .actor import JaegerActor
+
 
 NAME = "jaeger"
 
@@ -52,7 +57,7 @@ def start_file_loggers(start_log=True, start_can=True):
         can_log.start_file_logger(os.path.join(log_dir, "can.log"))
 
 
-actor_instance = None
+actor_instance: JaegerActor | None = None
 
 
 from .can import *

python/jaeger/actor/commands/__init__.py

@@ -19,6 +19,7 @@
 JaegerCommandType = Command[JaegerActor]
 
 
+from .alerts import *
 from .can import *
 from .chiller import *
 from .configuration import *

python/jaeger/actor/commands/alerts.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego (gallegoj@uw.edu)
+# @Date: 2022-01-27
+# @Filename: chiller.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from . import JaegerCommandType, command_parser
+
+
+if TYPE_CHECKING:
+    from jaeger import FPS
+
+
+__all__ = ["alerts"]
+
+
+@command_parser.group()
+def alerts(*args):
+    """Manages the alerts system."""
+
+    pass
+
+
+@alerts.command()
+async def status(command: JaegerCommandType, fps: FPS):
+    """Shows the status of the alerts."""
+
+    return command.finish(message={k: int(v) for k, v in fps.alerts.keywords.items()})
+
+
+@alerts.command()
+async def reset(command: JaegerCommandType, fps: FPS):
+    """Clears all the alerts. Needed after certain alerts have been raised."""
+
+    fps.alerts.reset()
+    return command.finish(message={k: int(v) for k, v in fps.alerts.keywords.items()})

python/jaeger/actor/commands/chiller.py

@@ -1,3 +1,11 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# @Author: José Sánchez-Gallego (gallegoj@uw.edu)
+# @Date: 2022-01-27
+# @Filename: chiller.py
+# @License: BSD 3-clause (http://www.opensource.org/licenses/BSD-3-Clause)
+
 from __future__ import annotations
 
 import asyncio

python/jaeger/actor/commands/positioner.py

@@ -264,7 +264,7 @@ async def initialise(command, fps, positioner_id, datums=False):
 
 @jaeger_parser.command()
 @click.argument("POSITIONERS", type=int, nargs=-1, required=False)
-async def status(command, fps, positioners):
+async def status(command: JaegerCommandType, fps: FPS, positioners):
     """Reports the position and status bit of a list of positioners."""
 
     positioner_ids = positioners or list(fps.positioners.keys())
@@ -277,6 +277,7 @@ async def status(command, fps, positioners):
         command.info(locked=fps.locked)
         command.info(n_positioners=len(fps.positioners))
         command.info(fps_status=f"0x{fps.status.value:x}")
+        command.info(message={k: int(v) for k, v in fps.alerts.keywords.items()})
 
     try:
         await fps.update_status(positioner_ids=0)