object detection: revise

opencv-yolo/web/web.py

@@ -72,9 +72,13 @@ API_CAMERA_CFG_URI = os.getenv("API_CAMERA_CFG_URI", "/video/analysis/camera/{}/
 RABBITMQ_URI = os.getenv("RABBITMQ_URI", "amqp://guest:guest@localhost:5672/")
 RABBITMQ_TOPIC = os.getenv("RABBITMQ_TOPIC", "*.camera.model")
 RABBITMQ_EXHANGE = os.getenv("RABBITMQ_EXHANGE", "video")
+AI_QUEUE_TASK = os.getenv('AI_TASK', 'videoai.v1.task')
+AI_QUEUE_RESULT = os.getenv('AI_RESULT', 'videoai.v1.result')
+AI_EXCHANGE = os.getenv('AI_XHG', 'videoai')
+
 CAMERA_CONFIG_MAP = {}
 
-def rabCallback(ch, method, prop, body):
+def rabModelCallback(ch, method, prop, body):
     try:
         print("received rabbitmsg. routing key: {} , body: {}".format(method.routing_key, body))
         rk = method.routing_key.split('.')
@@ -85,25 +89,56 @@ def rabCallback(ch, method, prop, body):
             CAMERA_CONFIG_MAP[sn] = json.loads(body)
             print("CAMERA_CONFIG_MAP: \n{}".format(json.dumps(CAMERA_CONFIG_MAP)))
     except Exception as e:
+        #extype, value, tb = sys.exc_info()
         #traceback.print_exc()
         #pdb.post_mortem(tb)
         print("failed to process callback {}: {} - {}".format(e, method, body))
 
-def initRabbit(): 
-    rabChan = None
+def rabAiTaskCallback(ch, method, prop, body):
+    try:
+        print("received rabbitmsg. routing key: {} , body: {}".format(method.routing_key, body))
+        take_task(json.loads(body))
+    except Exception as e:
+        extype, value, tb = sys.exc_info()
+        traceback.print_exc()
+        pdb.post_mortem(tb)
+        print("failed to process callback {}: {} - {}".format(e, method, body))
+
+def rabbit_camera_model(): 
     if RABBITMQ_URI:
         print("connecting rabbitmq {}".format(RABBITMQ_URI))
         parameters = pika.URLParameters(RABBITMQ_URI) #("amqp://ilabservice:iLabServiceOps123456@40.73.40.246:5672/")
         connection = pika.BlockingConnection(parameters)
-        rabChan = connection.channel()
-        rabChan.exchange_declare(exchange=RABBITMQ_EXHANGE, exchange_type="topic")
-        qr = rabChan.queue_declare('', durable=True)
-        rabChan.queue_bind(exchange=RABBITMQ_EXHANGE,
-                        queue=qr.method.queue, routing_key=RABBITMQ_TOPIC)
-        rabChan.basic_consume(
-            queue=qr.method.queue, on_message_callback=rabCallback, auto_ack=True)
-    return rabChan
-    
+        print("rabbitmq connection established")
+
+        # camera model channel
+        rabChanModel = connection.channel()
+        rabChanModel.exchange_declare(exchange=RABBITMQ_EXHANGE, exchange_type="topic")
+        modelQueue = rabChanModel.queue_declare('', durable=True)
+        rabChanModel.queue_bind(exchange=RABBITMQ_EXHANGE,
+                        queue=modelQueue.method.queue, routing_key=RABBITMQ_TOPIC)
+        rabChanModel.basic_consume(
+            queue=modelQueue.method.queue, on_message_callback=rabModelCallback, auto_ack=True)
+        print("bound camera model")
+        rabChanModel.start_consuming()
+
+def rabbit_ai_task(): 
+    if RABBITMQ_URI:
+        print("connecting rabbitmq {}".format(RABBITMQ_URI))
+        parameters = pika.URLParameters(RABBITMQ_URI) #("amqp://ilabservice:iLabServiceOps123456@40.73.40.246:5672/")
+        connection = pika.BlockingConnection(parameters)
+
+        # ai analysis task channel
+        rabChanAiTask = connection.channel()
+        if AI_EXCHANGE:
+            rabChanAiTask.exchange_declare(exchange=AI_EXCHANGE, exchange_type="direct")
+        rabChanAiTask.queue_declare(queue=AI_QUEUE_TASK, durable=True)
+        rabChanAiTask.queue_bind(exchange=AI_EXCHANGE,
+                        queue=AI_QUEUE_TASK, routing_key=AI_QUEUE_TASK)
+        rabChanAiTask.basic_consume(
+            queue=AI_QUEUE_TASK, on_message_callback=rabAiTaskCallback, auto_ack=True)
+        print("bound ai task")
+        rabChanAiTask.start_consuming()
 
 print("CONFIG: \n\tMQTT: {}:{}\n\tBIN_NAME: {}".format(
     MQTT_HOST, MQTT_PORT, binName))
@@ -128,6 +163,10 @@ def uploadFile(ipcSn, dirName, fileName, srcPath):
 
 
 class VAMMQTTClient:
+    rabChanModel = None
+    rabChanAiTask = None
+    th1 =None
+    th2 = None
     # The callback for when the client receives a CONNACK response from the server.
     @staticmethod
     def on_connect(client, userdata, flags, rc):
@@ -173,8 +212,9 @@ class VAMMQTTClient:
         self.client.on_message = VAMMQTTClient.on_message
         self.client.connect_async(host, port, 30)
         self.client.loop_start()
-        self.rabChan = initRabbit()
-        th = threading.Thread(target=lambda: self.rabChan.start_consuming()).start()
+
+        self.th1 = threading.Thread(target=rabbit_camera_model).start()
+        self.th2 = threading.Thread(target=rabbit_ai_task).start()
 
 app = Flask(__name__,
             static_url_path='',
@@ -195,22 +235,23 @@ worker.conf.update(
 def getRegionConfig(ipcSN):
     region = None
     res = None
-    reqUrl = API_HOST + API_CAMERA_CFG_URI.format(ipcSN)
-    try:
-        if ipcSN in CAMERA_CONFIG_MAP and type(CAMERA_CONFIG_MAP[ipcSN]) is dict:
-            region = CAMERA_CONFIG_MAP[ipcSN]['region']
-        elif CAMERA_CONFIG_MAP.get(ipcSN) is None and API_HOST:
-            res = requests.get(reqUrl)
-            if res.status_code == 200:
-                ret = res.json()
-                if 'data' in ret and 'region' in ret['data']:
-                    region = ret["data"]["region"]
-                    CAMERA_CONFIG_MAP[ipcSN] = ret["data"]
-            if CAMERA_CONFIG_MAP.get(ipcSN) is None:
-                CAMERA_CONFIG_MAP[ipCSN] = 0
-        print("getRegionConfig: {}, {}, {}: {}".format(ipcSN, reqUrl, res.text if res and (text in res) else None, json.dumps(region)))
-    except Exception as e:
-            print("failed to get camera config {}: {}".format(ipcSN, e))
+    if API_HOST:
+        reqUrl = API_HOST + API_CAMERA_CFG_URI.format(ipcSN)
+        try:
+            if ipcSN in CAMERA_CONFIG_MAP and type(CAMERA_CONFIG_MAP[ipcSN]) is dict:
+                region = CAMERA_CONFIG_MAP[ipcSN]['region']
+            elif CAMERA_CONFIG_MAP.get(ipcSN) is None and API_HOST:
+                res = requests.get(reqUrl)
+                if res.status_code == 200:
+                    ret = res.json()
+                    if 'data' in ret and 'region' in ret['data']:
+                        region = ret["data"]["region"]
+                        CAMERA_CONFIG_MAP[ipcSN] = ret["data"]
+                if CAMERA_CONFIG_MAP.get(ipcSN) is None:
+                    CAMERA_CONFIG_MAP[ipCSN] = 0
+            print("getRegionConfig: {}, {}, {}: {}".format(ipcSN, reqUrl, res.text if res and ('text' in res) else None, json.dumps(region)))
+        except Exception as e:
+                print("failed to get camera config {}: {}".format(ipcSN, e))
     return region
 
 def take_task(task):
@@ -316,11 +357,24 @@ def video_analysis(data):
         mc.username_pw_set(username=MQTT_USER,password=MQTT_PASSWORD)
     mc.connect(MQTT_HOST, MQTT_PORT)
     mc.publish('video.ai/v1.0/result', json.dumps(ret), qos=1)
+    print("mqtt message published")
+    # rabbitmq
+    if RABBITMQ_URI:
+        parameters = pika.URLParameters(RABBITMQ_URI)
+        connection = pika.BlockingConnection(parameters)
+        rabChanResult = connection.channel()
+        rabChanResult.exchange_declare(exchange=AI_EXCHANGE, exchange_type="direct")
+        rabChanResult.queue_declare(queue=AI_QUEUE_RESULT, durable=True)
+        rabChanResult.queue_bind(exchange=AI_EXCHANGE,
+                        queue=AI_QUEUE_RESULT, routing_key=AI_QUEUE_RESULT)
+        rabChanResult.basic_publish(exchange =AI_EXCHANGE,routing_key=AI_QUEUE_RESULT,body=json.dumps(ret),
+            properties=pika.BasicProperties(delivery_mode = 2))
+        print("rabbitmq message published")
+        rabChanResult.close()
     try:
         os.system('rm -fr ' + downloadDir)
     except Exception as e:
         print('cascaded exception in va: {}'.format(e))
-        #raise Exception()
 
     return ret