Merge pull request #10 from alercebroker/feat/metrics

feat(custom metrics): add ra,dec to metrics

settings.py

@@ -58,11 +58,58 @@
     "STEP_COMMENTS": os.getenv("STEP_COMMENTS", ""),
 }
 
+METRICS_CONFIG = {
+    "CLASS": "apf.metrics.KafkaMetricsProducer",
+    "EXTRA_METRICS": [
+        {"key": "candid", "format": lambda x: str(x)},
+        {"key": "candid", "format": lambda x: str(x)},
+    ],
+    "PARAMS": {
+        "PARAMS": {
+            "bootstrap.servers": os.environ["METRICS_HOST"],
+            "auto.offset.reset": "smallest"},
+        "TOPIC": os.environ["METRICS_TOPIC"],
+        "SCHEMA": {
+            "$schema": "http://json-schema.org/draft-07/schema",
+            "$id": "http://example.com/example.json",
+            "type": "object",
+            "title": "The root schema",
+            "description": "The root schema comprises the entire JSON document.",
+            "default": {},
+            "examples": [
+                {"timestamp_sent": "2020-09-01", "timestamp_received": "2020-09-01"}
+            ],
+            "required": ["timestamp_sent", "timestamp_received"],
+            "properties": {
+                "timestamp_sent": {
+                    "$id": "#/properties/timestamp_sent",
+                    "type": "string",
+                    "title": "The timestamp_sent schema",
+                    "description": "Timestamp sent refers to the time at which a message is sent.",
+                    "default": "",
+                    "examples": ["2020-09-01"],
+                },
+                "timestamp_received": {
+                    "$id": "#/properties/timestamp_received",
+                    "type": "string",
+                    "title": "The timestamp_received schema",
+                    "description": "Timestamp received refers to the time at which a message is received.",
+                    "default": "",
+                    "examples": ["2020-09-01"],
+                },
+            },
+            "additionalProperties": True,
+        },
+    },
+}
+
+
 # Step Configuration
 STEP_CONFIG = {
     "DB_CONFIG": DB_CONFIG,
     "CONSUMER_CONFIG": CONSUMER_CONFIG,
     "PRODUCER_CONFIG": PRODUCER_CONFIG,
     "N_PROCESS": os.getenv("N_PROCESS"),
     "STEP_METADATA": STEP_METADATA,
+    "METRICS_CONFIG": METRICS_CONFIG,
 }

sorting_hat_step/step.py

@@ -48,6 +48,13 @@ def produce(self, alerts: pd.DataFrame) -> None:
         alerts = alerts.replace(
             {np.nan: None}
         )  # transform np.nan to None (only for produce proposals)
+
+        self.metrics["ra"] = alerts["ra"].tolist()
+        self.metrics["dec"] = alerts["dec"].tolist()
+        self.metrics["oid"] = alerts["oid"].tolist()
+        self.metrics["tid"] = alerts["tid"].tolist()
+        self.metrics["aid"] = alerts["aid"].tolist()
+
         for index, alert in alerts.iterrows():
             alert = alert.to_dict()
             alert["rfid"] = None if alert["rfid"] is None else int(alert["rfid"])