leda_python/deviceMbus.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-


# ====#====#====#====
# file: deviceMbus
# time: 1/30/18
# ====#====#====#====

from . import ledaException as exception
from gi.repository import GLib
import dbus.mainloop.glib
from . import mbusConfig
from .refactoring import dbus_service
import threading
import logging
import json
from . import mbus
import os
import re
from . import json_coder

_logger = logging.getLogger(__name__)


class SyncMsg_Event(object):
    def __init__(self):
        self.event = threading.Event()
        self.clear()
        self.msg = None

    def clear(self):
        self.msg = None
        self.event.clear()

    def wait(self, time_s):
        self.event.wait(time_s)

        return self.msg

    def set(self, msg):
        self.msg = msg
        self.event.set()


mbusLoopFlag = False


def mbus_loop():
    try:
        _logger.debug("mbus main looping...")

        mainloop = GLib.MainLoop()
        mainloop.run()
    except:
        _logger.debug(">>>>>>>>>>>>> driver existed<<<<<<<<<<<<<<<")
        os._exit(0)


class device_callback(object):

    def callService(self, name, input):
        '''
		:param name[string]: method name
		:param input[string]: formate: key-value json-string ,eg:
			{
				"params":{
					"args1": xxx,
					"args2": yyy
				}
			}
		:return:
			code[int]         : 消息编码和msg对应
			msg[string]        : 返回的执行状态
			output[string]    : 方法返回的输出数据
				 {
					"key1": xxx,
					"key2": yyy,
					 ...
				}
		'''
        raise exception.LedaCallBackException("callService is empty")

    def getprofile_cb(self):
        '''
		:return: profile[string]: 设备三要素模型
		'''
        raise exception.LedaCallBackException("getprofile_cb is empty")

    def getProperties(self, input):
        '''
		:param input[string]: 格式json中的数组类型：[property1,property2]
		:return: 格式为key-value形式的json串,如：{property1:xxx,property2:yyy}
		'''
        raise exception.LedaCallBackException("getProperties is empty")

    def setProperties(self, input):
        '''
		:param input[string]:属性列表，其格式为key-value形式的json串,如：{property1:xxx,property2:yyy}
		:return: key-value json-string, if no data to return ,you should return "{}"
		'''
        raise exception.LedaCallBackException("setProperties is empty")


class device_service(object):
    def __init__(self, cloud_id, device_name, product_key, bus_callback_object):

        self.cloud_id = cloud_id
        self.device_name = device_name
        self.product_key = product_key
        self.bus_callback_object = bus_callback_object
        self.deviceMbusHandle = None
        self.deviceMbusObject = None

    def get_cloud_id(self):
        return self.cloud_id

    def _getDeviceProfile(self, interface):

        @dbus.service.method(interface, out_signature='s')
        def getDeviceProfile(self):
            _logger.debug("cloud_id:%s, method: getDeviceProfile", self.cloud_id)

            try:
                profile = self.callback_funs.getprofile_cb()
                if (False == isinstance(profile, str)):
                    _logger.warning("getprofile_cb(cloud_id:%s) return args type is invalid: %s", self.cloud_id,
                                    type(profile))
                    retDict = {
                        "code": exception.LEDA_ERROR_INVAILD_PARAM,  # params invalid
                        "message": "getprofile_cb(cloud_id:%s) return args type is invalid: %s" % (
                        self.cloud_id, type(profile))
                    }
                    _logger.debug("getDeviceProfile(cloud_id:%s): retMsg: %s", self.cloud_id, retDict)
                    return json.dumps(retDict, ensure_ascii = False)
                profile_dict = json.loads(profile)

            except(AttributeError, ValueError, TypeError) as err:
                _logger.warning('%s', err)

                retDict = {
                    "code": exception.LEDA_ERROR_INVAILD_PARAM,  # params invalid
                    "message": "%s" % (err)
                }
                _logger.debug("getDeviceProfile(cloud_id:%s): retMsg: %s", self.cloud_id, retDict)
                return json.dumps(retDict, ensure_ascii = False)

            except:
                _logger.exception("Err")

                retDict = {
                    "code": exception.LEDA_ERROR_FAILED,  # params invalid
                    "message": "unkonwn error"
                }
                _logger.debug("getDeviceProfile(cloud_id:%s): retMsg: %s", self.cloud_id, retDict)
                return json.dumps(retDict, ensure_ascii = False)

            retDict = {
                "code": exception.LEDA_SUCCESS,
                "message": "successfully",
                "params": {
                    "deviceProfile": profile_dict
                }
            }
            _logger.debug("getDeviceProfile(cloud_id:%s): retMsg: %s", self.cloud_id, retDict)
            return json.dumps(retDict, ensure_ascii = False)

        return getDeviceProfile

    def _callServices(self, interface):

        @dbus.service.method(interface, in_signature='ss', out_signature='s')
        def callServices(self, methodName, inMsg):
            _logger.debug("callServices(cloud_id:%s) in params: method: %s, args: %s", self.cloud_id, methodName, inMsg)

            codeInfoDict = {
                exception.LEDA_SUCCESS: 'successfully',
                exception.LEDA_ERROR_INVAILD_PARAM: 'invalid params',
                exception.LEDA_ERROR_FAILED: 'exec failed',
                exception.LEDA_ERROR_TIMEOUT: 'timeout',
                exception.LEDA_ERROR_NOT_SUPPORT: 'not support',
                exception.LEDA_ERROR_PROPERTY_NOT_EXIST: 'property not exist',
                exception.LEDA_ERROR_PROPERTY_READ_ONLY: 'property read only',
                exception.LEDA_ERROR_PROPERTY_WRITE_ONLY: 'property write only',
                exception.LEDA_ERROR_SERVICE_NOT_EXIST: 'service not exist',
                exception.LEDA_ERROR_SERVICE_INPUT_PARAM: 'invalid service input params'
            }

            try:
                inArgs = json.loads(inMsg)["params"]
                if (methodName == "get"):
                    code, retInfo = self.callback_funs.getProperties(inArgs)
                    if (False == isinstance(retInfo, dict) or (False == isinstance(code, int))):
                        _logger.warning("get(cloud_id:%s) return args type is invalid: %s", self.cloud_id,
                                        type(retInfo))
                        retDict = {
                            "code": exception.LEDA_ERROR_INVAILD_PARAM,  # params invalid
                            "message": "get(cloud_id:%s) return args type is invalid: %s" % (
                            self.cloud_id, type(retInfo)),
                            "params": {}
                        }
                        _logger.debug("get(cloud_id:%s): retMsg: %s", self.cloud_id, retDict)
                        return json.dumps(retDict, ensure_ascii = False)

                    if (exception.LEDA_SUCCESS != code):
                        retDict = {
                            "code": exception.LEDA_ERROR_FAILED,
                            "message": "get(cloud_id:%s) exec failed" % (self.cloud_id),
                            "params": {}
                        }
                        _logger.debug("get(cloud_id:%s): retMsg: %s", self.cloud_id, retDict)
                        return json.dumps(retDict, ensure_ascii = False)

                    retDict = {
                        "code": code,
                        "message": codeInfoDict[code],
                        "params": retInfo
                    }
                    _logger.debug("get(cloud_id:%s): retMsg: %s", self.cloud_id, retDict)
                    return json.dumps(retDict, cls=json_coder.Json_Encoder, ensure_ascii = False)

                elif (methodName == "set"):
                    code, retInfo = self.callback_funs.setProperties(inArgs)
                    if (False == isinstance(retInfo, dict) or (False == isinstance(code, int))):
                        _logger.warning("set(cloud_id:%s) return args type is invalid: %s", self.cloud_id,
                                        type(retInfo))

                        retDict = {
                            "code": exception.LEDA_ERROR_INVAILD_PARAM,  # params invalid
                            "message": "set(cloud_id:%s) return args type is invalid: %s" % (
                                self.cloud_id, type(retInfo)),
                            "params": {}
                        }
                        _logger.debug("set(cloud_id:%s): retMsg: %s", self.cloud_id, retDict)
                        return json.dumps(retDict, ensure_ascii = False)

                    if (exception.LEDA_SUCCESS != code):
                        retDict = {
                            "code": exception.LEDA_ERROR_FAILED,
                            "message": "set(cloud_id:%s) exec failed" % (self.cloud_id),
                            "params": {}
                        }
                        _logger.debug("set(cloud_id:%s): retMsg: %s", self.cloud_id, retDict)
                        return json.dumps(retDict, ensure_ascii = False)

                    retDict = {
                        "code": code,
                        "message": codeInfoDict[code],
                        "params": {}
                    }
                    _logger.debug("set(cloud_id:%s): retMsg: %s", self.cloud_id, retDict)
                    return json.dumps(retDict, ensure_ascii = False)

                else:
                    code, output = self.callback_funs.callService(methodName, inArgs)
                    if ((False == isinstance(code, int)) or (False == isinstance(output, dict))):
                        _logger.warning("callService(cloud_id:%s) return args type is invalid", self.cloud_id)
                        retDict = {
                            "code": exception.LEDA_ERROR_INVAILD_PARAM,  # params invalid
                            "message": "callService(cloud_id:%s) return args type is invalid" % (self.cloud_id),
                            "params": {
                                "code": exception.LEDA_ERROR_INVAILD_PARAM,
                                "message": "callService(cloud_id:%s) return args type is invalid" % (self.cloud_id),
                                "data": {}
                            }
                        }
                        _logger.debug("callServices(cloud_id:%s): %s retMsg: %s", self.cloud_id, methodName, retDict)
                        return json.dumps(retDict, ensure_ascii = False)

                    if (exception.LEDA_SUCCESS != code):
                        retDict = {
                            "code": exception.LEDA_ERROR_FAILED,
                            "message": "callService(cloud_id:%s) exec failed" % (self.cloud_id),
                            "params": {
                                "code": exception.LEDA_ERROR_FAILED,
                                "message": "callService(cloud_id:%s) exec failed" % (self.cloud_id),
                                "data": {}
                            }
                        }
                        _logger.debug("callServices(cloud_id:%s): %s retMsg: %s", self.cloud_id, methodName,
                                      retDict)
                        return json.dumps(retDict, ensure_ascii = False)

                    data = output

            except(AttributeError, ValueError, TypeError, KeyError) as err:
                _logger.exception('Err')
                _logger.warning('%s', err)
                retDict = {
                    "code": exception.LEDA_ERROR_INVAILD_PARAM,  # params invalid
                    "message": "%s" % (err),
                    "params": {}
                }
                _logger.debug("callServices(cloud_id:%s): %s retMsg: %s", self.cloud_id, methodName, retDict)
                return json.dumps(retDict, ensure_ascii = False)

            except:
                _logger.exception("Err")

                retDict = {
                    "code": exception.LEDA_ERROR_FAILED,  # params invalid
                    "message": "unkonwn error",
                    "params": {}
                }
                _logger.warning("callServices(cloud_id:%s): %s retMsg: %s", self.cloud_id, methodName, retDict)
                return json.dumps(retDict, ensure_ascii = False)

            retDict = {
                "code": exception.LEDA_SUCCESS,
                "message": "successfully",
                "params": {
                    "code": code,
                    "message": codeInfoDict[code],
                    "data": data
                }
            }
            _logger.debug("callServices(cloud_id:%s): %s retMsg: %s", self.cloud_id, methodName, retDict)
            return json.dumps(retDict, ensure_ascii = False)

        return callServices

    def _createMbusDynamicObject(self):

        interface = mbusConfig.CMP_DEVICE_WKN_PREFIX + self.cloud_id

        attrDict = {
            "callback_funs": self.bus_callback_object,
            "cloud_id": self.cloud_id,

            "callServices": self._callServices(interface),
            "getDeviceProfile": self._getDeviceProfile(interface)
        }

        DemoObjectClass = type("Device_" + self.product_key + self.device_name, (dbus_service.DbusObject,), attrDict)
        objPath = '/' + interface.replace('.', '/')
        objectHandle = self.deviceMbusHandle.createObject(DemoObjectClass, objPath)

        return objectHandle

    def releaseMbusObject(self):
        if (self.deviceMbusHandle):
            bus = self.deviceMbusHandle.getBus()
            wellKnownName = mbusConfig.CMP_DEVICE_WKN_PREFIX + self.cloud_id
            objectPath = '/' + wellKnownName.replace('.', '/')
            if (None != self.deviceMbusObject):
                self.deviceMbusObject.remove_from_connection(bus, objectPath)
                self.deviceMbusObject = None

            self.deviceMbusHandle.releaseName()
            self.deviceMbusHandle = None

    def device_report_property(self, report_info):
        '''
		:param report_info: report_info 上报信息,其格式为key-value形式的json串，如上报属性：
			{
			 "property1": {
				"value" : "xxx",
				"time" : 1524448722000
			 },
			 "property1": {
				"value" : "yyy",
				"time" : 1524448722000
			 }
			 ...
		}
		:return:
		'''

        if (None == self.deviceMbusHandle):
            _logger.warning("device(%s) can't report property unless online ", self.cloud_id)
            raise exception.LedaReportPropertyException(
                "device(%s) can't report property unless online" % (self.cloud_id))

        if (False == isinstance(report_info, str)):
            raise exception.LedaReportPropertyException(
                "device(%s):device_report_property,params type is invalid: %s" % (self.cloud_id, type(report_info)))

        srcWKN = self.deviceMbusHandle.getName()
        srcInterface = srcWKN
        srcObjectPath = "/" + srcInterface.replace(".", "/")
        self.deviceMbusHandle.unicastSignal(srcObjectPath, srcInterface, mbusConfig.DMP_SUB_WKN, 's',
                                            "propertiesChanged", report_info)

        _logger.info("Device(%s): report properties: %s" % (self.cloud_id, report_info))

    def device_report_event(self, name, report_info):
        '''
		:param name: 事件名称
		:param report_info: 携带信息，其格式为key-value形式的json串， 如上报事件：
			{
				"params": {
					"value" : {
					  "key1":"value1",
					  "key2":"value2"
					},
					"time" : 1524448722000
				  }
			 }
		:return:
		'''

        if (None == self.deviceMbusHandle):
            _logger.warning("device(%s) can't report event unless online ", self.cloud_id)
            raise exception.LedaReportEventException("device(%s) can't report event unless online" % (self.cloud_id))

        if ((False == isinstance(report_info, str)) or (False == isinstance(name, str))):
            raise exception.LedaReportEventException(
                "device(%s):device_report_event,params type is invalid" % (self.cloud_id))

        if (len(name) > mbusConfig.STRING_NAME_MAX_LEN):
            raise exception.LedaReportEventException(
                "device(%s):device_report_event,params name is too long(%s)" % (len(name)))

        srcWKN = self.deviceMbusHandle.getName()
        srcInterface = srcWKN
        srcObjectPath = "/" + srcInterface.replace(".", "/")

        self.deviceMbusHandle.unicastSignal(srcObjectPath, srcInterface, mbusConfig.DMP_SUB_WKN, 's', name, report_info)

        _logger.info("Device(%s): report event(%s): %s" % (self.cloud_id, name, report_info))


class driver_service(object):
    def __init__(self):
        self.driverMbusHandle = None
        self.driverMbusObject = None
        self.device_service_dict = {}
        self.deviceServiceDictLock = threading.Lock()
        self.driver_name = None
        self.driver_id = None

    def _exit(self, connection):
        _logger.warning("the connection is Abnormal(closed or bus daemon crashed),the process exited automatically")
        os._exit(0)

    def _notify_config(self, interface):

        @dbus.service.method(interface, in_signature='ss', out_signature='i')
        def notify_config(self, key, value):
            if (self.config_callback_obj):
                t = threading.Thread(target=self.config_callback_obj.deviceConfigCB, args=(key, value))
                t.setDaemon(True)
                t.start()

            return exception.LEDA_SUCCESS

        return notify_config

    def _getDeviceList(self, interface):

        @dbus.service.method(interface, in_signature='s', out_signature='s')
        def getDeviceList(self, deviceState):

            devNum = 0
            devList = []

            with self.deviceServiceDictLock:

                try:

                    if (deviceState == ""):
                        for pk_dn in self.device_service_dict:
                            devNum += 1
                            devList.append(self.device_service_dict[pk_dn][0])  # cloud_id
                    elif (deviceState == "online"):
                        for pk_dn in self.device_service_dict:
                            if (None != self.device_service_dict[pk_dn][1].deviceMbusHandle):
                                devNum += 1
                                devList.append(self.device_service_dict[pk_dn][0])  # cloud_id
                    elif (deviceState == "offline"):
                        for pk_dn in self.device_service_dict:
                            if (None == self.device_service_dict[pk_dn][1].deviceMbusHandle):
                                devNum += 1
                                devList.append(self.device_service_dict[pk_dn][0])  # cloud_id
                    else:
                        _logger.warning("method: getDeviceList inparams is invalid")

                except:
                    _logger.exception("Err")

            outMsg = {
                "params": {
                    "devNum": devNum,
                    "devList": devList
                }
            }

            return json.dumps(outMsg, ensure_ascii = False)

        return getDeviceList

    def _releaseMbusObject(self):
        if (self.driverMbusHandle):

            with self.deviceServiceDictLock:

                for pk_dn in self.device_service_dict:
                    self.device_service_dict[pk_dn][0] = None  # clean cloud_id
                    self.device_service_dict[pk_dn][1].device_disconnect()

                self.device_service_dict = {}
                try:
                    bus = self.driverMbusHandle.getBus()
                except:
                    _logger.exception('Err')
                    return

                wellKnownName = mbusConfig.CMP_DRIVER_WKN_PREFIX + self.driver_id

                objectPath = '/' + wellKnownName.replace('.', '/')
                if (None != self.driverMbusObject):
                    self.driverMbusObject.remove_from_connection(bus, objectPath)
                    self.driverMbusObject = None

                self.driverMbusHandle.releaseName()

    def _createMbusDynamicObject(self, driver_name):

        interface = mbusConfig.CMP_DRIVER_WKN_PREFIX + driver_name

        attrDict = {
            "device_service_dict": self.device_service_dict,
            "deviceServiceDictLock": self.deviceServiceDictLock,
            "config_callback_obj": None,

            "getDeviceList": self._getDeviceList(interface),
            "notify_config": self._notify_config(interface)
        }

        DemoObjectClass = type("Driver_" + driver_name, (dbus.service.Object,), attrDict)
        objPath = '/' + interface.replace('.', '/')
        objectHandle = self.driverMbusHandle.createObject(DemoObjectClass, objPath)

        return objectHandle

    def _openMonitorMbusDaemon(self):
        bus = self.driverMbusHandle.getBus()
        bus.call_on_disconnection(self._exit)

    def driver_init_with_driverId(self, driver_id):
        global mbusLoopFlag

        if (False == mbusLoopFlag):
            dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

        if (False == isinstance(driver_id, str)):
            raise exception.LedaParamsException("driver_init_with_driverId: driver_id type is invalid")

        try:
            wkn = mbusConfig.CMP_DRIVER_WKN_PREFIX + driver_id
            self.driverMbusHandle = mbus.MbusConnect(wkn)
            self._openMonitorMbusDaemon()
            self.driverMbusObject = self._createMbusDynamicObject(driver_id)
            self.driver_id = driver_id
        except SystemExit:
            raise exception.LedaException("mbus existed")
        except:
            _logger.exception("Err")
            raise exception.LedaException("mbus connect failed")

        if (False == mbusLoopFlag):
            t = threading.Thread(target=mbus_loop, name="mbusLoop")
            t.setDaemon(True)
            t.start()

            mbusLoopFlag = True

    def driver_init(self, driver_name):
        global mbusLoopFlag

        if (False == mbusLoopFlag):
            dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

        if (False == isinstance(driver_name, str)):
            raise exception.LedaParamsException("driver_init: driver_name type is invalid")

        try:
            wkn = mbusConfig.CMP_DRIVER_WKN_PREFIX + driver_name
            self.driverMbusHandle = mbus.MbusConnect(wkn)
            self._openMonitorMbusDaemon()
            self.driverMbusObject = self._createMbusDynamicObject(driver_name)
            self.driver_name = driver_name
        except SystemExit:
            raise exception.LedaException("mbus existed")
        except:
            _logger.exception("Err")
            raise exception.LedaException("mbus connect failed")

        if (False == mbusLoopFlag):
            t = threading.Thread(target=mbus_loop, name="mbusLoop")
            t.setDaemon(True)
            t.start()

            mbusLoopFlag = True

    def driver_set_watchdog(self, thread_name, count_down):
        '''
		:param thread_name: 需要保活的线程名称
		:param count_down: 倒计时时间，-1表示停止保活
		:return:
		'''
        if (None == self.driverMbusHandle):
            raise exception.LedaBusHandleException("mbus Handle is None")

        if (False == isinstance(thread_name, str)):
            raise exception.LedaRPCMethodException("driver_set_watchdog: thread_name is valid:%s" % (thread_name))
        else:
            if ((len(thread_name) > mbusConfig.STRING_NAME_MAX_LEN) or (len(thread_name) == 0)):
                raise exception.LedaRPCMethodException("driver_set_watchdog: thread_name is valid:%s" % (thread_name))

        if (False == isinstance(count_down, int)):
            raise exception.LedaRPCMethodException("driver_set_watchdog: count_down is valid:%s" % (count_down))
        else:
            if ((count_down == 0) or (count_down < -2)):
                raise exception.LedaRPCMethodException("driver_set_watchdog: count_down is valid:%s" % (count_down))

        try:
            self.driverMbusHandle.feedDog(thread_name, count_down)

        except:
            _logger.exception("Err")
            raise exception.LedaFeedDogException("feed dog failed")

    def driver_exit(self):
        if (None == self.driverMbusHandle):
            _logger.debug("driverMbusHandle is None")
        else:

            self._releaseMbusObject()

            self.driverMbusHandle.close()
            self.driverMbusHandle = None