Pipfile

@@ -5,5 +5,6 @@ name = "pypi"
 
 [packages]
 hbmqtt = "*"
+paho-mqtt = "*"
 
 [dev-packages]

bumper.py

@@ -3,6 +3,7 @@
 import logging
 import bumper
 import sys, socket
+import time
 
 logging.basicConfig(level=logging.DEBUG,
                         format='%(asctime)s %(levelname)-8s %(message)s')
@@ -14,11 +15,20 @@ xmpp_address = ("0.0.0.0", 5223)
 #mqtt_address = (socket.gethostbyname(socket.gethostname()), 8883)
 mqtt_address = ("0.0.0.0", 8883)
 
-# start conf server (async)
+# A default bot could be set here to automatically add it as available
-conf_server = bumper.ConfServer(conf_address, usessl=True, run_async=True)
+# dbot = bumper.VacBotDevice("did", "class", "resource", "name","nick" )
-# start mqtt server (async)
+# bclient = bumper.bumper_clients_var
-mqtt_server = bumper.MQTTServer(mqtt_address, run_async=True)
+# bclienttemp = bclient.get()
+# bclienttemp.append(dbot.asdict())
+# bclient.set(bclienttemp)
 
+# start mqtt server (async)
+mqtt_server = bumper.MQTTServer(mqtt_address, run_async=True,bumper_clients=bumper.bumper_clients_var)
+time.sleep(1.5) #Wait for broker startup
+# start mqtt server (async)
+mqtt_helperbot = bumper.MQTTHelperBot(mqtt_address, run_async=True,bumper_clients=bumper.bumper_clients_var)
+# start conf server (async)
+conf_server = bumper.ConfServer(conf_address, usessl=True, run_async=True, bumper_clients=bumper.bumper_clients_var, helperbot=mqtt_helperbot)
 # start xmpp server (sync)
 xmpp_server = bumper.XMPPServer(xmpp_address)
 

bumper/__init__.py

@@ -2,4 +2,22 @@
 
 from .confserver import ConfServer
 from .mqttserver import MQTTServer
+from .mqttserver import MQTTHelperBot
 from .xmppserver import XMPPServer
+import asyncio
+import contextvars
+
+bumper_clients_var = contextvars.ContextVar('bumper_clients', default=[])
+
+class VacBotDevice(object):
+    def __init__(self,did="", vac_bot_device_class="",resource="" , name="", nick="", company="eco-ng"):        
+        self.vac_bot_device_class = vac_bot_device_class
+        self.company = company
+        self.did = did
+        self.name = name
+        self.nick = nick
+        self.resource = resource
+
+    def asdict(self):
+        return {"class": self.vac_bot_device_class, "company": self.company,
+            "did": self.did, "name": self.name, "nick": self.nick, "resource": self.resource}

bumper/confserver.py

@@ -6,12 +6,19 @@ from http.server import BaseHTTPRequestHandler
 from http import HTTPStatus
 from threading import Thread
 import socket, logging, ssl, json, sys
+import asyncio
+import contextvars
+import bumper
+import string
+import random
 
 
 class RequestHandler(BaseHTTPRequestHandler):
-#class RequestHandler(SimpleHTTPRequestHandler):
+    bumper_clients = contextvars.ContextVar
+    helperbot = object
+
     def do_POST(self):
-        try:
+        try:            
             self.protocol_version = 'HTTP/1.1'
             content_length = int(self.headers['Content-Length'])
             post_data = self.rfile.read(content_length)
@@ -23,6 +30,10 @@ class RequestHandler(BaseHTTPRequestHandler):
                 body = {"code":0,"data":[{"classid":"dl8fht","product":{"_id":"5acb0fa87c295c0001876ecf","name":"DEEBOT 600 Series","icon":"5acc32067c295c0001876eea","UILogicId":"dl8fht","ota":False,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5acc32067c295c0001876eea"}},{"classid":"02uwxm","product":{"_id":"5ae1481e7ccd1a0001e1f69e","name":"DEEBOT OZMO Slim10 Series","icon":"5b1dddc48bc45700014035a1","UILogicId":"02uwxm","ota":False,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b1dddc48bc45700014035a1"}},{"classid":"y79a7u","product":{"_id":"5b04c0227ccd1a0001e1f6a8","name":"DEEBOT OZMO 900","icon":"5b04c0217ccd1a0001e1f6a7","UILogicId":"y79a7u","ota":True,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b04c0217ccd1a0001e1f6a7"}},{"classid":"jr3pqa","product":{"_id":"5b43077b8bc457000140363e","name":"DEEBOT 711","icon":"5b5ac4cc8d5a56000111e769","UILogicId":"jr3pqa","ota":True,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b5ac4cc8d5a56000111e769"}},{"classid":"uv242z","product":{"_id":"5b5149b4ac0b87000148c128","name":"DEEBOT 710","icon":"5b5ac4e45f21100001882bb9","UILogicId":"uv242z","ota":True,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5b5ac4e45f21100001882bb9"}},{"classid":"ls1ok3","product":{"_id":"5b6561060506b100015c8868","name":"DEEBOT 900 Series","icon":"5ba4a2cb6c2f120001c32839","UILogicId":"ls1ok3","ota":True,"iconUrl":"https://portal-ww.ecouser.net/api/pim/file/get/5ba4a2cb6c2f120001c32839"}}]}
             elif "notify_engine.do" in str(self.path):
                 body = {"ret":"ok"}
+            elif "iot/devmanager.do" in str(self.path): #Handle rest commands
+                randomid = ''.join(random.sample(string.ascii_letters,6))
+                retcmd = self.helperbot.send_command(json_body, randomid)
+                body = retcmd
             else:
                 todo = json_body['todo']
                 if todo == 'FindBest':
@@ -40,19 +51,9 @@ class RequestHandler(BaseHTTPRequestHandler):
                             "token": json_body["token"]
                             }
                 elif todo == 'GetDeviceList':
-                    #Find a way to handle this automatically
+                    active_bots = self.bumper_clients.get()
-                    #Maybe keep a list of devices from those that have checked into MQTT/XMPP/etc and return them
                     body = {
-                            "devices": [
+                            "devices": active_bots,
-                            {
-                            "class": "ls1ok3",
-                            "company": "eco-ng",
-                            "did": "tmpDeviceID",
-                            "name": "tmpName",
-                            "nick": "tmpNick",
-                            "resource": "tmpResource"
-                            }
-                            ],
                             "result": "ok",
                             "todo": "result"
                             }
@@ -73,8 +74,7 @@ class RequestHandler(BaseHTTPRequestHandler):
             self.protocol_version = 'HTTP/1.1'
             logging.debug("Path: " + self.path)
             
-            if "/login?" in str(self.path):
+            if "/login?" in str(self.path):              
-              
                 #Could implement basic auth if you wanted, or just accept anything
                 #Next up parse the path for variables
                 body = {
@@ -136,11 +136,18 @@ class RequestHandler(BaseHTTPRequestHandler):
 
 
 class HTTPServerThread(HTTPServer, Thread):
-    def __init__(self, server_address):
+    bumper_clients = contextvars.ContextVar
+    def __init__(self, server_address, bumper_clients, helperbot):
         Thread.__init__(self)
         self.server_address = server_address
+        self.bumper_clients = bumper_clients
+        self.helperbot = helperbot
         self.handler = RequestHandler
+        self.handler.bumper_clients = self.bumper_clients
+        self.handler.helperbot = self.helperbot
+        
         self.exit_flag = False
+        
         HTTPServer.__init__(self, self.server_address, self.handler)
         
     def handle_error(self, request, client_address):
@@ -161,14 +168,18 @@ class HTTPServerThread(HTTPServer, Thread):
 
 
 class ConfServer():
-    def __init__(self, address, usessl=False, run_async=True):
+    bumper_clients = contextvars.ContextVar
-        self.run_async = run_async
+   
-        self.server = HTTPServerThread(address)
+    def __init__(self, address, usessl=False, run_async=True, bumper_clients=contextvars.ContextVar, helperbot=None):
+        self.run_async = run_async        
+        self.server = HTTPServerThread(address, bumper_clients, helperbot)
+        
         try:            
             if usessl:
                 self.server.socket = ssl.wrap_socket(self.server.socket, keyfile='./certs/key.pem', certfile='./certs/cert.pem', server_side=True)
             if self.run_async:
                 self.server.start()
+                
             else:
                 try:
                     self.server.run()
@@ -176,6 +187,7 @@ class ConfServer():
                     self.disconnect()
         except Exception as e:
             logging.error('ConfServer: {}'.format(e))
+
     def disconnect(self):
         logging.info('ConfServer: shutting down...')
         self.server.disconnect()

bumper/mqttserver.py

@@ -4,37 +4,178 @@ import logging
 import asyncio
 import os
 from hbmqtt.broker import Broker
+from hbmqtt.client import MQTTClient
 import hbmqtt
 import pkg_resources
+import contextvars
+import time
+import bumper
+from threading import Thread
+import ssl
+from paho.mqtt.client import Client as ClientMQTT
+from paho.mqtt import publish as MQTTPublish
+from paho.mqtt import subscribe as MQTTSubscribe
 
 
-class BumperMQTTPlugin:
+class BumperMQTTPlugin:    
     def __init__(self, context):
         self.context = context        
+        try:
+            self.bots = self.context.config['bots']
+        except KeyError:
+             self.context.logger.warning("'bots' section not found in context configuration")
         logging.debug('Bumper Plugin Initialized')
 
-    @asyncio.coroutine
+    async def on_broker_client_connected(self, client_id):
-    def on_broker_client_connected(self, client_id):
         logging.debug('Bumper Connection: %s connected' % client_id)
-        #yield from self.context.broadcast_message('location/%s' % client_id, b'home')
+        connected_bots = self.bots['connected_bots'].get()                
+        didsplit = str(client_id).split("@")
+        #If this isn't a fake user (fuid) then add as a bot
+        if not (str(didsplit[0]).startswith("fuid") or str(didsplit[0]).startswith("helper")):
+            tmpbotdetail = str(didsplit[1]).split("/")        
+            newbot = bumper.VacBotDevice()
+            newbot.did = didsplit[0]        
+            newbot.vac_bot_device_class = tmpbotdetail[0]
+            newbot.resource = tmpbotdetail[1]
+            botactive = False
+            for bot in connected_bots:
+                if bot['did'] == newbot.did:
+                    botactive = True
+            
+            if botactive == False:
+                connected_bots.append(newbot.asdict())
+            
+            self.bots['connected_bots'].set(connected_bots)
 
-    @asyncio.coroutine
+        logging.debug('Connected Bots: %s' %self.bots['connected_bots'].get())
-    def on_broker_client_disconnected(self, client_id):
+
+    async def on_broker_client_disconnected(self, client_id):
         logging.debug('Bumper Connection: %s disconnected' % client_id)
-        #yield from self.context.broadcast_message('location/%s' % client_id, b'not_home')            
+        #To do handle removing bot
+
+class MQTTHelperBot(ClientMQTT):
 
 
-class MQTTServer():
+    def __init__(self, address, run_async=False, bumper_clients=contextvars.ContextVar):
-    clients = []
+        ClientMQTT.__init__(self)
+        self.address = address
+        self._client_id = "helper1@bumper/helper1"
+        self.command_responses = contextvars.ContextVar('command_responses', default=[])
+
+        try:            
+            if run_async: 
+                hloop = asyncio.new_event_loop()                
+                logging.debug("Starting MQTT HelperBot Thread: 1")
+                mserver = Thread(name="MQTTHelperBot_Thread",target=self.run_helperbot, args=(hloop,))
+                mserver.setDaemon(True)
+                mserver.start()                    
+                
+            else:
+                self.run_helperbot()
+
+        except:
+            logging.exception("Exception")
+            pass    
+
+    def run_helperbot(self, loop):           
+        formatter = "[%(asctime)s] :: %(levelname)s :: %(name)s :: %(message)s"
+        logging.basicConfig(level=logging.DEBUG, format=formatter)   
+        asyncio.set_event_loop(loop)
+        loop.run_until_complete(self.start_helper_bot())      
+        loop.run_forever() 
+    
+    async def start_helper_bot(self):        
+        #self._on_log = self.on_log #This provides more logging than needed, even for debug
+        self._on_message = self.get_msg
+        self._on_connect = self.on_connect        
+
+        #TODO: This is pretty insecure and accepts any cert, maybe actually check?
+        ssl_ctx = ssl.create_default_context()
+        ssl_ctx.check_hostname = False
+        ssl_ctx.verify_mode = ssl.CERT_NONE
+        self.tls_set_context(ssl_ctx)
+        self.tls_insecure_set(True)
+
+        self.connect(self.address[0], self.address[1])
+        self.loop_start()        
+        
+
+    def on_connect(self, client, userdata, flags, rc):        
+        if rc != 0:
+            logging.error("HelperBot - error connecting with MQTT Return {}".format(rc))
+            raise RuntimeError("HelperBot - error connecting with MQTT Return {}".format(rc))
+                 
+        else:
+            logging.debug("HelperBot - Connected with result code "+str(rc))
+            logging.debug("HelperBot - Subscribing to all")        
+
+        
+            self.subscribe('iot/p2p/+/+/+/+/helper1/bumper/helper1/+/+/+', qos=0)
+            self.subscribe('iot/p2p/+', qos=0)
+                                                      
+
+    def get_msg(self, client, userdata, message):
+        logging.debug("HelperBot MQTT Received Message on Topic: {} - Message: {}".format(message.topic, str(message.payload.decode("utf-8"))))
+        print(str(message.payload.decode("utf-8")))
+        cresp = self.command_responses.get()
+        cresp.append({"topic": message.topic,"payload":str(message.payload.decode("utf-8"))})
+        self.command_responses.set(cresp)                          
+
+    async def wait_for_resp(self, requestid):               
+        t_end = time.time() + 10
+        while time.time() < t_end:
+            await asyncio.sleep(0.2)
+            responses = self.command_responses.get()   
+            if len(responses) > 0:    
+                for msg in responses:
+                    topic = str(msg['topic']).split("/")
+                    if (topic[6] == "helper1" and topic[10] == requestid):
+                        logging.debug('Vac Responses MQTT: Topic: %s Payload: %s' % (msg['topic'], msg['payload']))
+                        resp = {
+                            "id": requestid,
+                            "ret": "ok",
+                            "resp": msg['payload']
+                        }
+                        cresp = self.command_responses.get()
+                        cresp.remove(msg)                    
+                        self.command_responses.set(cresp)                             
+                        return resp
+
+        return { "id": requestid, "errno": "timeout", "ret": "fail" }
+
+    def send_command(self, cmdjson, requestid):
+        ttopic = "iot/p2p/{}/helper1/bumper/helper1/{}/{}/{}/q/{}/{}".format(cmdjson["cmdName"],
+        cmdjson["toId"], cmdjson["toType"], cmdjson["toRes"], requestid, cmdjson["payloadType"])
+        self.publish(ttopic, cmdjson["payload"])
+
+        loop = asyncio.new_event_loop()
+        resp = loop.run_until_complete(self.wait_for_resp(requestid))     
+        
+        print(resp)        
+        return resp
+        
+
+class MQTTServer():    
     exit_flag = False
     default_config = {}  
+    bumper_clients = []
 
-    @asyncio.coroutine
+    async def broker_coro(self):         
-    def broker_coro(self):         
+        broker = hbmqtt.broker.Broker(config=self.default_config) 
-        broker = hbmqtt.broker.Broker(config=self.default_config)
+        for plugin in broker.plugins_manager.plugins:
-        yield from broker.start()  
+            if plugin.name == 'broker_sys':
+                broker.plugins_manager.plugins.remove(plugin)            
+            if plugin.name == 'packet_logger_plugin':
+                broker.plugins_manager.plugins.remove(plugin)        
+        
+        await broker.start()  
 
-    def __init__(self, address, run_async=False):
+    async def active_bot_listing(self):               
+        while True:
+            await asyncio.sleep(5)
+            logging.debug('Connected bots: %s' % self.bumper_clients.get())     
+
+    def __init__(self, address, run_async=False, bumper_clients=contextvars.ContextVar):
         #The below adds a plugin to the hbmqtt.broker.plugins without having to futz with setup.py
         distribution = pkg_resources.Distribution("hbmqtt.broker.plugins")
         bumper_plugin = pkg_resources.EntryPoint.parse('bumper = bumper.mqttserver:BumperMQTTPlugin', dist=distribution)        
@@ -42,6 +183,8 @@ class MQTTServer():
         pkg_resources.working_set.add(distribution)
         #for entry_point in pkg_resources.iter_entry_points("hbmqtt.broker.plugins"):
         #   print(entry_point)
+
+        self.bumper_clients = bumper_clients
         try:            
             # Initialize bot server
             self.default_config = {
@@ -64,23 +207,33 @@ class MQTTServer():
                         '' #No plugins == no auth
                     ]
                 },
-                'broker': {
-                    'plugins': [
-                        'bumper'
-                    ]
-                },
                 'topic-check': {
                     'enabled': False
+                },
+                'bots':{
+                    'connected_bots': bumper_clients
                 }
             }
             if run_async:
-                self.run()
+                sloop = asyncio.new_event_loop()                
-        finally:
+                logging.debug("Starting MQTTServer Thread: 1")
-            print("Done")     
+                mserver = Thread(name="MQTTServer_Thread",target=self.run_server, args=(sloop,))
+                mserver.setDaemon(True)
+                mserver.start()                   
+                
+            else:
+                self.run_server()
 
-    def run(self):
+        except:
+            logging.exception("Exception")
+            pass              
+    
+
+    def run_server(self, loop):           
         formatter = "[%(asctime)s] :: %(levelname)s :: %(name)s :: %(message)s"
-        logging.basicConfig(level=logging.DEBUG, format=formatter)
+        logging.basicConfig(level=logging.DEBUG, format=formatter)   
-        eloop = asyncio.get_event_loop().run_until_complete(self.broker_coro())        
+        asyncio.set_event_loop(loop)
-        asyncio.get_event_loop().run_forever()
+        loop.run_until_complete(self.broker_coro())
+        #loop.run_until_complete(self.active_bot_listing())                     
+        loop.run_forever()