Draft: Ingestion of SVOM ECLAIRS notices into LLAI workflow

This MR will ingest SVOM GRBs alerts detected by ECLAIRS instrument.

gwcelery/tasks/external_triggers.py

@@ -723,8 +723,7 @@ def _create_replace_external_event_and_skymap(
 def _kafka_to_voevent(alert):
     """Parse an alert sent via Kafka from a MOU partner in our joint
     subthreshold targeted search and convert to an equivalent XML string
-    GCN VOEvent. Also, include handling for SVOM/ECLAIRs mission alerts
-    with initial and update types.
+    GCN VOEvent. Also, include handling for SVOM/ECLAIRs mission alert.
 
     Parameters
     ----------
@@ -738,11 +737,11 @@ def _kafka_to_voevent(alert):
 
     """
     # Define basic values
-    # Support for both key types
     pipeline = alert['mission']
     start_time = alert['trigger_time']
     alert_time = alert['alert_datetime']
     far = alert['far']
+    snr = alert['rate_snr']
     duration = alert['rate_duration']
     id = '_'.join(str(x) for x in alert['id'])
     # Use central time since starting time is not well measured
@@ -755,7 +754,7 @@ def _kafka_to_voevent(alert):
     # sky localization may not be available
     ra = alert.get('ra')
     dec = alert.get('dec')
-    # Try to get dec first then ra, None if both misssing
+    # Try to get dec first then ra, None if both missing
     error = alert.get('dec_uncertainty')
     if error is None:
         error = alert.get('ra_uncertainty')
@@ -766,22 +765,34 @@ def _kafka_to_voevent(alert):
     if ra is None or dec is None or error is None:
         ra, dec, error = 0., 0., 0.
 
-    # Load template
+    # Load template based on the mission
     # FIXME: Need to adjust for targeted Fermi-Swift and SVOM/ECLAIRS GRBs
-    fname = str(Path(__file__).parent /
-                '../tests/data/{}_subgrbtargeted_template.xml'.format(
-                    pipeline.lower()))
-    root = etree.parse(fname)
+    if pipeline.lower() in ['fermi', 'swift']:
+        fname = f'{pipeline.lower()}_subgrbtargeted_template.xml'
+        ivorn_suffix = 'targeted_subthreshold'
+    else:
+        fname = 'svom_grb_template.xml'
+        ivorn_suffix = 'grb'
+
+    # Construct the full path to the template
+    template_path = Path(__file__).parent / f'../tests/data/{fname}'
+    if not template_path.is_file():
+        raise FileNotFoundError(f'Template file {fname} not found '
+                                f'in ../tests/data')
+    root = etree.parse(str(template_path))
 
     # Update template values
-    # Change ivorn to indicate this is a subthreshold targeted event
-    root.xpath('.')[0].attrib['ivorn'] = \
-        'ivo://lvk.internal/{0}#targeted_subthreshold-{1}'.format(
-            pipeline.lower(), trigger_time).encode()
+    # Change ivorn to indicate this is subthreshold targeted event and SVOM grb
+    ivorn = \
+        f'ivo://lvk.internal/{pipeline.lower()}#{ivorn_suffix}-{trigger_time}'
+
+    root.xpath('.')[0].attrib['ivorn'] = ivorn
 
     # Update ID
-    root.find("./What/Param[@name='TrigID']").attrib['value'] = \
-        id.encode()
+    if pipeline.lower() in ['fermi', 'swift']:
+        root.find("./What/Param[@name='TrigID']").attrib['value'] = str(id)
+    else:
+        root.find("./What/Param[@name='Burst_Id']").attrib['value'] = str(id)
 
     # Change times to chosen time
     root.find("./Who/Date").text = str(alert_time).encode()
@@ -789,11 +800,21 @@ def _kafka_to_voevent(alert):
                "ObservationLocation/AstroCoords/Time/TimeInstant/"
                "ISOTime")).text = str(trigger_time).encode()
 
-    root.find("./What/Param[@name='FAR']").attrib['value'] = \
-        str(far).encode()
+    if pipeline.lower() in ['fermi', 'swift']:
+        root.find("./What/Param[@name='FAR']").attrib['value'] = \
+            (str(far).encode())
+    else:
+        root.find("./Group[@name='Detection_Info']"
+                  "/Param[@name='SNR']").attrib['value'] = \
+            (str(snr).encode())
 
-    root.find("./What/Param[@name='Integ_Time']").attrib['value'] = \
-        str(duration).encode()
+    if pipeline.lower() in ['fermi', 'swift']:
+        root.find("./What/Param[@name='Integ_Time']").attrib['value'] = \
+            str(duration).encode()
+    else:
+        root.find("./Group[@name='Svom_Identifiers']"
+                  "/Param[@name='Timescale']").attrib['value'] = \
+            (str(duration).encode())
 
     # Sky position
     root.find(("./WhereWhen/ObsDataLocation/"