Integrate dummy capture loop and publish events to MQTT topic (#2)

* integrate dummy capture loop and publish events to MQTT topic

* adjust test data

* rename topic

* refactor broker configs

Author: whummer

Makefile

@@ -12,10 +12,10 @@ install:        ## Install dependencies in local virtualenv
 	test ! -e requirements.txt || ($(VENV_RUN); $(PIP_CMD) install -r requirements.txt)
 
 test:           ## Run local unit and integration tests
-	$(VENV_RUN); PYTHONPATH=`pwd` nosetests $(NOSE_ARGS) --with-timer --with-coverage --logging-level=WARNING --nocapture --no-skip --exe --cover-erase --cover-tests --cover-inclusive --cover-package=cpopserver --with-xunit --exclude='$(VENV_DIR).*' $(TEST_PATH)
+	$(VENV_RUN); PYTHONPATH=`pwd` nosetests $(NOSE_ARGS) --with-timer --with-coverage --logging-level=WARNING --nocapture --no-skip --exe --cover-erase --cover-tests --cover-inclusive --cover-package=cpopservice --with-xunit --exclude='$(VENV_DIR).*' $(TEST_PATH)
 
 start:          ## Start up the CPOP server
-	$(VENV_RUN); PYTHONPATH=`pwd` python cpopserver/server.py
+	$(VENV_RUN); PYTHONPATH=`pwd` python cpopservice/server.py
 
 lint:           ## Run code linter to check code style
 	($(VENV_RUN); flake8 --inline-quotes=single --show-source --max-line-length=120 --ignore=E128,W504 --exclude=$(VENV_DIR)* .)

adabins.py

@@ -21,4 +21,4 @@
     depth_img = pred[0, 0]/10
     cv2.imshow('frame', depth_img)
     if cv2.waitKey(1) & 0xFF == ord('q'):
-        break
+        break

cpopserver/config.py

@@ -0,0 +1,30 @@
+import cv2
+import torch
+from time import time
+from PIL import Image
+from adabins.infer import InferenceHelper
+from util.timer import Timer
+
+
+def run_capture_loop(capture, timer=None, result_queue=None, headless=True):
+    # infer_helper = InferenceHelper()
+    while True:
+        ret, frame = capture.read()
+        frame_rgb = frame[:, :, ::-1]
+        im_pil = Image.fromarray(frame_rgb)
+        timer and timer.start()
+        # centers, pred = inferHelper.predict_pil(im_pil)
+        timer and timer.stop()
+        # depth_img = pred[0, 0]/10
+        if result_queue:
+            message = {
+                'Timestamp': time(),
+                'Type': 'person',
+                'Position': {'X': 1, 'Y': 2, 'Z': 3},
+                'Shape': []
+            }
+            result_queue.put(message)
+        if not headless:
+            cv2.imshow('frame', depth_img)
+            if cv2.waitKey(1) & 0xFF == ord('q'):
+                break

cpopserver/server.py

@@ -0,0 +1,18 @@
+import os
+import logging
+import tempfile
+
+
+# host and port of local MQTT broker
+BROKER_HOST = os.environ.get('BROKER_HOST') or 'localhost'
+BROKER_PORT = int(os.environ.get('BROKER_PORT') or 1883)
+BROKER_STARTUP = os.environ.get('BROKER_STARTUP') not in ['false', '0', False]
+
+# MQTT topic name
+MQTT_TOPIC_NAME = 'cpop'
+
+# local folder to store temporary files
+TMP_FOLDER = os.path.join(tempfile.gettempdir(), 'cpopserver')
+
+# configure logging
+logging.basicConfig(level=logging.INFO)

cpopserver/__init__.py cpopservice/__init__.py

@@ -2,6 +2,9 @@
 
 
 class BaseModel(object):
+    def __init__(self, params={}):
+        self.__dict__ = params
+
     def to_bson(self):
         result = bson.dumps(self.__dict__)
         return result
@@ -20,6 +23,14 @@ def __hash__(self, other):
         # TODO: implement hash function if needed
         return 0
 
+    def __repr__(self):
+        return '%s(%s)' % (self.__class__.__name__, self.__dict__)
+
 
 class Event(BaseModel):
+    # attributes: timestamp, shape (list of Coordinates), position (Coordinates)
+    pass
+
+
+class Coordinates(BaseModel):
     pass

cpopservice/capture_loop.py

@@ -0,0 +1,37 @@
+import logging
+import paho.mqtt.client as mqtt
+from cpopservice import config
+
+LOG = logging.getLogger(__name__)
+
+
+class CPOPPublisher:
+    """ Publisher that publishes CPOP events to the pub/sub broker """
+
+    def publish_event(self, event):
+        raise Exception('Not implemented')
+
+    @staticmethod
+    def get(impl_type=None):
+        subclasses = CPOPPublisher.__subclasses__()
+        subclasses = {subclass.name(): subclass for subclass in subclasses}
+        if not impl_type and len(subclasses) != 1:
+            raise Exception('Multiple CPOPPublisher implemtations found and type not specified')
+        subclass = subclasses.get(impl_type) or list(subclasses.values())[0]
+        return subclass()
+
+
+class CPOPPublisherMQTT(CPOPPublisher):
+    """ Publisher based on MQTT broker """
+
+    @staticmethod
+    def name():
+        return 'mqtt'
+
+    def __init__(self):
+        self.client = mqtt.Client()
+        self.client.connect(config.BROKER_HOST, config.BROKER_PORT)
+
+    def publish_event(self, event):
+        LOG.info('Publishing message to topic %s: %s' % (config.MQTT_TOPIC_NAME, event))
+        self.client.publish(config.MQTT_TOPIC_NAME, event.to_bson())

cpopservice/config.py

@@ -0,0 +1,114 @@
+import os
+import glob
+import time
+import queue
+import logging
+import cv2
+from cpopservice import config
+from cpopservice.constants import MOSQUITTO_URL_LINUX
+from cpopservice.core.models import Event
+from cpopservice.core.pubsub import CPOPPublisher
+from cpopservice.capture_loop import run_capture_loop
+from cpopservice.utils.common import (
+    ShellCommandThread, FuncThread, get_os_type, mkdir, download, untar, find_command, sleep_forever)
+
+LOG = logging.getLogger(__name__)
+THREADS = []
+LOCAL_HOSTS = ['localhost', '0.0.0.0', '127.0.0.1']
+
+
+class CPOPServer(FuncThread):
+    def __init__(self):
+        FuncThread.__init__(self, self.run_loop)
+        self.publisher = CPOPPublisher.get()
+
+    def run_loop(self):
+        LOG.info('Starting CPOP server capture loop')
+        result_queue = queue.Queue()
+        capture = get_capture_device()
+        QueueSubscriber(result_queue, self.publisher).start()
+        run_capture_loop(capture, result_queue=result_queue)
+
+
+class QueueSubscriber(FuncThread):
+    def __init__(self, queue, publisher):
+        self.queue = queue
+        self.publisher = publisher
+        FuncThread.__init__(self, self.run_loop)
+
+    def run_loop(self):
+        LOG.info('Starting CPOP queue subscriber loop')
+        while True:
+            message = self.queue.get()
+            message = self._prepare_message(message)
+            self.publisher.publish_event(message)
+
+    def _prepare_message(self, message):
+        # TODO: for testing only
+        if isinstance(message, dict):
+            import time
+            time.sleep(0.5)
+            message = Event(message)
+        return message
+
+
+def get_capture_device():
+    # TODO review this
+    capture = cv2.VideoCapture(0)
+    return capture
+
+
+def start_mqtt_broker():
+    broker_host = config.BROKER_HOST
+    if not config.BROKER_STARTUP or broker_host not in LOCAL_HOSTS:
+        return
+    mosquitto_bin = install_mqtt()
+    LOG.info('Starting MQTT broker on port %s' % config.BROKER_PORT)
+    t = ShellCommandThread('%s -p %s' % (mosquitto_bin, config.BROKER_PORT))
+    t.start()
+    time.sleep(1)
+    THREADS.append(t)
+
+
+def install_mqtt():
+    bin_path = find_command('mosquitto')
+    if bin_path:
+        return bin_path
+    os_type = get_os_type()
+    target_dir = os.path.join(config.TMP_FOLDER, 'mosquitto')
+    archive = os.path.join(target_dir, 'archive.tgz')
+    mkdir(target_dir)
+    if not os.path.exists(archive):
+        if os_type == 'linux':
+            url = MOSQUITTO_URL_LINUX
+        else:
+            raise Exception('Unsupported OS type: %s' % os_type)
+        download(url, archive)
+        untar(archive, target_dir)
+    result = glob.glob('%s/**/*bin/mosquitto' % target_dir, recursive=True)
+    return result[0]
+
+
+def startup_servers():
+    start_mqtt_broker()
+    CPOPServer().start()
+
+
+def shutdown_servers():
+    LOG.info('Shutting down server threads ...')
+    while THREADS:
+        t = THREADS[0]
+        del THREADS[0]
+        t.stop()
+
+
+def main():
+    startup_servers()
+    try:
+        sleep_forever()
+    except KeyboardInterrupt:
+        shutdown_servers()
+
+
+if __name__ == '__main__':
+    main()

cpopserver/constants.py cpopservice/constants.py

@@ -28,6 +28,7 @@ pandas
 # extras --------------------------------------
 thop  # FLOPS computation
 pycocotools>=2.0  # COCO mAP
+opencv-python
 
 # CPOP Server dependencies
 bson

cpopserver/core/__init__.py cpopservice/core/__init__.py

@@ -1,10 +1,10 @@
 import queue
 import unittest
 import paho.mqtt.client as mqtt
-from cpopserver import config
-from cpopserver.core import models
-from cpopserver.server import startup_servers, shutdown_servers
-from cpopserver.utils.common import short_uid, FuncThread, retry
+from cpopservice import config
+from cpopservice.core import models
+from cpopservice.server import startup_servers, shutdown_servers
+from cpopservice.utils.common import short_uid, FuncThread, retry
 
 
 class TestCPOPServer(unittest.TestCase):