Support for LSST (#230)

* Add multi-suvey for the configuration file

* PEP8

* Rewrite log

* Update doc and test

* Propagate survey to the consumer. Display only relevant topics per service

* Propagate survey to bin

* Properly decode payload based on the survey

* Fix WARN messahe

* Fix writing avro function

* Update setup

* Update tools for visualisation

* DOc

* Data transfer ready for LSST

* Documentation

* Doc

* Fix test

* Add utilities to read parquet files

* Add survey

* Fix test

* 10.0-rc0

* Fix name of argument

* Rename argument to be consistent with other tools. Make argument rquired.

Author: JulienPeloton

.ruff.toml

@@ -43,7 +43,6 @@ ignore = [
     "D400", "D401", "D100", "D102", "D103", "D104", "D415", "D419", 
     "E731",
     "N812", "N806", "N803", 
-    "PD901", 
     "PLR0913"
 ]
 

README.md

@@ -39,10 +39,10 @@ In order to connect and poll alerts from Fink, you need to get your credentials:
 1. Subscribe to one or more Fink streams by filling this [form](https://forms.gle/2td4jysT4e9pkf889).
 2. After filling the form, we will send your credentials. Register them on your laptop by simply running:
   ```
-  fink_client_register -username <USERNAME> -group_id <GROUP_ID> -servers <SERVERS> ...
+  fink_client_register -survey SURVEY -username USERNAME -group_id GROUP_ID -servers SERVERS ...
   ```
 
-In case of doubt, run `fink_client_register -h`. You can also inspect the configuration file on disk:
+Note that `SURVEY` is among `ztf` or `lsst`. In case of doubt, run `fink_client_register -h`. You can also inspect the configuration file on disk:
 
 ```bash
 cat ~/.finkclient/credentials.yml
@@ -54,77 +54,104 @@ Once you have your credentials, you are ready to poll streams! You can easily ac
 
 ```bash
 fink_consumer -h
-usage: fink_consumer [-h] [--display] [--display_statistics] [-limit LIMIT]
-                     [--available_topics] [--save] [-outdir OUTDIR]
-                     [-schema SCHEMA] [--dump_schema] [-start_at START_AT]
+usage: fink_consumer    [-h] -survey SURVEY [--display]
+                        [--display_statistics] [-limit LIMIT]
+                        [--available_topics] [--save]
+                        [-outdir OUTDIR] [-schema SCHEMA]
+                        [--dump_schema] [-start_at START_AT]
 
-Kafka consumer to listen and archive Fink streams from the Livestream service
+Kafka consumer to listen and archive Fink streams from the
+Livestream service
 
-optional arguments:
+options:
   -h, --help            show this help message and exit
-  --display             If specified, print on screen information about
-                        incoming alert.
-  --display_statistics  If specified, print on screen information about queues,
-                        and exit.
-  -limit LIMIT          If specified, download only `limit` alerts. Default is
-                        None.
-  --available_topics    If specified, print on screen information about
-                        available topics.
-  --save                If specified, save alert data on disk (Avro). See also
-                        -outdir.
-  -outdir OUTDIR        Folder to store incoming alerts if --save is set. It
-                        must exist.
-  -schema SCHEMA        Avro schema to decode the incoming alerts. Default is
-                        None (version taken from each alert)
-  --dump_schema         If specified, save the schema on disk (json file)
-  -start_at START_AT    If specified, reset offsets to 0 (`earliest`) or empty
-                        queue (`latest`).
+  -survey SURVEY        Survey name among ztf or lsst. Note that
+                        each survey will have its own configuration
+                        file.
+  --display             If specified, print on screen information
+                        about incoming alert.
+  --display_statistics  If specified, print on screen information
+                        about queues, and exit.
+  -limit LIMIT          If specified, download only `limit` alerts.
+                        Default is None.
+  --available_topics    If specified, print on screen information
+                        about available topics.
+  --save                If specified, save alert data on disk
+                        (Avro). See also -outdir.
+  -outdir OUTDIR        Folder to store incoming alerts if --save is
+                        set. It must exist.
+  -schema SCHEMA        Avro schema to decode the incoming alerts.
+                        Default is None (version taken from each
+                        alert)
+  --dump_schema         If specified, save the schema on disk (json
+                        file)
+  -start_at START_AT    If specified, reset offsets to 0
+                        (`earliest`) or empty queue (`latest`).
 ```
 
 You can also look at an alert on the disk:
 
 ```bash
 fink_alert_viewer -h
-usage: fink_alert_viewer [-h] [-filename FILENAME]
+usage: fink_alert_viewer [-h] [-f F] [-s S]
 
-Display cutouts and lightcurve from a ZTF alert
+Display cutouts and lightcurve from an alert
 
-optional arguments:
-  -h, --help          show this help message and exit
-  -filename FILENAME  Path to an alert data file (avro format)
+options:
+  -h, --help  show this help message and exit
+  -f F        Path to an alert data file (avro format)
+  -s S        Survey name among ztf or lsst.
 ```
 
-More information at [docs/livestream](https://fink-broker.readthedocs.io/en/latest/services/livestream).
+More information at [docs/livestream](https://doc.lsst.fink-broker.org/en/latest/services/livestream).
 
 ## Data Transfer usage
 
-If you requested data using the [Data Transfer service](https://fink-portal.org/download), you can easily poll your stream using:
+If you requested data using the Data Transfer service ([ZTF](https://ztf.fink-portal.org/download) or [LSST](https://lsst.fink-portal.org/download)), you can easily poll your stream using:
 
 ```bash
-usage: fink_datatransfer.py [-h] [-topic TOPIC] [-limit LIMIT] [-outdir OUTDIR] [-partitionby PARTITIONBY] [-batchsize BATCHSIZE] [-nconsumers NCONSUMERS]
-                            [-maxtimeout MAXTIMEOUT] [-number_partitions NUMBER_PARTITIONS] [--restart_from_beginning] [--verbose]
+usage: fink_datatransfer [-h] -survey SURVEY [-topic TOPIC] [-limit LIMIT]
+                            [-outdir OUTDIR] [-partitionby PARTITIONBY]
+                            [-batchsize BATCHSIZE] [-nconsumers NCONSUMERS]
+                            [-maxtimeout MAXTIMEOUT]
+                            [-number_partitions NUMBER_PARTITIONS]
+                            [--restart_from_beginning] [--dump_schema] [--verbose]
 
 Kafka consumer to listen and archive Fink streams from the data transfer service
 
-optional arguments:
+options:
   -h, --help            show this help message and exit
+  -survey SURVEY        Survey name among ztf or lsst. Note that each survey will have its
+                        own configuration file.
   -topic TOPIC          Topic name for the stream that contains the data.
-  -limit LIMIT          If specified, download only `limit` alerts from the stream. Default is None, that is download all alerts.
-  -outdir OUTDIR        Folder to store incoming alerts. It will be created if it does not exist.
+  -limit LIMIT          If specified, download only `limit` alerts from the stream.
+                        Default is None, that is download all alerts.
+  -outdir OUTDIR        Folder to store incoming alerts. It will be created if it does not
+                        exist.
   -partitionby PARTITIONBY
-                        Partition data by `time` (year=YYYY/month=MM/day=DD), or `finkclass` (finkclass=CLASS), or `tnsclass` (tnsclass=CLASS). `classId` is
-                        also available for ELASTiCC data. Default is time.
-  -batchsize BATCHSIZE  Maximum number of alert within the `maxtimeout` (see conf). Default is 1000 alerts.
+                        If specified, partition data when writing alerts on disk.
+                        Available options: - `time`: year=YYYY/month=MM/day=DD (ztf and
+                        lsst) - `finkclass`: finkclass=CLASS (ztf only) - `tnsclass`:
+                        tnsclass=CLASS (ztf only) - `classId`: classId=CLASSID (ELASTiCC
+                        only) Default is None, that is no partitioning is applied (all
+                        parquet files in the `outdir` folder).
+  -batchsize BATCHSIZE  Maximum number of alert within the `maxtimeout` (see conf).
+                        Default is 1000 alerts.
   -nconsumers NCONSUMERS
-                        Number of parallel consumer to use. Default (-1) is the number of logical CPUs in the system.
+                        Number of parallel consumer to use. Default (-1) is the number of
+                        logical CPUs in the system.
   -maxtimeout MAXTIMEOUT
-                        Overwrite the default timeout (in seconds) from user configuration. Default is None.
+                        Overwrite the default timeout (in seconds) from user
+                        configuration. Default is None.
   -number_partitions NUMBER_PARTITIONS
-                        Number of partitions for the topic in the distant Kafka cluster. Do not touch unless you know what your are doing. Default is 10
+                        Number of partitions for the topic in the distant Kafka cluster.
+                        Do not touch unless you know what your are doing. Default is 10
                         (Fink Kafka cluster)
   --restart_from_beginning
-                        If specified, restart downloading from the 1st alert in the stream. Default is False.
+                        If specified, restart downloading from the 1st alert in the
+                        stream. Default is False.
+  --dump_schema         If specified, save the schema on disk (json file)
   --verbose             If specified, print on screen information about the consuming.
 ```
 
-More information at [docs/datatransfer](https://fink-broker.readthedocs.io/en/latest/services/data_transfer/).
+More information at [docs/datatransfer](https://doc.ztf.fink-broker.org/en/latest/services/data_transfer/).

bin/fink_client_test.sh

@@ -27,7 +27,7 @@ done
 TEST_DIR=${FINK_CLIENT_HOME}/tests
 
 # Fake credentials
-fink_client_register -username test -password None \
+fink_client_register -survey ztf -username test -password None \
   -servers 'localhost:9093, localhost:9094, localhost:9095' \
   -mytopics rrlyr -group_id test_group -maxtimeout 10 --tmp
 

fink_client/__init__.py

@@ -12,5 +12,5 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-__version__ = "9.2"
+__version__ = "10.0-rc0"
 __schema_version__ = "distribution_schema_fink_ztf_{}.avsc"

fink_client/avro_utils.py

@@ -230,9 +230,9 @@ def write_alert(
     alert: dict,
     schema: str,
     path: str,
-    overwrite: bool = False,
-    id1: str = "",
+    id1: str,
     id2: str = "",
+    overwrite: bool = False,
 ):
     """Write avro alert on disk
 
@@ -245,12 +245,12 @@ def write_alert(
     path: str
         Folder that will contain the alert. The filename will always be
         <objectID>.avro
+    id1: str
+        Prefix for alert name
+    id2: str, optional
+        Second prefix for alert name if need be (id1_id2.avro)
     overwrite: bool, optional
         If True, overwrite existing alert. Default is False.
-    id1: str, optional
-        First prefix for alert name: {id1}_{id2}.avro
-    id2: str, optional
-        Second prefix for alert name: {id1}_{id2}.avro
 
     Examples
     --------
@@ -270,7 +270,10 @@ def write_alert(
       ...
     OSError: ./ZTF19acihgng_1060135832015015002.avro already exists!
     """
-    alert_filename = os.path.join(path, "{}_{}.avro".format(alert[id1], alert[id2]))
+    if id2 is not None:
+        alert_filename = os.path.join(path, "{}_{}.avro".format(alert[id1], alert[id2]))
+    else:
+        alert_filename = os.path.join(path, "{}.avro".format(alert[id1]))
 
     if isinstance(schema, str):
         schema = _get_alert_schema(schema)

fink_client/configuration.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python
-# Copyright 2019-2020 AstroLab Software
+# Copyright 2019-2026 AstroLab Software
 # Author: Julien Peloton
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -16,13 +16,25 @@
 import yaml
 import os
 
+from fink_client.logger import get_fink_logger
 from fink_client.tester import regular_unit_tests
 
+_LOG = get_fink_logger()
 _ROOTDIR = os.path.join(os.environ["HOME"], ".finkclient")
-_CREDNAME = "credentials.yml"
+_CREDNAME = "{}_credentials.yml"
 
 
-def write_credentials(dict_file: dict, verbose: bool = False, tmp: bool = False):
+def check_survey_exists(survey):
+    """Check survey exists"""
+    if survey not in ["ztf", "lsst"]:
+        raise KeyError(
+            "-survey must be one of ['ztf', 'lsst']. {} is not recognized.".format(
+                survey
+            )
+        )
+
+
+def write_credentials(dict_file: dict, log_level: str = "WARN", tmp: bool = False):
     """Store user credentials on the computer.
 
     To get your credentials, contact Fink admins or fill the registration form:
@@ -32,56 +44,63 @@ def write_credentials(dict_file: dict, verbose: bool = False, tmp: bool = False)
     ----------
     dict_file: dict
         Dictionnary containing user credentials.
-    verbose: bool, optional
-        If True, print the credentials location. Default is False.
+    log_level: str, optional
+        Level of verbosity. Default is WARN.
     tmp: bool, optional
         If True, store the credentials under /tmp. Default is False.
 
     Examples
     --------
     >>> conf = {
+    ...   'survey': 'lsst',
     ...   'username': 'test',
     ...   'password': None,
     ...   'mytopics': ['rrlyr'],
     ...   'servers': 'localhost:9093',
     ...   'group_id': 'test_group',
     ...   'maxtimeout': 10
     ... }
-    >>> write_credentials(conf, verbose=False, tmp=True)
+    >>> write_credentials(conf, log_level="WARN", tmp=True)
     """
+    _LOG.setLevel(log_level)
     if tmp:
         ROOTDIR = "/tmp"
     else:
         ROOTDIR = _ROOTDIR
 
     # check there are no missing information
     mandatory_keys = [
+        "survey",
         "username",
-        "password",
         "group_id",
-        "mytopics",
         "servers",
-        "maxtimeout",
     ]
     for k in mandatory_keys:
         assert k in dict_file.keys(), "You need to specify {}".format(k)
 
+    check_survey_exists(dict_file["survey"])
+
     # Create the folder if it does not exist
     os.makedirs(ROOTDIR, exist_ok=True)
 
     # Store data into yml file
-    with open(os.path.join(ROOTDIR, _CREDNAME), "w") as f:
+    with open(os.path.join(ROOTDIR, _CREDNAME.format(dict_file["survey"])), "w") as f:
         yaml.dump(dict_file, f)
 
-    if verbose:
-        print("Credentials stored at {}/{}".format(ROOTDIR, _CREDNAME))
+    _LOG.info(
+        "Credentials stored at {}/{}".format(
+            ROOTDIR, _CREDNAME.format(dict_file["survey"])
+        )
+    )
 
 
-def load_credentials(tmp: bool = False) -> dict:
+def load_credentials(survey: str, tmp: bool = False) -> dict:
     """Load fink-client credentials.
 
     Parameters
     ----------
+    survey: str
+        Survey name, among ztf or lsst
     tmp: bool, optional
         If True, load the credentials from /tmp. Default is False.
 
@@ -93,45 +112,47 @@ def load_credentials(tmp: bool = False) -> dict:
     Examples
     --------
     >>> conf_in = {
+    ...   'survey': 'lsst',
     ...   'username': 'test',
     ...   'password': None,
     ...   'mytopics': ['rrlyr'],
     ...   'servers': 'localhost:9093',
     ...   'group_id': 'test_group',
     ...   'maxtimeout': 10
     ... }
-    >>> write_credentials(conf_in, verbose=False, tmp=True)
-    >>> conf_out = load_credentials(tmp=True)
+    >>> write_credentials(conf_in, tmp=True)
+    >>> conf_out = load_credentials(survey=conf_in["survey"], tmp=True)
 
     If, however the credentials do not exist yet
-    >>> os.remove('/tmp/credentials.yml')
-    >>> conf = load_credentials(tmp=True) # doctest: +NORMALIZE_WHITESPACE, +ELLIPSIS
+    >>> os.remove('/tmp/lsst_credentials.yml')
+    >>> conf = load_credentials(survey="lsst", tmp=True) # doctest: +NORMALIZE_WHITESPACE, +ELLIPSIS
     Traceback (most recent call last):
      ...
-    OSError: No credentials found, did you register?
+    OSError: No credentials found for survey lsst, did you register?
     To get your credentials, and use fink-client you need to:
       1. subscribe to one or more Fink streams at
         https://forms.gle/2td4jysT4e9pkf889
       2. run `fink_client_register` to register
-    See https://fink-broker.readthedocs.io/en/latest/services/data_transfer/
+    See https://doc.lsst.fink-broker.org/en/latest/services/data_transfer/
 
     """
+    check_survey_exists(survey)
     if tmp:
         ROOTDIR = "/tmp"
     else:
         ROOTDIR = _ROOTDIR
 
-    path = os.path.join(ROOTDIR, _CREDNAME)
+    path = os.path.join(ROOTDIR, _CREDNAME.format(survey))
 
     if not os.path.exists(path):
         msg = """
-        No credentials found, did you register?
+        No credentials found for survey {}, did you register?
         To get your credentials, and use fink-client you need to:
           1. subscribe to one or more Fink streams at
             https://forms.gle/2td4jysT4e9pkf889
           2. run `fink_client_register` to register
-        See https://fink-broker.readthedocs.io/en/latest/services/data_transfer/
-        """
+        See https://doc.{}.fink-broker.org/en/latest/services/data_transfer/
+        """.format(survey, survey)
         raise IOError(msg)
 
     with open(path) as f: