add query_ref_cat

csst_common/pipeline.py

@@ -23,6 +23,7 @@ import warnings
 from typing import Callable, NamedTuple, Optional, Any, Union
 from astropy.time import Time, TimeDelta
 from astropy.io import fits
+from astropy import table
 import joblib
 
 import csst_dfs_client
@@ -488,6 +489,47 @@ class Pipeline:
             hdulist.flush()
         return meta
 
+    def query_ref_cat(
+        self, file_path, catalog_name: str = "trilegal_093"
+    ) -> None | str:
+        """
+        References
+        ----------
+        https://gea.esac.esa.int/archive/documentation/GDR3/Gaia_archive/chap_datamodel/sec_dm_main_source_catalogue/ssec_dm_gaia_source.html
+        """
+        image = File(file_path)
+        header = fits.getheader(image.file_path)
+        pointing_ra = header["RA_OBJ"]
+        pointing_dec = header["DEC_OBJ"]
+        print(f"Query reference catalog RA={pointing_ra}, Dec={pointing_dec} ...")
+        rec_cat = self.retry(
+            csst_dfs_client.catalog.search,
+            n_try=3,
+            ra=pointing_ra,
+            dec=pointing_dec,
+            catalog_name=catalog_name,
+            radius=2.0,
+            columns=["*"],
+            min_mag=0,
+            max_mag=30,
+            obstime=-1,
+            limit=-1,
+        )
+        assert rec_cat.success, rec_cat
+
+        # convert pandas data frame to table
+        ref_cat = table.Table.from_pandas(rec_cat["data"])
+        self.logger.info(f"{len(ref_cat)} rows in reference catalog")
+        if len(ref_cat) < 10:
+            return None
+        # rename columns to lower case
+        ref_cat.rename_columns(
+            ref_cat.colnames, [colname.lower() for colname in ref_cat.colnames]
+        )
+        self.logger.info(f"Writing to file: {ref_cat} ...")
+        ref_cat.write(image.derive1(ext="cat_ref.fits"), overwrite=True)
+        return image.derive1(ext="cat_ref.fits")
+
 
 class ModuleResult(NamedTuple):
     module: str = ""