try predict

src/online_prediction.py

 import json
 from datetime import datetime
 
+import tensorflow as tf
 from kafka import KafkaConsumer
 
+from main_portrait import offline_predict_diary, offline_predict_tractate
+from models.esmm.diary_model import model_predict_diary
+from models.esmm.fe import device_fe, diary_fe, tractate_fe
+from models.esmm.tractate_model import model_predict_tractate
+from utils.cache import get_essm_model_save_path
 
-def consume_kafka():
+
+def consume_kafka(device_dict, diary_dict, tractate_dict, diary_predict_fn, tractate_predict_fn):
     consumer = KafkaConsumer(bootstrap_servers="172.16.44.25:9092,172.16.44.31:9092,172.16.44.45:9092",
                              group_id="essm_online_prediction",
                              enable_auto_commit=True,
@@ -15,8 +22,30 @@ def consume_kafka():
     for msg in consumer:
         data = json.loads(msg.value)
         device_id = data.get("device_id", "")
-        print(device_id + " " + str(datetime.now()))
+        if device_id == "androidid_a25a1129c0b38f7b":
+            print(device_id + " " + str(datetime.now()))
+            offline_predict_diary(device_id, device_dict, diary_dict, diary_predict_fn)
+            offline_predict_tractate(device_id, device_dict, tractate_dict, tractate_predict_fn)
 
 
 if __name__ == "__main__":
-    consume_kafka()
+    tf.compat.v1.logging.set_verbosity(tf.compat.v1.logging.INFO)
+
+    device_dict = device_fe.get_device_dict_from_redis()
+    diary_dict = diary_fe.get_diary_dict_from_redis()
+    tractate_dict = tractate_fe.get_tractate_dict_from_redis()
+    print("redis data: " + str(len(device_dict)) + " " + str(len(diary_dict)) + " " + str(len(tractate_dict)))
+
+    diary_save_path = get_essm_model_save_path("diary")
+    if not diary_save_path:
+        diary_save_path = "/home/gmuser/data/models/diary/1596509008"
+        print(diary_save_path + "!!!!!!!!!!!!!!!!!!!!!!!!!!!")
+    diary_predict_fn = tf.contrib.predictor.from_saved_model(diary_save_path)
+
+    tractate_save_path = get_essm_model_save_path("tractate")
+    if not tractate_save_path:
+        tractate_save_path = "/home/gmuser/data/models/tractate/1596509299"
+        print(tractate_save_path + "!!!!!!!!!!!!!!!!!!!!!!!!!!!")
+    tractate_predict_fn = tf.contrib.predictor.from_saved_model(tractate_save_path)
+
+    consume_kafka(device_dict, diary_dict, tractate_dict, diary_predict_fn, tractate_predict_fn)