add test for fink_fat_1.0

.gitignore

@@ -187,4 +187,23 @@ fink_fat_test/
 fink_fat_notebook/fink_fat_experiments/confirmed_all_fink_fat/
 fink_fat_notebook/fink_fat_experiments/confirmed_mops_fink_fat/
 fink_fat_notebook/fink_fat_experiments/data/
-fink_fat_notebook/parameters_selection/res_orbit_nb_point/
+fink_fat_notebook/parameters_selection/res_orbit_nb_point/
+dummy_kf.pkl
+fink_fat/Readme.md
+fink_fat/test/cli_test/fink_fat_out_test/mpc/mpc/
+fink_fat/test/pipeline_analysis.py
+fink_fat/test/pipeline_test.py
+fink_fat/test/run_roid.py
+fink_fat/test/sso_test_data/
+gen_ff_data_test.py
+kbo_neo_issue_ff_output/
+local.conf
+orbit_test_data/
+pipeline.conf
+pipeline_output/
+pipeline_test.logs
+pipeline_test_1.logs
+save_ff_output_kbo_neo_issue/
+sso_data_norm.parquet
+test_assoc_kalman.py
+test_assoc_orbit.py

fink_fat/command_line/association_cli.py

@@ -355,19 +355,39 @@ def get_last_sso_alert(object_class, date, verbose=False):
 
 def get_last_roid_streaming_alert(
     config: configparser.ConfigParser,
-    is_mpc: bool,
-    mode: str,
-    read_path: str,
-    output_path: str = None,
+    last_night: str,
+    output_path: str,
+    is_mpc: bool
 ):
+
+    input_path = config["OUTPUT"]["roid_module_output"]
+    split_night = last_night.split("-")
+    input_path = os.path.join(
+        input_path,
+        f"year={split_night[0]}",
+        f"month={split_night[1]}",
+        f"day={split_night[2]}",
+    )
+
+    mode = str(config["OUTPUT"]["roid_path_mode"])
+
     if mode == "local":
         # load alerts from local
-        sso_night = pd.read_parquet(read_path)
-
+        sso_night = pd.read_parquet(input_path)
+        if "candidate" in sso_night:
+            candidate_pdf = pd.json_normalize(sso_night["candidate"]).drop("candid", axis=1)
+            sso_night = pd.concat(
+                [sso_night, candidate_pdf],
+                axis=1,
+            )
     elif mode == "spark":
-        assert (
-            output_path is not None
-        ), "The argument 'output_path' is None.\nYou must set an output_path when loading data with spark"
+
+        output_path_spark = os.path.join(
+            output_path, 
+            f"year={split_night[0]}",
+            f"month={split_night[1]}",
+            f"day={split_night[2]}"
+        )
 
         # load alerts from spark
         master_manager = config["SOLVE_ORBIT_PARAMS"]["manager"]
@@ -387,7 +407,8 @@ def get_last_roid_streaming_alert(
         )
 
         application += " " + master_manager
-        application += " " + read_path
+        application += " " + input_path
+        application += " " + output_path_spark
         application += " " + is_mpc
 
         spark_submit = "spark-submit \
@@ -420,8 +441,8 @@ def get_last_roid_streaming_alert(
             logger.info(process.stdout)
             exit()
 
-        sso_night = pd.read_parquet(output_path)
-        os.remove(output_path)
+        sso_night = pd.read_parquet(output_path_spark)
+        os.remove(output_path_spark)
 
     else:
         raise ValueError(f"mode {mode} not exist")

fink_fat/command_line/cli_main/associations.py

@@ -48,7 +48,7 @@ def cli_associations(arguments, config, output_path):
 
     if object_class == "SSO fitroid":
         fitroid_associations(arguments, config, logger, output_path)
-        exit()
+        return
 
     tr_df_path = os.path.join(output_path, "trajectory_df.parquet")
     obs_df_path = os.path.join(output_path, "old_obs.parquet")

fink_fat/command_line/cli_main/fitroid.py

@@ -102,23 +102,22 @@ def fitroid_associations(
     path_fit_roid = os.path.join(output_path, "fit_roid.parquet")
     path_trajectory_df = os.path.join(output_path, "trajectory_df.parquet")
 
-    input_path = config["OUTPUT"]["roid_module_output"]
-    split_night = last_night.split("-")
-    input_path = os.path.join(
-        input_path,
-        f"year={split_night[0]}",
-        f"month={split_night[1]}",
-        f"day={split_night[2]}",
-    )
-
-    output_path_spark = None
-    if config["OUTPUT"]["roid_path_mode"] == "spark":
-        output_path_spark = os.path.join(output_path, "spark_alerts_night.parquet")
-
     # load the alerts from the last streaming night (roid science module with fink-fat must have been run)
     alerts_night = get_last_roid_streaming_alert(
-        config["OUTPUT"]["roid_path_mode"], is_mpc, input_path, output_path_spark
+        config, last_night, output_path, is_mpc
     )
+    if arguments["--verbose"]:
+        logger.info(
+            """
+STATISTICS - STREAMING NIGHT
+-----------------------
+                    
+roid count:
+{}
+                    """.format(
+                alerts_night["roid"].value_counts().sort_index()
+            )
+        )
 
     trajectory_orb, orbits = get_default_input()
 

fink_fat/test/cli_test/fitroid_test.conf

@@ -0,0 +1,73 @@
+[TW_PARAMS]
+# all params are in days
+trajectory_keep_limit=15
+old_observation_keep_limit=2
+trajectory_2_points_keep_limit=8
+orbit_keep_limit=30
+predict_function_keep_limit=10
+
+[ASSOC_PARAMS]
+# arcsecond
+intra_night_separation=120
+
+intra_night_magdiff_limit_same_fid=0.2
+intra_night_magdiff_limit_diff_fid=0.8
+
+# deg/day
+inter_night_separation=0.3
+
+inter_night_magdiff_limit_same_fid=0.1
+inter_night_magdiff_limit_diff_fid=0.5
+
+# degree
+maximum_angle=1
+
+# arcmin
+error_radius=15
+
+# arcsecond
+orbit_assoc_radius=600.0
+
+# if true, use dbscan for intra_night associations, use legacy fink_fat otherwise
+use_dbscan=False
+
+# roid_mpc = (True | False) if true load mpc alerts for associations with the fink alert stream.
+#                           load sso candidates otherwise (for fink_fat version >= 1.0)
+roid_mpc=True
+
+[ASSOC_PERF]
+store_kd_tree=false
+
+[SOLVE_ORBIT_PARAMS]
+n_triplets=30
+noise_ntrials=20
+prop_epoch=None
+orbfit_verbose=3
+
+orbfit_limit=6
+cpu_count=1
+ram_dir=""
+manager=local[4]
+principal=lsst
+secret=secret
+role=lsst
+exec_env='/home/julien.peloton'
+driver_memory=4
+executor_memory=16
+max_core=64
+executor_core=8
+orbfit_path=$ORBFIT_HOME
+
+
+[ASSOC_SYSTEM]
+tracklets_with_trajectories=true
+trajectories_with_new_observations=true
+tracklets_with_old_observations=true
+new_observations_with_old_observations=true
+
+[OUTPUT]
+association_output_file=fink_fat/test/cli_test/fink_fat_out
+
+# roid_path_mode = ('local' | 'spark')
+roid_path_mode=local
+roid_module_output=fink_fat/test/cli_test/roid_module_output

fink_fat/test/cli_test/run_roid.py

@@ -0,0 +1,69 @@
+from pyspark.sql import functions as F
+import os
+import sys
+
+from fink_utils.broker.sparkUtils import init_sparksession
+from fink_utils.spark.utils import concat_col
+
+from fink_science.asteroids.processor import roid_catcher
+
+from fink_fat.command_line.utils_cli import init_cli
+from pyspark.sql import SparkSession
+
+
+def addFileToSpark(spark: SparkSession, fitroid_path, orbit_path):
+    if os.path.exists(orbit_path):
+        spark.sparkContext.addFile(orbit_path)
+    if os.path.exists(fitroid_path):
+        spark.sparkContext.addFile(fitroid_path)
+
+
+if __name__ == "__main__":
+    datapath = "fink_fat/test/cli_test/small_sso_dataset"
+
+    year, month, day = sys.argv[1:]
+    config, output_path = init_cli({"--config": "fink_fat/test/cli_test/fitroid_test.conf"})
+
+    path_orbit = os.path.join(output_path, "orbital.parquet")
+    path_fit_roid = os.path.join(output_path, "fit_roid.parquet")
+
+    path_sso = os.path.join(datapath, f"year={int(year):04d}/month={int(month):02d}/day={int(day):02d}")
+
+    spark = init_sparksession("fink_fat_roid")
+    addFileToSpark(spark, path_fit_roid, path_orbit)
+
+    df = spark.read.load(path_sso)
+    what = ["jd", "magpsf"]
+    prefix = "c"
+    what_prefix = [prefix + i for i in what]
+    for colname in what:
+        df = concat_col(df, colname, prefix=prefix)
+
+    args = [
+        "candidate.ra",
+        "candidate.dec",
+        "candidate.jd",
+        "candidate.magpsf",
+        "candidate.candid",
+        "cjd",
+        "cmagpsf",
+        "candidate.fid",
+        "candidate.ndethist",
+        "candidate.sgscore1",
+        "candidate.ssdistnr",
+        "candidate.distpsnr1",
+        F.lit(float(config["ASSOC_PARAMS"]["error_radius"])),
+        F.lit(float(config["ASSOC_PARAMS"]["inter_night_magdiff_limit_same_fid"])),
+        F.lit(float(config["ASSOC_PARAMS"]["inter_night_magdiff_limit_diff_fid"])),
+        F.lit(int(config["TW_PARAMS"]["orbit_keep_limit"])),
+        F.lit(float(config["ASSOC_PARAMS"]["orbit_assoc_radius"])),
+        F.lit(True),
+    ]
+    df = df.withColumn("ff_roid", roid_catcher(*args))
+    df = df.drop(*what_prefix)
+
+    df = df.withColumn("year", F.lit(f"{int(year):04d}"))
+    df = df.withColumn("month", F.lit(f"{int(month):02d}"))
+    df = df.withColumn("day", F.lit(f"{int(day):02d}"))
+
+    df.write.partitionBy("year", "month", "day").parquet(config["OUTPUT"]["roid_module_output"])