Updating moc routes in valis

python/valis/routes/mocs.py

@@ -4,8 +4,10 @@
 #
 
 import orjson
+import os
 import pathlib
-from typing import List, Dict
+import re
+from typing import List, Dict, Annotated
 from pydantic import BaseModel, Field
 from fastapi import APIRouter, HTTPException, Query
 from fastapi_restful.cbv import cbv
@@ -15,12 +17,20 @@
 
 from sdss_access.path import Path
 
-def read_json(path):
+def read_json(path: str) -> dict:
+    """ Read a MOC.json file """
     with open(path, 'r') as f:
         lines = f.readlines()
         first = lines[0]
-        mocorder = int(first.split('\n')[0].split('#MOCORDER ')[-1])
-        data = orjson.loads("\n".join(lines[1:]))
+        if "MOCORDER" in first:
+            # written by Hipsgen-cat
+            mocorder = int(first.split('\n')[0].split('#MOCORDER ')[-1])
+            sub = lines[1:]
+        else:
+            # written by MOCpy
+            mocorder = int(max(map(int,re.findall(r'"(.*?)":', '\n'.join(lines)))))
+            sub = lines
+        data = orjson.loads("\n".join(sub))
         return {'order': mocorder, 'moc': data}
 
 
@@ -34,7 +44,6 @@ class MocModel(BaseModel):
 @cbv(router)
 class Mocs(Base):
     """ Endpoints for interacting with SDSS MOCs """
-    name: str = 'sdss_moc'
 
     def check_path_name(self, path, name: str):
         """ temp function until sort out directory org for """
@@ -48,32 +57,39 @@ def check_path_exists(self, spath, path: str):
             raise HTTPException(status_code=422, detail=f'path {path} does not exist on disk.')
 
     @router.get('/preview', summary='Preview an individual survey MOC', response_class=RedirectResponse)
-    async def get_moc(self, survey: str):
+    async def get_moc(self, survey: Annotated[str, Query(..., description='The SDSS survey name')] = 'manga'):
         """ Preview an individual survey MOC """
         return f'/static/mocs/{self.release.lower()}/{survey}/'
 
     @router.get('/json', summary='Get the MOC file in JSON format')
-    async def get_json(self, survey: str = Query(..., description='The SDSS survey name', example='manga')) -> MocModel:
+    async def get_json(self, survey: Annotated[str, Query(..., description='The SDSS survey name')] = 'manga') -> MocModel:
         """ Get the MOC file in JSON format """
         # temporarily affixing the access path to sdss5 sandbox until
         # we decide on real org for DRs, etc
-        spath = Path(release='sdss5')
+        spath = Path(release='sdsswork')
 
-        self.check_path_name(spath, self.name)
-        path = spath.full(self.name, release=self.release.lower(), survey=survey, ext='json')
+        self.check_path_name(spath, 'sdss_moc')
+        path = spath.full('sdss_moc', release=self.release.lower(), survey=survey, ext='json')
         self.check_path_exists(spath, path)
         return ORJSONResponseCustom(content=read_json(path), option=orjson.OPT_SERIALIZE_NUMPY)
 
     @router.get('/fits', summary='Download the MOC file in FITs format')
-    async def get_fits(self, survey: str = Query(..., description='The SDSS survey name', example='manga')):
+    async def get_fits(self, survey: Annotated[str, Query(..., description='The SDSS survey name')] = 'manga'):
         """ Download the MOC file in FITs format """
         # temporarily affixing the access path to sdss5 sandbox
         # we decide on real org for DRs, etc
-        spath = Path(release='sdss5')
+        spath = Path(release='sdsswork')
 
-        self.check_path_name(spath, self.name)
-        path = spath.full(self.name, release=self.release.lower(), survey=survey.lower(), ext='fits')
+        self.check_path_name(spath, 'sdss_moc')
+        path = spath.full('sdss_moc', release=self.release.lower(), survey=survey.lower(), ext='fits')
         self.check_path_exists(spath, path)
         pp = pathlib.Path(path)
         name = f'{survey.lower()}_{pp.name}'
-        return FileResponse(path, filename=name, media_type='application/fits')
+        return FileResponse(path, filename=name, media_type='application/fits')
+
+    @router.get('/list', summary='List the available MOCs')
+    async def list_mocs(self) -> list[str]:
+        """ List the available MOCs """
+        Path(release='sdsswork')
+        mocs = sorted(set([':'.join(i.parent.parts[-2:]) for i in pathlib.Path(os.getenv("SDSS_HIPS")).rglob('Moc.fits')]))
+        return mocs

tests/conftest.py

@@ -99,17 +99,33 @@ def testfile(setup_sas):
     return path
 
 
+def mocfits():
+    """ create a test moc fits """
+    # create the FITS HDUList
+    header = fits.Header([('MOCTOOL', 'CDSjavaAPI-4.1', 'Name of the MOC generator'),
+                          ('MOCORDER', 29, 'MOC resolution (best order)')])
+    primary = fits.PrimaryHDU()
+    cols = [fits.Column(name='NPIX', format='K', array=np.arange(5))]
+    bindata = fits.BinTableHDU.from_columns(cols, name='MOC', header=header)
+    return fits.HDUList([primary, bindata])
+
+
 @pytest.fixture()
 def testmoc(setup_sas):
     moc = os.getenv("SDSS_HIPS", "")
     path = pathlib.Path(moc) / 'dr17/manga/Moc.json'
     if not path.exists():
         path.parent.mkdir(parents=True, exist_ok=True)
 
+    # create fake json
     with open(path, 'w') as f:
         f.write("#MOCORDER 10\n")
         f.write("""{"9":[224407,224413,664253,664290,664292]}\n""")
 
+    # create fake fits
+    moc = mocfits()
+    moc.writeto(path.with_suffix('.fits'), overwrite=True)
+
 
 class MockTree(Tree):
     """ mock out the Tree class to insert test file """

tests/test_mocs.py

@@ -1,6 +1,7 @@
 # encoding: utf-8
 #
-
+import io
+from astropy.io import fits
 
 def get_data(response):
     assert response.status_code == 200
@@ -13,3 +14,19 @@ def test_moc_json(client, testmoc):
 
     assert data['order'] == 10
     assert data['moc']['9'] == [224407,224413,664253,664290,664292]
+
+
+def test_moc_fits(client, testmoc):
+    response = client.get("/mocs/fits?survey=manga&release=DR17")
+    assert response.status_code == 200
+    with fits.open(io.BytesIO(response.content)) as hdu:
+        assert hdu[1].header['EXTNAME'] == 'MOC'
+        assert hdu[1].header['MOCORDER'] == 29
+        assert "NPIX" in hdu[1].data.columns.names
+
+
+def test_list_mocs(client, testmoc):
+    response = client.get("/mocs/list")
+    data = get_data(response)
+
+    assert 'dr17:manga' in data