13725459265

Committed 07 Mar 2025 04:58PM UTC coverage: 71.901% (-1.7%) from 73.615%

Build # 13725459265

Build Type

push

github

Committed by

web-flow

Commit Message

Add Kowalski Backend (#490)

Run Details

471 of 640 new or added lines in 26 files covered. (73.59%)

15 existing lines in 4 files now uncovered.

1978 of 2751 relevant lines covered (71.9%)

0.72 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

75.0

/nuztf/cat_match.py

#!/usr/bin/env python
# coding: utf-8

import json
import logging
import warnings

from astropy import units as u
from astropy.coordinates import SkyCoord
from astropy.utils.exceptions import AstropyWarning
from astroquery.exceptions import RemoteServiceError
from astroquery.ipac.irsa import Irsa
from astroquery.ipac.ned import Ned

from nuztf.ampel import ampel_api_catalog, ampel_api_name
from nuztf.paths import CROSSMATCH_CACHE


def query_ned_for_z(
    ra_deg: float, dec_deg: float, searchradius_arcsec: float = 20, logger=None
):
    """Function to obtain redshifts from NED (via the AMPEL API)"""

    z = None
    dist_arcsec = None

    query = ampel_api_catalog(
        catalog="NEDz_extcats",
        catalog_type="extcats",
        ra_deg=ra_deg,
        dec_deg=dec_deg,
        search_radius_arcsec=searchradius_arcsec,
        search_type="nearest",
        logger=logger,
    )

    if query:
        z = query["body"]["z"]
        dist_arcsec = query["dist_arcsec"]

    return z, dist_arcsec


def query_ned_astroquery(
    ra_deg: float, dec_deg: float, searchradius_arcsec: float = 0.5
):
    """
    Function to obtain NED crossmatches via astroquery
    """
    c = SkyCoord(ra_deg, dec_deg, unit=u.deg, frame="icrs")

    r = searchradius_arcsec * u.arcsecond

    try:
        return Ned.query_region(c, radius=r)
    except RemoteServiceError:
        return None


def query_wise_astroquery(
    ra_deg: float, dec_deg: float, searchradius_arcsec: float = 3.0
):
    """
    Function to obtain WISE crossmatches via astroquery

    :param ra_deg: Right ascension (deg)
    :param dec_deg: Declination (deg)
    :param searchradius_arcsec: Search radius (arcsec)
    :return: result of query
    """
    c = SkyCoord(ra_deg, dec_deg, unit=u.deg, frame="icrs")

    r = searchradius_arcsec * u.arcsecond

    with warnings.catch_warnings():
        warnings.simplefilter("ignore", AstropyWarning)
        allwise = Irsa.query_region(c, catalog="allwise_p3as_psd", radius=r)
    return allwise


def ampel_api_tns(
    ra_deg: float, dec_deg: float, searchradius_arcsec: float = 3, logger=None
):
    """Function to query TNS via the AMPEL API"""

    full_name = None
    discovery_date = None
    source_group = None

    res = ampel_api_catalog(
        catalog="TNS",
        catalog_type="extcats",
        ra_deg=ra_deg,
        dec_deg=dec_deg,
        search_radius_arcsec=searchradius_arcsec,
        search_type="nearest",
        logger=logger,
    )

    if res:
        response_body = res["body"]
        name = response_body["objname"]
        prefix = response_body["name_prefix"]
        full_name = prefix + name
        discovery_date = response_body["discoverydate"]
        if "source_group" in response_body.keys():
            source_group = response_body["source_group"]["group_name"]

    return full_name, discovery_date, source_group


def get_cross_match_info(raw: dict, logger=None) -> str:
    """
    Function to get cross-match info for a given alert

    :param raw: Raw alert data
    :param logger: Logger
    :return: String with cross-match info
    """
    cache_file = CROSSMATCH_CACHE.joinpath(f"{raw['objectId']}.json")

    if cache_file.exists():
        with open(cache_file) as f:
            res = json.load(f)
            label = res["data"]
            return label

    alert = raw["candidate"]

    label = ""

    if logger is None:
        logger = logging.getLogger()

    # Check if known variable star (https://arxiv.org/pdf/1405.4290.pdf)

    res = ampel_api_catalog(
        catalog="CRTS_DR1",
        catalog_type="extcats",
        ra_deg=alert["ra"],
        dec_deg=alert["dec"],
        search_radius_arcsec=5.0,
        logger=logger,
    )
    if res is not None:
        if logger:
            logger.info(res)
        label = (
            f"[CRTS variable star: "
            f"{res[0]['body']['name']} ({res[0]['dist_arcsec']:.2f} arsec)]"
        )

    # Check if known QSO/AGN

    res = ampel_api_catalog(
        catalog="milliquas",
        catalog_type="extcats",
        ra_deg=alert["ra"],
        dec_deg=alert["dec"],
        search_radius_arcsec=1.5,
        logger=logger,
    )
    if res is not None:
        if len(res) == 1:
            if "q" in res[0]["body"]["broad_type"]:
                label = (
                    f"[MILLIQUAS: {res[0]['body']['name']} - "
                    f"Likely QSO (prob = {res[0]['body']['qso_prob']}%) "
                    f"({res[0]['dist_arcsec']:.2f} arsec)]"
                )
            else:
                label = (
                    f"[MILLIQUAS: {res[0]['body']['name']} - "
                    f"'{res[0]['body']['broad_type']}'-type source "
                    f"({res[0]['dist_arcsec']:.2f} arsec)]"
                )
        else:
            label = "[MULTIPLE MILLIQUAS MATCHES]"

    # Check if measured parallax in Gaia (i.e galactic)

    if label == "":
        res = ampel_api_catalog(
            catalog="GAIADR2",
            catalog_type="catsHTM",
            ra_deg=alert["ra"],
            dec_deg=alert["dec"],
            search_radius_arcsec=5.0,
            logger=logger,
        )
        if res is not None:

            if res[0]["body"]["Plx"] is not None:
                plx_sig = res[0]["body"]["Plx"] / res[0]["body"]["ErrPlx"]
                if plx_sig > 3.0:
                    label = (
                        f"[GAIADR2: {plx_sig:.1f}-sigma parallax "
                        f"({res[0]['dist_arcsec']:.2f} arsec)]"
                    )

            elif res[0]["body"]["PMRA"] is not None:
                pmra_sig = abs(res[0]["body"]["PMRA"] / res[0]["body"]["ErrPMRA"])
                if pmra_sig > 3.0:
                    label = (
                        f"[GAIADR2: {pmra_sig:.1f}-sigma proper motion (RA) "
                        f"({res[0]['dist_arcsec']:.2f} arsec)]"
                    )
                elif res[0]["body"]["PMDec"] is not None:
                    pmdec_sig = abs(
                        res[0]["body"]["PMDec"] / res[0]["body"]["ErrPMDec"]
                    )
                    if pmdec_sig > 3.0:
                        label = (
                            f"[GAIADR2: {pmdec_sig:.1f}-sigma proper motion (Dec) "
                            f"({res[0]['dist_arcsec']:.2f} arsec)]"
                        )

    # Check if classified as probable star in SDSS

    if label == "":
        res = ampel_api_catalog(
            catalog="SDSSDR10",
            catalog_type="catsHTM",
            ra_deg=alert["ra"],
            dec_deg=alert["dec"],
            search_radius_arcsec=1.5,
            logger=logger,
        )
        if res is not None:
            if len(res) == 1:
                if float(res[0]["body"]["type"]) == 6.0:
                    label = (
                        f"[SDSS Morphology: 'Star'-type source "
                        f"({res[0]['dist_arcsec']:.2f} arsec)]"
                    )
            else:
                label = "[MULTIPLE SDSS MATCHES]"

    # WISE colour cuts (https://iopscience.iop.org/article/10.3847/1538-4365/)

    if label == "":
        res = query_wise_astroquery(
            ra_deg=alert["ra"],
            dec_deg=alert["dec"],
            searchradius_arcsec=3.0,
        )
        if res is not None:
            if len(res) > 0:
                w1mw2 = res["w1mpro"][0] - res["w2mpro"][0]

                if w1mw2 > 0.8:
                    label = (
                        f"[Probable WISE-selected quasar:W1-W2={w1mw2:.2f}>0.8  "
                        f"({res[0]['dist']:.2f} arsec)]"
                    )
                elif w1mw2 > 0.5:
                    label = (
                        f"[Possible WISE-selected quasar:W1-W2={w1mw2:.2f}>0.5  "
                        f"({res[0]['dist']:.2f} arsec)]"
                    )
                else:
                    label = (
                        f"WISE DETECTION: W1-W2={w1mw2:.2f} "
                        f"({res[0]['dist']:.2f} arsec)"
                    )

                if len(res) > 1:
                    label += "[MULTIPLE WISE MATCHES]"

    # Just check NED

    if label == "":
        res = query_ned_astroquery(
            ra_deg=alert["ra"],
            dec_deg=alert["dec"],
            searchradius_arcsec=3.0,
        )

        if res is not None:
            if len(res) == 1:
                label = (
                    f"{res['Object Name'][0]} ['{res['Type'][0]}'-type source "
                    f"({res['Separation'][0]:.2f} arsec)]"
                )
            elif len(res) > 1:
                label += "[MULTIPLE NED MATCHES]"
                logger.debug(f"{res}")

    # Extra check to TNS, append to other info

    full_name, _, _ = ampel_api_tns(
        ra_deg=alert["ra"],
        dec_deg=alert["dec"],
    )

    if full_name is not None:
        label += f" [TNS NAME={full_name}]"

    with open(cache_file, "w") as f:
        json.dump({"data": label}, f)

    return label


def check_cross_match_info_by_name(name: str, logger=None):
    """
    Utility function to check cross-match info for a given name

    :param name: ZTF name
    :param logger:
    :return:
    """
    return get_cross_match_info(
        raw=ampel_api_name(name, with_history=False, logger=logger)[0], logger=logger
    )

1	#!/usr/bin/env python
2	# coding: utf-8
3
4	import json	1✔
5	import logging	1✔
6	import warnings	1✔
7
8	from astropy import units as u	1✔
9	from astropy.coordinates import SkyCoord	1✔
10	from astropy.utils.exceptions import AstropyWarning	1✔
11	from astroquery.exceptions import RemoteServiceError	1✔
12	from astroquery.ipac.irsa import Irsa	1✔
13	from astroquery.ipac.ned import Ned	1✔
14
15	from nuztf.ampel import ampel_api_catalog, ampel_api_name	1✔
16	from nuztf.paths import CROSSMATCH_CACHE	1✔
17
18
19	def query_ned_for_z(	1✔
20	ra_deg: float, dec_deg: float, searchradius_arcsec: float = 20, logger=None
21	):
22	"""Function to obtain redshifts from NED (via the AMPEL API)"""
23
24	z = None	1✔
25	dist_arcsec = None	1✔
26
27	query = ampel_api_catalog(	1✔
28	catalog="NEDz_extcats",
29	catalog_type="extcats",
30	ra_deg=ra_deg,
31	dec_deg=dec_deg,
32	search_radius_arcsec=searchradius_arcsec,
33	search_type="nearest",
34	logger=logger,
35	)
36
37	if query:	1✔
38	z = query["body"]["z"]	1✔
39	dist_arcsec = query["dist_arcsec"]	1✔
40
41	return z, dist_arcsec	1✔
42
43
44	def query_ned_astroquery(	1✔
45	ra_deg: float, dec_deg: float, searchradius_arcsec: float = 0.5
46	):
47	"""
48	Function to obtain NED crossmatches via astroquery
49	"""
50	c = SkyCoord(ra_deg, dec_deg, unit=u.deg, frame="icrs")	1✔
51
52	r = searchradius_arcsec * u.arcsecond	1✔
53
54	try:	1✔
55	return Ned.query_region(c, radius=r)	1✔
56	except RemoteServiceError:	×
57	return None	×
58
59
60	def query_wise_astroquery(	1✔
61	ra_deg: float, dec_deg: float, searchradius_arcsec: float = 3.0
62	):
63	"""
64	Function to obtain WISE crossmatches via astroquery
65
66	:param ra_deg: Right ascension (deg)
67	:param dec_deg: Declination (deg)
68	:param searchradius_arcsec: Search radius (arcsec)
69	:return: result of query
70	"""
71	c = SkyCoord(ra_deg, dec_deg, unit=u.deg, frame="icrs")	1✔
72
73	r = searchradius_arcsec * u.arcsecond	1✔
74
75	with warnings.catch_warnings():	1✔
76	warnings.simplefilter("ignore", AstropyWarning)	1✔
77	allwise = Irsa.query_region(c, catalog="allwise_p3as_psd", radius=r)	1✔
78	return allwise	1✔
79
80
81	def ampel_api_tns(	1✔
82	ra_deg: float, dec_deg: float, searchradius_arcsec: float = 3, logger=None
83	):
84	"""Function to query TNS via the AMPEL API"""
85
86	full_name = None	1✔
87	discovery_date = None	1✔
88	source_group = None	1✔
89
90	res = ampel_api_catalog(	1✔
91	catalog="TNS",
92	catalog_type="extcats",
93	ra_deg=ra_deg,
94	dec_deg=dec_deg,
95	search_radius_arcsec=searchradius_arcsec,
96	search_type="nearest",
97	logger=logger,
98	)
99
100	if res:	1✔
101	response_body = res["body"]	1✔
102	name = response_body["objname"]	1✔
103	prefix = response_body["name_prefix"]	1✔
104	full_name = prefix + name	1✔
105	discovery_date = response_body["discoverydate"]	1✔
106	if "source_group" in response_body.keys():	1✔
107	source_group = response_body["source_group"]["group_name"]	×
108
109	return full_name, discovery_date, source_group	1✔
110
111
112	def get_cross_match_info(raw: dict, logger=None) -> str:	1✔
113	"""
114	Function to get cross-match info for a given alert
115
116	:param raw: Raw alert data
117	:param logger: Logger
118	:return: String with cross-match info
119	"""
120	cache_file = CROSSMATCH_CACHE.joinpath(f"{raw['objectId']}.json")	1✔
121
122	if cache_file.exists():	1✔
123	with open(cache_file) as f:	1✔
124	res = json.load(f)	1✔
125	label = res["data"]	1✔
126	return label	1✔
127
128	alert = raw["candidate"]	1✔
129
130	label = ""	1✔
131
132	if logger is None:	1✔
UNCOV 133	logger = logging.getLogger()	×
134
135	# Check if known variable star (https://arxiv.org/pdf/1405.4290.pdf)
136
137	res = ampel_api_catalog(	1✔
138	catalog="CRTS_DR1",
139	catalog_type="extcats",
140	ra_deg=alert["ra"],
141	dec_deg=alert["dec"],
142	search_radius_arcsec=5.0,
143	logger=logger,
144	)
145	if res is not None:	1✔
146	if logger:	×
147	logger.info(res)	×
148	label = (	×
149	f"[CRTS variable star: "
150	f"{res[0]['body']['name']} ({res[0]['dist_arcsec']:.2f} arsec)]"
151	)
152
153	# Check if known QSO/AGN
154
155	res = ampel_api_catalog(	1✔
156	catalog="milliquas",
157	catalog_type="extcats",
158	ra_deg=alert["ra"],
159	dec_deg=alert["dec"],
160	search_radius_arcsec=1.5,
161	logger=logger,
162	)
163	if res is not None:	1✔
164	if len(res) == 1:	1✔
165	if "q" in res[0]["body"]["broad_type"]:	1✔
166	label = (	×
167	f"[MILLIQUAS: {res[0]['body']['name']} - "
168	f"Likely QSO (prob = {res[0]['body']['qso_prob']}%) "
169	f"({res[0]['dist_arcsec']:.2f} arsec)]"
170	)
171	else:
172	label = (	1✔
173	f"[MILLIQUAS: {res[0]['body']['name']} - "
174	f"'{res[0]['body']['broad_type']}'-type source "
175	f"({res[0]['dist_arcsec']:.2f} arsec)]"
176	)
177	else:
178	label = "[MULTIPLE MILLIQUAS MATCHES]"	×
179
180	# Check if measured parallax in Gaia (i.e galactic)
181
182	if label == "":	1✔
183	res = ampel_api_catalog(	1✔
184	catalog="GAIADR2",
185	catalog_type="catsHTM",
186	ra_deg=alert["ra"],
187	dec_deg=alert["dec"],
188	search_radius_arcsec=5.0,
189	logger=logger,
190	)
191	if res is not None:	1✔
192
193	if res[0]["body"]["Plx"] is not None:	×
194	plx_sig = res[0]["body"]["Plx"] / res[0]["body"]["ErrPlx"]	×
195	if plx_sig > 3.0:	×
196	label = (	×
197	f"[GAIADR2: {plx_sig:.1f}-sigma parallax "
198	f"({res[0]['dist_arcsec']:.2f} arsec)]"
199	)
200
201	elif res[0]["body"]["PMRA"] is not None:	×
202	pmra_sig = abs(res[0]["body"]["PMRA"] / res[0]["body"]["ErrPMRA"])	×
203	if pmra_sig > 3.0:	×
204	label = (	×
205	f"[GAIADR2: {pmra_sig:.1f}-sigma proper motion (RA) "
206	f"({res[0]['dist_arcsec']:.2f} arsec)]"
207	)
208	elif res[0]["body"]["PMDec"] is not None:	×
209	pmdec_sig = abs(	×
210	res[0]["body"]["PMDec"] / res[0]["body"]["ErrPMDec"]
211	)
212	if pmdec_sig > 3.0:	×
213	label = (	×
214	f"[GAIADR2: {pmdec_sig:.1f}-sigma proper motion (Dec) "
215	f"({res[0]['dist_arcsec']:.2f} arsec)]"
216	)
217
218	# Check if classified as probable star in SDSS
219
220	if label == "":	1✔
221	res = ampel_api_catalog(	1✔
222	catalog="SDSSDR10",
223	catalog_type="catsHTM",
224	ra_deg=alert["ra"],
225	dec_deg=alert["dec"],
226	search_radius_arcsec=1.5,
227	logger=logger,
228	)
229	if res is not None:	1✔
230	if len(res) == 1:	1✔
231	if float(res[0]["body"]["type"]) == 6.0:	1✔
232	label = (	×
233	f"[SDSS Morphology: 'Star'-type source "
234	f"({res[0]['dist_arcsec']:.2f} arsec)]"
235	)
236	else:
237	label = "[MULTIPLE SDSS MATCHES]"	×
238
239	# WISE colour cuts (https://iopscience.iop.org/article/10.3847/1538-4365/)
240
241	if label == "":	1✔
242	res = query_wise_astroquery(	1✔
243	ra_deg=alert["ra"],
244	dec_deg=alert["dec"],
245	searchradius_arcsec=3.0,
246	)
247	if res is not None:	1✔
248	if len(res) > 0:	1✔
249	w1mw2 = res["w1mpro"][0] - res["w2mpro"][0]	1✔
250
251	if w1mw2 > 0.8:	1✔
252	label = (	×
253	f"[Probable WISE-selected quasar:W1-W2={w1mw2:.2f}>0.8 "
254	f"({res[0]['dist']:.2f} arsec)]"
255	)
256	elif w1mw2 > 0.5:	1✔
257	label = (	×
258	f"[Possible WISE-selected quasar:W1-W2={w1mw2:.2f}>0.5 "
259	f"({res[0]['dist']:.2f} arsec)]"
260	)
261	else:
262	label = (	1✔
263	f"WISE DETECTION: W1-W2={w1mw2:.2f} "
264	f"({res[0]['dist']:.2f} arsec)"
265	)
266
267	if len(res) > 1:	1✔
268	label += "[MULTIPLE WISE MATCHES]"	×
269
270	# Just check NED
271
272	if label == "":	1✔
273	res = query_ned_astroquery(	1✔
274	ra_deg=alert["ra"],
275	dec_deg=alert["dec"],
276	searchradius_arcsec=3.0,
277	)
278
279	if res is not None:	1✔
280	if len(res) == 1:	1✔
281	label = (	×
282	f"{res['Object Name'][0]} ['{res['Type'][0]}'-type source "
283	f"({res['Separation'][0]:.2f} arsec)]"
284	)
285	elif len(res) > 1:	1✔
286	label += "[MULTIPLE NED MATCHES]"	×
287	logger.debug(f"{res}")	×
288
289	# Extra check to TNS, append to other info
290
291	full_name, _, _ = ampel_api_tns(	1✔
292	ra_deg=alert["ra"],
293	dec_deg=alert["dec"],
294	)
295
296	if full_name is not None:	1✔
297	label += f" [TNS NAME={full_name}]"	1✔
298
299	with open(cache_file, "w") as f:	1✔
300	json.dump({"data": label}, f)	1✔
301
302	return label	1✔
303
304
305	def check_cross_match_info_by_name(name: str, logger=None):	1✔
306	"""
307	Utility function to check cross-match info for a given name
308
309	:param name: ZTF name
310	:param logger:
311	:return:
312	"""
313	return get_cross_match_info(	×
314	raw=ampel_api_name(name, with_history=False, logger=logger)[0], logger=logger
315	)

desy-multimessenger / nuztf / 13725459265

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous