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

52.73

/nuztf/ampel/ampel_lightcurve.py

import logging
from json import JSONDecodeError

import backoff
import requests
from astropy.io import fits  # type: ignore
from astropy.time import Time  # type: ignore

from nuztf.ampel.urls import API_ZTF_ARCHIVE_URL
from nuztf.ampel.utils import get_ampel_token, merge_alerts


@backoff.on_exception(
    backoff.expo,
    requests.exceptions.RequestException,
    max_time=600,
)
def ampel_api_name(
    ztf_name: str,
    with_history: bool = True,
    with_cutouts: bool = False,
    limit: int = 999999,
    logger=None,
) -> list:
    """Function to query ampel via name"""
    if logger is None:
        logger = logging.getLogger(__name__)

    if with_history:
        hist = "true"
    else:
        hist = "false"

    if with_cutouts:
        cutouts = "true"
    else:
        cutouts = "false"

    queryurl_ztf_name = (
        API_ZTF_ARCHIVE_URL
        + f"/object/{ztf_name}/alerts?with_history={hist}&with_cutouts={cutouts}&limit={limit}"
    )

    logger.debug(queryurl_ztf_name)

    headers = {"Authorization": f"Bearer {get_ampel_token()}"}

    response = requests.get(
        queryurl_ztf_name,
        headers=headers,
    )

    if response.status_code == 503:
        raise requests.exceptions.RequestException

    try:
        query_res = [i for i in response.json()]
        query_res = merge_alerts(query_res)

    except JSONDecodeError:
        if response.headers:
            logger.debug(response.headers)
        raise requests.exceptions.RequestException

    return query_res


@backoff.on_exception(
    backoff.expo,
    requests.exceptions.RequestException,
    max_time=600,
)
def ampel_api_lightcurve(
    ztf_name: str,
    t_min_jd=Time("2017-01-01T00:00:00.0", format="isot", scale="utc").jd,
    t_max_jd=Time.now().jd,
    program_id: int = None,
    logger=None,
) -> list:
    """
    Function to query ampel via name, returns a virtual alert
    constructed by AMPEL containing ALL photopoints and upper limits

    """

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

    if program_id is None:
        queryurl_lightcurve = (
            API_ZTF_ARCHIVE_URL + f"/object/{ztf_name}/photopoints?jd_start={t_min_jd}&"
            f"jd_end={t_max_jd}"
        )
    else:
        queryurl_lightcurve = (
            API_ZTF_ARCHIVE_URL + f"/object/{ztf_name}/photopoints?jd_start={t_min_jd}&"
            f"jd_end={t_max_jd}&programid={program_id}"
        )

    logger.debug(queryurl_lightcurve)

    headers = {"Authorization": f"Bearer {get_ampel_token()}"}

    response = requests.get(
        queryurl_lightcurve,
        headers=headers,
    )

    if response.status_code == 503:
        if response.headers:
            logger.debug(response.headers)
        raise requests.exceptions.RequestException

    try:
        query_res = [response.json()]

    except JSONDecodeError:
        if response.headers:
            logger.debug(response.headers)
        raise requests.exceptions.RequestException

    return query_res


@backoff.on_exception(
    backoff.expo,
    requests.exceptions.RequestException,
    max_time=600,
)
def ampel_api_alerts(
    ztf_name: str,
    t_min_jd=Time("2017-01-01T00:00:00.0", format="isot", scale="utc").jd,
    t_max_jd=Time.now().jd,
    program_id: int = None,
    logger=None,
) -> list:
    """
    Function to query ampel via name, returns a virtual alert
    constructed by AMPEL containing ALL photopoints and upper limits

    """

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

    if program_id is None:
        queryurl_lightcurve = (
            API_ZTF_ARCHIVE_URL + f"/object/{ztf_name}/alerts?jd_start={t_min_jd}&"
            f"jd_end={t_max_jd}&with_history=true"
        )
    else:
        queryurl_lightcurve = (
            API_ZTF_ARCHIVE_URL + f"/object/{ztf_name}/alerts?jd_start={t_min_jd}&"
            f"jd_end={t_max_jd}&programid={program_id}"
        )

    logger.debug(queryurl_lightcurve)

    headers = {"Authorization": f"Bearer {get_ampel_token()}"}

    response = requests.get(
        queryurl_lightcurve,
        headers=headers,
    )

    if response.status_code == 503:
        if response.headers:
            logger.debug(response.headers)
        raise requests.exceptions.RequestException

    try:
        query_res = [response.json()]

    except JSONDecodeError:
        if response.headers:
            logger.debug(response.headers)
        raise requests.exceptions.RequestException

    return query_res

1	import logging	1✔
2	from json import JSONDecodeError	1✔
3
4	import backoff	1✔
5	import requests	1✔
6	from astropy.io import fits # type: ignore	1✔
7	from astropy.time import Time # type: ignore	1✔
8
9	from nuztf.ampel.urls import API_ZTF_ARCHIVE_URL	1✔
10	from nuztf.ampel.utils import get_ampel_token, merge_alerts	1✔
11
12
13	@backoff.on_exception(	1✔
14	backoff.expo,
15	requests.exceptions.RequestException,
16	max_time=600,
17	)
18	def ampel_api_name(	1✔
19	ztf_name: str,
20	with_history: bool = True,
21	with_cutouts: bool = False,
22	limit: int = 999999,
23	logger=None,
24	) -> list:
25	"""Function to query ampel via name"""
26	if logger is None:	1✔
27	logger = logging.getLogger(__name__)	1✔
28
29	if with_history:	1✔
30	hist = "true"	1✔
31	else:
NEW 32	hist = "false"	×
33
34	if with_cutouts:	1✔
NEW 35	cutouts = "true"	×
36	else:
37	cutouts = "false"	1✔
38
39	queryurl_ztf_name = (	1✔
40	API_ZTF_ARCHIVE_URL
41	+ f"/object/{ztf_name}/alerts?with_history={hist}&with_cutouts={cutouts}&limit={limit}"
42	)
43
44	logger.debug(queryurl_ztf_name)	1✔
45
46	headers = {"Authorization": f"Bearer {get_ampel_token()}"}	1✔
47
48	response = requests.get(	1✔
49	queryurl_ztf_name,
50	headers=headers,
51	)
52
53	if response.status_code == 503:	1✔
54	raise requests.exceptions.RequestException
55
56	try:	1✔
57	query_res = [i for i in response.json()]	1✔
58	query_res = merge_alerts(query_res)	1✔
59
60	except JSONDecodeError:
61	if response.headers:
62	logger.debug(response.headers)
63	raise requests.exceptions.RequestException
64
65	return query_res	1✔
66
67
68	@backoff.on_exception(	1✔
69	backoff.expo,
70	requests.exceptions.RequestException,
71	max_time=600,
72	)
73	def ampel_api_lightcurve(	1✔
74	ztf_name: str,
75	t_min_jd=Time("2017-01-01T00:00:00.0", format="isot", scale="utc").jd,
76	t_max_jd=Time.now().jd,
77	program_id: int = None,
78	logger=None,
79	) -> list:
80	"""
81	Function to query ampel via name, returns a virtual alert
82	constructed by AMPEL containing ALL photopoints and upper limits
83
84	"""
85
NEW 86	if logger is None:	×
NEW 87	logger = logging.getLogger(__name__)	×
88
NEW 89	if program_id is None:	×
NEW 90	queryurl_lightcurve = (	×
91	API_ZTF_ARCHIVE_URL + f"/object/{ztf_name}/photopoints?jd_start={t_min_jd}&"
92	f"jd_end={t_max_jd}"
93	)
94	else:
NEW 95	queryurl_lightcurve = (	×
96	API_ZTF_ARCHIVE_URL + f"/object/{ztf_name}/photopoints?jd_start={t_min_jd}&"
97	f"jd_end={t_max_jd}&programid={program_id}"
98	)
99
NEW 100	logger.debug(queryurl_lightcurve)	×
101
NEW 102	headers = {"Authorization": f"Bearer {get_ampel_token()}"}	×
103
NEW 104	response = requests.get(	×
105	queryurl_lightcurve,
106	headers=headers,
107	)
108
NEW 109	if response.status_code == 503:	×
110	if response.headers:
111	logger.debug(response.headers)
112	raise requests.exceptions.RequestException
113
NEW 114	try:	×
NEW 115	query_res = [response.json()]	×
116
117	except JSONDecodeError:
118	if response.headers:
119	logger.debug(response.headers)
120	raise requests.exceptions.RequestException
121
NEW 122	return query_res	×
123
124
125	@backoff.on_exception(	1✔
126	backoff.expo,
127	requests.exceptions.RequestException,
128	max_time=600,
129	)
130	def ampel_api_alerts(	1✔
131	ztf_name: str,
132	t_min_jd=Time("2017-01-01T00:00:00.0", format="isot", scale="utc").jd,
133	t_max_jd=Time.now().jd,
134	program_id: int = None,
135	logger=None,
136	) -> list:
137	"""
138	Function to query ampel via name, returns a virtual alert
139	constructed by AMPEL containing ALL photopoints and upper limits
140
141	"""
142
NEW 143	if logger is None:	×
NEW 144	logger = logging.getLogger(__name__)	×
145
NEW 146	if program_id is None:	×
NEW 147	queryurl_lightcurve = (	×
148	API_ZTF_ARCHIVE_URL + f"/object/{ztf_name}/alerts?jd_start={t_min_jd}&"
149	f"jd_end={t_max_jd}&with_history=true"
150	)
151	else:
NEW 152	queryurl_lightcurve = (	×
153	API_ZTF_ARCHIVE_URL + f"/object/{ztf_name}/alerts?jd_start={t_min_jd}&"
154	f"jd_end={t_max_jd}&programid={program_id}"
155	)
156
NEW 157	logger.debug(queryurl_lightcurve)	×
158
NEW 159	headers = {"Authorization": f"Bearer {get_ampel_token()}"}	×
160
NEW 161	response = requests.get(	×
162	queryurl_lightcurve,
163	headers=headers,
164	)
165
NEW 166	if response.status_code == 503:	×
167	if response.headers:
168	logger.debug(response.headers)
169	raise requests.exceptions.RequestException
170
NEW 171	try:	×
NEW 172	query_res = [response.json()]	×
173
174	except JSONDecodeError:
175	if response.headers:
176	logger.debug(response.headers)
177	raise requests.exceptions.RequestException
178
NEW 179	return query_res	×

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