diff --git a/Pipfile b/Pipfile index cb8599b..fff0c8b 100644 --- a/Pipfile +++ b/Pipfile @@ -5,5 +5,6 @@ name = "pypi" [packages] hbmqtt = "*" +paho-mqtt = "*" [dev-packages] diff --git a/bumper.py b/bumper.py index 3a7bb61..efbcd8c 100644 --- a/bumper.py +++ b/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) -conf_server = bumper.ConfServer(conf_address, usessl=True, run_async=True) -# start mqtt server (async) -mqtt_server = bumper.MQTTServer(mqtt_address, run_async=True) +# A default bot could be set here to automatically add it as available +# dbot = bumper.VacBotDevice("did", "class", "resource", "name","nick" ) +# bclient = bumper.bumper_clients_var +# 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) diff --git a/bumper/__init__.py b/bumper/__init__.py index e75e9aa..574ad3a 100644 --- a/bumper/__init__.py +++ b/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} diff --git a/bumper/confserver.py b/bumper/confserver.py index 7651b47..d3edc3d 100644 --- a/bumper/confserver.py +++ b/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 - #Maybe keep a list of devices from those that have checked into MQTT/XMPP/etc and return them + active_bots = self.bumper_clients.get() body = { - "devices": [ - { - "class": "ls1ok3", - "company": "eco-ng", - "did": "tmpDeviceID", - "name": "tmpName", - "nick": "tmpNick", - "resource": "tmpResource" - } - ], + "devices": active_bots, "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): - self.run_async = run_async - self.server = HTTPServerThread(address) + bumper_clients = contextvars.ContextVar + + 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() diff --git a/bumper/mqttserver.py b/bumper/mqttserver.py index 1bf0374..5c52c57 100644 --- a/bumper/mqttserver.py +++ b/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 - def on_broker_client_connected(self, client_id): + async 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 - def on_broker_client_disconnected(self, client_id): + logging.debug('Connected Bots: %s' %self.bots['connected_bots'].get()) + + 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(): - clients = [] + def __init__(self, address, run_async=False, bumper_clients=contextvars.ContextVar): + 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 - def broker_coro(self): - broker = hbmqtt.broker.Broker(config=self.default_config) - yield from broker.start() + async def broker_coro(self): + broker = hbmqtt.broker.Broker(config=self.default_config) + for plugin in broker.plugins_manager.plugins: + 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() - finally: - print("Done") + sloop = asyncio.new_event_loop() + logging.debug("Starting MQTTServer Thread: 1") + 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) - eloop = asyncio.get_event_loop().run_until_complete(self.broker_coro()) - asyncio.get_event_loop().run_forever() + logging.basicConfig(level=logging.DEBUG, format=formatter) + asyncio.set_event_loop(loop) + loop.run_until_complete(self.broker_coro()) + #loop.run_until_complete(self.active_bot_listing()) + loop.run_forever()