add support for streaming ANTARES alerts

README.md

@@ -38,9 +38,10 @@ in the output.
 
 Each Kafka stream that your TOM listens to (via `readstreams`) will have a configuration dictionary
 in your `settings.py` `ALERT_STREAMS`. `ALERT_STREAMS` is a list of configuration dictionaries, one
-dictionary for each Kafka stream. Here's an example `ALERT_STREAMS` configuration for two Kafka streams:
-[SCiMMA Hopskotch](https://scimma.org/hopskotch.html) and
-[GCN Classic over Kafka](https://gcn.nasa.gov/quickstart).
+dictionary for each Kafka stream. Here's an example `ALERT_STREAMS` configuration for three Kafka streams:
+[SCiMMA Hopskotch](https://scimma.org/hopskotch.html),
+[GCN Classic over Kafka](https://gcn.nasa.gov/quickstart), and
+[ANTARES](https://nsf-noirlab.gitlab.io/csdc/antares/client/).
 
 ```python
 ALERT_STREAMS = [
@@ -83,6 +84,18 @@ ALERT_STREAMS = [
                 'gcn.classic.text.LVC_RETRACTION': 'tom_alertstreams.alertstreams.alertstream.alert_logger',
             },
         },
+    },
+    {
+        'ACTIVE': True,
+        'NAME': 'tom_alertstreams.alertstreams.antares.AntaresAlertStream',
+        'OPTIONS': {
+            'API_KEY': os.getenv('ANTARES_API_KEY'),
+            'API_SECRET': os.getenv('ANTARES_API_SECRET'),
+            'GROUP': os.getenv('ANTARES_GROUP_ID'),
+            'TOPIC_HANDLERS': {
+                'extragalactic_staging': 'tom_antares.alertstream_handlers.handle_alert',
+            }
+        },
     }
 ]
 ```

pyproject.toml

@@ -40,7 +40,8 @@ packages = [
 dependencies = [
     "psycopg2-binary >=2.9,<3.0",
     "gcn-kafka >=0.3,<1.0",
-    "hop-client >=0.10,<1.0"
+    "hop-client >=0.10,<1.0",
+    "antares-client",
 ]
 
 [tool.setuptools_scm]

tom_alertstreams/alertstreams/antares.py

@@ -0,0 +1,30 @@
+import logging
+from .alertstream import AlertStream
+from antares_client.stream import StreamingClient
+
+logger = logging.getLogger(__name__)
+
+
+class AntaresAlertStream(AlertStream):
+    """
+    Wrapper for the ANTARES broker streaming client. See https://nsf-noirlab.gitlab.io/csdc/antares/client/.
+    """
+    required_keys = ['API_KEY', 'API_SECRET', 'TOPIC_HANDLERS']
+    allowed_keys = ['API_KEY', 'API_SECRET', 'TOPIC_HANDLERS', 'GROUP', 'SSL_CA_LOCATION', 'ENABLE_AUTO_COMMIT']
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        logger.debug(f'AntaresAlertStream.__init__() kwargs: {kwargs}')
+        optional_keys = set(self.allowed_keys) - set(self.required_keys)
+        options = {key.lower(): kwargs[key] for key in optional_keys if key in kwargs}
+        self.stream = StreamingClient(
+            topics=self.topic_handlers.keys(),
+            api_key=self.api_key,
+            api_secret=self.api_secret,
+            **options
+        )
+
+    def listen(self):
+        for topic, locus in self.stream.iter():
+            base_topic = topic.removeprefix(self.stream._TOPIC_PREFIX)
+            logger.info(f"received {locus.locus_id} on {base_topic}")
+            self.alert_handler[base_topic](locus)

tom_alertstreams/alertstreams/hopskotch.py

@@ -52,7 +52,7 @@ def get_all_public_topics(self) -> list[str]:
         # include only topics that a) contain a '.'; b) don't start with '__' (excludes __consumer_offsets)
         publicly_readable_topics = [topic for topic in list_topics(self.url, hop_auth).keys()
                                     if not (topic.startswith('__') and (topic.count('.')==0))]
-        logger.info(f'publicly_readable_topics: {publicly_readable_topics}')
+        logger.debug(f'publicly_readable_topics: {publicly_readable_topics}')
 
         return publicly_readable_topics