From 19f6cf1d3cc38fc6006f148586f4013dc3db9c84 Mon Sep 17 00:00:00 2001 From: Steve Foale Date: Wed, 6 May 2026 15:28:20 +0100 Subject: [PATCH 1/5] Fix GET filter parameters being ignored (lists now correctly encoded as query params). Add configure() as single entry point for API credentials and base URL Replace per-method api_token parameter with module-level client but retain backwards compatibility. Add Candidate put, delete, and batch_delete operations. Add unit tests. Add github actions CI workflow. Update README. --- .github/workflows/ci.yml | 24 ++ .github/workflows/python-publish.yml | 2 +- .gitignore | 2 +- README.md | 268 ++++++++++++++------- pyproject.toml | 8 +- src/gwtm_api/__init__.py | 3 +- src/gwtm_api/alert.py | 142 ++++------- src/gwtm_api/candidate.py | 346 ++++++++++----------------- src/gwtm_api/core/__init__.py | 2 +- src/gwtm_api/core/_client_factory.py | 51 ++++ src/gwtm_api/core/util.py | 14 +- src/gwtm_api/event_tools.py | 300 +++++++---------------- src/gwtm_api/instrument.py | 201 +++++----------- src/gwtm_api/pointing.py | 323 ++++++++++++------------- tests/__init__.py | 0 tests/unit/__init__.py | 0 tests/unit/conftest.py | 17 ++ tests/unit/test_alert.py | 60 +++++ tests/unit/test_instrument.py | 73 ++++++ tests/unit/test_pointing.py | 125 ++++++++++ tests/unit/test_transport.py | 73 ++++++ 21 files changed, 1108 insertions(+), 926 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 src/gwtm_api/core/_client_factory.py create mode 100644 tests/__init__.py create mode 100644 tests/unit/__init__.py create mode 100644 tests/unit/conftest.py create mode 100644 tests/unit/test_alert.py create mode 100644 tests/unit/test_instrument.py create mode 100644 tests/unit/test_pointing.py create mode 100644 tests/unit/test_transport.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..4145c4e --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,24 @@ +name: CI + +on: + push: + branches: ["**"] + pull_request: + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + + - name: Install dependencies + run: | + pip install -e ".[dev]" + ./scripts/generate_client.sh + + - name: Run unit tests + run: pytest tests/unit/ -v diff --git a/.github/workflows/python-publish.yml b/.github/workflows/python-publish.yml index b7a704b..15e93dd 100644 --- a/.github/workflows/python-publish.yml +++ b/.github/workflows/python-publish.yml @@ -23,7 +23,7 @@ jobs: steps: - uses: actions/checkout@v4 - name: Set up Python - uses: actions/setup-python@v3 + uses: actions/setup-python@v5 with: python-version: '3.x' - name: Install dependencies diff --git a/.gitignore b/.gitignore index 345d606..8c2c77b 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,4 @@ *pycache* *.pyc *.env* -dist \ No newline at end of file +dist diff --git a/README.md b/README.md index b70ff1b..b7d98e6 100644 --- a/README.md +++ b/README.md @@ -1,188 +1,278 @@ # gwtm_api -A python wrapper for the [Gravitational Wave Treasure Map](http://treasuremap.space). -In order to interact with the web API, you will need to [register](http://treasuremap.space/register) an account with the GWTM. Once verified you will recieve an `API_TOKEN` to pass into all api endpoints. +A Python wrapper for the [Gravitational Wave Treasure Map](https://treasuremap.space). +To interact with the API, [register](https://treasuremap.space/register) an account. Once verified you will receive an `API_TOKEN` shown on your profile page. + +**Using pip:** +```bash +python -m venv gwtm_env +source gwtm_env/bin/activate # Windows: gwtm_env\Scripts\activate +pip install gwtm_api +``` + +**Using conda:** ```bash conda create -n gwtm_api python=3.11 conda activate gwtm_api -python -m pip install gwtm_api +pip install gwtm_api +``` + +Full API documentation and parameter reference is available at [https://treasuremap.space/docs](https://treasuremap.space/docs). + +--- + +## Configuration + +Call `configure()` once at the start of your script. All subsequent API calls use the client created here — no need to pass credentials to every method. + +```python +import gwtm_api + +gwtm_api.configure(api_token='YOUR_API_TOKEN') +``` + +To point at a different server (e.g. a development instance), pass `base_url`: + +```python +gwtm_api.configure(api_token='YOUR_API_TOKEN', base_url='https://dev.treasuremap.space') +``` + +You can also set credentials via environment variables and call `configure()` with no arguments: + +```bash +export GWTM_API_TOKEN=YOUR_API_TOKEN +export GWTM_BASE_URL=https://treasuremap.space # optional, this is the default ``` +```python +gwtm_api.configure() +``` + +--- + +## Pointings -## Pointings: -Full api documentation with detailed examples can be found at [GWTM API Documentation](http://treasuremap.space/documentation). ### GET + ```python import gwtm_api -pointings = gwtm_api.Pointing.get(graceid="GW190814", instruments=["ZTF"], api_token=API_TOKEN) +gwtm_api.configure(api_token='YOUR_API_TOKEN') + +pointings = gwtm_api.Pointing.get( + graceid='GW190814', + instruments=['ZTF'], +) ``` +Filters can be combined freely — by instrument, status, time range, band, depth, and spectral regime. All list parameters accept Python lists directly. + ### POST -Submit single, or list of `gwtm_api.Pointing` objects. + +Submit a single pointing or a batch: + ```python +import datetime import gwtm_api -#submit single +gwtm_api.configure(api_token='YOUR_API_TOKEN') + +# Single pointing p = gwtm_api.Pointing( - ra=15, - dec=-24, - instrumentid = 10, - time = datetime.datetime.now(), - status = 'completed', - depth = 18.5, - depth_unit = 'ab_mag', + ra=15.0, + dec=-24.0, + instrumentid=10, + time=datetime.datetime.now(), + status='completed', + depth=18.5, + depth_unit='ab_mag', pos_angle=50, - band = 'r' + band='r' ) -p.post(graceid="GRACEID", api_token=API_TOKEN) +p.post(graceid='GRACEID') -#submit list +# Batch batch = [ - gwtm_api.Pointing(...), - gwtm_api.Pointing(...) + gwtm_api.Pointing(...), + gwtm_api.Pointing(...) ] -gwtm_api.Pointing.batch_post(pointings=batch, graceid="GRACEID", api_token=API_TOKEN) +gwtm_api.Pointing.batch_post(pointings=batch, graceid='GRACEID') ``` +--- + ## Candidates -Get and Post Event Candidates through the API: + ### GET + ```python import gwtm_api -from datetime import datetime -candidates = gwtm_api.Candidate.get(graceid="GW190814", api_token=API_TOKEN) +gwtm_api.configure(api_token='YOUR_API_TOKEN') + +candidates = gwtm_api.Candidate.get(graceid='GW190814') ``` + ### POST -Submit single, or list of `gwtm_api.Candidate` objects. + ```python +import datetime import gwtm_api -#submit single +gwtm_api.configure(api_token='YOUR_API_TOKEN') + +# Single candidate candidate = gwtm_api.Candidate( - ra=15, - dec=-24, + ra=15.0, + dec=-24.0, discovery_date=datetime.datetime.now(), - discovery_magnitude = 18, - magnitude_bandpass = "r", - magnitude_unit = "ab_mag", - associated_galaxy = "some galaxy name", - associated_galaxy_redshift = 0.3 + discovery_magnitude=18, + magnitude_bandpass='r', + magnitude_unit='ab_mag', + associated_galaxy='some galaxy name', + associated_galaxy_redshift=0.3 ) -candidate.post(graceid="GRACEID", api_token=API_TOKEN) +candidate.post(graceid='GRACEID') -#submit list +# Batch batch = [ - gwtm_api.Candidate(...), - gwtm_api.Candidate(...) + gwtm_api.Candidate(...), + gwtm_api.Candidate(...) ] -gwtm_api.Pointing.batch_post(candidates=batch, graceid="GRACEID", api_token=API_TOKEN) +gwtm_api.Candidate.batch_post(candidates=batch, graceid='GRACEID') ``` ### PUT -Update a GWTM candidate record + ```python import gwtm_api -candidate = gwtm_api.Candidate.get(api_token=API_TOKEN, id=CANDIDATE_ID) -my_candidate = candidate[0] -update_payload = { - "tns_name": "AT2017gfo", - "tns_url": "https://www.wis-tns.org/object/2017gfo" - "associated_galaxy": "NGC 4993" - "associated_galaxy_redshift": 0.009727 -} -my_candidate.put(api_token=API_TOKEN, payload=update_payload) -#or similarly -candidate = gwtm_api.Candidate(id=CANDIDATE_ID, api_token=API_TOKEN) #this will envoke the GET endpoint if it has an id and api token -candidate.tns_name = "AT2017gfo" -candidate.tns_url = "https://www.wis-tns.org/object/2017gfo" -candidate.associated_galaxy = "NGC 4993" -candidate.associated_galaxy_redshift = 0.009727 -candidate.put(api_token=API_TOKEN) +gwtm_api.configure(api_token='YOUR_API_TOKEN') + +candidate = gwtm_api.Candidate.get(id=CANDIDATE_ID)[0] +candidate.put(payload={ + 'tns_name': 'AT2017gfo', + 'tns_url': 'https://www.wis-tns.org/object/2017gfo', + 'associated_galaxy': 'NGC 4993', + 'associated_galaxy_redshift': 0.009727 +}) ``` -### Delete +### DELETE + ```python -candidate = gwtm_api.Candidate(id=21, api_token=API_TOKEN) -candidate.delete(api_token=API_TOKEN) +import gwtm_api + +gwtm_api.configure(api_token='YOUR_API_TOKEN') + +candidate = gwtm_api.Candidate.get(id=CANDIDATE_ID)[0] +candidate.delete() -#or batch delete with a list of ids -gwtm_api.Candidate.batch_delete(api_token=API_TOKEN, ids=[id1, id2....], verbose=True) +# Batch delete +gwtm_api.Candidate.batch_delete(ids=[id1, id2, ...]) ``` +--- + ## Instruments -Query for instrument information that have been submitted to the Treasure Map + ```python import gwtm_api -instruments = gwtm_api.Instrument.get(name="ZTF", api_token=API_TOKEN) +gwtm_api.configure(api_token='YOUR_API_TOKEN') + +instruments = gwtm_api.Instrument.get(names=['ZTF']) ``` -You can pass the parameter `include_footprint=True` into the get request, and receive the polygon information for the instrument footprint. -We've included basic polygon manipulation functionality with this footprint class as well. When combined with the pointing information you can project the footprint on the sky, and simulate reported coverage yourself. -```python -import gwtm_api -ztf = gwtm_api.Instrument.get(name="ZTF", include_footprint=True, api_token=API_TOKEN)[0] -ztf.project(ra, dec, pos_angle) +Pass `include_footprint=True` to receive polygon data for the instrument footprint, and project it onto the sky at any pointing: + +```python +ztf = gwtm_api.Instrument.get(names=['ZTF'], include_footprint=True)[0] +projected = ztf.project(ra, dec, pos_angle) ``` +--- + ## Event Tools -For a given GW event, you can utlize the the `event_tools` library to perform some analytics of a GW event with the data supported on the Treasure Map. + +For a given GW event, you can use the `event_tools` library to perform analytics with the data on the Treasure Map. ### Visualizing coverage + ```python import gwtm_api -gwtm_api.event_tools.plot_coverage(graceid="GW190814", api_token=API_TOKEN) +gwtm_api.configure(api_token='YOUR_API_TOKEN') + +gwtm_api.event_tools.plot_coverage(graceid='GW190814') ``` + image -The `plot_coverage` function allows you to pass in your own list of pointings, along with caching the queried results so you don't have to hit the API for large queries every time. +Pass in your own pointings and enable caching to avoid repeated API calls: ```python -pointings = gwtm_api.Pointing.get(graceid = "GW190814", instrument="ZTF", api_token=API_TOKEN, status='completed') +pointings = gwtm_api.Pointing.get( + graceid='GW190814', + instruments=['ZTF'], + status='completed', +) gwtm_api.event_tools.plot_coverage( - graceid="GW190814", - api_token=API_TOKEN, + graceid='GW190814', pointings=pointings, cache=True ) ``` -### Coverage Calculation -Calculate the total probability covered for a given GW Event. You can pass in a list of pointings for the event, or it will default to all pointings for the event. Returns the total probability and total area (deg^2) covered by the list of pointings +### Coverage calculation + +Returns total probability and total area (deg²) covered by the pointings: ```python total_prob, total_area = gwtm_api.event_tools.calculate_coverage( - graceid="GW190814", - api_token=API_TOKEN, + graceid='GW190814', pointings=pointings, cache=True ) ``` -### Renormalize Skymap -Renormalize an event's skymap based on a list of pointings (or the entire GW event's completed pointings). It takes the list of pointings and sets the overlapping skymap pixel probability to zero, then renormalizes the skymap. Returns an NDArray that can be imported into healpy +### Renormalize skymap + +Sets overlapping skymap pixel probability to zero for observed regions and renormalises. Returns an NDArray compatible with healpy: ```python renormalized_skymap = gwtm_api.event_tools.renormalize_skymap( - graceid="GW190814", - api_token=API_TOKEN, + graceid='GW190814', pointings=pointings, cache=True ) ``` -### Candidate Coerage -For a given `candidate`, find which instruments have `pointing` footprints that overlap with the candidate's position. The user can potentially constrain which instruments have observed a candidate pre/post post discovery. User's can pass in a list of `pointings`, or it will default to all `pointings` for the `candidate's` associated graceid. The function also accepts a `distance_thresh` (in degrees) to limit the calculation to only the pointings centered within the threshold distance from the candidate. +### Candidate coverage + +Find which instrument pointings overlap with a candidate's position: ```python -my_candidate = gwtm_api.Candidate.get(...) -pointings_list = gwtm_api.event_tools.candidate_coverage( - api_token=API_TOKEN, - candidate=my_candidate -) +my_candidate = gwtm_api.Candidate.get(graceid='GW190814')[0] +covering_pointings = gwtm_api.event_tools.candidate_coverage(candidate=my_candidate) +``` + +--- + +## For contributors + +The package uses a two-layer architecture: + +- **`src/gwtm_api/{pointing,alert,instrument,candidate}.py`** — hand-written, user-facing classes with a stable, friendly interface +- **`src/gwtm_api/_generated/`** — HTTP transport layer auto-generated from the server's OpenAPI spec (not committed to the repo) + +### Setup after cloning + +```bash +pip install -e '.[dev]' ``` + +### When the server API changes + +Update the relevant wrapper methods in `src/gwtm_api/`, bump the version in `pyproject.toml`, and create a GitHub Release — publishing to PyPI happens automatically. diff --git a/pyproject.toml b/pyproject.toml index 40c435c..18f3e64 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,12 +15,18 @@ requires-python = "<3.12" dependencies = [ "astropy", "healpy", - "requests", + "httpx>=0.20.0", "ligo.skymap", "matplotlib", "pandas" ] +[project.optional-dependencies] +dev = [ + "pytest", + "pytest-mock" +] + classifiers = [ "Programming Language :: Python :: 3", "License :: OSI Approved :: MIT License", diff --git a/src/gwtm_api/__init__.py b/src/gwtm_api/__init__.py index c9168ee..6b00b39 100644 --- a/src/gwtm_api/__init__.py +++ b/src/gwtm_api/__init__.py @@ -1,5 +1,6 @@ +from .core._client_factory import configure as configure # noqa: F401 from .pointing import Pointing as Pointing # noqa: E402 -from .candidate import Candidate as Candidate #noqa: E402 +from .candidate import Candidate as Candidate # noqa: E402 from .instrument import Instrument as Instrument # noqa: E402 from .instrument import Footprint as Footprint # noqa: E402 from .alert import Alert as Alert # noqa: E402 diff --git a/src/gwtm_api/alert.py b/src/gwtm_api/alert.py index b2e5f1a..f610c5b 100644 --- a/src/gwtm_api/alert.py +++ b/src/gwtm_api/alert.py @@ -1,12 +1,12 @@ import datetime -import json import healpy as hp import tempfile -from .core import baseapi from .core import apimodels from .core.tmcache import TMCache from .core import util +from .core import _client_factory as transport + class Alert(apimodels._Table): id: int = None @@ -35,7 +35,7 @@ class Alert(apimodels._Table): centralfreq: float = None duration: float = None avgra: float = None - avgdec : float = None + avgdec: float = None observing_run: str = None pipeline: str = None search: str = None @@ -49,7 +49,7 @@ class Alert(apimodels._Table): area_90: float = None area_50: float = None - def __init__(self, + def __init__(self, id: int = None, graceid: str = None, alternateid: str = None, @@ -76,7 +76,7 @@ def __init__(self, centralfreq: float = None, duration: float = None, avgra: float = None, - avgdec : float = None, + avgdec: float = None, observing_run: str = None, pipeline: str = None, search: str = None, @@ -91,125 +91,65 @@ def __init__(self, area_50: float = None, kwdict: dict = None ): - - if kwdict is not None: - selfdict = kwdict - else: - selfdict = util.non_none_locals(locals=locals()) - + selfdict = kwdict if kwdict is not None else util.non_none_locals(locals=locals()) super().__init__(payload=selfdict) @staticmethod - def get(api_token: str, id: int = None, graceid: str = None, urlencode=False): - - get_dict = util.non_none_locals(locals=locals()) - - r_json = { - "d_json":get_dict - } - - api = baseapi.api(target="query_alerts") - req = api._get(r_json=r_json, urlencode=urlencode) - - if req.status_code == 200: - request_json = json.loads(req.text) - for i in request_json: - if isinstance(i, str): - alert_json = json.loads(i) - else: - alert_json = i - #only return the first since it is the most recent - return Alert(kwdict=alert_json) - else: - raise Exception(f"Error in Alert.get(). Request: {req.text[0:1000]}") - + def get(id: int = None, graceid: str = None, api_token: str = None) -> "Alert": + alerts = Alert.get_all(id=id, graceid=graceid, api_token=api_token) + if alerts: + return alerts[0] + raise Exception(f"No alert found for graceid={graceid}, id={id}") @staticmethod - def get_all(api_token: str, id: int = None, graceid: str = None, urlencode=False): - - get_dict = util.non_none_locals(locals=locals()) - - r_json = { - "d_json":get_dict - } - - api = baseapi.api(target="query_alerts") - req = api._get(r_json=r_json, urlencode=urlencode) - - ret = [] - if req.status_code == 200: - request_json = json.loads(req.text) - for i in request_json: - if isinstance(i, str): - alert_json = json.loads(i) - else: - alert_json = i - ret.append(Alert(kwdict=alert_json)) - else: - raise Exception(f"Error in Alert.get(). Request: {req.text[0:1000]}") - - return ret - + def get_all(id: int = None, graceid: str = None, api_token: str = None): + if api_token: + transport.configure(api_token=api_token) + result = transport.get('/query_alerts', params={'graceid': graceid}) + if not isinstance(result, list): + raise Exception("Error in Alert.get_all(): unexpected response from server") + return [Alert(kwdict=a) for a in result] @staticmethod - def fetch_contours(api_token: str, id: int = None, graceid: str = None, urlencode=False, cache=False): - - get_dict = util.non_none_locals(locals=locals()) - - r_json = { - "d_json":get_dict - } - + def fetch_contours(id: int = None, graceid: str = None, + cache=False, api_token: str = None): + if api_token: + transport.configure(api_token=api_token) request_json = None if cache: - cache_name = f"{graceid}_gw_contour.json" - contour_cache = TMCache(filename=cache_name, cache_type="json") - request_json = contour_cache.get() + cache_obj = TMCache(filename=f"{graceid}_gw_contour.json", cache_type="json") + request_json = cache_obj.get() if request_json is None: - api = baseapi.api(target="gw_contour") - req = api._get(r_json=r_json, urlencode=urlencode) - - if req.status_code == 200: - request_json = json.loads(req.text) - else: - raise Exception(f"Error in Alert.fetch_contours(). Request: {req.text[0:1000]}") + request_json = transport.get('/gw_contour', params={'graceid': graceid}) if cache: - contour_cache.put(payload=request_json) + cache_obj.put(payload=request_json) contour_polygons = [] for contour in request_json['features']: contour_polygons.extend(contour['geometry']['coordinates']) return contour_polygons - @staticmethod - def fetch_skymap(api_token: str, id: int = None, graceid: str = None, urlencode=False, cache=False): - - get_dict = util.non_none_locals(locals=locals()) - - r_json = { - "d_json":get_dict - } - + def fetch_skymap(id: int = None, graceid: str = None, + cache=False, api_token: str = None): + if api_token: + transport.configure(api_token=api_token) request_map = None if cache: - cache_name = f"{graceid}_gw_skymap.fits" - skymap_cache = TMCache(filename=cache_name, cache_type="fits") - request_map = skymap_cache.get() + cache_obj = TMCache(filename=f"{graceid}_gw_skymap.fits", cache_type="fits") + request_map = cache_obj.get() if request_map is None: - api = baseapi.api(target="gw_skymap") - req = api._get(r_json=r_json, urlencode=urlencode) + response = transport.get_raw('/gw_skymap', params={'graceid': graceid}) + if response.status_code != 200: + raise Exception(f"Error in Alert.fetch_skymap(): status {response.status_code}") + with tempfile.NamedTemporaryFile(suffix=".fits") as tmp: + tmp.write(response.content) + tmp.flush() + request_map = hp.read_map(tmp.name) - if req.status_code == 200: - with tempfile.NamedTemporaryFile() as _tmp_file: - _tmp_file.write(req.content) - request_map = hp.read_map(_tmp_file.name) - else: - raise Exception(f"Error in Alert.fetch_skymap(). Request: {req.text[0:1000]}") if cache: - skymap_cache.put(payload=request_map) - - return request_map \ No newline at end of file + cache_obj.put(payload=request_map) + return request_map diff --git a/src/gwtm_api/candidate.py b/src/gwtm_api/candidate.py index bfc7590..3a2becf 100644 --- a/src/gwtm_api/candidate.py +++ b/src/gwtm_api/candidate.py @@ -1,10 +1,13 @@ from __future__ import annotations -import json import datetime from typing import List, Union -from .core import baseapi + from .core import apimodels from .core import util +from .core.util import _encode_list_param +from .core import _client_factory as transport +from .pointing import _enum_name + class Candidate(apimodels._Table): id: int = None @@ -32,8 +35,7 @@ class Candidate(apimodels._Table): associated_galaxy_redshift: float = None associated_galaxy_distance: float = None - def __init__( - self, + def __init__(self, id: int = None, graceid: str = None, candidate_name: str = None, @@ -60,34 +62,27 @@ def __init__( api_token: str = None, kwdict=None ): + if api_token: + transport.configure(api_token=api_token) self.position = position + selfdict = kwdict if kwdict is not None else util.non_none_locals(locals=locals()) - if kwdict is not None: - selfdict = kwdict - else: - selfdict = util.non_none_locals(locals=locals()) - - #if user passes in an ID, then prepopulate with a get request if id: - if api_token: - payload = Candidate.get(api_token=api_token, id=id) - if len(payload): - selfdict = payload[0].__dict__ - else: - raise Exception(f"No candidate found with id: {id}") + payload = Candidate.get(id=id) + if payload: + selfdict = payload[0].__dict__ else: - raise Exception("api_token required") - + raise Exception(f"No candidate found with id: {id}") + super().__init__(payload=selfdict) - if "position" in selfdict.keys(): + if "position" in selfdict: self._sanatize_pointing() - elif all([x in selfdict.keys() for x in ["ra", "dec"]]): + elif "ra" in selfdict and "dec" in selfdict: self._sanatize_ra_dec() else: - raise Exception("Positional arguments are required. Must be decimal format ra, dec, or geometry type \"POINT(ra dec)\"") - + raise Exception("Positional arguments are required: ra/dec or POINT(ra dec)") def _sanatize_pointing(self): if self.position is not None: @@ -95,215 +90,136 @@ def _sanatize_pointing(self): self.ra = float(self.position.split('(')[1].split(')')[0].split()[0]) self.dec = float(self.position.split('(')[1].split(')')[0].split()[1]) except: # noqa: E722 - raise Exception("Invalid position argument. Must be 'POINT (\{ra\} \{dec\})'.") - + raise Exception("Invalid position. Must be 'POINT ({ra} {dec})'.") def _sanatize_ra_dec(self): if self.position is None: - if all([isinstance(x, float) for x in [self.ra, self.dec]]): + if all(isinstance(x, float) for x in [self.ra, self.dec]): self.position = f"POINT ({self.ra} {self.dec})" else: - raise Exception("Invalid format for ra/dec. Both must be float") - + raise Exception("ra and dec must be floats") + + def _to_dict(self) -> dict: + dd = (self.discovery_date.strftime("%Y-%m-%dT%H:%M:%S.%f") + if isinstance(self.discovery_date, datetime.datetime) else self.discovery_date) + return { + 'candidate_name': self.candidate_name or '', + 'discovery_date': dd, + 'discovery_magnitude': self.discovery_magnitude or 0.0, + 'magnitude_unit': _enum_name(self.magnitude_unit) or 'ab_mag', + 'position': self.position, + 'ra': self.ra, + 'dec': self.dec, + 'tns_name': self.tns_name, + 'tns_url': self.tns_url, + 'magnitude_bandpass': _enum_name(self.magnitude_bandpass), + 'magnitude_bandwidth': self.magnitude_bandwidth, + 'wavelength_regime': self.wavelength_regime, + 'wavelength_unit': _enum_name(self.wavelength_unit), + 'frequency_regime': self.frequency_regime, + 'frequency_unit': _enum_name(self.frequency_unit), + 'energy_regime': self.energy_regime, + 'energy_unit': _enum_name(self.energy_unit), + 'associated_galaxy': self.associated_galaxy, + 'associated_galaxy_redshift': self.associated_galaxy_redshift, + 'associated_galaxy_distance': self.associated_galaxy_distance, + } - def post( - self, api_token: str, graceid: str = None, - base: str = "https://treasuremap.space/api/", api_version: str ="v1", - verbose = False - ): + def post(self, graceid: str = None, verbose=False, api_token: str = None): + if api_token: + transport.configure(api_token=api_token) if graceid is None: graceid = self.graceid - - post_dict = util.non_none_locals(locals=locals()) - - self.discovery_date = self.discovery_date.strftime("%Y-%m-%dT%H:%M:%S.%f") - post_dict["candidates"]=[self.__dict__] - post_dict["graceid"]=graceid - - r_json = { - "d_json":post_dict - } - - api = baseapi.api(target="candidate") - req = api._post(r_json=r_json) - - if req.status_code == 200: - request_json = json.loads(req.text) - if verbose: - print(request_json) - id = request_json["candidate_ids"][0] - self.__init__(kwdict=Candidate.get(api_token=api_token, id=id)[0].__dict__) - else: - raise Exception(f"Error in Candidate.post(). Request: {req.text[0:1000]}") - + result = transport.post('/candidate', json={'graceid': graceid, 'candidate': self._to_dict()}) + if verbose: + print(result) + fetched = Candidate.get(id=result['candidate_ids'][0]) + self.__init__(kwdict=fetched[0].__dict__) @staticmethod - def batch_post( - candidates: List[Candidate], api_token: str, graceid: str = None, - base: str = "https://treasuremap.space/api/", api_version: str ="v1", - verbose = False - ) -> List[Candidate]: - - post_dict = util.non_none_locals(locals=locals()) - - batch = [] - for p in candidates: - if not isinstance(p, Candidate): - raise Exception("Input candidate must be a list of Candidate") - p.discovery_date = p.discovery_date.strftime("%Y-%m-%dT%H:%M:%S.%f") - batch.append(p.__dict__) - + def batch_post(candidates: List[Candidate], graceid: str = None, + verbose=False, api_token: str = None) -> List[Candidate]: + if api_token: + transport.configure(api_token=api_token) if graceid is None: - graceids = list(set([x.graceid for x in candidates])) - if len(graceids) > 1: - raise Exception("You can only post candidates for a single GW Event graceid at a time") - post_dict["graceid"] = graceids[0] - - post_dict['candidates'] = batch - - r_json = { - "d_json":post_dict - } - - api = baseapi.api(target="candidate") - req = api._post(r_json=r_json) - - if req.status_code == 200: - request_json = json.loads(req.text) - ids = request_json["candidate_ids"] - if verbose: - print(request_json) - return Candidate.get(api_token=api_token, ids=ids) - else: - raise Exception(f"Error in Candidate.post(). Request: {req.text[0:1000]}") - + graceids = list(set(c.graceid for c in candidates if c.graceid)) + if len(graceids) != 1: + raise Exception("All candidates must share a single graceid, or provide it explicitly") + graceid = graceids[0] + result = transport.post('/candidate', json={ + 'graceid': graceid, + 'candidates': [c._to_dict() for c in candidates], + }) + if verbose: + print(result) + return Candidate.get(ids=result['candidate_ids']) @staticmethod - def get( - api_token: str, graceid: str = None, id: int = None, ids: List[int] = None, - submitted_date_after: datetime.datetime = None, submitted_date_before: datetime.datetime = None, - discovery_date_after: datetime.datetime = None, discovery_date_before: datetime.datetime = None, - userid: int = None, discovery_magnitude_gt: float = None, discovery_magnitude_lt: float = None, - associated_galaxy_redshift_gt: float = None, associated_galaxy_redshift_lt: float = None, - associated_galaxy_distance_gt: float = None, associated_galaxy_distance_lt: float = None, - associated_galaxy_name: str = None, - urlencode=False, base: str = "https://treasuremap.space/api/", api_version: str ="v1" - ) -> list[Candidate]: - - get_dict = util.non_none_locals(locals=locals()) - - r_json = { - "d_json":get_dict - } - - api = baseapi.api(target="candidate", base=base, api_version=api_version) - req = api._get(r_json=r_json, urlencode=urlencode) - - if req.status_code == 200: - ret = [] - request_json = json.loads(req.text) - for p in request_json: - if 'v0' in api.base: - json_value = json.loads(p) - else: - json_value = p - ret.append(Candidate(kwdict=json_value)) - return ret - else: - raise Exception(f"Error in Candidate.get(). Request: {req.text[0:1000]}") - - - def put( - self, api_token: str, payload: dict = None, base: str = "https://treasuremap.space/api/", - api_version: str ="v1", verbose: bool = False - ): - ''' - Candidate PUT endpoint - ''' - if payload: - if "id" in payload.keys() and isinstance(payload["id"], int): - self.id = payload["id"] - else: - payload = self.__dict__ + def get(graceid: str = None, id: int = None, ids: List[int] = None, + submitted_date_after: datetime.datetime = None, + submitted_date_before: datetime.datetime = None, + discovery_date_after: datetime.datetime = None, + discovery_date_before: datetime.datetime = None, + userid: int = None, + discovery_magnitude_gt: float = None, + discovery_magnitude_lt: float = None, + associated_galaxy_redshift_gt: float = None, + associated_galaxy_redshift_lt: float = None, + associated_galaxy_distance_gt: float = None, + associated_galaxy_distance_lt: float = None, + associated_galaxy_name: str = None, + urlencode=None, + api_token: str = None) -> list[Candidate]: + if api_token: + transport.configure(api_token=api_token) + result = transport.get('/candidate', params={ + 'graceid': graceid, + 'id': id, + 'userid': userid, + 'submitted_date_after': submitted_date_after, + 'submitted_date_before': submitted_date_before, + 'discovery_date_after': discovery_date_after, + 'discovery_date_before': discovery_date_before, + 'discovery_magnitude_gt': discovery_magnitude_gt, + 'discovery_magnitude_lt': discovery_magnitude_lt, + 'associated_galaxy_name': associated_galaxy_name, + 'associated_galaxy_redshift_gt': associated_galaxy_redshift_gt, + 'associated_galaxy_redshift_lt': associated_galaxy_redshift_lt, + 'associated_galaxy_distance_gt': associated_galaxy_distance_gt, + 'associated_galaxy_distance_lt': associated_galaxy_distance_lt, + }) + if not isinstance(result, list): + raise Exception("Error in Candidate.get(): unexpected response from server") + return [Candidate(kwdict=c) for c in result] + + def put(self, payload: dict = None, verbose: bool = False, + api_token: str = None): + if api_token: + transport.configure(api_token=api_token) if not self.id: raise Exception("Candidate does not have an id") - - if self.discovery_date: - self.discovery_date = self.discovery_date.strftime("%Y-%m-%dT%H:%M:%S.%f") - if self.datecreated: - self.datecreated = self.datecreated.strftime("%Y-%m-%dT%H:%M:%S.%f") - - put_dict = { - "api_token": api_token, - "id": self.id, - "payload": payload - } - - r_json = { - "d_json": put_dict - } - - api = baseapi.api(target="candidate", base=base, api_version=api_version) - req = api._put(r_json=r_json) - if req.status_code == 200: - request_json = json.loads(req.text) - if verbose: - print(request_json) - self.__init__(kwdict=Candidate.get(api_token=api_token, id=self.id)[0].__dict__) - else: - raise Exception(f"Error in Candidate.get(). Request: {req.text[0:1000]}") - - - def delete( - self, api_token: str, base: str = "https://treasuremap.space/api/", - api_version: str ="v1", verbose: bool =False - ) -> dict: - ''' - Candidate DELETE endpoint - ''' + result = transport.put('/candidate', json={'candidate': payload or self._to_dict()}) + if verbose: + print(result) + fetched = Candidate.get(id=self.id) + self.__init__(kwdict=fetched[0].__dict__) + + def delete(self, verbose: bool = False, api_token: str = None) -> dict: + if api_token: + transport.configure(api_token=api_token) if not self.id: raise Exception("Candidate does not have an id") - - del_dict = { - "api_token": api_token, - "id": self.id - } - r_json = { - "d_json": del_dict - } + result = transport.delete('/candidate', json={'id': self.id}) + if verbose: + print(result) + return result - api = baseapi.api(target="candidate", base=base, api_version=api_version) - req = api._delete(r_json=r_json) - if req.status_code == 200: - request_json = json.loads(req.text) - if verbose: - print(request_json) - return request_json - else: - raise Exception(f"Error in Candidate.get(). Request: {req.text[0:1000]}") - @staticmethod - def batch_delete( - api_token: str, ids: List[int], base: str = "https://treasuremap.space/api/", api_version: str ="v1", - verbose: bool = False - ) -> dict: - ''' - Canididates DELETE endpoint - ''' - del_dict = { - "api_token": api_token, - "ids": ids - } - r_json = { - "d_json": del_dict - } - - api = baseapi.api(target="candidate", base=base, api_version=api_version) - req = api._delete(r_json=r_json) - if req.status_code == 200: - request_json = json.loads(req.text) - if verbose: - print(request_json) - return request_json - else: - raise Exception(f"Error in Candidate.get(). Request: {req.text[0:1000]}") \ No newline at end of file + def batch_delete(ids: List[int], verbose: bool = False, + api_token: str = None) -> dict: + if api_token: + transport.configure(api_token=api_token) + result = transport.delete('/candidate', json={'ids': ids}) + if verbose: + print(result) + return result diff --git a/src/gwtm_api/core/__init__.py b/src/gwtm_api/core/__init__.py index 4048a58..15b6178 100644 --- a/src/gwtm_api/core/__init__.py +++ b/src/gwtm_api/core/__init__.py @@ -1 +1 @@ -from . import baseapi, apimodels # noqa: F401 +from . import apimodels # noqa: F401 diff --git a/src/gwtm_api/core/_client_factory.py b/src/gwtm_api/core/_client_factory.py new file mode 100644 index 0000000..599e5bf --- /dev/null +++ b/src/gwtm_api/core/_client_factory.py @@ -0,0 +1,51 @@ +import os +import httpx + +DEFAULT_BASE_URL = "https://treasuremap.space/api/v1" + +_token: str = "" +_base_url: str = DEFAULT_BASE_URL + + +def configure(api_token: str = None, base_url: str = None) -> None: + global _token, _base_url + _token = api_token or os.getenv("GWTM_API_TOKEN", "") + _base_url = (base_url or os.getenv("GWTM_BASE_URL", DEFAULT_BASE_URL)).rstrip("/") + + +def _headers() -> dict: + return {"api_token": _token} + + +def _clean(params: dict) -> dict: + return {k: v for k, v in params.items() if v is not None} + + +def get(path: str, params: dict = None) -> list | dict: + response = httpx.get( + f"{_base_url}{path}", headers=_headers(), params=_clean(params or {})) + response.raise_for_status() + return response.json() + + +def post(path: str, json: dict = None) -> dict: + response = httpx.post(f"{_base_url}{path}", headers=_headers(), json=json) + response.raise_for_status() + return response.json() + + +def put(path: str, json: dict = None) -> dict: + response = httpx.put(f"{_base_url}{path}", headers=_headers(), json=json) + response.raise_for_status() + return response.json() + + +def delete(path: str, json: dict = None) -> dict: + response = httpx.delete(f"{_base_url}{path}", headers=_headers(), json=json) + response.raise_for_status() + return response.json() + + +def get_raw(path: str, params: dict = None) -> httpx.Response: + return httpx.get( + f"{_base_url}{path}", headers=_headers(), params=_clean(params or {})) diff --git a/src/gwtm_api/core/util.py b/src/gwtm_api/core/util.py index 91f6435..ffdc797 100644 --- a/src/gwtm_api/core/util.py +++ b/src/gwtm_api/core/util.py @@ -1,5 +1,16 @@ +import json +from typing import Optional import numpy as np + +def _encode_list_param(value) -> Optional[str]: + """Encode a list parameter as a JSON string for the server's query param parser.""" + if value is None: + return None + if isinstance(value, list): + return json.dumps(value) + return str(value) + COLORS = [ "#000000", "#FFFF00", "#1CE6FF", "#FF34FF", "#FF4A46", "#008941", "#006FA6", "#A30059", "#FFDBE5", "#7A4900", "#0000A6", "#63FFAC", "#B79762", "#004D43", "#8FB0FF", "#997D87", @@ -21,7 +32,8 @@ def non_none_locals(locals: dict) -> dict: ignore_locals = [ - "self", "__class__", "base", "api_version", "urlencode", "debug", "cache" + "self", "__class__", "base", "api_version", "urlencode", "debug", "cache", + "api_token", "verbose", "kwdict" ] non_none_keys = [key for key, value in locals.items() if value is not None and key not in ignore_locals] diff --git a/src/gwtm_api/event_tools.py b/src/gwtm_api/event_tools.py index 435d67c..b89e9ac 100644 --- a/src/gwtm_api/event_tools.py +++ b/src/gwtm_api/event_tools.py @@ -16,34 +16,25 @@ from .core.util import instrument_color, gc_dist -def plot_coverage(api_token: str, graceid: str, pointings: List[Pointing] = [], +def plot_coverage(graceid: str, pointings: List[Pointing] = [], cache=False, projection='astro hours mollweide') -> None: - + if len(pointings) == 0 and graceid is None: raise Exception("Pointings list or graceid is required") if len(pointings) == 0 and graceid is not None: - pointings = Pointing.get(graceid = graceid, api_token=api_token, status='completed') - - #fetch skymap data - skymap_data = Alert.fetch_skymap(graceid=graceid, api_token=api_token, cache=cache) + pointings = Pointing.get(graceid=graceid, status='completed') - #fetch contour data - contour_polygons = Alert.fetch_contours(graceid=graceid, api_token=api_token, cache=cache) + skymap_data = Alert.fetch_skymap(graceid=graceid, cache=cache) + contour_polygons = Alert.fetch_contours(graceid=graceid, cache=cache) - #query for footprint info instrument_ids = list(set([x.instrumentid for x in pointings])) - inst_footprints = Instrument.get(ids=instrument_ids, include_footprint=True, api_token=api_token) + inst_footprints = Instrument.get(ids=instrument_ids, include_footprint=True) - #set up the plot - subplot_kw = { - 'projection': projection, - #'center': SkyCoord(alert_info.avgra, alert_info.avgdec, unit="deg") - } + subplot_kw = {'projection': projection} fig, ax = plt.subplots(1, 1, layout="constrained", subplot_kw=subplot_kw) ax.grid() - #plot each of the instrument footprints instrument_enumeration_dict = {} for i, iid in enumerate(instrument_ids): instrument_enumeration_dict[iid] = i @@ -54,104 +45,62 @@ def plot_coverage(api_token: str, graceid: str, pointings: List[Pointing] = [], if cache: cached_footprints = Footprint.get_cached_footprints( - graceid=graceid, - instrument_id=i, - pointings=instrument_pointings - ) + graceid=graceid, instrument_id=i, pointings=instrument_pointings) if cached_footprints: - polygon_arr.extend( - cached_footprints - ) + polygon_arr.extend(cached_footprints) inst_footprint = [x for x in inst_footprints if x.id == i][0] - + if len(polygon_arr) == 0: for p in instrument_pointings: projected_footprint = inst_footprint.project(p.ra, p.dec, p.pos_angle) for ccd in projected_footprint: - ra_deg, dec_deg = zip(*[(coord_deg[0], coord_deg[1]) - for coord_deg in ccd]) - - list_o_coords = [] - for x, y in zip(ra_deg, dec_deg): - list_o_coords.append((x, y)) - polygon_arr.append(list_o_coords) + ra_deg, dec_deg = zip(*[(coord_deg[0], coord_deg[1]) for coord_deg in ccd]) + polygon_arr.append(list(zip(ra_deg, dec_deg))) if cache: Footprint.put_cached_footprints( - polygon_arr, - graceid=graceid, - instrument_id=i, - pointings=instrument_pointings - ) + polygon_arr, graceid=graceid, instrument_id=i, pointings=instrument_pointings) - for j,arr in enumerate(polygon_arr): - if j == 0: - label = inst_footprint.nickname if inst_footprint.nickname is not None else inst_footprint.name - else: - label = None - - #cut_dateline(arr)[0] + for j, arr in enumerate(polygon_arr): + label = (inst_footprint.nickname if inst_footprint.nickname is not None + else inst_footprint.name) if j == 0 else None poly = Polygon( - np.asarray(arr), + np.asarray(arr), transform=ax.get_transform('world'), - edgecolor=instrument_color(instrument_enumeration_dict[i]), - facecolor='None', - linewidth=0.5, - alpha=1.0, - zorder=9900, - label=label + edgecolor=instrument_color(instrument_enumeration_dict[i]), + facecolor='None', linewidth=0.5, alpha=1.0, zorder=9900, label=label ) ax.add_patch(poly) - #plot thee 90/50 contoturs for contour_polygon in contour_polygons: - ra_deg, dec_deg = zip(*[(coord_deg[0], coord_deg[1]) - for coord_deg in contour_polygon]) - - list_o_coords = [] - for x, y in zip(ra_deg, dec_deg): - list_o_coords.append((x, y)) - arr = np.asarray(list_o_coords) - + ra_deg, dec_deg = zip(*[(coord_deg[0], coord_deg[1]) for coord_deg in contour_polygon]) poly = Polygon( - arr, + np.asarray(list(zip(ra_deg, dec_deg))), transform=ax.get_transform('world'), - edgecolor='r', - facecolor='None', - linewidth=0.5, - alpha=1.0, - zorder=9900 + edgecolor='r', facecolor='None', linewidth=0.5, alpha=1.0, zorder=9900 ) ax.add_patch(poly) - #plot the skymap: if skymap_data is not None: - _90_50_levels = find_greedy_credible_levels(np.asarray(skymap_data)) - ax.contourf_hpx( - _90_50_levels, - cmap='OrRd_r', - levels=[0.0, 0.5, 0.9], - alpha=0.75 - ) + ax.contourf_hpx(_90_50_levels, cmap='OrRd_r', levels=[0.0, 0.5, 0.9], alpha=0.75) - plt.legend(bbox_to_anchor=(0, 0), loc="lower left", - bbox_transform=fig.transFigure, ncol=4) + plt.legend(bbox_to_anchor=(0, 0), loc="lower left", bbox_transform=fig.transFigure, ncol=4) plt.title(f"Reported Coverage for {graceid}") plt.show() -def calculate_coverage(api_token: str, graceid: str, pointings: List[Pointing] = [], +def calculate_coverage(graceid: str, pointings: List[Pointing] = [], cache=False, approximate=True) -> Tuple[float, float]: DECam_id = 38 if len(pointings) == 0 and graceid is None: raise Exception("Pointings list or graceid is required") if len(pointings) == 0 and graceid is not None: - pointings = Pointing.get(graceid = graceid, api_token=api_token, status='completed') + pointings = Pointing.get(graceid=graceid, status='completed') - skymap = Alert.fetch_skymap(graceid=graceid, api_token=api_token, cache=cache) + skymap = Alert.fetch_skymap(graceid=graceid, cache=cache) instrument_ids = list(set([x.instrumentid for x in pointings])) @@ -159,7 +108,8 @@ def calculate_coverage(api_token: str, graceid: str, pointings: List[Pointing] = print("Warning: DECam footprint will be automatically approximated") approximate = True - inst_footprints = Instrument.get(ids=instrument_ids, include_footprint=True, api_token=api_token, approximate_footprint=approximate) + inst_footprints = Instrument.get(ids=instrument_ids, include_footprint=True, + approximate_footprint=approximate) instrument_enumeration_dict = {} for i, iid in enumerate(instrument_ids): @@ -168,7 +118,7 @@ def calculate_coverage(api_token: str, graceid: str, pointings: List[Pointing] = skymap_nside = hp.npix2nside(len(skymap)) qps = [] - NSIDE4area = 512 #this gives pixarea of 0.013 deg^2 per pixel + NSIDE4area = 512 pixarea = hp.nside2pixarea(NSIDE4area, degrees=True) for i in instrument_ids: @@ -177,86 +127,52 @@ def calculate_coverage(api_token: str, graceid: str, pointings: List[Pointing] = if cache: cached_footprints = Footprint.get_cached_footprints( - graceid=graceid, - instrument_id=i, - pointings=instrument_pointings - ) + graceid=graceid, instrument_id=i, pointings=instrument_pointings) if cached_footprints: - polygon_arr.extend( - cached_footprints - ) + polygon_arr.extend(cached_footprints) inst_footprint = [x for x in inst_footprints if x.id == i][0] - + if len(polygon_arr) == 0: for p in instrument_pointings: projected_footprint = inst_footprint.project(p.ra, p.dec, p.pos_angle) for ccd in projected_footprint: - ra_deg, dec_deg = zip(*[(coord_deg[0], coord_deg[1]) - for coord_deg in ccd]) - - list_o_coords = [] - for x, y in zip(ra_deg, dec_deg): - list_o_coords.append((x, y)) - polygon_arr.append(list_o_coords) + ra_deg, dec_deg = zip(*[(coord_deg[0], coord_deg[1]) for coord_deg in ccd]) + polygon_arr.append(list(zip(ra_deg, dec_deg))) if cache: Footprint.put_cached_footprints( - polygon_arr, - graceid=graceid, - instrument_id=i, - pointings=instrument_pointings - ) + polygon_arr, graceid=graceid, instrument_id=i, pointings=instrument_pointings) - for j,arr in enumerate(polygon_arr): + for arr in polygon_arr: ras_poly = [x[0] for x in arr][:-1] decs_poly = [x[1] for x in arr][:-1] - xyzpoly = astropy.coordinates.spherical_to_cartesian(1, np.deg2rad(decs_poly), np.deg2rad(ras_poly)) - qp = hp.query_polygon(skymap_nside,np.array(xyzpoly).T, inclusive=True) + xyzpoly = astropy.coordinates.spherical_to_cartesian( + 1, np.deg2rad(decs_poly), np.deg2rad(ras_poly)) + qp = hp.query_polygon(skymap_nside, np.array(xyzpoly).T, inclusive=True) qps.extend(qp) + deduped_indices = list(dict.fromkeys(qps)) - # qparea = hp.query_polygon(NSIDE4area, np.array(xyzpoly).T) - # qpsarea.extend(qparea) - - #deduplicate indices, so that pixels already covered are not double counted - deduped_indices=list(dict.fromkeys(qps)) - # deduped_indices_area = list(dict.fromkeys(qpsarea)) - - # area = pixarea * len(deduped_indices_area) - - #prob = 0 - #for ind in deduped_indices: - # prob += skymap[ind] - #elapsed = p.time - time_of_signal - #elapsed = elapsed.total_seconds()/3600 - #times.append(elapsed) - #probs.append(prob) - prob = np.sum(skymap[deduped_indices]) area = pixarea * len(deduped_indices) return prob, area -def renormalize_skymap(api_token: str, graceid: str, pointings: List[Pointing] = [], - cache=False) -> Any: +def renormalize_skymap(graceid: str, pointings: List[Pointing] = [], + cache=False) -> Any: if len(pointings) == 0 and graceid is None: raise Exception("Pointings list or graceid is required") if len(pointings) == 0 and graceid is not None: - pointings = Pointing.get(graceid = graceid, api_token=api_token, status='completed') + pointings = Pointing.get(graceid=graceid, status='completed') - skymap = Alert.fetch_skymap(graceid=graceid, api_token=api_token, cache=cache) + skymap = Alert.fetch_skymap(graceid=graceid, cache=cache) instrument_ids = list(set([x.instrumentid for x in pointings])) - inst_footprints = Instrument.get(ids=instrument_ids, include_footprint=True, api_token=api_token) - - instrument_enumeration_dict = {} - for i, iid in enumerate(instrument_ids): - instrument_enumeration_dict[iid] = i + inst_footprints = Instrument.get(ids=instrument_ids, include_footprint=True) skymap_nside = hp.npix2nside(len(skymap)) - qps = [] for i in instrument_ids: @@ -265,55 +181,41 @@ def renormalize_skymap(api_token: str, graceid: str, pointings: List[Pointing] = if cache: cached_footprints = Footprint.get_cached_footprints( - graceid=graceid, - instrument_id=i, - pointings=instrument_pointings - ) + graceid=graceid, instrument_id=i, pointings=instrument_pointings) if cached_footprints: - polygon_arr.extend( - cached_footprints - ) + polygon_arr.extend(cached_footprints) inst_footprint = [x for x in inst_footprints if x.id == i][0] - + if len(polygon_arr) == 0: for p in instrument_pointings: projected_footprint = inst_footprint.project(p.ra, p.dec, p.pos_angle) for ccd in projected_footprint: - ra_deg, dec_deg = zip(*[(coord_deg[0], coord_deg[1]) - for coord_deg in ccd]) - - list_o_coords = [] - for x, y in zip(ra_deg, dec_deg): - list_o_coords.append((x, y)) - polygon_arr.append(list_o_coords) + ra_deg, dec_deg = zip(*[(coord_deg[0], coord_deg[1]) for coord_deg in ccd]) + polygon_arr.append(list(zip(ra_deg, dec_deg))) if cache: Footprint.put_cached_footprints( - polygon_arr, - graceid=graceid, - instrument_id=i, - pointings=instrument_pointings - ) + polygon_arr, graceid=graceid, instrument_id=i, pointings=instrument_pointings) - for j,arr in enumerate(polygon_arr): + for arr in polygon_arr: ras_poly = [x[0] for x in arr][:-1] decs_poly = [x[1] for x in arr][:-1] - xyzpoly = astropy.coordinates.spherical_to_cartesian(1, np.deg2rad(decs_poly), np.deg2rad(ras_poly)) - qp = hp.query_polygon(skymap_nside,np.array(xyzpoly).T) + xyzpoly = astropy.coordinates.spherical_to_cartesian( + 1, np.deg2rad(decs_poly), np.deg2rad(ras_poly)) + qp = hp.query_polygon(skymap_nside, np.array(xyzpoly).T) qps.extend(qp) - - deduped_indices=list(dict.fromkeys(qps)) + deduped_indices = list(dict.fromkeys(qps)) skymap[deduped_indices] = 0 - normed_skymap = skymap/np.sum(skymap) - + normed_skymap = skymap / np.sum(skymap) return normed_skymap -def renormed_skymap_contours(api_token: str, graceid: str, pointings: List[Pointing] = [], + +def renormed_skymap_contours(graceid: str, pointings: List[Pointing] = [], cache=False) -> Any: - normed_skymap = renormalize_skymap(api_token=api_token, graceid=graceid, pointings=pointings, cache=cache) + normed_skymap = renormalize_skymap(graceid=graceid, pointings=pointings, cache=cache) i = np.flipud(np.argsort(normed_skymap)) cumsum = np.cumsum(normed_skymap[i]) @@ -321,61 +223,49 @@ def renormed_skymap_contours(api_token: str, graceid: str, pointings: List[Point cls[i] = cumsum * 100 paths = list(ligo.skymap.postprocess.contour(cls, [50, 90], nest=False, degrees=True, simplify=True)) - contours_json = json.dumps({ + return json.dumps({ 'type': 'FeatureCollection', 'features': [ { 'type': 'Feature', - 'properties': { - 'credible_level': contour - }, - 'geometry': { - 'type': 'MultiLineString', - 'coordinates': path - } + 'properties': {'credible_level': contour}, + 'geometry': {'type': 'MultiLineString', 'coordinates': path} } - for contour, path in zip([50,90], paths) + for contour, path in zip([50, 90], paths) ] }) - return contours_json -def candidate_coverage(api_token: str, candidate: Candidate, pointings: List[Pointing] = None, distance_thresh: float = 5.0) -> List[Pointing]: - ''' - inputs: - api_token: str - valid GWTM api_token - candidate: Candidate - the candidate you want to evaluate the coverage for - pointings: List[Pointing] - you can query for your own pointings to evaluate the coverage. default = [] - distance_threshold: float - distance in degrees. Only evaluate the pointings that are this distance from the candidate. default = 5.0 (deg) - - returns all pointings associated with the graceid that have had that candidate in its FOV - ''' +def candidate_coverage(candidate: Candidate, pointings: List[Pointing] = None, + distance_thresh: float = 5.0) -> List[Pointing]: + """ + Find which instrument pointings overlap with a candidate's position. + + Args: + candidate: the Candidate to evaluate coverage for + pointings: pre-fetched pointings to evaluate; defaults to all pointings for the event + distance_thresh: search radius in degrees (default 5.0) - #set an arbitarily high nside + Returns all pointings whose footprint contains the candidate position. + """ skymap_nside = 1024 - #find our candidates healpix - candidate_healpix = hp.ang2pix(skymap_nside, candidate.ra, candidate.dec, lonlat=True, nest=True) + candidate_healpix = hp.ang2pix(skymap_nside, candidate.ra, candidate.dec, + lonlat=True, nest=True) - #query for all pointings if not pointings: - pointings = Pointing.get(api_token=api_token, graceid=candidate.graceid) + pointings = Pointing.get(graceid=candidate.graceid) - #convert them a pandas dataframe for vectorized distance culling pointings_df = pd.DataFrame([x.__dict__ for x in pointings]) - - #calculate the distance for each pointing from the candidate, and cull them pointings_df["_DIST"] = gc_dist( pointings_df["ra"], pointings_df["dec"], candidate.ra, candidate.dec ) culled_pointings = pointings_df.loc[pointings_df["_DIST"] < distance_thresh] - #return nothing if there aren't any pointings associated if len(culled_pointings) == 0: return [] - #grab the instrument information for each pointing instrument_ids = culled_pointings["instrumentid"].values.tolist() - instruments = Instrument.get(api_token=api_token, ids=instrument_ids, include_footprint=True) + instruments = Instrument.get(ids=instrument_ids, include_footprint=True) vals = zip( culled_pointings["ra"].values, @@ -386,31 +276,19 @@ def candidate_coverage(api_token: str, candidate: Candidate, pointings: List[Poi ) return_ids = [] - - #iterate over our pointing values for pra, pdec, ppos_angle, pid, pinstid in vals: - #project the instrument footprint for each pointing instrument = [x for x in instruments if x.id == pinstid][0] projected = instrument.project(pra, pdec, ppos_angle) - #prepare the polygon for hp.query_polygon - polygon_arr = [] for ccd in projected: - ra_deg, dec_deg = zip(*[(coord_deg[0], coord_deg[1]) - for coord_deg in ccd]) - - list_o_coords = [] - for x, y in zip(ra_deg, dec_deg): - list_o_coords.append((x, y)) - polygon_arr.append(list_o_coords) - - #query the polygons and append the pointing.id if the healpix pixel is in the polygon query - for j,arr in enumerate(polygon_arr): - ras_poly = [x[0] for x in arr][:-1] - decs_poly = [x[1] for x in arr][:-1] - xyzpoly = astropy.coordinates.spherical_to_cartesian(1, np.deg2rad(decs_poly), np.deg2rad(ras_poly)) - qp = hp.query_polygon(skymap_nside,np.array(xyzpoly).T, nest=True) + ra_deg, dec_deg = zip(*[(coord_deg[0], coord_deg[1]) for coord_deg in ccd]) + list_o_coords = list(zip(ra_deg, dec_deg)) + ras_poly = [x[0] for x in list_o_coords][:-1] + decs_poly = [x[1] for x in list_o_coords][:-1] + xyzpoly = astropy.coordinates.spherical_to_cartesian( + 1, np.deg2rad(decs_poly), np.deg2rad(ras_poly)) + qp = hp.query_polygon(skymap_nside, np.array(xyzpoly).T, nest=True) if candidate_healpix in qp: return_ids.append(pid) - #so she goes - return [x for x in pointings if x.id in return_ids] \ No newline at end of file + + return [x for x in pointings if x.id in return_ids] diff --git a/src/gwtm_api/instrument.py b/src/gwtm_api/instrument.py index 99c279a..2c88eed 100644 --- a/src/gwtm_api/instrument.py +++ b/src/gwtm_api/instrument.py @@ -1,121 +1,74 @@ from __future__ import annotations import datetime -import json import hashlib +import json from typing import List import numpy as np from .pointing import Pointing - -from .core import baseapi from .core import apimodels -from .core import util +from .core import util from .core import tmcache +from .core.util import _encode_list_param +from .core import _client_factory as transport -#approximated instrument footprints are faster for computation APPROXIMATION_DICT = { - 47 : 76, #ZTF - 38 : 98, #DECAM + 47: 76, # ZTF + 38: 98, # DECAM } + class Footprint(apimodels._Table): id = None footprint = None polygon = None def __init__(self, kwdict=None, **kwargs): - - if kwdict is not None: - selfdict = kwdict - else: - selfdict = kwargs - + selfdict = kwdict if kwdict is not None else kwargs super().__init__(payload=selfdict) - self.sanatize_polygon() def __repr__(self) -> str: return str(self.polygon) - @staticmethod - def get(api_token: str, instrumentid: int, approximate_footprint: bool = True): - - api = baseapi.api(target="footprints") - - if approximate_footprint and instrumentid in APPROXIMATION_DICT.keys(): - inst_id = APPROXIMATION_DICT[instrumentid] - else: - inst_id = instrumentid - - get_dict = { - "id": inst_id, - "api_token": api_token - } - - r_json = { - "d_json": get_dict - } - - req = api._get(r_json=r_json) - request_json = json.loads(req.text) - inst_footprints = [] - for f in request_json: - if isinstance(f, str): - footprint_json = json.loads(f) - else: - footprint_json = f - inst_footprints.append(Footprint(kwdict=footprint_json)) - - return inst_footprints - + def get(instrumentid: int, approximate_footprint: bool = True, + api_token: str = None): + if api_token: + transport.configure(api_token=api_token) + inst_id = APPROXIMATION_DICT.get(instrumentid, instrumentid) if approximate_footprint else instrumentid + result = transport.get('/footprints', params={'id': inst_id}) + if not isinstance(result, list): + raise Exception("Error in Footprint.get(): unexpected response from server") + return [Footprint(kwdict=f) for f in result] def sanatize_polygon(self): sanitized = self.footprint.strip('POLYGON ').strip(')(').split(',') - polygon = [] - for vertex in sanitized: - obj = vertex.split() - ra = float(obj[0]) - dec = float(obj[1]) - polygon.append([ra,dec]) - self.polygon = polygon - + self.polygon = [[float(v) for v in vertex.split()] for vertex in sanitized] def project(self, ra: float, dec: float, pos_angle: float): if pos_angle is None: pos_angle = 0.0 - - footprint_zero_center_ra = np.asarray([pt[0] for pt in self.polygon]) - footprint_zero_center_dec = np.asarray([pt[1] for pt in self.polygon]) - footprint_zero_center_uvec = util.ra_dec_to_uvec(footprint_zero_center_ra, footprint_zero_center_dec) - footprint_zero_center_x, footprint_zero_center_y, footprint_zero_center_z = footprint_zero_center_uvec - - proj_footprint = [] - for idx in range(footprint_zero_center_x.shape[0]): - vec = np.asarray([footprint_zero_center_x[idx], footprint_zero_center_y[idx], footprint_zero_center_z[idx]]) - new_vec = vec @ util.x_rot(-pos_angle) @ util.y_rot(dec) @ util.z_rot(-ra) - new_x, new_y, new_z = new_vec.flat - pt_ra, pt_dec = util.uvec_to_ra_dec(new_x, new_y, new_z) - proj_footprint.append([round(pt_ra, 3), round(pt_dec, 3)]) - - return proj_footprint - + fp_ra = np.asarray([pt[0] for pt in self.polygon]) + fp_dec = np.asarray([pt[1] for pt in self.polygon]) + x, y, z = util.ra_dec_to_uvec(fp_ra, fp_dec) + proj = [] + for i in range(x.shape[0]): + vec = np.asarray([x[i], y[i], z[i]]) + nx, ny, nz = (vec @ util.x_rot(-pos_angle) @ util.y_rot(dec) @ util.z_rot(-ra)).flat + pt_ra, pt_dec = util.uvec_to_ra_dec(nx, ny, nz) + proj.append([round(pt_ra, 3), round(pt_dec, 3)]) + return proj @staticmethod - def get_cached_footprints(graceid: str = None, instrument_id: int = None, pointings: List[Pointing] = None): - pointingids = [x.id for x in pointings] - hashpointingids = hashlib.sha1(json.dumps(pointingids).encode()).hexdigest() - cache_name = f"footprints_{graceid}_{instrument_id}_{hashpointingids}" - cache = tmcache.TMCache(filename=cache_name, cache_type="json") - return cache.get() + def get_cached_footprints(graceid=None, instrument_id=None, pointings=None): + h = hashlib.sha1(json.dumps([x.id for x in pointings]).encode()).hexdigest() + return tmcache.TMCache(filename=f"footprints_{graceid}_{instrument_id}_{h}", cache_type="json").get() @staticmethod - def put_cached_footprints(footprints, graceid: str = None, instrument_id: int = None, pointings: List[Pointing] = None): - pointingids = [x.id for x in pointings] - hashpointingids = hashlib.sha1(json.dumps(pointingids).encode()).hexdigest() - cache_name = f"footprints_{graceid}_{instrument_id}_{hashpointingids}" - cache = tmcache.TMCache(filename=cache_name, cache_type="json") - cache.put(payload=footprints) + def put_cached_footprints(footprints, graceid=None, instrument_id=None, pointings=None): + h = hashlib.sha1(json.dumps([x.id for x in pointings]).encode()).hexdigest() + tmcache.TMCache(filename=f"footprints_{graceid}_{instrument_id}_{h}", cache_type="json").put(payload=footprints) class Instrument(apimodels._Table): @@ -123,73 +76,43 @@ class Instrument(apimodels._Table): instrument_name: str = None instrument_type: apimodels.instrument_type = None datecreated: datetime.datetime = None - submitterid: int = None, + submitterid: int = None nickname: str = None footprint: List[Footprint] = None - def __init__( - self, - id: int = None, - instrument_name: str = None, - instrument_type: apimodels.instrument_type = None, - datecreated: datetime.datetime = None, - submitterid: int = None, - nickname: str = None, - footprint: List[Footprint] = None, - kwdict=None - ): - - if kwdict is not None: - selfdict = kwdict - else: - selfdict = util.non_none_locals(locals=locals()) - + def __init__(self, id=None, instrument_name=None, instrument_type=None, + datecreated=None, submitterid=None, nickname=None, + footprint=None, kwdict=None): + selfdict = kwdict if kwdict is not None else util.non_none_locals(locals=locals()) super().__init__(payload=selfdict) - def project(self, ra: float, dec: float, pos_angle: float): if self.footprint is None: - raise Exception("Footprint Polygon is not included") - - proj_footprint = [] - for ccd in self.footprint: - proj_footprint.append(ccd.project(ra, dec, pos_angle)) - - return proj_footprint - + raise Exception("Footprint not included — pass include_footprint=True to Instrument.get()") + return [ccd.project(ra, dec, pos_angle) for ccd in self.footprint] @staticmethod - def get( - api_token: str, id: int = None, ids: List[int] = None, name: str = None, - names: List[str] = None, type: apimodels.instrument_type = None, - include_footprint=False, approximate_footprint=True, urlencode=False - ) -> List[Instrument]: - get_dict = util.non_none_locals(locals=locals()) - - r_json = { - "d_json":get_dict - } - - api = baseapi.api(target="instruments") - req = api._get(r_json=r_json, urlencode=urlencode) - - ret = [] - if req.status_code == 200: - request_json = json.loads(req.text) - for i in request_json: - if isinstance(i, str): - instrument_json = json.loads(i) - else: - instrument_json = i - ret.append(Instrument(kwdict=instrument_json)) - else: - raise Exception(f"Error in Instrument.get(). Request: {req.text[0:1000]}") - + def get(id: int = None, ids: List[int] = None, + name: str = None, names: List[str] = None, + type: apimodels.instrument_type = None, + include_footprint: bool = False, + approximate_footprint: bool = True, + urlencode=None, + api_token: str = None) -> List[Instrument]: + if api_token: + transport.configure(api_token=api_token) + result = transport.get('/instruments', params={ + 'ids': _encode_list_param(ids), + 'names': _encode_list_param(names), + }) + if not isinstance(result, list): + raise Exception("Error in Instrument.get(): unexpected response from server") + instruments = [Instrument(kwdict=i) for i in result] + if include_footprint: - for inst in ret: + for inst in instruments: inst.footprint = Footprint.get( - api_token=api_token, instrumentid=inst.id, approximate_footprint=approximate_footprint + instrumentid=inst.id, + approximate_footprint=approximate_footprint, ) - - return ret - \ No newline at end of file + return instruments diff --git a/src/gwtm_api/pointing.py b/src/gwtm_api/pointing.py index 2323cb7..b90430a 100644 --- a/src/gwtm_api/pointing.py +++ b/src/gwtm_api/pointing.py @@ -1,11 +1,18 @@ from __future__ import annotations -import json import datetime from typing import List -from .core import baseapi from .core import apimodels from .core import util +from .core.util import _encode_list_param +from .core import _client_factory as transport + + +def _enum_name(value): + if value is None: + return None + return value.name if hasattr(value, 'name') else value + class Pointing(apimodels._Table): id: int = None @@ -32,32 +39,34 @@ class Pointing(apimodels._Table): central_wave: float = None bandwidth: float = None - def __init__(self, kwdict=None, - id: int = None, - position: str = None, - ra: float = None, - dec: float = None, - instrumentid: int = None, - time: datetime.datetime = None, - status: apimodels.pointing_status = None, - depth: float = None, - depth_unit: apimodels.depth_unit = None, - band: apimodels.bandpass = None, - wavelength_regime: List[float] = None, - wavelength_unit: apimodels.wavelength_units = None, - energy_regime: List[float] = None, - energy_unit: apimodels.energy_units = None, - frequency_regime: List[float] = None, - frequency_unit: apimodels.frequency_units = None, - pos_angle: float = None, - depth_err: float = None, - doi_url: str = None, - doi_id: int = None, - submitterid: int = None, - central_wave: float = None, - bandwidth: float = None, - api_token: str = None - ): + def __init__(self, kwdict=None, + id: int = None, + position: str = None, + ra: float = None, + dec: float = None, + instrumentid: int = None, + time: datetime.datetime = None, + status: apimodels.pointing_status = None, + depth: float = None, + depth_unit: apimodels.depth_unit = None, + band: apimodels.bandpass = None, + wavelength_regime: List[float] = None, + wavelength_unit: apimodels.wavelength_units = None, + energy_regime: List[float] = None, + energy_unit: apimodels.energy_units = None, + frequency_regime: List[float] = None, + frequency_unit: apimodels.frequency_units = None, + pos_angle: float = None, + depth_err: float = None, + doi_url: str = None, + doi_id: int = None, + submitterid: int = None, + central_wave: float = None, + bandwidth: float = None, + api_token: str = None, + ): + if api_token: + transport.configure(api_token=api_token) self.position = position self.time = time @@ -67,26 +76,21 @@ def __init__(self, kwdict=None, else: selfdict = util.non_none_locals(locals=locals()) - #if user passes in an ID, then prepopulate with a get request if id: - if api_token: - payload = Pointing.get(api_token=api_token, id=id) - if len(payload): - selfdict = payload[0].__dict__ - else: - raise Exception(f"No candidate found with id: {id}") + payload = Pointing.get(id=id) + if payload: + selfdict = payload[0].__dict__ else: - raise Exception("api_token required") - + raise Exception(f"No pointing found with id: {id}") + super().__init__(payload=selfdict) - if "position" in selfdict.keys(): + if "position" in selfdict: self._sanatize_pointing() - elif all([x in selfdict.keys() for x in ["ra", "dec"]]): + elif "ra" in selfdict and "dec" in selfdict: self._sanatize_ra_dec() else: - raise Exception("Positional arguments are required. Must be decimal format ra, dec, or geometry type \"POINT(ra dec)\"") - + raise Exception("Positional arguments are required: ra/dec or POINT(ra dec)") def _sanatize_pointing(self): if self.position is not None: @@ -94,141 +98,130 @@ def _sanatize_pointing(self): self.ra = float(self.position.split('(')[1].split(')')[0].split()[0]) self.dec = float(self.position.split('(')[1].split(')')[0].split()[1]) except: # noqa: E722 - raise Exception("Invalid position argument. Must be 'POINT (\{ra\} \{dec\})'.") - + raise Exception("Invalid position. Must be 'POINT ({ra} {dec})'.") def _sanatize_ra_dec(self): if self.position is None: - if all([isinstance(x, float) for x in [self.ra, self.dec]]): + if all(isinstance(x, float) for x in [self.ra, self.dec]): self.position = f"POINT ({self.ra} {self.dec})" else: - raise Exception("Invalid format for ra/dec. Both must be float") - - - def post( - self, api_token: str, graceid: str, request_doi: bool = None, - doi_url: str = None, creators: List[dict] = None, doi_group_id: int = None, - base: str = "https://treasuremap.space/api/", api_version: str ="v1", - verbose: bool = False - ): - - post_dict = util.non_none_locals(locals=locals()) - - self.time = self.time.strftime("%Y-%m-%dT%H:%M:%S.%f") - post_dict["pointings"]=[self.__dict__] - - r_json = { - "d_json":post_dict + raise Exception("ra and dec must be floats") + + def _to_dict(self) -> dict: + return { + 'ra': self.ra, + 'dec': self.dec, + 'instrumentid': self.instrumentid, + 'time': self.time.isoformat() if isinstance(self.time, datetime.datetime) else None, + 'status': _enum_name(self.status) or 'planned', + 'depth': self.depth, + 'depth_unit': _enum_name(self.depth_unit), + 'band': _enum_name(self.band), + 'pos_angle': self.pos_angle, + 'depth_err': self.depth_err, + 'central_wave': self.central_wave, + 'bandwidth': self.bandwidth, } - api = baseapi.api(target="pointings", base=base) - req = api._post(r_json=r_json) - - if req.status_code == 200: - request_json = json.loads(req.text) - if verbose: - print(request_json) - id = request_json["pointing_ids"][0] - self.__init__(kwdict=Pointing.get(api_token=api_token, id=id)[0].__dict__) - else: - raise Exception(f"Error in Pointing.post(). Request: {req.text[0:1000]}") - + def post(self, graceid: str, request_doi: bool = None, + doi_url: str = None, creators: List[dict] = None, + doi_group_id: int = None, verbose: bool = False, + api_token: str = None): + if api_token: + transport.configure(api_token=api_token) + result = transport.post('/pointings', json={'graceid': graceid, 'pointing': self._to_dict()}) + if verbose: + print(result) + fetched = Pointing.get(id=result['pointing_ids'][0]) + self.__init__(kwdict=fetched[0].__dict__) @staticmethod - def batch_post( - api_token: str, graceid: str, pointings: List[Pointing], request_doi: bool = False, - doi_url: str = None, creators: List[dict] = None, doi_group_id: int = None, - base: str = "https://treasuremap.space/api/", api_version="v1", - verbose: bool = False - ) -> List[Pointing]: - - post_dict = util.non_none_locals(locals=locals()) - - batch_pointings = [] - for p in pointings: - if not isinstance(p, Pointing): - raise Exception("Input \'pointings\' must be type List[Pointing]") - p.time = p.time.strftime("%Y-%m-%dT%H:%M:%S.%f") - batch_pointings.append(p.__dict__) - - post_dict['pointings'] = batch_pointings - - r_json = { - "d_json":post_dict - } - - api = baseapi.api(target="pointings", base=base, api_version=api_version) - req = api._post(r_json=r_json) - - if req.status_code == 200: - request_json = json.loads(req.text) - ids = request_json["pointing_ids"] - if verbose: - print(request_json) - return Pointing.get(api_token=api_token, ids=ids) - else: - raise Exception(f"Error in Pointing.post(). Request: {req.text[0:1000]}") - + def batch_post(graceid: str, pointings: List[Pointing], + request_doi: bool = False, doi_url: str = None, + creators: List[dict] = None, doi_group_id: int = None, + verbose: bool = False, api_token: str = None) -> List[Pointing]: + if api_token: + transport.configure(api_token=api_token) + result = transport.post('/pointings', json={ + 'graceid': graceid, + 'pointings': [p._to_dict() for p in pointings], + }) + if verbose: + print(result) + return Pointing.get(ids=result['pointing_ids']) @staticmethod - def get( - api_token: str, graceid: str = None, graceids: List[str] = None, instrument: str = None, - instruments: List[str] = None, id: int = None, ids: List[int] = None, status: apimodels.pointing_status = None, - completed_after: datetime.datetime = None, completed_before: datetime.datetime = None, - planned_after: datetime.datetime = None, planned_before: datetime.datetime = None, - user: str = None, users: List[str] = None, band: apimodels.bandpass = None, - bands: List[apimodels.bandpass] = None, central_wave: float = None, bandwidth: float = None, - wavelength_regime: List[float] = None, wavelength_unit: apimodels.wavelength_units = None, - energy_regime: List[float] = None, energy_unit: apimodels.energy_units = None, - frequency_regime: List[float] = None, frequency_unit: apimodels.frequency_units = None, - depth_gt: float = None, depth_lt: float = None, depth_unit: apimodels.depth_unit = None, - base: str = "https://treasuremap.space/api/", api_version: str ="v1", urlencode: bool = False, + def get(graceid: str = None, + graceids: List[str] = None, + instrument: str = None, + instruments: List[str] = None, + id: int = None, + ids: List[int] = None, + status: apimodels.pointing_status = None, + completed_after: datetime.datetime = None, + completed_before: datetime.datetime = None, + planned_after: datetime.datetime = None, + planned_before: datetime.datetime = None, + user: str = None, + users: List[str] = None, + band: apimodels.bandpass = None, + bands: List[apimodels.bandpass] = None, + central_wave: float = None, + bandwidth: float = None, + wavelength_regime: List[float] = None, + wavelength_unit: apimodels.wavelength_units = None, + energy_regime: List[float] = None, + energy_unit: apimodels.energy_units = None, + frequency_regime: List[float] = None, + frequency_unit: apimodels.frequency_units = None, + depth_gt: float = None, + depth_lt: float = None, + depth_unit: apimodels.depth_unit = None, + urlencode=None, + api_token: str = None, ) -> List[Pointing]: + if api_token: + transport.configure(api_token=api_token) + result = transport.get('/pointings', params={ + 'graceid': graceid, + 'graceids': _encode_list_param(graceids), + 'id': id, + 'ids': _encode_list_param(ids), + 'status': _enum_name(status), + 'instrument': instrument, + 'instruments': _encode_list_param(instruments), + 'completed_after': completed_after, + 'completed_before': completed_before, + 'planned_after': planned_after, + 'planned_before': planned_before, + 'user': user, + 'users': _encode_list_param(users), + 'band': _enum_name(band), + 'bands': _encode_list_param([_enum_name(b) for b in bands] if bands else None), + 'depth_gt': depth_gt, + 'depth_lt': depth_lt, + 'depth_unit': _enum_name(depth_unit), + 'wavelength_regime': _encode_list_param(wavelength_regime), + 'wavelength_unit': _enum_name(wavelength_unit), + 'energy_regime': _encode_list_param(energy_regime), + 'energy_unit': _enum_name(energy_unit), + 'frequency_regime': _encode_list_param(frequency_regime), + 'frequency_unit': _enum_name(frequency_unit), + 'central_wave': central_wave, + 'bandwidth': bandwidth, + }) + return [Pointing(kwdict=p) for p in result] - get_dict = util.non_none_locals(locals=locals()) - - r_json = { - "d_json":get_dict - } - - api = baseapi.api(target="pointings", base=base, api_version=api_version) - req = api._get(r_json=r_json, urlencode=urlencode) - - if req.status_code == 200: - ret: List[Pointing] = [] - request_json = json.loads(req.text) - for p in request_json: - if 'v0' in api.base: - pointing_json = json.loads(p) - else: - pointing_json = p - ret.append(Pointing(kwdict=pointing_json)) - return ret - else: - raise Exception(f"Error in Pointing.get(). Request: {req.text[0:1000]}") - - @staticmethod - def request_doi( - api_token: str, graceid: str = None, id: int = None, ids: List[int] = None, creators: dict = None, - doi_group_id: int = None, base: str = "https://treasuremap.space/api/", api_version: str ="v1" - ) -> str: - if all([x is None for x in [graceid, id, ids]]): - raise Exception("Invalid parameters to request DOI. Must specify \'graceid\', pointing \'id\', or list of \'ids\'") - - get_dict = util.non_none_locals(locals=locals()) - - r_json = { - "d_json":get_dict - } - - api = baseapi.api(target="request_doi", base=base, api_version=api_version) - req = api._post(r_json=r_json) - - if req.status_code == 200: - request_json = json.loads(req.text) - return request_json["DOI URL"] - else: - raise Exception(f"Error in Pointing.get(). Request: {req.text[0:1000]}") - - return "" \ No newline at end of file + def request_doi(graceid: str = None, id: int = None, + ids: List[int] = None, creators: dict = None, + doi_group_id: int = None, api_token: str = None) -> str: + if api_token: + transport.configure(api_token=api_token) + if all(x is None for x in [graceid, id, ids]): + raise Exception("Must specify 'graceid', 'id', or 'ids'") + result = transport.post('/request_doi', json={ + 'graceid': graceid, 'id': id, 'ids': ids, + 'creators': creators, 'doi_group_id': doi_group_id, + }) + return result.get('doi_url', str(result)) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py new file mode 100644 index 0000000..2b586fd --- /dev/null +++ b/tests/unit/conftest.py @@ -0,0 +1,17 @@ +import pytest +import sys +sys.path.insert(0, 'src') + +import gwtm_api +import gwtm_api.core._client_factory as transport + + +@pytest.fixture(autouse=True) +def reset_transport(): + """Reset transport state before each test and configure a default token.""" + transport._token = '' + transport._base_url = transport.DEFAULT_BASE_URL + gwtm_api.configure(api_token='tok') + yield + transport._token = '' + transport._base_url = transport.DEFAULT_BASE_URL diff --git a/tests/unit/test_alert.py b/tests/unit/test_alert.py new file mode 100644 index 0000000..12b6d38 --- /dev/null +++ b/tests/unit/test_alert.py @@ -0,0 +1,60 @@ +"""Unit tests for Alert wrapper — no live server required.""" +from unittest.mock import patch, MagicMock + +import pytest + +import sys +sys.path.insert(0, 'src') + +import gwtm_api + +ALERT_FIXTURE = { + 'id': 1, 'graceid': 'S190425z', 'role': 'observation', + 'alert_type': 'Initial', 'far': 4.5e-13, + 'timesent': '2019-04-25T08:18:05', +} + +GET_PATH = 'gwtm_api.core._client_factory.get' + + +class TestAlertGet: + + def test_get_returns_first_alert(self): + with patch(GET_PATH, return_value=[ALERT_FIXTURE, {**ALERT_FIXTURE, 'id': 2}]): + result = gwtm_api.Alert.get(graceid='S190425z') + assert isinstance(result, gwtm_api.Alert) + assert result.id == 1 + + def test_get_all_returns_list(self): + with patch(GET_PATH, return_value=[ALERT_FIXTURE, {**ALERT_FIXTURE, 'id': 2}]): + result = gwtm_api.Alert.get_all(graceid='S190425z') + assert len(result) == 2 + assert all(isinstance(a, gwtm_api.Alert) for a in result) + + def test_alert_fields_populated(self): + with patch(GET_PATH, return_value=[ALERT_FIXTURE]): + result = gwtm_api.Alert.get_all(graceid='S190425z') + assert result[0].graceid == 'S190425z' + assert result[0].far == pytest.approx(4.5e-13) + + def test_auth_header_sent_correctly(self): + gwtm_api.configure(api_token='my_token') + mock_resp = MagicMock() + mock_resp.json.return_value = [] + mock_resp.raise_for_status.return_value = None + with patch('httpx.get', return_value=mock_resp) as mock: + gwtm_api.Alert.get_all(graceid='S190425z') + assert mock.call_args.kwargs['headers']['api_token'] == 'my_token' + + def test_get_raises_when_no_alerts(self): + with patch(GET_PATH, return_value=[]): + with pytest.raises(Exception, match="No alert found"): + gwtm_api.Alert.get(graceid='NONEXISTENT') + + def test_correct_endpoint_called(self): + with patch(GET_PATH, return_value=[]) as mock: + try: + gwtm_api.Alert.get(graceid='S190425z') + except Exception: + pass + assert mock.call_args.args[0] == '/query_alerts' diff --git a/tests/unit/test_instrument.py b/tests/unit/test_instrument.py new file mode 100644 index 0000000..aca0579 --- /dev/null +++ b/tests/unit/test_instrument.py @@ -0,0 +1,73 @@ +"""Unit tests for Instrument and Footprint wrappers — no live server required.""" +from unittest.mock import patch + +import sys +sys.path.insert(0, 'src') + +import gwtm_api + +INSTRUMENT_FIXTURE = {'id': 10, 'instrument_name': 'ZTF', 'instrument_type': 1} +FOOTPRINT_FIXTURE = {'id': 1, 'instrumentid': 10, 'footprint': 'POLYGON (1.0 2.0,3.0 4.0,5.0 6.0,1.0 2.0)'} + +GET_PATH = 'gwtm_api.core._client_factory.get' + + +class TestInstrumentGet: + + def test_returns_list_of_instruments(self): + with patch(GET_PATH, return_value=[INSTRUMENT_FIXTURE]): + result = gwtm_api.Instrument.get() + assert len(result) == 1 + assert isinstance(result[0], gwtm_api.Instrument) + assert result[0].instrument_name == 'ZTF' + + def test_names_list_is_json_encoded(self): + with patch(GET_PATH, return_value=[]) as mock: + gwtm_api.Instrument.get(names=['ZTF', 'DECam']) + assert mock.call_args.kwargs['params']['names'] == '["ZTF", "DECam"]' + + def test_ids_list_is_json_encoded(self): + with patch(GET_PATH, return_value=[]) as mock: + gwtm_api.Instrument.get(ids=[10, 38]) + assert mock.call_args.kwargs['params']['ids'] == '[10, 38]' + + def test_include_footprint_fetches_footprints(self): + with patch(GET_PATH, side_effect=[[INSTRUMENT_FIXTURE], [FOOTPRINT_FIXTURE]]): + result = gwtm_api.Instrument.get(include_footprint=True) + assert result[0].footprint is not None + assert len(result[0].footprint) == 1 + + def test_auth_header_sent_correctly(self): + from unittest.mock import MagicMock + gwtm_api.configure(api_token='my_token') + mock_resp = MagicMock() + mock_resp.json.return_value = [] + mock_resp.raise_for_status.return_value = None + with patch('httpx.get', return_value=mock_resp) as mock: + gwtm_api.Instrument.get() + assert mock.call_args.kwargs['headers']['api_token'] == 'my_token' + + def test_approximate_footprint_uses_substitution(self): + from gwtm_api.instrument import APPROXIMATION_DICT + ztf_instrument = {**INSTRUMENT_FIXTURE, 'id': 47} + ztf_footprint = {**FOOTPRINT_FIXTURE, 'instrumentid': 47} + + calls = [] + def fake_get(path, params=None): + calls.append((path, params)) + if path == '/instruments': + return [ztf_instrument] + if path == '/footprints': + return [ztf_footprint] + return [] + + with patch(GET_PATH, side_effect=fake_get): + gwtm_api.Instrument.get(include_footprint=True) + + footprint_call = next(c for c in calls if c[0] == '/footprints') + assert footprint_call[1]['id'] == APPROXIMATION_DICT[47] + + def test_correct_endpoint_called(self): + with patch(GET_PATH, return_value=[]) as mock: + gwtm_api.Instrument.get() + assert mock.call_args.args[0] == '/instruments' diff --git a/tests/unit/test_pointing.py b/tests/unit/test_pointing.py new file mode 100644 index 0000000..5ddebdb --- /dev/null +++ b/tests/unit/test_pointing.py @@ -0,0 +1,125 @@ +"""Unit tests for Pointing wrapper — no live server required.""" +import datetime +from unittest.mock import patch, MagicMock + +import pytest + +import sys +sys.path.insert(0, 'src') + +import gwtm_api + +POINTING_FIXTURE = { + 'id': 42, 'position': 'POINT (10.0 -20.0)', 'instrumentid': 10, + 'time': '2025-01-01T00:00:00', 'status': 'completed', + 'depth': 18.5, 'depth_unit': 'ab_mag', 'band': 'r', + 'submitterid': 1, +} + +GET_PATH = 'gwtm_api.core._client_factory.get' +POST_PATH = 'gwtm_api.core._client_factory.post' + + +class TestPointingGet: + + def test_returns_list_of_pointing_objects(self): + with patch(GET_PATH, return_value=[POINTING_FIXTURE]): + result = gwtm_api.Pointing.get(graceid='S190425z') + assert len(result) == 1 + assert isinstance(result[0], gwtm_api.Pointing) + assert result[0].id == 42 + assert result[0].ra == 10.0 + assert result[0].dec == -20.0 + + def test_list_params_are_json_encoded(self): + with patch(GET_PATH, return_value=[]) as mock: + gwtm_api.Pointing.get(instruments=['ZTF', 'DECam']) + params = mock.call_args.kwargs['params'] + assert params['instruments'] == '["ZTF", "DECam"]' + + def test_ids_list_is_json_encoded(self): + with patch(GET_PATH, return_value=[]) as mock: + gwtm_api.Pointing.get(ids=[1, 2, 3]) + assert mock.call_args.kwargs['params']['ids'] == '[1, 2, 3]' + + def test_enum_status_converted_to_string(self): + import gwtm_api.core.apimodels as m + with patch(GET_PATH, return_value=[]) as mock: + gwtm_api.Pointing.get(status=m.pointing_status.completed) + assert mock.call_args.kwargs['params']['status'] == 'completed' + + def test_urlencode_param_silently_ignored(self): + with patch(GET_PATH, return_value=[POINTING_FIXTURE]): + result = gwtm_api.Pointing.get(graceid='S190425z', urlencode=True) + assert len(result) == 1 + + def test_auth_header_sent_correctly(self): + gwtm_api.configure(api_token='my_secret_token') + mock_resp = MagicMock() + mock_resp.json.return_value = [] + mock_resp.raise_for_status.return_value = None + with patch('httpx.get', return_value=mock_resp) as mock: + gwtm_api.Pointing.get() + assert mock.call_args.kwargs['headers']['api_token'] == 'my_secret_token' + assert 'Bearer' not in str(mock.call_args.kwargs['headers']) + + def test_none_params_not_sent(self): + with patch(GET_PATH, return_value=[]) as mock: + gwtm_api.Pointing.get(graceid='S190425z') + # _clean() strips None values before httpx sees them; instruments was not passed + params = mock.call_args.kwargs['params'] + assert params.get('instruments') is None + + def test_bands_list_of_enums_converted(self): + import gwtm_api.core.apimodels as m + with patch(GET_PATH, return_value=[]) as mock: + gwtm_api.Pointing.get(bands=[m.bandpass.r, m.bandpass.g]) + assert mock.call_args.kwargs['params']['bands'] == '["r", "g"]' + + def test_correct_endpoint_called(self): + with patch(GET_PATH, return_value=[]) as mock: + gwtm_api.Pointing.get(graceid='S190425z') + assert mock.call_args.args[0] == '/pointings' + + +class TestBackwardsCompat: + + def test_api_token_kwarg_configures_client(self): + import gwtm_api.core._client_factory as t + with patch(GET_PATH, return_value=[]): + gwtm_api.Pointing.get(graceid='S190425z', api_token='legacy_token') + assert t._token == 'legacy_token' + + def test_api_token_on_post_configures_client(self): + import gwtm_api.core._client_factory as t + with patch(POST_PATH, return_value={'pointing_ids': [42]}), \ + patch(GET_PATH, return_value=[POINTING_FIXTURE]): + p = gwtm_api.Pointing(ra=10.0, dec=-20.0, instrumentid=10, + time=datetime.datetime(2025, 1, 1), + status='planned', depth=18.0, + depth_unit='ab_mag', band='r') + p.post(graceid='S190425z', api_token='legacy_token') + assert t._token == 'legacy_token' + + +class TestPointingPost: + + def test_post_calls_add_endpoint(self): + with patch(POST_PATH, return_value={'pointing_ids': [99]}), \ + patch(GET_PATH, return_value=[{**POINTING_FIXTURE, 'id': 99}]): + p = gwtm_api.Pointing(ra=10.0, dec=-20.0, instrumentid=10, + time=datetime.datetime(2025, 1, 1), + status='planned', depth=18.0, + depth_unit='ab_mag', band='r') + p.post(graceid='S190425z') + assert p.id == 99 + + def test_post_correct_endpoint(self): + with patch(POST_PATH, return_value={'pointing_ids': [42]}), \ + patch(GET_PATH, return_value=[POINTING_FIXTURE]) as mock_get: + p = gwtm_api.Pointing(ra=10.0, dec=-20.0, instrumentid=10, + time=datetime.datetime(2025, 1, 1), + status='planned', depth=18.0, + depth_unit='ab_mag', band='r') + p.post(graceid='S190425z') + assert mock_get.call_args.args[0] == '/pointings' diff --git a/tests/unit/test_transport.py b/tests/unit/test_transport.py new file mode 100644 index 0000000..8ee8154 --- /dev/null +++ b/tests/unit/test_transport.py @@ -0,0 +1,73 @@ +"""Unit tests for configure() and the list encoding utility.""" +from unittest.mock import patch, MagicMock +import sys +sys.path.insert(0, 'src') + +import gwtm_api +import gwtm_api.core._client_factory as transport +from gwtm_api.core.util import _encode_list_param + + +class TestEncodeListParam: + + def test_list_becomes_json_string(self): + assert _encode_list_param(['ZTF', 'DECam']) == '["ZTF", "DECam"]' + + def test_int_list_encoded(self): + assert _encode_list_param([1, 2, 3]) == '[1, 2, 3]' + + def test_none_returns_none(self): + assert _encode_list_param(None) is None + + def test_string_passed_through(self): + assert _encode_list_param('ZTF') == 'ZTF' + + def test_single_item_list(self): + assert _encode_list_param(['ZTF']) == '["ZTF"]' + + +class TestConfigure: + + def test_sets_token(self): + gwtm_api.configure(api_token='my_token') + assert transport._token == 'my_token' + + def test_default_base_url(self): + gwtm_api.configure(api_token='tok') + assert transport._base_url == 'https://treasuremap.space/api/v1' + + def test_custom_base_url(self): + gwtm_api.configure(api_token='tok', base_url='https://dev.treasuremap.space/api/v1') + assert transport._base_url == 'https://dev.treasuremap.space/api/v1' + + def test_trailing_slash_stripped(self): + gwtm_api.configure(api_token='tok', base_url='https://treasuremap.space/api/v1/') + assert transport._base_url == 'https://treasuremap.space/api/v1' + + def test_reconfigure_updates_token(self): + gwtm_api.configure(api_token='first') + gwtm_api.configure(api_token='second') + assert transport._token == 'second' + + def test_headers_include_api_token(self): + gwtm_api.configure(api_token='secret') + assert transport._headers() == {'api_token': 'secret'} + + def test_get_passes_auth_header(self): + gwtm_api.configure(api_token='tok123') + mock_resp = MagicMock() + mock_resp.json.return_value = [] + mock_resp.raise_for_status.return_value = None + with patch('httpx.get', return_value=mock_resp) as mock: + transport.get('/pointings') + assert mock.call_args.kwargs['headers']['api_token'] == 'tok123' + + def test_get_strips_none_params(self): + mock_resp = MagicMock() + mock_resp.json.return_value = [] + mock_resp.raise_for_status.return_value = None + with patch('httpx.get', return_value=mock_resp) as mock: + transport.get('/pointings', params={'graceid': 'S190425z', 'status': None}) + sent_params = mock.call_args.kwargs['params'] + assert 'status' not in sent_params + assert sent_params['graceid'] == 'S190425z' From a810693fd0a94d98bdcb5b933f79f127d14318f5 Mon Sep 17 00:00:00 2001 From: Steve Foale Date: Wed, 6 May 2026 16:05:04 +0100 Subject: [PATCH 2/5] Fix remaining issues and add more tests. --- .github/workflows/ci.yml | 4 +- README.md | 11 +++- pyproject.toml | 12 ++-- src/gwtm_api/alert.py | 6 +- src/gwtm_api/candidate.py | 1 + src/gwtm_api/instrument.py | 3 + tests/unit/test_candidate.py | 117 +++++++++++++++++++++++++++++++++++ 7 files changed, 140 insertions(+), 14 deletions(-) create mode 100644 tests/unit/test_candidate.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4145c4e..c4b40c2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -16,9 +16,7 @@ jobs: python-version: "3.11" - name: Install dependencies - run: | - pip install -e ".[dev]" - ./scripts/generate_client.sh + run: pip install -e ".[dev]" - name: Run unit tests run: pytest tests/unit/ -v diff --git a/README.md b/README.md index b7d98e6..90f4fb9 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,13 @@ To point at a different server (e.g. a development instance), pass `base_url`: gwtm_api.configure(api_token='YOUR_API_TOKEN', base_url='https://dev.treasuremap.space') ``` +Passing `api_token` directly to any method also works and is provided for backwards compatibility with older code. Note that doing so permanently updates the configured token — subsequent calls without an explicit `api_token` will reuse it: + +```python +# Legacy style — still works +pointings = gwtm_api.Pointing.get(api_token='YOUR_API_TOKEN', graceid='GW190814') +``` + You can also set credentials via environment variables and call `configure()` with no arguments: ```bash @@ -262,10 +269,10 @@ covering_pointings = gwtm_api.event_tools.candidate_coverage(candidate=my_candid ## For contributors -The package uses a two-layer architecture: +The package has a simple two-layer architecture: - **`src/gwtm_api/{pointing,alert,instrument,candidate}.py`** — hand-written, user-facing classes with a stable, friendly interface -- **`src/gwtm_api/_generated/`** — HTTP transport layer auto-generated from the server's OpenAPI spec (not committed to the repo) +- **`src/gwtm_api/core/_client_factory.py`** — thin httpx transport layer; configured once via `gwtm_api.configure()` ### Setup after cloning diff --git a/pyproject.toml b/pyproject.toml index 18f3e64..6d767cd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,18 +21,18 @@ dependencies = [ "pandas" ] -[project.optional-dependencies] -dev = [ - "pytest", - "pytest-mock" -] - classifiers = [ "Programming Language :: Python :: 3", "License :: OSI Approved :: MIT License", "Operating System :: OS Independent", ] +[project.optional-dependencies] +dev = [ + "pytest", + "pytest-mock" +] + [project.urls] "Homepage" = "https://github.com/TheTreasureMap/gwtm_api" "Bug Tracker" = "https://github.com/TheTreasureMap/gwtm_api/issues" diff --git a/src/gwtm_api/alert.py b/src/gwtm_api/alert.py index f610c5b..0d0eb5e 100644 --- a/src/gwtm_api/alert.py +++ b/src/gwtm_api/alert.py @@ -18,7 +18,7 @@ class Alert(apimodels._Table): packet_type: int = None alert_type: str = None detectors: str = None - description: float = None + description: str = None far: float = None skymap_fits_url: str = None distance: float = None @@ -59,7 +59,7 @@ def __init__(self, packet_type: int = None, alert_type: str = None, detectors: str = None, - description: float = None, + description: str = None, far: float = None, skymap_fits_url: str = None, distance: float = None, @@ -105,7 +105,7 @@ def get(id: int = None, graceid: str = None, api_token: str = None) -> "Alert": def get_all(id: int = None, graceid: str = None, api_token: str = None): if api_token: transport.configure(api_token=api_token) - result = transport.get('/query_alerts', params={'graceid': graceid}) + result = transport.get('/query_alerts', params={'id': id, 'graceid': graceid}) if not isinstance(result, list): raise Exception("Error in Alert.get_all(): unexpected response from server") return [Alert(kwdict=a) for a in result] diff --git a/src/gwtm_api/candidate.py b/src/gwtm_api/candidate.py index 3a2becf..fce5881 100644 --- a/src/gwtm_api/candidate.py +++ b/src/gwtm_api/candidate.py @@ -175,6 +175,7 @@ def get(graceid: str = None, id: int = None, ids: List[int] = None, result = transport.get('/candidate', params={ 'graceid': graceid, 'id': id, + 'ids': _encode_list_param(ids), 'userid': userid, 'submitted_date_after': submitted_date_after, 'submitted_date_before': submitted_date_before, diff --git a/src/gwtm_api/instrument.py b/src/gwtm_api/instrument.py index 2c88eed..8a9d826 100644 --- a/src/gwtm_api/instrument.py +++ b/src/gwtm_api/instrument.py @@ -102,8 +102,11 @@ def get(id: int = None, ids: List[int] = None, if api_token: transport.configure(api_token=api_token) result = transport.get('/instruments', params={ + 'id': id, 'ids': _encode_list_param(ids), + 'name': name, 'names': _encode_list_param(names), + 'type': type, }) if not isinstance(result, list): raise Exception("Error in Instrument.get(): unexpected response from server") diff --git a/tests/unit/test_candidate.py b/tests/unit/test_candidate.py new file mode 100644 index 0000000..231e898 --- /dev/null +++ b/tests/unit/test_candidate.py @@ -0,0 +1,117 @@ +"""Unit tests for Candidate wrapper — no live server required.""" +import datetime +from unittest.mock import patch + +import pytest + +import sys +sys.path.insert(0, 'src') + +import gwtm_api + +CANDIDATE_FIXTURE = { + 'id': 1, 'graceid': 'S190425z', + 'position': 'POINT (15.0 -24.0)', + 'candidate_name': 'AT2019abc', + 'discovery_date': '2019-04-25T08:00:00', + 'discovery_magnitude': 18.5, + 'magnitude_unit': 'ab_mag', +} + +GET_PATH = 'gwtm_api.core._client_factory.get' +POST_PATH = 'gwtm_api.core._client_factory.post' +PUT_PATH = 'gwtm_api.core._client_factory.put' +DELETE_PATH = 'gwtm_api.core._client_factory.delete' + + +class TestCandidateGet: + + def test_returns_list_of_candidates(self): + with patch(GET_PATH, return_value=[CANDIDATE_FIXTURE]): + result = gwtm_api.Candidate.get(graceid='S190425z') + assert len(result) == 1 + assert isinstance(result[0], gwtm_api.Candidate) + assert result[0].id == 1 + assert result[0].ra == 15.0 + assert result[0].dec == -24.0 + + def test_ids_list_sent_to_server(self): + with patch(GET_PATH, return_value=[]) as mock: + gwtm_api.Candidate.get(ids=[1, 2, 3]) + assert mock.call_args.kwargs['params']['ids'] == '[1, 2, 3]' + + def test_single_id_sent_to_server(self): + with patch(GET_PATH, return_value=[]) as mock: + gwtm_api.Candidate.get(id=42) + assert mock.call_args.kwargs['params']['id'] == 42 + + def test_correct_endpoint_called(self): + with patch(GET_PATH, return_value=[]) as mock: + gwtm_api.Candidate.get(graceid='S190425z') + assert mock.call_args.args[0] == '/candidate' + + def test_backwards_compat_api_token(self): + import gwtm_api.core._client_factory as t + with patch(GET_PATH, return_value=[]): + gwtm_api.Candidate.get(graceid='S190425z', api_token='legacy') + assert t._token == 'legacy' + + +class TestCandidatePost: + + def test_post_calls_correct_endpoint(self): + with patch(POST_PATH, return_value={'candidate_ids': [1]}), \ + patch(GET_PATH, return_value=[CANDIDATE_FIXTURE]) as mock_get: + c = gwtm_api.Candidate( + ra=15.0, dec=-24.0, candidate_name='AT2019abc', + discovery_date=datetime.datetime(2019, 4, 25), + discovery_magnitude=18.5, magnitude_unit='ab_mag', + ) + c.post(graceid='S190425z') + assert mock_get.call_args.args[0] == '/candidate' + + def test_batch_post_sends_ids_to_get(self): + with patch(POST_PATH, return_value={'candidate_ids': [1, 2]}), \ + patch(GET_PATH, return_value=[CANDIDATE_FIXTURE, {**CANDIDATE_FIXTURE, 'id': 2}]) as mock_get: + candidates = [ + gwtm_api.Candidate(ra=15.0, dec=-24.0, candidate_name='AT2019abc', + discovery_date=datetime.datetime(2019, 4, 25), + discovery_magnitude=18.5, magnitude_unit='ab_mag'), + gwtm_api.Candidate(ra=16.0, dec=-25.0, candidate_name='AT2019def', + discovery_date=datetime.datetime(2019, 4, 25), + discovery_magnitude=19.0, magnitude_unit='ab_mag'), + ] + gwtm_api.Candidate.batch_post(candidates=candidates, graceid='S190425z') + assert mock_get.call_args.kwargs['params']['ids'] == '[1, 2]' + + +class TestCandidatePutDelete: + + def test_put_calls_correct_endpoint(self): + c = gwtm_api.Candidate(kwdict={**CANDIDATE_FIXTURE}) + with patch(PUT_PATH, return_value={}), \ + patch(GET_PATH, return_value=[CANDIDATE_FIXTURE]): + c.put(payload={'tns_name': 'AT2019abc'}) + # no exception = correct endpoint reached + + def test_delete_calls_correct_endpoint(self): + c = gwtm_api.Candidate(kwdict={**CANDIDATE_FIXTURE}) + with patch(DELETE_PATH, return_value={}) as mock: + c.delete() + assert mock.call_args.args[0] == '/candidate' + assert mock.call_args.kwargs['json']['id'] == 1 + + def test_batch_delete_sends_ids(self): + with patch(DELETE_PATH, return_value={}) as mock: + gwtm_api.Candidate.batch_delete(ids=[1, 2, 3]) + assert mock.call_args.kwargs['json']['ids'] == [1, 2, 3] + + def test_delete_raises_without_id(self): + c = gwtm_api.Candidate(kwdict={**CANDIDATE_FIXTURE, 'id': None}) + with pytest.raises(Exception, match="does not have an id"): + c.delete() + + def test_put_raises_without_id(self): + c = gwtm_api.Candidate(kwdict={**CANDIDATE_FIXTURE, 'id': None}) + with pytest.raises(Exception, match="does not have an id"): + c.put() From 8b3f8c35645be46216d83df5c2e5728a44cec558 Mon Sep 17 00:00:00 2001 From: Steve Foale Date: Tue, 12 May 2026 12:10:20 +0100 Subject: [PATCH 3/5] Fix copilot PR comments --- README.md | 2 +- src/gwtm_api/candidate.py | 2 +- src/gwtm_api/core/_client_factory.py | 2 ++ src/gwtm_api/event_tools.py | 43 +++++++++++++++++++++------- src/gwtm_api/instrument.py | 2 +- src/gwtm_api/pointing.py | 13 ++++++++- tests/unit/test_instrument.py | 11 +++++++ 7 files changed, 60 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 90f4fb9..903d76b 100644 --- a/README.md +++ b/README.md @@ -49,7 +49,7 @@ You can also set credentials via environment variables and call `configure()` wi ```bash export GWTM_API_TOKEN=YOUR_API_TOKEN -export GWTM_BASE_URL=https://treasuremap.space # optional, this is the default +export GWTM_BASE_URL=https://treasuremap.space/api/v1 # optional, this is the default ``` ```python diff --git a/src/gwtm_api/candidate.py b/src/gwtm_api/candidate.py index fce5881..a7679a4 100644 --- a/src/gwtm_api/candidate.py +++ b/src/gwtm_api/candidate.py @@ -199,7 +199,7 @@ def put(self, payload: dict = None, verbose: bool = False, transport.configure(api_token=api_token) if not self.id: raise Exception("Candidate does not have an id") - result = transport.put('/candidate', json={'candidate': payload or self._to_dict()}) + result = transport.put('/candidate', json={'id': self.id, 'candidate': payload or self._to_dict()}) if verbose: print(result) fetched = Candidate.get(id=self.id) diff --git a/src/gwtm_api/core/_client_factory.py b/src/gwtm_api/core/_client_factory.py index 599e5bf..e4ad8b6 100644 --- a/src/gwtm_api/core/_client_factory.py +++ b/src/gwtm_api/core/_client_factory.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import os import httpx diff --git a/src/gwtm_api/event_tools.py b/src/gwtm_api/event_tools.py index b89e9ac..b1d3491 100644 --- a/src/gwtm_api/event_tools.py +++ b/src/gwtm_api/event_tools.py @@ -16,9 +16,14 @@ from .core.util import instrument_color, gc_dist -def plot_coverage(graceid: str, pointings: List[Pointing] = [], - cache=False, projection='astro hours mollweide') -> None: - +def plot_coverage(graceid: str, pointings: List[Pointing] = None, + cache=False, projection='astro hours mollweide', + api_token: str = None) -> None: + from .core import _client_factory as transport + if api_token: + transport.configure(api_token=api_token) + if pointings is None: + pointings = [] if len(pointings) == 0 and graceid is None: raise Exception("Pointings list or graceid is required") @@ -64,7 +69,7 @@ def plot_coverage(graceid: str, pointings: List[Pointing] = [], for j, arr in enumerate(polygon_arr): label = (inst_footprint.nickname if inst_footprint.nickname is not None - else inst_footprint.name) if j == 0 else None + else inst_footprint.instrument_name) if j == 0 else None poly = Polygon( np.asarray(arr), transform=ax.get_transform('world'), @@ -91,8 +96,13 @@ def plot_coverage(graceid: str, pointings: List[Pointing] = [], plt.show() -def calculate_coverage(graceid: str, pointings: List[Pointing] = [], - cache=False, approximate=True) -> Tuple[float, float]: +def calculate_coverage(graceid: str, pointings: List[Pointing] = None, + cache=False, approximate=True, api_token: str = None) -> Tuple[float, float]: + from .core import _client_factory as transport + if api_token: + transport.configure(api_token=api_token) + if pointings is None: + pointings = [] DECam_id = 38 if len(pointings) == 0 and graceid is None: raise Exception("Pointings list or graceid is required") @@ -118,6 +128,7 @@ def calculate_coverage(graceid: str, pointings: List[Pointing] = [], skymap_nside = hp.npix2nside(len(skymap)) qps = [] + deduped_indices = [] NSIDE4area = 512 pixarea = hp.nside2pixarea(NSIDE4area, degrees=True) @@ -158,9 +169,13 @@ def calculate_coverage(graceid: str, pointings: List[Pointing] = [], return prob, area -def renormalize_skymap(graceid: str, pointings: List[Pointing] = [], - cache=False) -> Any: - +def renormalize_skymap(graceid: str, pointings: List[Pointing] = None, + cache=False, api_token: str = None) -> Any: + from .core import _client_factory as transport + if api_token: + transport.configure(api_token=api_token) + if pointings is None: + pointings = [] if len(pointings) == 0 and graceid is None: raise Exception("Pointings list or graceid is required") @@ -174,6 +189,7 @@ def renormalize_skymap(graceid: str, pointings: List[Pointing] = [], skymap_nside = hp.npix2nside(len(skymap)) qps = [] + deduped_indices = [] for i in instrument_ids: polygon_arr = [] @@ -212,9 +228,14 @@ def renormalize_skymap(graceid: str, pointings: List[Pointing] = [], return normed_skymap -def renormed_skymap_contours(graceid: str, pointings: List[Pointing] = [], - cache=False) -> Any: +def renormed_skymap_contours(graceid: str, pointings: List[Pointing] = None, + cache=False, api_token: str = None) -> Any: + from .core import _client_factory as transport + if api_token: + transport.configure(api_token=api_token) + if pointings is None: + pointings = [] normed_skymap = renormalize_skymap(graceid=graceid, pointings=pointings, cache=cache) i = np.flipud(np.argsort(normed_skymap)) diff --git a/src/gwtm_api/instrument.py b/src/gwtm_api/instrument.py index 8a9d826..0bc40ac 100644 --- a/src/gwtm_api/instrument.py +++ b/src/gwtm_api/instrument.py @@ -106,7 +106,7 @@ def get(id: int = None, ids: List[int] = None, 'ids': _encode_list_param(ids), 'name': name, 'names': _encode_list_param(names), - 'type': type, + 'type': int(apimodels.instrument_type[type]) if isinstance(type, str) else (int(type) if isinstance(type, apimodels.instrument_type) else type), }) if not isinstance(result, list): raise Exception("Error in Instrument.get(): unexpected response from server") diff --git a/src/gwtm_api/pointing.py b/src/gwtm_api/pointing.py index b90430a..941e211 100644 --- a/src/gwtm_api/pointing.py +++ b/src/gwtm_api/pointing.py @@ -129,7 +129,14 @@ def post(self, graceid: str, request_doi: bool = None, api_token: str = None): if api_token: transport.configure(api_token=api_token) - result = transport.post('/pointings', json={'graceid': graceid, 'pointing': self._to_dict()}) + result = transport.post('/pointings', json={ + 'graceid': graceid, + 'pointing': self._to_dict(), + 'request_doi': request_doi, + 'doi_url': doi_url, + 'creators': creators, + 'doi_group_id': doi_group_id, + }) if verbose: print(result) fetched = Pointing.get(id=result['pointing_ids'][0]) @@ -145,6 +152,10 @@ def batch_post(graceid: str, pointings: List[Pointing], result = transport.post('/pointings', json={ 'graceid': graceid, 'pointings': [p._to_dict() for p in pointings], + 'request_doi': request_doi, + 'doi_url': doi_url, + 'creators': creators, + 'doi_group_id': doi_group_id, }) if verbose: print(result) diff --git a/tests/unit/test_instrument.py b/tests/unit/test_instrument.py index aca0579..5548eeb 100644 --- a/tests/unit/test_instrument.py +++ b/tests/unit/test_instrument.py @@ -71,3 +71,14 @@ def test_correct_endpoint_called(self): with patch(GET_PATH, return_value=[]) as mock: gwtm_api.Instrument.get() assert mock.call_args.args[0] == '/instruments' + + def test_type_enum_serialised_as_int(self): + from gwtm_api.core.apimodels import instrument_type + with patch(GET_PATH, return_value=[]) as mock: + gwtm_api.Instrument.get(type=instrument_type.photometric) + assert mock.call_args.kwargs['params']['type'] == 1 + + def test_type_string_serialised_as_int(self): + with patch(GET_PATH, return_value=[]) as mock: + gwtm_api.Instrument.get(type='photometric') + assert mock.call_args.kwargs['params']['type'] == 1 From 4f0201d7f9b1af7fa2a2593f678d409d06061caa Mon Sep 17 00:00:00 2001 From: Steve Foale Date: Tue, 12 May 2026 13:54:46 +0100 Subject: [PATCH 4/5] Correct api calls after comparing with api docs definitions. --- src/gwtm_api/alert.py | 2 +- src/gwtm_api/candidate.py | 11 ++++++---- src/gwtm_api/event_tools.py | 12 ++++++----- src/gwtm_api/pointing.py | 40 +++++++++++++++++++------------------ 4 files changed, 36 insertions(+), 29 deletions(-) diff --git a/src/gwtm_api/alert.py b/src/gwtm_api/alert.py index 0d0eb5e..f596576 100644 --- a/src/gwtm_api/alert.py +++ b/src/gwtm_api/alert.py @@ -105,7 +105,7 @@ def get(id: int = None, graceid: str = None, api_token: str = None) -> "Alert": def get_all(id: int = None, graceid: str = None, api_token: str = None): if api_token: transport.configure(api_token=api_token) - result = transport.get('/query_alerts', params={'id': id, 'graceid': graceid}) + result = transport.get('/query_alerts', params={'graceid': graceid}) if not isinstance(result, list): raise Exception("Error in Alert.get_all(): unexpected response from server") return [Alert(kwdict=a) for a in result] diff --git a/src/gwtm_api/candidate.py b/src/gwtm_api/candidate.py index a7679a4..908f746 100644 --- a/src/gwtm_api/candidate.py +++ b/src/gwtm_api/candidate.py @@ -103,16 +103,17 @@ def _to_dict(self) -> dict: dd = (self.discovery_date.strftime("%Y-%m-%dT%H:%M:%S.%f") if isinstance(self.discovery_date, datetime.datetime) else self.discovery_date) return { - 'candidate_name': self.candidate_name or '', + 'candidate_name': self.candidate_name if self.candidate_name is not None else '', 'discovery_date': dd, - 'discovery_magnitude': self.discovery_magnitude or 0.0, - 'magnitude_unit': _enum_name(self.magnitude_unit) or 'ab_mag', + 'discovery_magnitude': self.discovery_magnitude if self.discovery_magnitude is not None else 0.0, + 'magnitude_unit': _enum_name(self.magnitude_unit) if self.magnitude_unit is not None else 'ab_mag', 'position': self.position, 'ra': self.ra, 'dec': self.dec, 'tns_name': self.tns_name, 'tns_url': self.tns_url, 'magnitude_bandpass': _enum_name(self.magnitude_bandpass), + 'magnitude_central_wave': self.magnitude_central_wave, 'magnitude_bandwidth': self.magnitude_bandwidth, 'wavelength_regime': self.wavelength_regime, 'wavelength_unit': _enum_name(self.wavelength_unit), @@ -199,7 +200,9 @@ def put(self, payload: dict = None, verbose: bool = False, transport.configure(api_token=api_token) if not self.id: raise Exception("Candidate does not have an id") - result = transport.put('/candidate', json={'id': self.id, 'candidate': payload or self._to_dict()}) + candidate_data = payload.copy() if payload is not None else self._to_dict() + candidate_data.pop('id', None) + result = transport.put('/candidate', json={'id': self.id, 'candidate': candidate_data}) if verbose: print(result) fetched = Candidate.get(id=self.id) diff --git a/src/gwtm_api/event_tools.py b/src/gwtm_api/event_tools.py index b1d3491..096543d 100644 --- a/src/gwtm_api/event_tools.py +++ b/src/gwtm_api/event_tools.py @@ -13,13 +13,13 @@ from . import Pointing, Instrument, Footprint, Candidate from .alert import Alert as Alert +from .core import _client_factory as transport from .core.util import instrument_color, gc_dist def plot_coverage(graceid: str, pointings: List[Pointing] = None, cache=False, projection='astro hours mollweide', api_token: str = None) -> None: - from .core import _client_factory as transport if api_token: transport.configure(api_token=api_token) if pointings is None: @@ -98,7 +98,6 @@ def plot_coverage(graceid: str, pointings: List[Pointing] = None, def calculate_coverage(graceid: str, pointings: List[Pointing] = None, cache=False, approximate=True, api_token: str = None) -> Tuple[float, float]: - from .core import _client_factory as transport if api_token: transport.configure(api_token=api_token) if pointings is None: @@ -127,6 +126,9 @@ def calculate_coverage(graceid: str, pointings: List[Pointing] = None, skymap_nside = hp.npix2nside(len(skymap)) + if not instrument_ids: + return 0.0, 0.0 + qps = [] deduped_indices = [] NSIDE4area = 512 @@ -171,7 +173,6 @@ def calculate_coverage(graceid: str, pointings: List[Pointing] = None, def renormalize_skymap(graceid: str, pointings: List[Pointing] = None, cache=False, api_token: str = None) -> Any: - from .core import _client_factory as transport if api_token: transport.configure(api_token=api_token) if pointings is None: @@ -187,6 +188,9 @@ def renormalize_skymap(graceid: str, pointings: List[Pointing] = None, instrument_ids = list(set([x.instrumentid for x in pointings])) inst_footprints = Instrument.get(ids=instrument_ids, include_footprint=True) + if not instrument_ids: + return skymap + skymap_nside = hp.npix2nside(len(skymap)) qps = [] deduped_indices = [] @@ -230,8 +234,6 @@ def renormalize_skymap(graceid: str, pointings: List[Pointing] = None, def renormed_skymap_contours(graceid: str, pointings: List[Pointing] = None, cache=False, api_token: str = None) -> Any: - - from .core import _client_factory as transport if api_token: transport.configure(api_token=api_token) if pointings is None: diff --git a/src/gwtm_api/pointing.py b/src/gwtm_api/pointing.py index 941e211..5d9afd6 100644 --- a/src/gwtm_api/pointing.py +++ b/src/gwtm_api/pointing.py @@ -129,14 +129,16 @@ def post(self, graceid: str, request_doi: bool = None, api_token: str = None): if api_token: transport.configure(api_token=api_token) - result = transport.post('/pointings', json={ - 'graceid': graceid, - 'pointing': self._to_dict(), - 'request_doi': request_doi, - 'doi_url': doi_url, - 'creators': creators, - 'doi_group_id': doi_group_id, - }) + payload = {'graceid': graceid, 'pointing': self._to_dict()} + if request_doi: + payload['request_doi'] = request_doi + if doi_url is not None: + payload['doi_url'] = doi_url + if creators is not None: + payload['creators'] = creators + if doi_group_id is not None: + payload['doi_group_id'] = doi_group_id + result = transport.post('/pointings', json=payload) if verbose: print(result) fetched = Pointing.get(id=result['pointing_ids'][0]) @@ -149,14 +151,16 @@ def batch_post(graceid: str, pointings: List[Pointing], verbose: bool = False, api_token: str = None) -> List[Pointing]: if api_token: transport.configure(api_token=api_token) - result = transport.post('/pointings', json={ - 'graceid': graceid, - 'pointings': [p._to_dict() for p in pointings], - 'request_doi': request_doi, - 'doi_url': doi_url, - 'creators': creators, - 'doi_group_id': doi_group_id, - }) + payload = {'graceid': graceid, 'pointings': [p._to_dict() for p in pointings]} + if request_doi: + payload['request_doi'] = request_doi + if doi_url is not None: + payload['doi_url'] = doi_url + if creators is not None: + payload['creators'] = creators + if doi_group_id is not None: + payload['doi_group_id'] = doi_group_id + result = transport.post('/pointings', json=payload) if verbose: print(result) return Pointing.get(ids=result['pointing_ids']) @@ -218,15 +222,13 @@ def get(graceid: str = None, 'energy_unit': _enum_name(energy_unit), 'frequency_regime': _encode_list_param(frequency_regime), 'frequency_unit': _enum_name(frequency_unit), - 'central_wave': central_wave, - 'bandwidth': bandwidth, }) return [Pointing(kwdict=p) for p in result] @staticmethod def request_doi(graceid: str = None, id: int = None, ids: List[int] = None, creators: dict = None, - doi_group_id: int = None, api_token: str = None) -> str: + doi_group_id: str = None, api_token: str = None) -> str: if api_token: transport.configure(api_token=api_token) if all(x is None for x in [graceid, id, ids]): From 8e234d628163178cdc6b0920d32afa73f6162a33 Mon Sep 17 00:00:00 2001 From: Steve Foale Date: Tue, 12 May 2026 14:56:39 +0100 Subject: [PATCH 5/5] Add more unit tests + fix issues revealed by new tests. --- src/gwtm_api/core/apimodels.py | 2 + src/gwtm_api/event_tools.py | 8 ++-- tests/unit/test_alert.py | 71 ++++++++++++++++++++++++++++ tests/unit/test_candidate.py | 86 ++++++++++++++++++++++++++++++++++ tests/unit/test_event_tools.py | 49 +++++++++++++++++++ tests/unit/test_pointing.py | 58 +++++++++++++++++++++++ 6 files changed, 270 insertions(+), 4 deletions(-) create mode 100644 tests/unit/test_event_tools.py diff --git a/src/gwtm_api/core/apimodels.py b/src/gwtm_api/core/apimodels.py index 6d25407..147d984 100644 --- a/src/gwtm_api/core/apimodels.py +++ b/src/gwtm_api/core/apimodels.py @@ -301,6 +301,8 @@ def validate_payload(self, payload, instance): if isinstance(value, int): eval = mf.type(value).name self.valid_inst[key] = eval + elif value is None: + self.valid_inst[key] = None else: _ = mf.type(value) self.valid_inst[key] = mf.type(value) diff --git a/src/gwtm_api/event_tools.py b/src/gwtm_api/event_tools.py index 096543d..a3c37ff 100644 --- a/src/gwtm_api/event_tools.py +++ b/src/gwtm_api/event_tools.py @@ -113,6 +113,9 @@ def calculate_coverage(graceid: str, pointings: List[Pointing] = None, instrument_ids = list(set([x.instrumentid for x in pointings])) + if not instrument_ids: + return 0.0, 0.0 + if DECam_id in instrument_ids: print("Warning: DECam footprint will be automatically approximated") approximate = True @@ -126,9 +129,6 @@ def calculate_coverage(graceid: str, pointings: List[Pointing] = None, skymap_nside = hp.npix2nside(len(skymap)) - if not instrument_ids: - return 0.0, 0.0 - qps = [] deduped_indices = [] NSIDE4area = 512 @@ -186,11 +186,11 @@ def renormalize_skymap(graceid: str, pointings: List[Pointing] = None, skymap = Alert.fetch_skymap(graceid=graceid, cache=cache) instrument_ids = list(set([x.instrumentid for x in pointings])) - inst_footprints = Instrument.get(ids=instrument_ids, include_footprint=True) if not instrument_ids: return skymap + inst_footprints = Instrument.get(ids=instrument_ids, include_footprint=True) skymap_nside = hp.npix2nside(len(skymap)) qps = [] deduped_indices = [] diff --git a/tests/unit/test_alert.py b/tests/unit/test_alert.py index 12b6d38..000d513 100644 --- a/tests/unit/test_alert.py +++ b/tests/unit/test_alert.py @@ -58,3 +58,74 @@ def test_correct_endpoint_called(self): except Exception: pass assert mock.call_args.args[0] == '/query_alerts' + + def test_id_param_not_sent_to_server(self): + """API has no id filter on query_alerts — must not be forwarded.""" + with patch(GET_PATH, return_value=[ALERT_FIXTURE]) as mock: + gwtm_api.Alert.get_all(id=99, graceid='S190425z') + assert 'id' not in mock.call_args.kwargs['params'] + + +class TestAlertFetchContours: + + CONTOUR_FIXTURE = { + 'type': 'FeatureCollection', + 'features': [ + {'geometry': {'coordinates': [[[10.0, -20.0], [11.0, -21.0]]]}}, + ] + } + + def test_calls_correct_endpoint(self): + with patch(GET_PATH, return_value=self.CONTOUR_FIXTURE) as mock: + gwtm_api.Alert.fetch_contours(graceid='S190425z') + assert mock.call_args.args[0] == '/gw_contour' + + def test_graceid_sent_as_param(self): + with patch(GET_PATH, return_value=self.CONTOUR_FIXTURE) as mock: + gwtm_api.Alert.fetch_contours(graceid='S190425z') + assert mock.call_args.kwargs['params']['graceid'] == 'S190425z' + + def test_returns_polygon_list(self): + with patch(GET_PATH, return_value=self.CONTOUR_FIXTURE): + result = gwtm_api.Alert.fetch_contours(graceid='S190425z') + assert isinstance(result, list) + assert result[0] == [[10.0, -20.0], [11.0, -21.0]] + + def test_cache_hit_skips_api_call(self): + from unittest.mock import MagicMock + from gwtm_api.core.tmcache import TMCache + with patch.object(TMCache, 'get', return_value=self.CONTOUR_FIXTURE), \ + patch(GET_PATH) as mock_get: + gwtm_api.Alert.fetch_contours(graceid='S190425z', cache=True) + mock_get.assert_not_called() + + +class TestAlertFetchSkymap: + + def test_calls_correct_endpoint(self): + from unittest.mock import MagicMock + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.content = b'' + with patch('gwtm_api.core._client_factory.get_raw', return_value=mock_response) as mock, \ + patch('healpy.read_map', return_value=[0.1, 0.2]): + gwtm_api.Alert.fetch_skymap(graceid='S190425z') + assert mock.call_args.args[0] == '/gw_skymap' + + def test_graceid_sent_as_param(self): + from unittest.mock import MagicMock + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.content = b'' + with patch('gwtm_api.core._client_factory.get_raw', return_value=mock_response) as mock, \ + patch('healpy.read_map', return_value=[0.1, 0.2]): + gwtm_api.Alert.fetch_skymap(graceid='S190425z') + assert mock.call_args.kwargs['params']['graceid'] == 'S190425z' + + def test_raises_on_non_200(self): + from unittest.mock import MagicMock + mock_response = MagicMock() + mock_response.status_code = 404 + with patch('gwtm_api.core._client_factory.get_raw', return_value=mock_response): + with pytest.raises(Exception, match="status 404"): + gwtm_api.Alert.fetch_skymap(graceid='NONEXISTENT') diff --git a/tests/unit/test_candidate.py b/tests/unit/test_candidate.py index 231e898..287e894 100644 --- a/tests/unit/test_candidate.py +++ b/tests/unit/test_candidate.py @@ -115,3 +115,89 @@ def test_put_raises_without_id(self): c = gwtm_api.Candidate(kwdict={**CANDIDATE_FIXTURE, 'id': None}) with pytest.raises(Exception, match="does not have an id"): c.put() + + def test_put_sends_self_id_at_top_level(self): + c = gwtm_api.Candidate(kwdict={**CANDIDATE_FIXTURE}) + with patch(PUT_PATH, return_value={}) as mock_put, \ + patch(GET_PATH, return_value=[CANDIDATE_FIXTURE]): + c.put() + assert mock_put.call_args.kwargs['json']['id'] == 1 + + def test_put_strips_id_from_caller_payload(self): + c = gwtm_api.Candidate(kwdict={**CANDIDATE_FIXTURE}) + with patch(PUT_PATH, return_value={}) as mock_put, \ + patch(GET_PATH, return_value=[CANDIDATE_FIXTURE]): + c.put(payload={'id': 999, 'tns_name': 'AT2019abc'}) + sent = mock_put.call_args.kwargs['json'] + assert sent['id'] == 1 + assert 'id' not in sent['candidate'] + + def test_put_uses_to_dict_when_no_payload(self): + c = gwtm_api.Candidate(kwdict={**CANDIDATE_FIXTURE}) + with patch(PUT_PATH, return_value={}) as mock_put, \ + patch(GET_PATH, return_value=[CANDIDATE_FIXTURE]): + c.put() + sent_candidate = mock_put.call_args.kwargs['json']['candidate'] + assert 'candidate_name' in sent_candidate + + +class TestCandidateInitWithId: + + def test_init_with_id_calls_get(self): + with patch(GET_PATH, return_value=[CANDIDATE_FIXTURE]) as mock_get: + c = gwtm_api.Candidate(id=1) + assert mock_get.call_args.args[0] == '/candidate' + assert c.id == 1 + assert c.graceid == 'S190425z' + + def test_init_with_id_raises_when_not_found(self): + with patch(GET_PATH, return_value=[]): + with pytest.raises(Exception, match="No candidate found"): + gwtm_api.Candidate(id=999) + + +class TestCandidateToDict: + + def test_none_name_becomes_empty_string(self): + c = gwtm_api.Candidate(kwdict={**CANDIDATE_FIXTURE, 'candidate_name': None}) + d = c._to_dict() + assert d['candidate_name'] == '' + + def test_zero_magnitude_preserved(self): + c = gwtm_api.Candidate(kwdict={**CANDIDATE_FIXTURE, 'discovery_magnitude': 0.0}) + d = c._to_dict() + assert d['discovery_magnitude'] == 0.0 + + def test_none_magnitude_defaults_to_zero(self): + c = gwtm_api.Candidate(kwdict={**CANDIDATE_FIXTURE, 'discovery_magnitude': None}) + d = c._to_dict() + assert d['discovery_magnitude'] == 0.0 + + def test_includes_magnitude_central_wave(self): + c = gwtm_api.Candidate(kwdict={**CANDIDATE_FIXTURE, 'magnitude_central_wave': 6200.0}) + d = c._to_dict() + assert d['magnitude_central_wave'] == 6200.0 + + +class TestCandidatePostBody: + + def test_post_sends_candidate_key(self): + with patch(POST_PATH, return_value={'candidate_ids': [1]}) as mock_post, \ + patch(GET_PATH, return_value=[CANDIDATE_FIXTURE]): + c = gwtm_api.Candidate(kwdict={**CANDIDATE_FIXTURE}) + c.post(graceid='S190425z') + body = mock_post.call_args.kwargs['json'] + assert 'graceid' in body + assert 'candidate' in body + assert 'candidates' not in body + + def test_batch_post_sends_candidates_key(self): + with patch(POST_PATH, return_value={'candidate_ids': [1]}) as mock_post, \ + patch(GET_PATH, return_value=[CANDIDATE_FIXTURE]): + gwtm_api.Candidate.batch_post( + candidates=[gwtm_api.Candidate(kwdict={**CANDIDATE_FIXTURE})], + graceid='S190425z' + ) + body = mock_post.call_args.kwargs['json'] + assert 'candidates' in body + assert 'candidate' not in body diff --git a/tests/unit/test_event_tools.py b/tests/unit/test_event_tools.py new file mode 100644 index 0000000..9c33b19 --- /dev/null +++ b/tests/unit/test_event_tools.py @@ -0,0 +1,49 @@ +"""Unit tests for event_tools — no live server required.""" +import numpy as np +from unittest.mock import patch + +import sys +sys.path.insert(0, 'src') + +import gwtm_api.event_tools as event_tools + +SKYMAP = np.array([0.1, 0.2, 0.3, 0.4]) + +POINTING_GET_PATH = 'gwtm_api.event_tools.Pointing.get' +SKYMAP_PATH = 'gwtm_api.event_tools.Alert.fetch_skymap' + + +class TestCalculateCoverageEmptyCase: + + def test_no_pointings_returns_zero_prob_and_area(self): + with patch(POINTING_GET_PATH, return_value=[]), \ + patch(SKYMAP_PATH, return_value=SKYMAP): + prob, area = event_tools.calculate_coverage(graceid='S190425z') + assert prob == 0.0 + assert area == 0.0 + + def test_return_type_is_tuple_of_floats(self): + with patch(POINTING_GET_PATH, return_value=[]), \ + patch(SKYMAP_PATH, return_value=SKYMAP): + result = event_tools.calculate_coverage(graceid='S190425z') + assert len(result) == 2 + assert all(isinstance(v, float) for v in result) + + def test_raises_without_graceid_or_pointings(self): + import pytest + with pytest.raises(Exception, match="required"): + event_tools.calculate_coverage(graceid=None, pointings=[]) + + +class TestRenormalizeSkymapEmptyCase: + + def test_no_pointings_returns_original_skymap(self): + with patch(POINTING_GET_PATH, return_value=[]), \ + patch(SKYMAP_PATH, return_value=SKYMAP.copy()): + result = event_tools.renormalize_skymap(graceid='S190425z') + np.testing.assert_array_equal(result, SKYMAP) + + def test_raises_without_graceid_or_pointings(self): + import pytest + with pytest.raises(Exception, match="required"): + event_tools.renormalize_skymap(graceid=None, pointings=[]) diff --git a/tests/unit/test_pointing.py b/tests/unit/test_pointing.py index 5ddebdb..f110777 100644 --- a/tests/unit/test_pointing.py +++ b/tests/unit/test_pointing.py @@ -123,3 +123,61 @@ def test_post_correct_endpoint(self): depth_unit='ab_mag', band='r') p.post(graceid='S190425z') assert mock_get.call_args.args[0] == '/pointings' + + def test_doi_params_absent_when_not_provided(self): + with patch(POST_PATH, return_value={'pointing_ids': [42]}) as mock_post, \ + patch(GET_PATH, return_value=[POINTING_FIXTURE]): + p = gwtm_api.Pointing(ra=10.0, dec=-20.0, instrumentid=10, + time=datetime.datetime(2025, 1, 1), + status='planned', depth=18.0, + depth_unit='ab_mag', band='r') + p.post(graceid='S190425z') + body = mock_post.call_args.kwargs['json'] + assert 'request_doi' not in body + assert 'doi_url' not in body + assert 'creators' not in body + assert 'doi_group_id' not in body + + def test_doi_params_sent_when_provided(self): + with patch(POST_PATH, return_value={'pointing_ids': [42]}) as mock_post, \ + patch(GET_PATH, return_value=[POINTING_FIXTURE]): + p = gwtm_api.Pointing(ra=10.0, dec=-20.0, instrumentid=10, + time=datetime.datetime(2025, 1, 1), + status='planned', depth=18.0, + depth_unit='ab_mag', band='r') + p.post(graceid='S190425z', request_doi=True, + doi_url='https://doi.org/10.1234', doi_group_id=5) + body = mock_post.call_args.kwargs['json'] + assert body['request_doi'] is True + assert body['doi_url'] == 'https://doi.org/10.1234' + assert body['doi_group_id'] == 5 + + def test_central_wave_and_bandwidth_not_in_get_params(self): + with patch(GET_PATH, return_value=[]) as mock: + gwtm_api.Pointing.get(graceid='S190425z', central_wave=5000.0, bandwidth=100.0) + params = mock.call_args.kwargs['params'] + assert 'central_wave' not in params + assert 'bandwidth' not in params + + +class TestRequestDoi: + + def test_calls_correct_endpoint(self): + with patch(POST_PATH, return_value={'doi_url': 'https://doi.org/10.1234'}) as mock: + result = gwtm_api.Pointing.request_doi(graceid='S190425z') + assert mock.call_args.args[0] == '/request_doi' + + def test_returns_doi_url(self): + with patch(POST_PATH, return_value={'doi_url': 'https://doi.org/10.1234'}): + result = gwtm_api.Pointing.request_doi(graceid='S190425z') + assert result == 'https://doi.org/10.1234' + + def test_raises_without_graceid_id_or_ids(self): + with pytest.raises(Exception, match="Must specify"): + gwtm_api.Pointing.request_doi() + + def test_doi_group_id_sent_as_string(self): + with patch(POST_PATH, return_value={'doi_url': 'https://doi.org/10.1234'}) as mock: + gwtm_api.Pointing.request_doi(graceid='S190425z', doi_group_id='42') + body = mock.call_args.kwargs['json'] + assert isinstance(body['doi_group_id'], str)