Merge pull request #267 from alercebroker/refactor/mjd-filter-skip

feat: created a flag to make the mjd filtering optional in lc step
alercebroker · Nov 6, 2023 · 1825633 · 1825633
2 parents be25ab3 + e4b228f
commit 1825633
Show file tree

Hide file tree

Showing 5 changed files with 13 additions and 5 deletions.
diff --git a/charts/lightcurve-step/templates/configmap.yaml b/charts/lightcurve-step/templates/configmap.yaml
@@ -20,3 +20,4 @@ data:
   sql-secret-name: {{ .Values.configmap.psqlSecretName }}
   mongo-secret: {{ .Values.configmap.mongoSecret }}
   logging-debug: "{{ .Values.configmap.loggingDebug }}"
+  skip-mjd-filter: "{{ .Values.configmap.skipMjdFilter }}"
diff --git a/charts/lightcurve-step/templates/deployment.yaml b/charts/lightcurve-step/templates/deployment.yaml
@@ -101,6 +101,11 @@ spec:
                 configMapKeyRef:
                   name: {{ include "lightcurve-step.fullname" . }}
                   key: logging-debug
+            - name: SKIP_MJD_FILTER
+              valueFrom:
+                configMapKeyRef:
+                  name: {{ include "lightcurve-step.fullname" . }}
+                  key: skip-mjd-filter
             {{- if .Values.secrets.kafkaAuth.consumer.enabled }}
             - name: CONSUMER_KAFKA_USERNAME
               valueFrom:

diff --git a/charts/lightcurve-step/values.yaml b/charts/lightcurve-step/values.yaml
@@ -58,6 +58,7 @@ configmap:
   useProfiling: ""
   mongoSecret: ""
   pyroscopeServer: "http://pyroscope.pyroscope:4040"
+  skipMjdFilter: ""
 
 secrets:
   kafkaAuth:

diff --git a/correction_step/correction/_step/step.py b/correction_step/correction/_step/step.py
@@ -11,6 +11,7 @@
 
 from pprint import pprint
 
+
 class CorrectionStep(GenericStep):
     """Step that applies magnitude correction to new alert and previous candidates.
 
@@ -90,13 +91,10 @@ def pre_execute(cls, messages: list[dict]) -> dict:
 
     @classmethod
     def execute(cls, message: dict) -> dict:
-        pprint(message)
         corrector = Corrector(message["detections"])
         detections = corrector.corrected_as_records()
         non_detections = pd.DataFrame(message["non_detections"]).drop_duplicates(["oid", "fid", "mjd"])
         coords = corrector.coordinates_as_records()
-        # delet dis
-        pprint(coords)
         return {
             "detections": detections,
             "non_detections": non_detections.to_dict("records"),

diff --git a/lightcurve-step/lightcurve_step/step.py b/lightcurve-step/lightcurve_step/step.py
@@ -1,5 +1,6 @@
 import logging
 import numpy as np
+import os
 import pandas as pd
 import pickle
 from typing import List
@@ -53,6 +54,7 @@ def pre_execute(cls, messages: List[dict]) -> dict:
             last_mjds[aid] = max(last_mjds.get(aid, 0), msg["detections"][0]["mjd"])
             detections.extend([det | {"new": True} for det in msg["detections"]])
             non_detections.extend(msg["non_detections"])
+
         logger = logging.getLogger("alerce.LightcurveStep")
         logger.debug(f"Received {len(detections)} detections from messages")
         return {
@@ -231,8 +233,9 @@ def serialize_dia_object(ef: dict):
             except KeyError:
                 nd = []
             dets["extra_fields"] = dets["extra_fields"].apply(serialize_dia_object)
-            mjds = result["last_mjds"]
-            dets = dets[dets["mjd"] <= mjds[aid]]
+            if not os.getenv("SKIP_MJD_FILTER", "false") == "true":
+                mjds = result["last_mjds"]
+                dets = dets[dets["mjd"] <= mjds[aid]]
             output.append(
                 {
                     "aid": aid,