Merge branch 'master' into issue/529/tracklet

bin/daily_stats.py

@@ -20,8 +20,6 @@
 import numpy as np
 import pandas as pd
 
-import pyspark.sql.functions as F
-
 from fink_broker.sparkUtils import init_sparksession
 from fink_broker.sparkUtils import load_parquet_files
 
@@ -111,13 +109,6 @@ def main():
 
     out_dic['simbad_gal'] = n_simbad_gal
 
-    # to account for schema migration
-    if 'knscore' not in df_sci.columns:
-        df_sci = df_sci.withColumn('knscore', F.lit(-1.0))
-    # 12/08/2021
-    if 'tracklet' not in df_sci.columns:
-        df_sci = df_sci.withColumn('tracklet', F.lit(''))
-
     df_class = df_sci.withColumn(
         'class',
         extract_fink_classification(

bin/index_archival.py

@@ -21,7 +21,6 @@
 3. Construct HBase catalog
 4. Push data (single shot)
 """
-import pyspark.sql.functions as F
 from pyspark.sql.functions import lit, concat_ws, col
 from pyspark.sql.functions import arrays_zip, explode
 from pyspark.sql.functions import pandas_udf, PandasUDFType
@@ -45,9 +44,6 @@
 
 from fink_filters.classification import extract_fink_classification
 
-from fink_science.utilities import concat_col
-from fink_science.asteroids.processor import roid_catcher
-
 from fink_tns.utils import download_catalog
 
 from astropy.coordinates import SkyCoord
@@ -89,26 +85,6 @@ def main():
     names = [col(i) for i in columns]
     index_name = '.' + columns[0]
 
-    if columns[0] == 'class':
-        # There was a bug in the SSO classification prior to 02/2021
-        # This piece of code is temporary, just the time to recompute data
-        df = df.drop('roid')
-
-        # Retrieve time-series information
-        to_expand = ['jd', 'magpsf']
-
-        # Append temp columns with historical + current measurements
-        prefix = 'c'
-        for colname in to_expand:
-            df = concat_col(df, colname, prefix=prefix)
-
-        # recompute asteroid classification
-        args_roid = [
-            'cjd', 'cmagpsf',
-            'candidate.ndethist', 'candidate.sgscore1',
-            'candidate.ssdistnr', 'candidate.distpsnr1']
-        df = df.withColumn('roid', roid_catcher(*args_roid))
-
     # Drop partitioning columns
     df = df.drop('year').drop('month').drop('day')
 
@@ -186,17 +162,9 @@ def main():
             ] + common_cols
         )
     elif columns[0] == 'ssnamenr':
-        # TODO: Computation of SSO flags was bugged.
-        # Ideally, we would have to filter on the `roid==3` field, but
-        # there was a bug in its computation (see https://github.com/astrolabsoftware/fink-science/issues/85)
-        # Hence, as long as we the data is not recomputed, we use this condition
-        # to flag known SSO (which is actually used since 02/2021).
+        # Flag only objects with likely counterpart in MPC
         df_index = df\
-            .filter(df['ssnamenr'] != 'null')\
-            .filter(df['ssdistnr'] >= 0)\
-            .filter(df['ssdistnr'] < 5)\
-            .filter((F.abs(df['distpsnr1']) - df['ssdistnr']) > 0.0)\
-            .filter(df['ndethist'] <= 2)\
+            .filter(df['roid'] == 3)\
             .select(
                 [
                     concat_ws('_', *names).alias(index_row_key_name)

fink_broker/sparkUtils.py

@@ -17,7 +17,6 @@
 from pyspark.sql import DataFrame
 from pyspark.sql.column import Column, _to_java_column
 from pyspark.sql.types import StructType
-import pyspark.sql.functions as F
 
 import os
 import json
@@ -356,13 +355,6 @@ def load_parquet_files(path: str) -> DataFrame:
         .option('mergeSchema', "true") \
         .load(path)
 
-    # to account for schema migration
-    if 'knscore' not in df.columns:
-        df = df.withColumn('knscore', F.lit(-1.0))
-    # 12/08/2021
-    if 'tracklet' not in df.columns:
-        df = df.withColumn('tracklet', F.lit(''))
-
     return df
 
 def get_schemas_from_avro(