Merge pull request #225 from lsst-uk/hotfix/api_alerce_fink

Hotfix/api alerce fink
lsst-uk · Apr 16, 2024 · 6154392 · 6154392
2 parents 2187a8d + 7e34d9d
commit 6154392
Show file tree

Hide file tree

Showing 3 changed files with 42 additions and 39 deletions.
diff --git a/services/externalBrokers/alerce/consume_alerce.py b/services/externalBrokers/alerce/consume_alerce.py
@@ -47,8 +47,8 @@ def make_lc_annotation(record):
 conf = {
     'bootstrap.servers': settings.ALERCE_KAFKA,
     'group.id'         : settings.ALERCE_GROUP_ID,
-    'security.protocol': 'SASL_PLAINTEXT',
-    'sasl.mechanism'   : 'SCRAM-SHA-256',
+    'security.protocol': 'SASL_SSL',
+    'sasl.mechanism'   : 'SCRAM-SHA-512',
     'sasl.username'    : 'lasair',
     'sasl.password'    : settings.ALERCE_PASSWORD,
     'auto.offset.reset': 'earliest',
@@ -73,14 +73,18 @@ def make_lc_annotation(record):
         print('Unknown classifier, quitting')
         sys.exit()
 
-    L = lasair.lasair_client(settings.API_TOKEN, endpoint='https://lasair-ztf.lsst.ac.uk/api')
-#    L = lasair.lasair_client(settings.API_TOKEN)
+    L = lasair.lasair_client(settings.API_TOKEN, endpoint='https://' + settings.LASAIR_URL + '/api')
 
     streamReader.subscribe([topic])
     while 1:
         msg = streamReader.poll(timeout=5)
         if msg == None: 
             break
+        if msg.error():
+            print('ERROR in ingest/poll: ' +  str(msg.error()))
+            break
+        print('message:', msg.value())
+        sys.exit()
         fo = io.BytesIO(msg.value())
         for record in reader(fo):
             r = make_annotation(record)

diff --git a/services/externalBrokers/fink/get_fink_annotate.py b/services/externalBrokers/fink/get_fink_annotate.py
@@ -1,4 +1,5 @@
 import os, sys
+import json
 from datetime import datetime
 import lasair
 
@@ -9,14 +10,19 @@
 sys.path.append('fink-client')
 from fink_client.consumer import AlertConsumer
 
+TESTMODE = (len(sys.argv) > 1 and sys.argv[1] == 'TEST')
+
 nid  = date_nid.nid_now()
 date = date_nid.nid_to_date(nid)
-logfile = settings.SERVICES_LOG +'/'+ date + '.log'
-logf = open(logfile, 'a')
+
+if TESTMODE:
+    logf = sys.stdout
+else:
+    logfile = settings.SERVICES_LOG +'/'+ date + '.log'
+    logf = open(logfile, 'a')
 
 # Lasair client
-L = lasair.lasair_client(settings.API_TOKEN, endpoint='https://lasair-ztf.lsst.ac.uk/api')
-#L = lasair.lasair_client(settings.API_TOKEN')
+L = lasair.lasair_client(settings.API_TOKEN, endpoint='https://' + settings.LASAIR_URL + '/api')
 topic_out = 'fink'
 
 # Fink configuration
@@ -29,11 +35,13 @@
 # Instantiate a consumer
 consumer = AlertConsumer(settings.FINK_TOPICS, fink_config)
 
-#d = consumer.available_topics()
-#topics = list(d.keys())
-#topics.sort()
-#for topic in topics: 
-#    print(topic)
+if TESTMODE:
+    d = consumer.available_topics()
+    topics = list(d.keys())
+    topics.sort()
+    for topic in topics: 
+        if topic.startswith('fink'):
+            print('Found topic:', topic)
 
 nalert = {}
 n = 0
@@ -42,34 +50,19 @@
     (topic, alert, version) = consumer.poll(maxtimeout)
     if topic is None:
         break
-    print(topic)
-
-    objectId = alert['objectId']
-    classification = topic[:16]
-    classdict = {}
-
-    try: classdict['rf_snia_vs_nonia']:   float(alert['rf_snia_vs_nonia']) 
-    except: pass
-
-    try: classdict['snn_snia_vs_nonia']:  float(alert['snn_snia_vs_nonia']) 
-    except: pass
-
-    try: classdict['snn_sn_vs_all']:      float(alert['snn_sn_vs_all']) 
-    except: pass
-
-    try: classdict['rf_kn_vs_nonkn']:     float(alert['rf_kn_vs_nonkn']) 
-    except: pass
+    if TESTMODE:
+        print(alert['objectId'], topic)
 
-    try: classdict['anomaly_score']:      float(alert['anomaly_score']) 
-    except: pass
+    if not topic in settings.FINK_TOPICS:
+        continue
 
     L.annotate(
         topic_out,
-        objectId,
-        classification,
+        alert['objectId'],
+        topic[:16],
         version='0.1',
         explanation='',
-        classdict=classdict,
+        classdict={'classification': topic},
         url='')
 
     n += 1

diff --git a/webserver/lasairapi/serializers.py b/webserver/lasairapi/serializers.py
@@ -98,7 +98,10 @@ def save(self):
 
         result = []
         for objectId in olist:
-            result.append(objjson(objectId))
+            try:
+                result.append(objjson(objectId))
+            except Exception as e:
+                result.append({'error': str(e)})
         return result
 
 
@@ -331,9 +334,12 @@ def save(self):
 
         lightcurves = []
         for objectId in olist:
-            candidates = LF.fetch(objectId)
-            fpcandidates = FLF.fetch(objectId)
-            lightcurves.append({'objectId':objectId, 'candidates':candidates, 'forcedphot': fpcandidates})
+            try:
+                candidates = LF.fetch(objectId)
+                fpcandidates = FLF.fetch(objectId)
+                lightcurves.append({'objectId':objectId, 'candidates':candidates, 'forcedphot': fpcandidates})
+            except Exception as e:
+                lightcurves.append({'error': str(e)})
 
         LF.close()
         FLF.close()