diff --git a/README.md b/README.md index b7d8816..e130ba0 100644 --- a/README.md +++ b/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', + } + }, } ] ``` 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..bd17fa4 --- /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](locus) 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