Merge b79145e into 8890b9e

nuztf/neutrino_scanner.py

@@ -4,15 +4,14 @@
 import logging
 import os
 
+import healpy as hp
 import numpy as np
 import yaml
 from astropy.time import Time
-from tqdm import tqdm
-from ztfquery.io import LOCALSOURCE
-
-import healpy as hp
 from nuztf.base_scanner import BaseScanner
 from nuztf.parse_nu_gcn import find_gcn_no, get_latest_gcn, parse_gcn_circular
+from tqdm import tqdm
+from ztfquery.io import LOCALSOURCE
 
 nu_candidate_output_dir = os.path.join(LOCALSOURCE, "neutrino_candidates")
 

nuztf/parse_nu_gcn.py

@@ -192,22 +192,34 @@ def parse_radec(string: str):
 
 
 def parse_gcn_circular(gcn_number: int):
-    """ """
-    url = f"https://gcn.gsfc.nasa.gov/gcn3/{gcn_number}.gcn3"
-    response = requests.get(url)
+    """
+    Parses the handwritten text of a given GCN;
+    extracts author, time and RA/Dec (with errors)
+    """
+
     returndict = {}
     mainbody_starts_here = 999
-    splittext = response.text.splitlines()
+
+    endpoint = f"https://gcn.nasa.gov/circulars/{gcn_number}/json"
+    res = requests.get(endpoint)
+    res_json = res.json()
+
+    subject = res_json.get("subject")
+    submitter = res_json.get("submitter")
+    body = res_json.get("body")
+
+    base = submitter.split("at")[0].split(" ")
+    author = [x for x in base if x != ""][1]
+    returndict.update({"author": author})
+
+    name = subject.split(" - ")[0]
+    returndict.update({"name": name})
+
+    splittext = body.splitlines()
     splittext = list(filter(None, splittext))
+
     for i, line in enumerate(splittext):
-        if "SUBJECT" in line:
-            name = line.split(" - ")[0].split(": ")[1]
-            returndict.update({"name": name})
-        elif "FROM" in line:
-            base = line.split("at")[0].split(": ")[1].split(" ")
-            author = [x for x in base if x != ""][1]
-            returndict.update({"author": author})
-        elif (
+        if (
             ("RA" in line or "Ra" in line)
             and ("DEC" in splittext[i + 1] or "Dec" in splittext[i + 1])
             and i < mainbody_starts_here

nuztf/skymap_scanner.py

@@ -12,12 +12,15 @@
 import yaml
 from astropy.time import Time
 from astropy_healpix import HEALPix
-from tqdm import tqdm
-from ztfquery.io import LOCALSOURCE
-
-from nuztf.ampel_api import ampel_api_lightcurve, ampel_api_skymap
+from nuztf.ampel_api import (
+    ampel_api_acknowledge_chunk,
+    ampel_api_lightcurve,
+    ampel_api_skymap,
+)
 from nuztf.base_scanner import BaseScanner
 from nuztf.skymap import Skymap
+from tqdm import tqdm
+from ztfquery.io import LOCALSOURCE
 
 
 class RetractionError(Exception):
@@ -94,11 +97,15 @@ def get_alerts(self):
         self.queue = []
 
         resume = True
-        chunk_size = 8000
+        chunk_size = 2000
         resume_token = None
 
+        i = 0
+        total_chunks = 0
+        t0 = time.time()
+
         while resume:
-            query_res, resume_token = ampel_api_skymap(
+            res, resume_token, chunk_id, remaining_chunks = ampel_api_skymap(
                 pixels=self.cone_ids,
                 nside=self.cone_nside,
                 t_min_jd=self.t_min.jd,
@@ -108,15 +115,31 @@ def get_alerts(self):
                 resume_token=resume_token,
                 warn_exceeding_chunk=False,
             )
-            self.queue.extend(query_res)
+            self.queue.extend(res)
+
+            ampel_api_acknowledge_chunk(resume_token=resume_token, chunk_id=chunk_id)
 
-            if len(query_res) < chunk_size:
+            if i == 0:
+                total_chunks = remaining_chunks + 1
+                self.logger.info(f"Total chunks: {total_chunks}")
+
+            if remaining_chunks % 50 == 0 and remaining_chunks != 0:
+                t1 = time.time()
+                processed_chunks = total_chunks - remaining_chunks
+                time_per_chunk = (t1 - t0) / processed_chunks
+                remaining_time = time_per_chunk * remaining_chunks
+                self.logger.info(
+                    f"Remaining chunks: {remaining_chunks}. Estimated time to finish: {remaining_time/60:.0f} min"
+                )
+
+            if len(res) < chunk_size:
                 resume = False
                 self.logger.info("Done.")
             else:
-                self.logger.info(
+                self.logger.debug(
                     f"Chunk size reached ({chunk_size}), commencing next query."
                 )
+            i += 1
 
         time_healpix_end = time.time()
         time_healpix = time_healpix_end - time_healpix_start