Ad/multiple_models

fink_filters/filter_anomaly_notification/filter.py

@@ -25,16 +25,10 @@
 
 
 def anomaly_notification_(
-    df_proc,
-    threshold=10,
-    send_to_tg=False,
-    channel_id=None,
-    send_to_slack=False,
-    channel_name=None,
-    trick_par=10,
-    cut_coords=False,
-    history_period=90,
-):
+        df_proc, threshold=10,
+        send_to_tg=False, channel_id=None,
+        send_to_slack=False, channel_name=None,
+        trick_par=10, cut_coords=False, history_period=90, send_to_anomaly_base=False, model=''):
     """ Create event notifications with a high `anomaly_score` value
 
     Notes
@@ -74,6 +68,15 @@ def anomaly_notification_(
     history_period: int
             Time period in days for which the number
             of references is calculated
+    model: str
+        Name of the model used.
+        Name must start with a ‘_’ and be ‘_{user_name}’,
+        where user_name is the user name of the model at https://anomaly.fink-portal.org/.
+    send_to_anomaly_base: bool
+        If True, notifications are uploaded to
+        https://anomaly.fink-portal.org/ in the selected model's
+        account. Only works for model != ‘’
+
 
     Returns
     ----------
@@ -88,7 +91,7 @@ def anomaly_notification_(
     >>> from fink_science.anomaly_detection.processor import anomaly_score
 
     >>> df = spark.read.format('parquet').load('datatest/regular')
-
+    >>> MODELS = ['', '_beta'] # '' corresponds to the model for a telegram channel
     >>> what = [
     ...     'jd', 'fid', 'magpsf', 'sigmapsf',
     ...     'magnr', 'sigmagnr', 'isdiffpos', 'distnr']
@@ -108,15 +111,22 @@ def anomaly_notification_(
     ...     'cdistnr', 'cmagnr', 'csigmagnr', 'cisdiffpos']
 
     >>> df = df.withColumn('lc_features', extract_features_ad(*ad_args))
-    >>> df = df.withColumn("anomaly_score", anomaly_score("lc_features"))
+    >>> for model in MODELS:
+    ...     df = df.withColumn(f'anomaly_score{model}', anomaly_score("lc_features", F.lit(model)))
 
-    >>> df_proc = df.select(
-    ...     'objectId', 'candidate.ra',
-    ...     'candidate.dec', 'candidate.rb',
-    ...     'anomaly_score', 'timestamp')
-    >>> df_out = anomaly_notification_(df_proc)
+    >>> for model in MODELS:
+    ...     df_proc = df.select(
+    ...         'objectId', 'candidate.ra',
+    ...         'candidate.dec', 'candidate.rb',
+    ...         f'anomaly_score{model}', 'timestamp')
+    ...     df_out = anomaly_notification_(df_proc, send_to_tg=False,
+    ...     send_to_slack=False, send_to_anomaly_base=True, model=model)
 
     # Disable communication
+    >>> df_proc = df.select(
+    ...         'objectId', 'candidate.ra',
+    ...         'candidate.dec', 'candidate.rb',
+    ...         'anomaly_score', 'timestamp')
     >>> pdf_anomalies = anomaly_notification_(df_proc, threshold=10,
     ...     send_to_tg=False, channel_id=None,
     ...     send_to_slack=False, channel_name=None)
@@ -140,39 +150,37 @@ def anomaly_notification_(
             return pd.DataFrame()
 
     # Compute the median for the night
-    med = df_proc.select("anomaly_score").approxQuantile("anomaly_score", [0.5], 0.05)
+    buf_df = df_proc.select(f'anomaly_score{model}')
+    med = buf_df.approxQuantile(f'anomaly_score{model}', [0.5], 0.05)
     med = round(med[0], 2)
 
     # Extract anomalous objects
 
-    pdf_anomalies_ext = (
-        df_proc.sort(["anomaly_score"], ascending=True)
-        .limit(trick_par * threshold)
-        .toPandas()
-    )
-    pdf_anomalies_ext = pdf_anomalies_ext.drop_duplicates(["objectId"])
-    upper_bound = np.max(pdf_anomalies_ext["anomaly_score"].values[:threshold])
-    pdf_anomalies = pdf_anomalies_ext[pdf_anomalies_ext["anomaly_score"] <= upper_bound]
+    pdf_anomalies_ext = df_proc.sort([f'anomaly_score{model}'], ascending=True).limit(trick_par * threshold).toPandas()
+    pdf_anomalies_ext = pdf_anomalies_ext.drop_duplicates(['objectId'])
+    upper_bound = np.max(pdf_anomalies_ext[f'anomaly_score{model}'].values[:threshold])
+    pdf_anomalies = pdf_anomalies_ext[pdf_anomalies_ext[f'anomaly_score{model}'] <= upper_bound].head(10)
 
     history_objects = filter_utils.get_an_history(history_period)
 
-    tg_data, slack_data = [], []
+    tg_data, slack_data, base_data = [], [], []
+
     for _, row in pdf_anomalies.iterrows():
         gal = SkyCoord(
             ra=row.ra * u.degree, dec=row.dec * u.degree, frame="icrs"
         ).galactic
         oid = filter_utils.get_OID(row.ra, row.dec)
-        t1a = f"ID: [{row.objectId}](https://fink-portal.org/{row.objectId})"
-        t1b = f"ID: <https://fink-portal.org/{row.objectId}|{row.objectId}>"
-        t_oid_1a = f"DR OID (<1''): [{oid}](https://ztf.snad.space/view/{oid})"
+        t1a = f'**ID**: [{row.objectId}](https://fink-portal.org/{row.objectId})'
+        t1b = f'ID: <https://fink-portal.org/{row.objectId}|{row.objectId}>'
+        t_oid_1a = f"**DR OID (<1'')**: [{oid}](https://ztf.snad.space/view/{oid})"
         t_oid_1b = f"DR OID (<1''): <https://ztf.snad.space/view/{oid}|{oid}>"
-        t2_ = f"GAL coordinates: {round(gal.l.deg, 6)},   {round(gal.b.deg, 6)}"
-        t_ = f"""
-EQU: {row.ra},   {row.dec}"""
+        t2_ = f'**GAL coordinates**: {round(gal.l.deg, 6)},   {round(gal.b.deg, 6)}'
+        t_ = f'''
+**EQU**: {row.ra},   {row.dec}'''
         t2_ += t_
-        t3_ = f"UTC: {str(row.timestamp)[:-3]}"
-        t4_ = f"Real bogus: {round(row.rb, 2)}"
-        t5_ = f"Anomaly score: {round(row.anomaly_score, 2)}"
+        t3_ = f'**UTC**: {str(row.timestamp)[:-3]}'
+        t4_ = f'**Real bogus**: {round(row.rb, 2)}'
+        t5_ = f'**Anomaly score**: {round(row[f"anomaly_score{model}"], 2)}'
         if row.objectId in history_objects:
             t5_ += f"""
 Detected as top-{threshold} in the last {history_period} days: {history_objects[row.objectId]} {'times' if history_objects[row.objectId] > 1 else 'time'}."""
@@ -183,39 +191,43 @@ def anomaly_notification_(
         cutout.seek(0)
         cutout_perml = f"<{cutout_perml}|{' '}>"
         curve_perml = f"<{curve_perml}|{' '}>"
-        tg_data.append(
-            (
-                f"""{t1a}
+        if model == '':
+            tg_data.append((f'''{t1a}
 {t_oid_1a}
 {t2_}
 {t3_}
 {t4_}
-{t5_}""",
-                cutout,
-                curve,
-            )
-        )
-        slack_data.append(f"""==========================
+{t5_}''', cutout, curve))
+            slack_data.append(f'''==========================
 {t1b}
 {t_oid_1b}
 {t2_}
 {t3_}
 {t4_}
 {t5_}
-{cutout_perml}{curve_perml}""")
-    init_msg = f"Median anomaly score overnight: {med}."
-    if cut_coords:
+{cutout_perml}{curve_perml}''')
+        base_data.append((row.objectId, f'''{t1a}
+{t_oid_1a}
+{t2_}
+{t3_}
+{t4_}
+{t5_}'''.replace('\n', '  \n'), cutout, curve))
+
+    init_msg = f'Median anomaly score overnight: {med}.'
+    if cut_coords and model == '':
         init_msg += f"""
 (of the objects in the sky area)
 Sky area:
-    1) delta <= 20°
-    2) alpha ∈ (160°, 240°)
+1) delta <= 20°
+2) alpha ∈ (160°, 240°)
 Total number of objects per night in the area: {cut_count}.
 """
     if send_to_slack:
         filter_utils.msg_handler_slack(slack_data, channel_name, init_msg)
     if send_to_tg:
         filter_utils.msg_handler_tg(tg_data, channel_id, init_msg)
+    if model != '':
+        filter_utils.load_to_anomaly_base(base_data, model)
     return pdf_anomalies