add einstein probe support

fink_mm/__init__.py

@@ -12,6 +12,6 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-__version__ = "0.21.1"
+__version__ = "0.22.0"
 __distribution_schema_version__ = "1.3"
 __observatory_schema_version__ = "1.1"

fink_mm/conftest.py

@@ -143,10 +143,10 @@ def init_icecube(doctest_namespace):
 @pytest.fixture(autouse=True)
 def init_LVK(doctest_namespace):
     doctest_namespace["lvk_initial_path"] = (
-        "fink_mm/test/test_data/VODB/lvk/initial.txt"
+        "fink_mm/test/test_data/VODB/lvk/initial.json"
     )
-    doctest_namespace["lvk_update_path"] = "fink_mm/test/test_data/VODB/lvk/update.txt"
-    doctest_namespace["lvk_test_path"] = "fink_mm/test/test_data/VODB/lvk/test.txt"
+    doctest_namespace["lvk_update_path"] = "fink_mm/test/test_data/VODB/lvk/update.json"
+    doctest_namespace["lvk_test_path"] = "fink_mm/test/test_data/VODB/lvk/test.json"
 
     lvk_initial = json_to_class(
         load_json_from_path(doctest_namespace["lvk_initial_path"], logger)
@@ -164,6 +164,19 @@ def init_LVK(doctest_namespace):
     doctest_namespace["lvk_test"] = lvk_test
 
 
+@pytest.fixture(autouse=True)
+def init_EP(doctest_namespace):
+    doctest_namespace["ep_initial_path"] = (
+        "fink_mm/test/test_data/VODB/einsteinprobe/alert.example.json"
+    )
+
+    ep_alert = json_to_class(
+        load_json_from_path(doctest_namespace["ep_initial_path"], logger)
+    )
+
+    doctest_namespace["ep_alert"] = ep_alert
+
+
 @pytest.fixture(autouse=True, scope="session")
 def init_spark(doctest_namespace):
     from astropy.time import Time

fink_mm/observatory/EinsteinProbe/Einsteinprobe.py

@@ -0,0 +1,68 @@
+from fink_mm.observatory.observatory import Observatory
+import os.path as path
+from fink_mm.observatory import OBSERVATORY_PATH
+
+
+class Einsteinprobe(Observatory):
+
+    def __init__(self, notice: str):
+        """
+        Initialise a LVK class
+
+        Parameters
+        ----------
+        voevent: ObjectifiedElement
+
+        Example
+        -------
+        >>> ep_event = load_json_from_path(ep_initial_path, logger)
+        >>> ep_alert = json_to_class(ep_event)
+        >>> type(ep_alert)
+        <class 'Einsteinprobe.Einsteinprobe'>
+        """
+        super().__init__(
+            path.join(OBSERVATORY_PATH, "EinsteinProbe", "einsteinprobe.json"), notice
+        )
+
+    def is_observation(self) -> bool:
+        """
+        Test if the json is of type observation.
+
+        Parameters
+        ----------
+        voevent : voevent object
+            The voevent object.
+
+        Returns
+        -------
+        is_observation : boolean
+            Return True if the voevent is of observation type, otherwise return False
+
+        Examples
+        --------
+        >>> ep_alert.is_observation()
+        True
+        """
+        return True
+
+    def get_trigger_id(self):
+        """
+        Get the triggerId of the voevent
+
+        Example
+        -------
+        >>> ep_alert.get_trigger_id()
+        '01708973486'
+        """
+        return self.voevent["id"][0]
+
+    def err_to_arcminute(self):
+        """
+        Return the error radius of the voevent in arcminute
+
+        Example
+        -------
+        >>> ep_alert.err_to_arcminute()
+        1.2
+        """
+        return self.voevent["ra_dec_error"] * 60

fink_mm/observatory/EinsteinProbe/einsteinprobe.json

@@ -0,0 +1,10 @@
+{
+    "name": "EinsteinProbe",
+    "//gcn_description": "https://gcn.nasa.gov/missions/einstein-probe",
+    "gcn_file_format": "json",
+    "packet_type": ["WXT"],
+    "kafka_topics": [
+        "gcn.notices.einstein_probe.wxt.alert"
+    ],
+    "grb_detection_rate": -1.0
+}

fink_mm/observatory/__init__.py

@@ -187,6 +187,8 @@ def json_to_class(gcn: dict) -> observatory.Observatory:
     """
     if "superevent_id" in gcn:
         return __OBS_CLASS["lvk"](gcn)
+    elif gcn["instrument"] == "WXT":
+        return __OBS_CLASS["einsteinprobe"](gcn)
     else:
         raise Exception("unknown json format")
 

fink_mm/observatory/observatory.py

@@ -142,7 +142,6 @@ def detect_instruments(self) -> str:
         >>> fermi_gbm.detect_instruments()
         'GBM'
         """
-
         return self.voevent.attrib["ivorn"].split("#")[1].split("_")[0]
 
     @abstractmethod

fink_mm/test/test_data/VODB/einsteinprobe/alert.example.json

@@ -0,0 +1,13 @@
+{
+  "$schema": "https://gcn.nasa.gov/schema/main/gcn/notices/einstein_probe/wxt/alert.schema.json",
+  "instrument": "WXT",
+  "trigger_time": "2024-03-01T21:46:05.13Z",
+  "id": ["01708973486"],
+  "ra": 120,
+  "dec": 40,
+  "ra_dec_error": 0.02,
+  "image_energy_range": [0.5, 4],
+  "net_count_rate": 1,
+  "image_snr": 1,
+  "additional_info": "The net count rate is derived from an accumulated image (up to 20 min) in 0.5-4 keV, assuming a constant flux. However, it can be significantly lower than the actual count rate of a burst with a duration much shorter than 20 min."
+}