Add ZTF crossmatching (#987)

mirar/catalog/kowalski/__init__.py

@@ -6,3 +6,4 @@
 from mirar.catalog.kowalski.gaia import Gaia, GaiaBright
 from mirar.catalog.kowalski.ps1 import PS1, PS1STRM, PS1SGSc
 from mirar.catalog.kowalski.tmass import TMASS
+from mirar.catalog.kowalski.ztf import ZTF

mirar/catalog/kowalski/base_kowalski_catalog.py

@@ -79,6 +79,31 @@ def get_kowalski() -> Kowalski:
     return kowalski_instance
 
 
+def flatten_kowalski_data(matches: list[dict]) -> list[dict]:
+    """
+    Flatten a Kowalski data dict
+
+    :param matches: List of matches
+    :return: Flattened list of depth-1 dictionaries
+    """
+
+    new = []
+
+    if len(matches) > 0:
+        for match in matches:
+            new_dict = {}
+            if isinstance(match, dict):
+                for key, val in match.items():
+                    if isinstance(val, dict):
+                        for subkey, subval in val.items():
+                            new_dict[f"{key}.{subkey}"] = subval
+                    else:
+                        new_dict[key] = val
+            new.append(new_dict)
+
+    return new
+
+
 class BaseKowalskiXMatch(BaseXMatchCatalog, ABC):
     """
     Base class for a catalog using Kowalski
@@ -133,7 +158,14 @@ def near_query_kowalski(self, coords: dict) -> dict:
         response = self.kowalski.query(query=query)
         data = response.get("default").get("data")
 
-        return data[self.catalog_name]
+        res = {}
+
+        # Flatten if Kowalski data is nested
+        for key, matches in data[self.catalog_name].items():
+            new = flatten_kowalski_data(matches)
+            res[key] = new
+
+        return res
 
     def query(self, coords) -> dict:
         """

mirar/catalog/kowalski/ztf.py

@@ -0,0 +1,37 @@
+"""
+Module for querying ZTF using Kowalski
+"""
+
+from mirar.catalog.kowalski.base_kowalski_catalog import BaseKowalskiXMatch
+
+
+class ZTF(BaseKowalskiXMatch):
+    """
+    ZTF Kowalski catalog
+    """
+
+    catalog_name = "ZTF_alerts"
+    abbreviation = "ztf"
+    projection = {
+        "_id": 1,
+        "objectId": 1,
+        "candidate.ra": 1,
+        "candidate.dec": 1,
+    }
+
+    column_names = {
+        "_id": "ztfid",
+        "objectId": "ztfname",
+        "candidate.ra": f"{abbreviation}ra",
+        "candidate.dec": f"{abbreviation}dec",
+    }
+
+    column_dtypes = {
+        "ztfid": str,
+        "ztfname": str,
+        "ztfra": float,
+        "ztfdec": float,
+    }
+
+    ra_column_name = f"{abbreviation}ra"
+    dec_column_name = f"{abbreviation}dec"

mirar/pipelines/winter/blocks.py

@@ -7,7 +7,7 @@
 
 from winterrb.model import WINTERNet
 
-from mirar.catalog.kowalski import PS1, PS1STRM, TMASS, Gaia, GaiaBright, PS1SGSc
+from mirar.catalog.kowalski import PS1, PS1STRM, TMASS, ZTF, Gaia, GaiaBright, PS1SGSc
 from mirar.downloader.get_test_data import get_test_data_dir
 from mirar.paths import (
     BASE_NAME_KEY,
@@ -715,6 +715,7 @@
     XMatch(catalog=PS1STRM(num_sources=3, search_radius_arcmin=0.5)),
     XMatch(catalog=Gaia(num_sources=1, search_radius_arcmin=1.5)),
     XMatch(catalog=GaiaBright(num_sources=1, search_radius_arcmin=1.5)),
+    XMatch(catalog=ZTF(num_sources=1, search_radius_arcmin=2.0 / 60.0)),
     CustomSourceTableModifier(
         modifier_function=winter_candidate_avro_fields_calculator
     ),

mirar/pipelines/winter/config/files/avro_schema/winter.alert.avsc

@@ -4,7 +4,7 @@
     "name": "alert",
     "doc": "avro alert schema for WINTER",
     "fields": [
-        {"name": "schemavsn", "type": "string", "doc": "schema version used", "default": "0.3"},
+        {"name": "schemavsn", "type": "string", "doc": "schema version used", "default": "0.4"},
         {"name": "publisher", "type": "string", "doc": "origin of alert packet", "default": "WINTER-mirar"},
         {"name": "objectid", "type": "string", "doc": "name of parent source, e.g 'WNTR24aaaab'"},
         {"name": "candid", "type": "long"},

mirar/pipelines/winter/config/files/avro_schema/winter.alert.candidate.avsc

@@ -127,6 +127,8 @@
 		{"name": "ruwegaia", "type": ["float", "null"], "doc": "Random Unit Weighted Error of closest source from Gaia EDR3 catalog; if exists within 90 arcsec [mas]"},
 		{"name": "distgaiabright", "type": ["float", "null"], "doc": "Distance to closest source from Gaia DR3 catalog brighter than magnitude 14; if exists within 90 arcsec [arcsec]"},
 		{"name": "plxgaiabright", "type": ["float", "null"], "doc": "Absolute (parallax/sigma) of closest source from Gaia EDR3 catalog brighter than magnitude 14; if exists within 90 arcsec [mag]"},
-		{"name": "ruwegaiabright", "type": ["float", "null"], "doc": "Random Unit Weighted Error of closest source from Gaia EDR3 catalog brighter than magnitude 14; if exists within 90 arcsec [mas]"}
+		{"name": "ruwegaiabright", "type": ["float", "null"], "doc": "Random Unit Weighted Error of closest source from Gaia EDR3 catalog brighter than magnitude 14; if exists within 90 arcsec [mas]"},
+		{"name": "ztfname", "type": ["string", "null"], "doc": "ZTF name of candidate if exists within 2 arcsec, e.g. ZTF20abzqkxg"},
+		{"name": "distztf", "type": ["float", "null"], "doc": "Distance to closest source from ZTF alert catalog; if exists within 2 arcsec [arcsec]"}
 	]
 }

mirar/pipelines/winter/generator/candidates.py

@@ -228,6 +228,9 @@ def winter_candidate_avro_fields_calculator(source_table: SourceBatch) -> Source
         src_df["plxgaiabright"] = src_df["gaiabright_parallax_over_error1"]
         src_df["ruwegaiabright"] = src_df["gaiabright_ruwe1"]
 
+        src_df["distztf"] = src_df["distztfnr1"]
+        src_df["ztfname"] = src_df["ztfname1"]
+
         source.set_data(src_df)
         new_batch.append(source)
 

mirar/pipelines/winter/models/_candidates.py

@@ -209,6 +209,10 @@ class CandidatesTable(WinterBase):  # pylint: disable=too-few-public-methods
     plxgaiabright = Column(Float, nullable=True)
     ruwegaiabright = Column(Float, nullable=True)
 
+    # ZTF properties
+    ztfname = Column(VARCHAR(12), nullable=True)
+    distztf = Column(Float, nullable=True)
+
 
 class Candidate(BaseDB):
     """
@@ -343,6 +347,9 @@ class Candidate(BaseDB):
     plxgaiabright: float | None = Field(default=None)
     ruwegaiabright: float | None = Field(ge=0, default=None)
 
+    ztfname: str | None = Field(default=None, max_length=12)
+    distztf: float | None = Field(ge=0, default=None)
+
     def insert_entry(
         self, duplicate_protocol, returning_key_names=None
     ) -> pd.DataFrame:

mirar/pipelines/winter/winter_pipeline.py

@@ -26,6 +26,7 @@
     load_final_stack,
     load_raw,
     load_skyportal,
+    load_sources,
     load_test,
     mask_and_split,
     mosaic,
@@ -92,6 +93,7 @@ class WINTERPipeline(Pipeline):
         + detect_candidates
         + process_candidates
         + avro_broadcast,
+        "recandidates": load_sources + process_candidates + avro_broadcast,
         "default": reduce
         + imsub
         + detect_candidates

mirar/processors/xmatch.py

@@ -6,6 +6,7 @@
 
 import astropy.units as u
 import numpy as np
+import pandas as pd
 from astropy.coordinates import SkyCoord
 
 from mirar.catalog.base.base_xmatch_catalog import BaseXMatchCatalog
@@ -89,7 +90,7 @@ def _apply_to_sources(
                 result_dec_colname = self.catalog.dec_column_name + f"{num + 1}"
                 dist_colname = f"dist{self.catalog.abbreviation}nr{num + 1}"
                 candidate_table[dist_colname] = np.array(np.nan, dtype=float)
-                crd_nanmask = np.invert(np.isnan(candidate_table[result_ra_colname]))
+                crd_nanmask = pd.notnull(candidate_table[result_ra_colname])
                 result_crds = SkyCoord(
                     ra=candidate_table[result_ra_colname][crd_nanmask],
                     dec=candidate_table[result_dec_colname][crd_nanmask],