#None: Update v0.4.20

Author: mickolaua

RELEASE_NOTES

@@ -189,6 +189,19 @@ in previous functions. This helps with allocating time slots. It is garantied
 that all the targets in the list can be observed. However, the time slots are currently 
 not send with observation programs.
 
+## 0.4.20
+
+- Added comment to `CfgOption`
+- Added comments for all options
+- Added QueueHub to allow use of the Telegram bot and the sockets at the same time
+- Added semaphore to limit number of consumer tasks
+- Added test for EP alert parser
+- Fixed /visplot command of the Telegram bot
+- Fixed issue when a current configuration file update changes values of already 
+- Improved format for distance in LVK alerts
+- Save configuration file with comments
+- Updated EP alert parser to use the id when possible
+existing options
 
 
 # v0.3.0

aware/__init__.py

@@ -11,9 +11,13 @@
 # ----------------------------------------------------------------------------
 """
 
+import asyncio
 import collections
 import traceback
 from types import ModuleType
+from ruamel.yaml import YAML
+
+import uvloop
 
 __modules__ = [
     "config",
@@ -119,6 +123,11 @@ def deep_update(src: dict, updates: dict):
                 # src[k] is not a dict, nothing to merge, so just set it,
                 # regardless if src[k] *was* a dict
                 src[k] = v
+                if src.get(k):
+                    if isinstance(src[k], tuple):
+                        src[k][1] = v[1]
+                else:
+                    src[k] = v                   
             else:
                 # note: updates[k] is a dict
                 if k not in src:
@@ -127,7 +136,11 @@ def deep_update(src: dict, updates: dict):
                 elif not isinstance(src[k], dict):
                     # src[k] is not a dict, so just set it to updates[k],
                     # overriding whatever it was
-                    src[k] = v
+                    if src.get(k):
+                        if isinstance(src[k], tuple):
+                            src[k][1] = v[1]
+                    else:
+                        src[k] = v   
                 else:
                     # both src[k] and updates[k] are dicts, push them on the stack
                     # to merge
@@ -140,3 +153,6 @@ def deep_update(src: dict, updates: dict):
 # # It is save to remove configuration here, because it is presented it config module
 # del cfg
 # del yaml_cfg
+
+
+asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

aware/__main__.py

@@ -47,11 +47,11 @@ def goodbye(start_time: float):
     "environment variable)",
 )
 def main(telegram: bool, sockserver: bool, mode: str, threads: int, dump: str):
-    if sockserver and telegram:
-        log.error(
-            "Simultaneous execution of Telegram bot and socket server not supported"
-        )
-        return 1
+    # if sockserver and telegram:
+    #     log.error(
+    #         "Simultaneous execution of Telegram bot and socket server not supported"
+    #     )
+    #     return 1
 
     if dump:
         if not os.getenv("AWARE_CONFIG_FILE", ""):

aware/__version__.py

@@ -1,2 +1,2 @@
-__version__ = (0, 4, 19)
+__version__ = (0, 4, 20)
 __strversion__ = "{}.{}.{}".format(__version__)

aware/alert/crossmatch.py

@@ -19,7 +19,12 @@
 
 
 # Maximal date difference to match a pair of alerts
-max_date_diff_min = CfgOption("max_date_diff_min", 5.0, float)
+max_date_diff_min = CfgOption(
+    "max_date_diff_min",
+    5.0,
+    float,
+    comment="Maximal date difference [min] to match a pair of alerts",
+)
 
 
 def crossmatch_alerts(

aware/alert/plugins/ep.py

@@ -7,6 +7,7 @@
 """
 
 from io import BytesIO, StringIO
+import numpy as np
 import orjson
 from astropy.time import Time
 from aware.logger import log
@@ -28,15 +29,26 @@ def parse_alert(
         except orjson.JSONDecodeError as e:
             log.error("Could not parse alert due to incorrect JSON encoding: %s", e)
             info = None
-        else:
+
+        try:
             ra = msg_content["ra"]
             dec = msg_content["dec"]
             radius = msg_content["ra_dec_error"]
+            snr = msg_content["image_snr"]
+
+            # Normalized so, at S/N = 5, importance is 90%
+            importance = np.tanh(snr / 3.35)
+
             trigger_date = Time(
                 msg_content["trigger_time"], format="isot"
             ).to_datetime()
-            importance = 0.9
-            event = f"J{ra:.3f}{dec:+.3f}"
+
+            # If there is no trigger id, then event name is derived from coordinates
+            if "id" in msg_content:
+                event = np.atleast_1d(msg_content["id"])[0]
+            else:
+                event = f"J{ra:.3f}{dec:+.3f}"
+
             localization = CircularSkyMap(ra_center=ra, dec_center=dec, radius=radius)
             info = TargetInfo(
                 localization=localization,
@@ -46,4 +58,10 @@ def parse_alert(
                 trigger_date=trigger_date,
                 importance=importance,
             )
+        except Exception as e:
+            if not isinstance(e, KeyboardInterrupt):
+                raise
+            else:
+                log.error("Failed to parse EP alert: %s", e, exc_info=e)
+
         return info

aware/alert/plugins/gbm.py

@@ -18,7 +18,12 @@
 
 
 FERMI_GBM = "Fermi (GBM)"
-fermi_gbm_sys_error = CfgOption("fermi_gbm_sys_error", 2.0, float)  # degrees
+fermi_gbm_sys_error = CfgOption(
+    "fermi_gbm_sys_error",
+    2.0,
+    float,
+    comment="Systematic error [deg] to be added to statistical one in quadrature",
+)  # degrees
 
 
 def parse_fermi_alert(

aware/alert/plugins/lvc.py

@@ -32,9 +32,24 @@
 ]
 
 
-has_ns_prob_thresh = CfgOption("has_ns_prob_thresh", 0.667, float)
-bbh_skymap_max_area = CfgOption("bbh_skymap_max_area", 30, float)
-far_threshold_global = CfgOption("far_threshold_global", 10, float)
+has_ns_prob_thresh = CfgOption(
+    "has_ns_prob_thresh",
+    0.667,
+    float,
+    comment="The probability threshold to consider LVK event to be either from NSBH or BNS merger",
+)
+bbh_skymap_max_area = CfgOption(
+    "bbh_skymap_max_area",
+    30,
+    float,
+    comment="The maximum area of BBH merger to be considered for planning",
+)
+far_threshold_global = CfgOption(
+    "far_threshold_global",
+    10,
+    float,
+    comment="The FAR threshold for an event to be considered as not a real event",
+)
 
 
 HZ_to_reciprocal_yr = 31557600
@@ -260,7 +275,9 @@ def parse_lvc_alert(
     if meta:
         # distmu_str = render_number_as_rich_utf8(distmu.value, 3)
         # distsigma_str = render_number_as_rich_utf8(2 * distsigma.value, 3)
-        description += f"Distance: {distmu.value:.1f} \u00B1 {distsigma.value:.1f} Mpc (2\u03C3)"
+        description += (
+            f"Distance: {distmu.value:.1f} \u00B1 {distsigma.value:.1f} Mpc (2\u03C3)"
+        )
     else:
         description += "Skymap is not available at the moment."
 

aware/app.py

@@ -8,14 +8,80 @@
 import matplotlib as mpl
 import uvloop
 
+from aware.logger import log
 from aware.service import ServiceFactory
+from aiomisc import Service
 
 from .config import dev
 from .sql.models import create_session
 from .sql.util import create_alert_tables
+from aware.data import AlertMessage, DataPackage
+from collections import deque
 
 
-asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+class QueueHub(Service):
+    """
+    QueueHub is performing distribution of items from a main queue between
+    queues owned by other clients. It is useful because when a multiple clients use
+    the same queue, the one which gets an item first, actually removes it from the
+    queue.
+    """
+
+    def __init__(
+        self,
+        master_queue: asyncio.Queue[AlertMessage | DataPackage],
+        queues: list[asyncio.Queue[AlertMessage | DataPackage]] = None,
+        **kwargs,
+    ):
+        self.master_queue = master_queue
+        self.queues = deque(queues or [])
+
+    async def distribute(self):
+        """
+        Perform actual distribution of items between master queue and other queues
+
+        Raises
+        ------
+        ValueError
+            if the queue is empty
+        """
+        if self.queues is None:
+            raise ValueError(
+                "No queues to distribute; use add_queue method to add queues"
+            )
+
+        while self.queues:
+            try:
+                item = await self.master_queue.get()
+            except asyncio.QueueEmpty as e:
+                log.error(
+                    "Error occurred while sending item to hub memebers", exc_info=e
+                )
+            else:
+                for q in self.queues:
+                    await q.put(item)
+
+    async def start(self):
+        """Start the hub."""
+        await self.distribute()
+
+    async def add_queue(self, q: asyncio.Queue):
+        """Add a queue to the hub
+
+        Parameters
+        ----------
+        q : asyncio.Queue
+            a queue to add
+        """
+        async with asyncio.Lock():
+            self.queues.append(q)
+
+    async def cleanup(self):
+        """
+        Cleanup the hub
+        """
+        async with asyncio.Lock():
+            self.queues.clear()
 
 
 class Application:
@@ -63,16 +129,37 @@ def _create_services(self):
         self.services = []
         factory = ServiceFactory(self._que)
 
+        # if self.mode == "test":
+        #     self.services.append(factory.create_test_consumer())
+        # elif self.mode == "prod":
+        #     self.services.append(factory.create_prod_consumer())
+
+        # if self.run_telegram:
+        #     self.services.append(factory.create_telegram_bot())
+
+        # if self.run_socket_server:
+        #     self.services.append(factory.create_socket_server())
+
+        queues = []
+
+        # Consumer fills the master queue with data
         if self.mode == "test":
-            self.services.append(factory.create_test_consumer())
+            self.services.append(factory.create_test_consumer(queue=self._que))
         elif self.mode == "prod":
-            self.services.append(factory.create_prod_consumer())
+            self.services.append(factory.create_prod_consumer(queue=self._que))
 
         if self.run_telegram:
-            self.services.append(factory.create_telegram_bot())
+            tg_que = asyncio.Queue()
+            self.services.append(factory.create_telegram_bot(queue=tg_que))
+            queues.append(tg_que)
 
         if self.run_socket_server:
-            self.services.append(factory.create_socket_server())
+            socket_que = asyncio.Queue()
+            self.services.append(factory.create_socket_server(queue=socket_que))
+            queues.append(socket_que)
+
+        hub = QueueHub(self._que, queues)
+        self.services.insert(0, hub)
 
     def run(self) -> float:
         engine, _ = create_session()

aware/cache.py

@@ -18,9 +18,16 @@
 __all__ = ["cache_value"]
 
 
-cache_dir = CfgOption("cache_dir", ".cache", str)
-timeout = CfgOption("timeout", 60, float)
-expire = CfgOption("expire", 86400, float)
+cache_dir = CfgOption("cache_dir", ".cache", str, comment="Cache directory")
+timeout = CfgOption(
+    "timeout",
+    60,
+    float,
+    comment="Maximum number of seconds to wait for retrieval of a cached value",
+)
+expire = CfgOption(
+    "expire", 86400, float, comment="Cached value expiration time in seconds"
+)
 
 
 def __access_cache():
@@ -40,10 +47,11 @@ def set(self, *args, **kwargs): ...
 
             def get(self, *args, **kwargs): ...
 
-            def __enter__(self): return self
+            def __enter__(self):
+                return self
 
             def __exit__(self, *args, **kwargs): ...
-            
+
         cache = EmptyCache()
 
     return cache