Source code for snews_pt.snews_pt_utils

"""
Utility tools for snews_pt
"""

import json
import os
from collections import namedtuple

import click
import dotenv
from dotenv import load_dotenv

from .core.logging import getLogger

[docs] log = getLogger(__name__)
[docs] default_detector_file = ( os.path.dirname(__file__) + "/auxiliary/detector_properties.json" )
[docs] def set_env(env_path=None): """Set environment parameters Parameters ---------- env_path : str, (optional) path for the environment file. Use default settings if not given """ dirname = os.path.dirname(__file__) default_env_path = dirname + "/auxiliary/test-config.env" env = env_path or default_env_path load_dotenv(env)
[docs] def retrieve_detectors(detectors_path=default_detector_file): """Retrieve the name-ID-location of the participating detectors. Parameters ---------- detectors_path : str, optional path to detector properties. File needs to be in JSON format Returns ------- None """ if not os.path.isfile(detectors_path): os.system(f"python {os.path.dirname(__file__)}/auxiliary/make_detector_file.py") with open(detectors_path) as json_file: detectors = json.load(json_file) # make a namedtuple Detector = namedtuple("Detector", ["name", "id", "location"]) for k, v in detectors.items(): detectors[k] = Detector(v[0], v[1], v[2]) return detectors
[docs] def isnotebook(): """Tell if the script is running on a notebook""" try: shell = get_ipython().__class__.__name__ if shell == "ZMQInteractiveShell": return True # Jupyter notebook or qtconsole elif shell == "TerminalInteractiveShell": return False # Terminal running IPython else: return False # Other type (?) except NameError: return False # Probably standard Python interpreter
[docs] def display_gif(): """Some fun method to display an alert gif If running on notebook. """ if isnotebook(): from IPython.display import HTML, display giphy_snews = ( "https://raw.githubusercontent.com/SNEWS2/" "hop-SNalert-app/snews2_dev/hop_comms/auxiliary/snalert.gif" ) display(HTML(f"<img src={giphy_snews}>"))
[docs] def set_name(detector_name="TEST", _return=False): """set your detector's name. Messages sent with detector_name="TEST" will be ignored at the server Alerts can still be subscribed and listened as "TEST" """ envpath = os.path.join(os.path.dirname(__file__), "auxiliary/test-config.env") load_dotenv(envpath) detectors = list(retrieve_detectors().keys()) if detector_name == "TEST": if int(os.getenv("HAS_NAME_CHANGED")) == 0: for i, d in enumerate(detectors): click.secho(f"[{i:2d}] {d}") inp = input( click.secho("Please put select your detector's index\n", bold=True) ) detector_name = detectors[int(inp)] os.environ["DETECTOR_NAME"] = detector_name os.environ["HAS_NAME_CHANGED"] = "1" dotenv.set_key(envpath, "DETECTOR_NAME", os.environ["DETECTOR_NAME"]) dotenv.set_key(envpath, "HAS_NAME_CHANGED", os.environ["HAS_NAME_CHANGED"]) else: detector_name = os.environ["DETECTOR_NAME"] else: if detector_name not in detectors: raise KeyError( f"{detector_name} is not a valid detector. \nChoose from {detectors}" ) os.environ["DETECTOR_NAME"] = detector_name os.environ["HAS_NAME_CHANGED"] = "1" dotenv.set_key(envpath, "DETECTOR_NAME", os.environ["DETECTOR_NAME"]) dotenv.set_key(envpath, "HAS_NAME_CHANGED", os.environ["HAS_NAME_CHANGED"]) if _return: return detector_name else: click.secho(f'You are {os.environ["DETECTOR_NAME"]}')