1# (c) Stefan Countryman, 2019 

2 

3""" 

4Listen for new LVAlerts from GraceDB containing GW triggers. This is LIGO/Virgo's 

5internal infrastructure for distributing notifications about updates to all 

6gravitational wave (GW) triggers. 

7""" 

8 

9import os 

10import logging 

11from functools import partial 

12from argparse import Action 

13from llama.cli import ( 

14 Parsers, 

15 CliParser, 

16 get_logging_cli, 

17 log_exceptions_and_recover, 

18) 

19from llama.utils import LOGDIR 

20from llama.listen import Parsers as ListenParsers 

21from llama.listen.lvalert import ( 

22 get_client, 

23 process_alert_json, 

24 NODES, 

25 Client, 

26 SERVER, 

27 PLAYGROUND_SERVER, 

28) 

29 

30LOGGER = logging.getLogger(__name__) 

31LVALERT_LOGFILE = os.path.join(LOGDIR, 'lvalert.log') 

32 

33 

34class PrintNodesAction(Action): # pylint: disable=too-few-public-methods 

35 """Print available LVAlert nodes and quit.""" 

36 

37 def __call__(self, parser, namespace, values, option_string=None): 

38 """Only runs if the flag is provided.""" 

39 if not values: 

40 values = SERVER 

41 client = get_client(values) 

42 if client.connect(): 

43 client.process(block=False) 

44 print(f"Available nodes for {values}:") 

45 for node in client.get_nodes(): 

46 print(f" {node}") 

47 client.abort() 

48 exit() 

49 else: 

50 print("Could not fetch the node list.") 

51 client.abort() 

52 exit(1) 

53 

54 

55def get_parser(): 

56 """Get CLI.""" 

57 parser = CliParser( 

58 description=__doc__, 

59 parents=( 

60 get_logging_cli(LVALERT_LOGFILE, 'debug'), 

61 ListenParsers.rundir, 

62 Parsers.erralert 

63 ) 

64 ) 

65 parser.add_argument("-n", "--nodes", nargs="*", default=tuple(NODES), 

66 help=f""" 

67 The LVAlert nodes to subscribe to. Make sure to subscribe to all the 

68 searches you are interested in. (default: {NODES})""") 

69 parser.add_argument("-s", "--server", default=SERVER, help=f""" 

70 The server to subscribe to. (default: {SERVER})""") 

71 parser.add_argument("--list-nodes", action=PrintNodesAction, 

72 nargs='?', choices=(SERVER, PLAYGROUND_SERVER), 

73 help=f""" 

74 Print available LVAlert nodes for the given server and exit. (default: 

75 {SERVER})""") 

76 return parser 

77 

78 

79@log_exceptions_and_recover() 

80def listen(args): 

81 """Listen for new LVAlerts. Log and recover from errors and restart. 

82 

83 Parameters 

84 ---------- 

85 args : argparse.Namespace 

86 Arguments parsed using ``get_parser``. 

87 """ 

88 client = Client(server=args.server, 

89 processor=partial(process_alert_json, 

90 rundir=args.rundir), 

91 nodes=tuple(args.nodes)) 

92 # this should exit when the user requests it; if it just aborts, we 

93 # should restart with a freshly-initialized client (since we lost 

94 # subscriptions or failed a heartbeat test at time of writing) 

95 client.listen() 

96 LOGGER.error("RESTARTING after client.abort with a fresh client.") 

97 

98 

99def main(): 

100 """Subscribe to requested nodes on the requested server and start listening 

101 for new alerts, reacting to them with ``process_alert_json`` and saving 

102 things to the specified run directory.""" 

103 parser = get_parser() 

104 args = parser.parse_args() 

105 while True: 

106 listen(args) 

107 

108 

109if __name__ == '__main__': 

110 main()