From 6bdb8cf7e885a5d893711685e8d8d5a6460601a0 Mon Sep 17 00:00:00 2001 From: Griffin Hosseinzadeh <1976665+griffin-h@users.noreply.github.com> Date: Thu, 29 Jan 2026 17:00:49 -0800 Subject: [PATCH 1/5] add antares client --- pyproject.toml | 3 ++- tom_alertstreams/alertstreams/antares.py | 30 ++++++++++++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) create mode 100644 tom_alertstreams/alertstreams/antares.py diff --git a/pyproject.toml b/pyproject.toml index 8ce7b4b..ae0eeb5 100644 --- a/pyproject.toml +++ b/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] diff --git a/tom_alertstreams/alertstreams/antares.py b/tom_alertstreams/alertstreams/antares.py new file mode 100644 index 0000000..2df65f1 --- /dev/null +++ b/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](None, locus) From 02c199159cd4b164a09c0d5c93b766da352bca68 Mon Sep 17 00:00:00 2001 From: Griffin Hosseinzadeh <1976665+griffin-h@users.noreply.github.com> Date: Thu, 29 Jan 2026 17:03:12 -0800 Subject: [PATCH 2/5] do not print full list of publicly readable hopskotch topics in log --- tom_alertstreams/alertstreams/hopskotch.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tom_alertstreams/alertstreams/hopskotch.py b/tom_alertstreams/alertstreams/hopskotch.py index 9a649eb..3ec9518 100644 --- a/tom_alertstreams/alertstreams/hopskotch.py +++ b/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 From a23545959d5b4358a158d953d8fbdce213f9e67a Mon Sep 17 00:00:00 2001 From: Griffin Hosseinzadeh <1976665+griffin-h@users.noreply.github.com> Date: Thu, 29 Jan 2026 17:03:44 -0800 Subject: [PATCH 3/5] remove placeholder None --- tom_alertstreams/alertstreams/antares.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tom_alertstreams/alertstreams/antares.py b/tom_alertstreams/alertstreams/antares.py index 2df65f1..bd17fa4 100644 --- a/tom_alertstreams/alertstreams/antares.py +++ b/tom_alertstreams/alertstreams/antares.py @@ -27,4 +27,4 @@ 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](None, locus) + self.alert_handler[base_topic](locus) From 2c71b39e8347d19560e2557e64ad7e42057ecb96 Mon Sep 17 00:00:00 2001 From: Griffin Hosseinzadeh <1976665+griffin-h@users.noreply.github.com> Date: Tue, 10 Feb 2026 10:56:21 -0800 Subject: [PATCH 4/5] Update README.md --- README.md | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index b7d8816..a6ebb8e 100644 --- a/README.md +++ b/README.md @@ -39,8 +39,9 @@ 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). +[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', + } + }, } ] ``` From a58d9dc1b956696eea7c23882db40597a3df67ff Mon Sep 17 00:00:00 2001 From: Griffin Hosseinzadeh <1976665+griffin-h@users.noreply.github.com> Date: Tue, 10 Feb 2026 10:56:54 -0800 Subject: [PATCH 5/5] Update README.md --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index a6ebb8e..e130ba0 100644 --- a/README.md +++ b/README.md @@ -38,7 +38,7 @@ 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: +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/).