Merge pull request #3386 from esdc-esac-esa-int/ESA_euclid_EUCLIDMNGT-1449_x-match

Author: bsipocz

CHANGES.rst

@@ -33,6 +33,7 @@ esa.euclid
 
 - New method, get_scientific_product_list, to retrieve scientific LE3
   products. [#3313]
+- New cross-match method [#3386]
 
 gaia
 ^^^^

astroquery/esa/euclid/core.py

@@ -23,6 +23,7 @@
 from astroquery import log
 from astroquery.utils import commons
 from astroquery.utils.tap import TapPlus
+from astroquery.utils.tap import taputils
 from . import conf
 
 
@@ -116,6 +117,156 @@ def __init__(self, *, environment='PDR', tap_plus_conn_handler=None, datalink_ha
         if show_server_messages:
             self.get_status_messages()
 
+    def cross_match_basic(self, *, table_a_full_qualified_name, table_a_column_ra, table_a_column_dec,
+                          table_b_full_qualified_name=None, table_b_column_ra=None, table_b_column_dec=None,
+                          results_name=None,
+                          radius=1.0, background=False, verbose=False):
+        """Performs a positional cross-match between the specified tables.
+
+        This method carries out the following steps in one step:
+
+            1. updates the user table metadata to flag the positional RA/Dec columns;
+            2. launches a positional cross-match as an asynchronous query;
+            3. returns all the columns from both tables plus the angular distance (deg) for the cross-matched sources.
+
+        The result is a join table with the identifies of both tables and the distance (degrees), that is returned
+        without metadata units. If desired, units can be added using the Units package of Astropy as follows:
+        ``results[‘separation’].unit = u.degree``. To speed up the cross-match, pass the biggest table to the
+        ``table_b_full_qualified_name`` parameter.
+        TAP+ only
+
+        Parameters
+        ----------
+        table_a_full_qualified_name : str, mandatory
+            a full qualified table name (i.e. schema name and table name)
+        table_a_column_ra : str, mandatory
+            the ‘ra’ column in the table table_a_full_qualified_name
+        table_a_column_dec :  str, mandatory
+            the ‘dec’ column in the table table_a_full_qualified_name
+        table_b_full_qualified_name : str, optional, default the main_table associated to the selected environment
+            a full qualified table name (i.e. schema name and table name)
+        table_b_column_ra : str, optional, default the main_table_ra_column associated to the selected environment
+            the ‘ra’ column in the table table_b_full_qualified_name
+        table_b_column_dec :  str, default the main_table_dec_column associated to the selected environment
+            the ‘dec’ column in the table table_b_full_qualified_name
+        results_name : str, optional, default None
+            custom name defined by the user for the job that is going to be created
+        radius : float (arc. seconds), str or astropy.coordinate, optional, default 1.0
+            radius  (valid range: 0.1-10.0). For an astropy.coordinate any angular unit is valid, but its value in arc
+            sec must be contained within the valid range.
+        background : bool, optional, default 'False'
+            when the job is executed in asynchronous mode, this flag specifies
+            whether the execution will wait until results are available
+        verbose : bool, optional, default 'False'
+            flag to display information about the process
+
+        Returns
+        -------
+        A Job object
+        """
+
+        radius_quantity = self.__get_radius_as_quantity_arcsec(radius)
+
+        radius_arc_sec = radius_quantity.value
+
+        if radius_arc_sec < 0.1 or radius_arc_sec > 10.0:
+            raise ValueError(f"Invalid radius value. Found {radius_quantity}, valid range is: 0.1 to 10.0")
+
+        schema_a = self.__get_schema_name(table_a_full_qualified_name)
+        if not schema_a:
+            raise ValueError(f"Schema name is empty in full qualified table: '{table_a_full_qualified_name}'")
+
+        if table_b_full_qualified_name is None:
+            table_b_full_qualified_name = self.main_table
+            table_b_column_ra = self.main_table_ra
+            table_b_column_dec = self.main_table_dec
+        else:
+            if table_b_column_ra is None or table_b_column_dec is None:
+                raise ValueError(f"Invalid ra or dec column names: '{table_b_column_ra}' and '{table_b_column_dec}'")
+
+        schema_b = self.__get_schema_name(table_b_full_qualified_name)
+        if not schema_b:
+            raise ValueError(f"Schema name is empty in full qualified table: '{table_b_full_qualified_name}'")
+
+        table_metadata_a = self.__get_table_metadata(table_a_full_qualified_name, verbose)
+
+        table_metadata_b = self.__get_table_metadata(table_b_full_qualified_name, verbose)
+
+        self.__check_columns_exist(table_metadata_a, table_a_full_qualified_name, table_a_column_ra, table_a_column_dec)
+
+        self.__update_ra_dec_columns(table_a_full_qualified_name, table_a_column_ra, table_a_column_dec,
+                                     table_metadata_a, verbose)
+
+        self.__check_columns_exist(table_metadata_b, table_b_full_qualified_name, table_b_column_ra, table_b_column_dec)
+
+        self.__update_ra_dec_columns(table_b_full_qualified_name, table_b_column_ra, table_b_column_dec,
+                                     table_metadata_b, verbose)
+
+        query = (
+            f"SELECT a.*, DISTANCE(a.{table_a_column_ra}, a.{table_a_column_dec}, b.{table_b_column_ra}, "
+            f"b.{table_b_column_dec}) AS separation, b.* "
+            f"FROM {table_a_full_qualified_name} AS a JOIN {table_b_full_qualified_name} AS b "
+            f"ON DISTANCE(a.{table_a_column_ra}, a.{table_a_column_dec}, b.{table_b_column_ra}, b.{table_b_column_dec})"
+            f" < {radius_quantity.to(u.deg).value}")
+
+        return self.launch_job_async(query=query, name=results_name, output_file=None, output_format="votable_gzip",
+                                     verbose=verbose, dump_to_file=False, background=background, upload_resource=None,
+                                     upload_table_name=None)
+
+    def __get_radius_as_quantity_arcsec(self, radius):
+        """
+        transform the input radius into an astropy.Quantity in arc seconds
+        """
+        if not isinstance(radius, units.Quantity):
+            radius_quantity = Quantity(value=radius, unit=u.arcsec)
+        else:
+            radius_quantity = radius.to(u.arcsec)
+        return radius_quantity
+
+    def __update_ra_dec_columns(self, full_qualified_table_name, column_ra, column_dec, table_metadata, verbose):
+        """
+        Update table metadata for the ‘ra’ and the ‘dec’ columns in the input table
+        """
+        if full_qualified_table_name.startswith("user_"):
+            list_of_changes = list()
+            for column in table_metadata.columns:
+                if column.name == column_ra and column.flags != '1':
+                    list_of_changes.append([column_ra, "flags", "Ra"])
+                    list_of_changes.append([column_ra, "indexed", True])
+                if column.name == column_dec and column.flags != '2':
+                    list_of_changes.append([column_dec, "flags", "Dec"])
+                    list_of_changes.append([column_dec, "indexed", True])
+
+            if list_of_changes:
+                TapPlus.update_user_table(self, table_name=full_qualified_table_name, list_of_changes=list_of_changes,
+                                          verbose=verbose)
+
+    def __check_columns_exist(self, table_metadata_a, full_qualified_table_name, column_ra, column_dec):
+        """
+        Check whether the ‘ra’ and the ‘dec’ columns exists the input table
+        """
+        column_names = [column.name for column in table_metadata_a.columns]
+        if column_ra not in column_names or column_dec not in column_names:
+            raise ValueError(
+                f"Please check: columns {column_ra} or {column_dec} not available in the table '"
+                f"{full_qualified_table_name}'")
+
+    def __get_table_metadata(self, full_qualified_table_name, verbose):
+        """
+        Get the table metadata for the input table
+        """
+
+        return self.load_table(table=full_qualified_table_name, verbose=verbose)
+
+    def __get_schema_name(self, full_qualified_table_name):
+        """
+        Get the schema name from the full qualified table
+        """
+        schema = taputils.get_schema_name(full_qualified_table_name)
+        if schema is None:
+            raise ValueError(f"Not found schema name in full qualified table: '{full_qualified_table_name}'")
+        return schema
+
     def launch_job(self, query, *, name=None, dump_to_file=False, output_file=None, output_format="csv", verbose=False,
                    upload_resource=None, upload_table_name=None):
         """
@@ -282,10 +433,10 @@ def __cone_search(self, coordinate, radius, *, table_name=None, ra_column_name=N
             coordinates center point
         radius : astropy.units, mandatory
             radius
-        table_name : str, optional, default main gaia table name doing the cone search against
-        ra_column_name : str, optional, default ra column in main gaia table
+        table_name : str, optional, default main table name doing the cone search against
+        ra_column_name : str, optional, default ra column in main table
             ra column doing the cone search against
-        dec_column_name : str, optional, default dec column in main gaia table
+        dec_column_name : str, optional, default dec column in main table
             dec column doing the cone search against
         async_job : bool, optional, default 'False'
             executes the job in asynchronous/synchronous mode (default