driver based

Posted 2021-12-04 mftang2018

#!/usr/bin/env python3
import json
import binascii
import asn1tools
import paho.mqtt.client as mqtt
import requests
import logging
import sys,os
import time
import datetime
import random
from enum import Enum
from queue import Queue

#add path for import module
filePath = os.getcwd()
sys.path.append(filePath)

from Com_Utilities.Handler_asn1 import *

class BaseClient:
    APPLICATION_OCTET_STREAM = 42
    TOPIC_PREFIX = ‘olc‘

    def __init__(self, log, parser, trigger, mqclient, timeout, identity):
        self.parser = parser
        self.trigger = trigger
        self.mqclient = mqclient
        self.timeout = timeout
        self.identity = identity
        self.log = log
        identity = binascii.hexlify(identity.encode(encoding=‘utf-8‘))
        identity = identity.decode()
        self.topic_request = ‘/%s/%s/request‘ % (self.TOPIC_PREFIX, identity)
        self.topic_reqponse = ‘/%s/%s/response‘ % (self.TOPIC_PREFIX, identity)
        self.mqclient.subscribe(self.topic_request)

    def random_data(self, len):
        tmp = []
        for i in range(len):
            tmp.append(random.randint(0, 255))
        return bytes(tmp)

    def get_seconds_20000101(self, dt=None):
        if dt is None:
            dt = datetime.datetime.now()
        s = (dt - datetime.datetime(2000, 1, 1, 0, 0, 0)).total_seconds()
        return int(s)

    def send_resp(self, code, data=None, block2=None):
        resp = ResponseBean()
        resp.code = code
        if data is not None:
            resp.contentFormat = self.APPLICATION_OCTET_STREAM
            data = binascii.b2a_base64(data)
            data = str(data, ‘utf-8‘)
            data = data.replace(‘\n‘, ‘‘).replace(‘\r‘, ‘‘)
            resp.payload = data
        if block2 is not None:
            resp.block2 = block2
        self.log.info(‘send response: %s‘ % resp)
        blob = self.parser.encode_response(resp)
        self.mqclient.publish(self.topic_reqponse, blob)

    def send_resp_bean(self, code, name, bean):
        data = self.parser.encode_response_payload(name, bean)
        return self.send_resp(code, data)

    def send_resp_block2(self, code, data, m, szx, num):
        bob = BlockOptionBean()
        bob.m = m
        bob.szx = szx
        bob.num = num
        return self.send_resp(code, data, bob)

    def send_resp_sync(self, resource, sr, seconds=None):
        if seconds is None:
            seconds = self.get_seconds_20000101()
        resp = {‘resourcePaths‘: resource, ‘revision‘: sr, ‘currentUtcTime‘: {‘secondsSince20000101‘: seconds}}
        self.send_resp_bean(code=ResponseCode.CONTENT, name=‘SyncResponse2‘, bean=resp)

    def receive_req(self, timeout=None):
        if timeout is None:
            timeout = self.timeout
        msg = self.mqclient.receive_msg(timeout)
        req = self.parser.decode_request(msg)
        self.log.info(‘receive request: %s‘ % req)
        return req

    def do_resource_sync(self, resource, sr=None):
        self.trigger.trigger()
        while True:
            req = self.receive_req()
            if req.url.endswith(‘/s/2‘) and req.code == ‘POST‘:
                if sr is None:
                    asnobj = self.parser.decode_request_payload(req)
                    sr = int(asnobj[‘revision‘]) + 1
                self.send_resp_sync(resource, sr)
                return sr
            elif req.url.endswith(‘/s/2‘) and req.code == ‘GET‘ and sr is not None:
                self.send_resp_sync(resource, sr)
                return sr

    def do_empty_sync(self, sr=None):
        return self.do_resource_sync(resource=[], sr=sr)

    def test_case(self):
        pass

BaseClient

#!/usr/bin/env python3
import json
import binascii
import asn1tools
import paho.mqtt.client as mqtt
import requests
import logging
import serial
import sys,os
import threading
import time
import datetime
import random
from enum import Enum
from queue import Queue

#add path for import module
filePath = os.getcwd()
sys.path.append(filePath)

from Proxy_DUT.Coap_Proxy import *

class CaseSuites:

    def __init__(self, log, parser, mqclient, trigger, timeout, identity):
        self.log = log
        self.parser = parser
        self.mqclient = mqclient
        self.trigger = trigger
        self.timeout = timeout
        self.identity = identity

        self.basis = BaseClient(log = self.log,
                                parser = self.parser,
                                trigger=self.trigger,
                                mqclient=self.mqclient,
                                timeout=self.timeout,
                                identity=self.identity)

        self.ov_client = PostSyncClient(log = self.log,
                                        parser = self.parser,
                                        trigger=self.trigger,
                                        mqclient=self.mqclient,
                                        timeout=self.timeout,
                                        identity=self.identity,
                                        res_paths=[],
                                        res_responses={})

        self.ov_revision = random.randint(1, 3)
        self.unhappysync = UnhappySyncClient(log = self.log,
                                             parser = self.parser,
                                             trigger=self.trigger,
                                             mqclient=self.mqclient,
                                             timeout=self.timeout,
                                             identity=self.identity,
                                             error_code=None)

        self.emptysync = NilSyncClient( log = self.log,
                                        parser = self.parser,
                                        trigger=self.trigger,
                                        mqclient=self.mqclient,
                                        timeout=self.timeout,
                                        identity=self.identity)

        self.wrongsync = WrongSyncClient(log = self.log,
                                         parser = self.parser,
                                         trigger=self.trigger,
                                         mqclient=self.mqclient,
                                         timeout=self.timeout,
                                         identity=self.identity)

        self.messysync = MessySyncClient(log = self.log,
                                         parser = self.parser,
                                         trigger=self.trigger,
                                         mqclient=self.mqclient,
                                         timeout=self.timeout,
                                         identity=self.identity)

        self.extremitysync = ExtremitySyncClient(log = self.log,
                                                 parser = self.parser,
                                                 trigger=self.trigger,
                                                 mqclient=self.mqclient,
                                                 timeout=self.timeout,
                                                 identity=self.identity,
                                                 bean=None)

    def base_func(self):
        return self.basis

    def override(self, dim):
        self.ov_client.res_paths = [‘/l/ov/%d‘ % self.ov_revision]
        self.ov_client.res_responses = {‘/l/ov/%d/0‘ % self.ov_revision: (‘Override‘, {‘nominalLightLevel‘: dim, ‘enabled‘: 1})}
        self.ov_revision += 1
        return self.ov_client

    def ota(self):
        return OtaClient(log = self.log,
                         parser = self.parser,
                         trigger=self.trigger,
                         mqclient=self.mqclient,
                         timeout=self.timeout,
                         identity=self.identity)

    def unhappy_sync(self, error_code):
        self.unhappysync.error_code = error_code
        return self.unhappysync

    def empty_sync(self):
        return self.emptysync

    def wrong_sync(self):
        return self.wrongsync

    def messy_sync(self):
        return self.messysync

    def extremity_sync(self, bean):
        self.extremitysync.bean = bean
        return self.extremitysync

CaseSuites

#!/usr/bin/env python3
import json
import binascii
import asn1tools
import paho.mqtt.client as mqtt
import requests
import logging
import sys,os
import time
import datetime
import random
from enum import Enum
from queue import Queue

#add path for import module
filePath = os.getcwd()
sys.path.append(filePath)

from Com_Utilities.Handler_asn1 import *
from Proxy_DUT.Coap_Base import *

class PostSyncClient(BaseClient):

    def __init__(self, log, parser, trigger, mqclient, timeout, identity, res_paths, res_responses):
        super().__init__(log, parser,trigger, mqclient, timeout, identity)
        self.res_paths = res_paths
        self.res_responses = res_responses
        self.res_count = None
        self.log = log

    def test_case(self):
        time.sleep(2)
        sr = 0
        state = 100
        self.res_count = 0
        while True:
            if state == 100:
                self.trigger.syncTrigger_SYNC()
                self.log.info(‘trigger: syncTrigger_SYNC‘)
                state = 200

            elif state == 200:
                req = self.receive_req()
                if req.url.endswith(‘/s/2‘) and req.code == ‘POST‘:
                    asnobj = self.parser.decode_request_payload(req)
                    sr = int(asnobj[‘revision‘]) + 1
                    self.send_resp_sync(self.res_paths, sr)
                    state = 300
                else:
                    self.log.error(‘unknown resource request %s‘ % req.url)

            elif state == 300:
                req = self.receive_req()
                if req.url.endswith(‘/s/2‘) and req.code == ‘POST‘:
                    self.parser.decode_request_payload(req)
                    self.send_resp_sync([], sr)
                    state = 400
                else:
                    for p in self.res_responses:
                        if req.url.endswith(p):
                            name = self.res_responses[p][0]
                            resp = self.res_responses[p][1]
                            self.send_resp_bean(code=ResponseCode.CONTENT, name=name, bean=resp)
                            self.res_count += 1
                        else:
                            self.log.error(‘unknown resource request %s‘ % req.url)
            elif state == 400:
                # consume report log messages.
                try:
                    msg = self.receive_msg()
                except:
                    self.log.debug(‘finished‘)
                    break
                req = self.parse_req_msg(msg)
                self.log.info(‘receive report bean: %s‘ % req)
                self.send_resp(code=ResponseCode.CONTENT)


class OtaClient(BaseClient):

    def __init__(self,log,parser,trigger, mqclient, timeout, identity, filename):
        super().__init__(log, parser,trigger, mqclient, timeout, identity)
        f = open(filename, ‘rb‘)
        self.binfile = f.read()
        f.close()
        self.rr = binascii.hexlify(self.random_data(20))

    def test_case(self):
        time.sleep(2)
        state = 100
        sr = None
        while True:
            if state == 100:
                sr = self.do_resource_sync([‘/p/fu/%s‘ % self.rr])
                state = 300
            elif state == 300:
                req = self.receive_req()
                if req.url.endswith(‘/p/fu/%s/0‘ % self.rr) and req.code == ‘GET‘:
                    if req.block2 is None:
                        self.log.error(‘block2 is not found‘)
                        self.send_resp(ResponseCode.REQUEST_ENTITY_INCOMPLETE)
                    else:
                        sz = 1 << (req.block2.szx + 4)
                        a = sz * req.block2.num
                        b = a + sz
                        last_num = (len(self.binfile) + sz - 1) / sz
                        if a >= len(self.binfile):
                            self.log.error(‘the data range [%d:%d] is out of file length. ‘ % (a, b))
                            self.send_resp(ResponseCode.REQUEST_ENTITY_TOO_LARGE)
                        else:
                            if b >= self.binfile[a:b]:
                                data = self.binfile[a:]
                            else:
                                data = self.binfile[a:b]
                            self.send_resp_block2(ResponseCode.CONTENT, data=data, m=req.block2.num < last_num, szx=req.block2.szx,
                                                  num=req.block2.num)
                elif req.url.endswith(‘/s/2‘) and req.code == ‘POST‘:
                    self.parser.decode_request_payload(req)
                    self.send_resp_sync([], sr)
                elif req.url.endswith(‘/s/2‘) and req.code == ‘GET‘:
                    self.send_resp_sync([‘/p/fu/%s‘ % self.rr], sr)
                elif req.url.endswith(‘/p/fs/0‘) and req.code == ‘POST‘:
                    self.parser.decode_request_payload(req)
                    break


class UnhappySyncClient(BaseClient):

    def __init__(self, log, parser,trigger, mqclient, timeout, identity, error_code):
        super().__init__(log, parser,trigger, mqclient, timeout, identity)
        self.error_code = error_code

    def test_case(self):
        time.sleep(2)
        self.trigger.syncTrigger_SYNC()
        while True:
            req = self.receive_req()
            if req.url.endswith(‘/s/2‘) and req.code == ‘POST‘:
                self.parser.decode_request_payload(req)
                self.send_resp(code=self.error_code)
                time.sleep(60)
                break


class NilSyncClient(BaseClient):

    def __init__(self,log, parser, trigger, mqclient, timeout, identity):
        super().__init__(log,parser,trigger, mqclient, timeout, identity)

    def test_case(self):
        time.sleep(2)
        self.trigger.syncTrigger_SYNC()
        while True:
            req = self.receive_req()
            if req.url.endswith(‘/s/2‘) and req.code == ‘POST‘:
                self.parser.decode_request_payload(req)
                self.send_resp(code=ResponseCode.CONTENT)
                time.sleep(60)
                break


class WrongSyncClient(BaseClient):

    def __init__(self,log,parser, trigger, mqclient, timeout, identity):
        super().__init__(log,parser,trigger, mqclient, timeout, identity)

    def test_case(self):
        time.sleep(2)
        self.trigger.syncTrigger_SYNC()
        while True:
            req = self.receive_req()
            if req.url.endswith(‘/s/2‘) and req.code == ‘POST‘:
                self.parser.decode_request_payload(req)
                self.send_resp_bean(code=ResponseCode.CONTENT, name=‘Override‘, bean={‘nominalLightLevel‘: 10000, ‘enabled‘: 1})
                time.sleep(60)
                break


class MessySyncClient(BaseClient):

    def __init__(self,log,parser, trigger, mqclient, timeout, identity):
        super().__init__(log, parser, trigger, mqclient, timeout, identity)

    def test_case(self):
        time.sleep(2)
        self.trigger.syncTrigger_SYNC()
        while True:
            req = self.receive_req()
            if req.url.endswith(‘/s/2‘) and req.code == ‘POST‘:
                self.parser.decode_request_payload(req)
                data = self.random_data(20)
                self.send_resp(code=ResponseCode.CONTENT, data=data)
                time.sleep(60)
                break


class ExtremitySyncClient(BaseClient):

    def __init__(self,log, parser, trigger, mqclient, timeout, identity, bean):
        super().__init__(log, parser, trigger, mqclient, timeout, identity)
        self.bean = bean
        self.res_cnt = None
        self.res_abn_cnt = None
        self.log = log

    def test_case(self):
        time.sleep(2)
        state = 100
        self.res_cnt = 0
        self.res_abn_cnt = 0
        while True:
            if state == 100:
                self.trigger.syncTrigger_SYNC()
                state = 200
            elif state == 200:
                req = self.receive_req()
                if req.url.endswith(‘/s/2‘) and req.code == ‘POST‘:
                    self.parser.decode_request_payload(req)
                    self.send_resp_bean(code=ResponseCode.CONTENT, name=‘SyncResponse2‘, bean=self.bean)
                    if len(self.bean[‘resourcePaths‘]) == 0:
                        state = 300
                    else:
                        state = 250
            elif state == 250:
                req = self.receive_req()
                if req.url.endswith(‘/s/2‘) and req.code == ‘POST‘:
                    asnobj = self.parser.decode_request_payload(req)
                    sr = int(asnobj[‘revision‘]) + 1
                    self.send_resp_sync([], sr)
                    state = 300
                    time.sleep(60)
                else:
                    for r in self.bean[‘resourcePaths‘]:
                        if req.url.startwith(r):
                            self.send_resp(ResponseCode.NOT_FOUND)
                            self.res_cnt += 1
                        else:
                            self.log.error(‘unkown resource path %s‘ % req.url)
                            self.res_abn_cnt += 1
            elif state == 300:
                time.sleep(60)
                break

Proxy

#coding:utf-8
#!/usr/bin/python3
import json
import os
import re
import binascii
import asn1tools
import sys
import logging
import datetime
import random
import time
import threading
from enum import Enum
from queue import Queue

#add path for import module
filePath = os.getcwd()
sys.path.append(filePath)

from Com_Utilities.Logger import loggerInfor as log
from Com_Utilities.ResourceHandler import resourceHandler as Resource
from Com_Utilities.MQTT_Client import MQTTClient as mqttClient
from Com_Utilities.Handler_asn1 import ParserTool as parserTool
from Resource_Managers.Dali_Manager import DaliSniff as Dali
from Resource_Managers.Fake_Sync_Trigger import CFake_SyncTrigger
from Proxy_DUT.Coap_olc import CaseSuites as caseSuites

__all__ = ["WrapperProductTestDevice"]

class WrapperProductTestDevice():
      def __init__(self):
            self.log_Inial()
            #for fake trigger
            ctcMessage_path = os.path.dirname(os.path.realpath(__file__))
            scheme_file = ctcMessage_path+‘\\ctc-message-schema.asn1‘

            self.log = log()
            self.log.info(‘asn file path: %s‘ % scheme_file)
            self.parserTool = parserTool(scheme_file)
         
            self.resource = Resource()
            self.trigger = None
            self.mqclient = None
            self.casesuite = None
            self.dali = None
            self.daliAlive = True
  
      def Wrapper_Setup(self):
            self.log.info(‘Wrapper Setup‘)
            self.trigger = CFake_SyncTrigger("CT_Serial", "Serial")
            # server port message: [0]-port otr ip, [1]-baud or port

            serverPort = self.resource.get_SerialOrIP("CT_Server","CT_Server")
            ctcid = self.resource.get_DeviceID("olc","CT_CTC1");
            #self.dali = Dali() 
            #MQTT Client
            self.mqclient = mqttClient(host=serverPort[0], 
                                       port=int(serverPort[1]), 
                                       qos=2, 
                                       timeout=45, 
                                       log = self.log)
            self.mqclient.loop_start()

            #case Suites
            self.casesuite = caseSuites(self.log,
                                        self.parserTool,
                                        self.mqclient, 
                                        self.trigger, 
                                        45, 
                                        ctcid)

      def Wrapper_TearDown(self):
            os.system("taskkill /F /IM MutiOne.exe")
            self.daliAlive = False
            self.casesuite = None
            self.log.info(‘Wrapper TearDown‘)
            self.trigger.syncTrigger_Stop()
            self.mqclient.loop_stop()
            
      def log_Inial(self):
            now_time = datetime.datetime.now()
            filePath = os.getcwd() + "\\log\\" 
            if os.path.exists(filePath) is False:
                os.makedirs(filePath)

            logFile = filePath+‘test-%s.log‘ % datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
            logging.basicConfig(filename = logFile,
                                # stream=sys.stdout,
                                level=logging.INFO,
                                format=‘%(asctime)s - %(levelname)5s - %(lineno)4s - %(filename)s - %(message)s‘)

WrapperProductTestDevice

#coding:utf-8
#!/usr/bin/python3
import json
import os
import re
import binascii
import asn1tools
import sys
import logging
import datetime
import random
import time
import threading
from enum import Enum
from queue import Queue
#add path for import module
filePath = os.getcwd()
sys.path.append(filePath)

from Resource_Managers.Dali_Manager import DaliSniff as Dali
from Wrapper_Devices.Wrapper_Based import *

__all__ = ["Wrapper_ManualOverride"]

class Wrapper_ManualOverride(WrapperProductTestDevice):
    def __init__(self):
        WrapperProductTestDevice.__init__(self)

    def test(self, num, dimming,func_case):
        succ = 0
        fail = 0           
        d1 = datetime.datetime.now()
           
        for i in range(1, num + 1):
            try:
                self.log.info(‘====== run test count:  %d ======‘ % i)
                self.dali = Dali() 
                func_case(self.casesuite, i, num, dimming)
                succ += 1
                self.log.info(‘****** [%4d/%-4d]  OK ******‘ % (succ, i))

                time.sleep(15)
                self.dali.Dali_Stop()
                self.log.info(‘dimming: %d‘ % dimming)
                dimmingReport = self.dali.Dali_Value()

                if dimmingReport == 256:
                    Warning("Read dali value error by MultiOne!")
                    self.log.info("Read dali value error by MultiOne!")
                else:
                    self.log.info(‘dali multiOne: %d‘ % dimmingReport)
                    assert dimmingReport == dimming

            except Exception as e:
                fail += 1
                self.log.error(‘****** [%4d/%-4d] NOK ******: %s‘ % (fail, i, repr(e)))
                time.sleep(10)
        assert fail == 0

        d2 = datetime.datetime.now()
        self.log.info(‘start time: %s‘ % d1)

        self.log.info(‘end time: %s‘ % d2)
        self.log.info(‘successful cases: %d‘ % succ)
        self.log.info(‘failed cases: %d‘ % fail)

    def ManualOverid_interface(self, casesuite, sn, num, dim):
        count = 0
        while count < 3:
            caseobj = casesuite.override(dim*10)
            caseobj.test_case()
            if caseobj.res_count != 1:
                count += 1
            else:
                return

        raise Exception("ctc doesn‘t fetch ov resource.")


#only for test
if __name__ == ‘__main__‘:

    now_time = datetime.datetime.now()
    filePath = os.getcwd() + "\\log\\" 
    if os.path.exists(filePath) is False:
        os.makedirs(filePath)

    logFile = filePath+‘test-%s.log‘ % datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    logging.basicConfig(filename = logFile,
                        # stream=sys.stdout,
                        level=logging.INFO,
                        format=‘%(asctime)s - %(levelname)5s - %(lineno)4s - %(filename)s - %(message)s‘)

    test = Wrapper_ManualOverride()
    test.Wrapper_Setup()

    count = 0
    while count < 10:
        test.test(1, 10, test.ManualOverid_interface)
        time.sleep(10)
        test.test(1, 30, test.ManualOverid_interface)
        count += 1

    test.Wrapper_TearDown()

Wrapper_ManualOverride

#coding:utf-8
#!/usr/bin/python3
import sys,os
import re
import logging

#add path for import module
filePath = os.getcwd()
sys.path.append(filePath)
from Com_Utilities import FlexConfig as flexConfig

__all__ = ["resourceHandler"]

class resourceHandler(object):
    def __init__(self):
        logging.basicConfig(stream=sys.stdout, level=logging.DEBUG, format=‘%(asctime)s - %(levelname)5s - %(lineno)4s - %(filename)s - %(message)s‘)
        self.resourceFile = os.getcwd()+"\\resource_Managers\\Resources.xml"

    def get_SerialOrIP(self,userName, userID):
        self.portconfig = flexConfig.ResourcesConfig(self.resourceFile, userName,userID)        
        #serial port message: [0]-port, [1]-baud
        string = self.portconfig.getSubElement_Port()
        serialportMsg = re.split(‘,‘, string)
        return serialportMsg

    def get_DeviceID(self,userName, userID):
        self.portconfig = flexConfig.ResourcesConfig(self.resourceFile, userName,userID)
        string = self.portconfig.getSubElement_Device()
        return string

resourceHandler

#coding:utf-8
#!/usr/bin/python3
import json
import os
import binascii
import asn1tools
import sys
import paho.mqtt.client as mqtt
import requests
import logging
from enum import Enum
from queue import Queue

__all__ = ["MQTTClient"]

class MQTTClient:

    def __init__(self,host, port, qos, timeout, log):
        self.host = host
        self.port = port
        self.qos = qos
        self.queue = Queue()
        self.mqtt_client = None
        self.timeout = timeout
        self.log = log

    def on_message(self, client, userdata, msg):
        self.log.debug(‘get a message: %s‘ % msg)
        self.queue.put(msg)

    def subscribe(self, topic):
        self.mqtt_client.subscribe(topic, self.qos)
        self.log.debug(‘subscribe to %s‘ % topic)

    def unsubscribe(self, topic):
        self.mqtt_client.unsubscribe(topic)
        self.log.debug(‘unsubscribe %s‘ % topic)

    def receive_msg(self, timeout=None):
        self.log.debug(‘waiting for message.‘)
        if timeout is None:
            timeout = self.timeout
        return self.queue.get(timeout=timeout)

    def publish(self, topic, blob):
        self.mqtt_client.publish(topic, blob)

    def loop_start(self):
        if self.mqtt_client is None:
            self.mqtt_client = mqtt.Client()
            self.mqtt_client.on_message = self.on_message
            self.mqtt_client.connect(self.host, self.port, self.timeout)
            self.mqtt_client.loop_start()

    def loop_stop(self):
        if self.mqtt_client is not None:
            self.mqtt_client.loop_stop()
            self.mqtt_client.disconnect()
            self.mqtt_client = None

MQTTClient

#coding:utf-8
#!/usr/bin/python3
import json
import os
import logging
import asn1tools
import binascii
from enum import Enum

__all__ = ["ParserTool", "ResponseCode", "RequestBean", "ResponseBean", "BlockOptionBean"]

class ResponseCode(Enum):
    CREATED = 2 << 5 | 1
    DELETED = 2 << 5 | 2
    VALID = 2 << 5 | 3
    CHANGED = 2 << 5 | 4
    CONTENT = 2 << 5 | 5
    CONTINUE = 2 << 5 | 31

    BAD_REQUEST = 4 << 5 | 0
    UNAUTHORIZED = 4 << 5 | 1
    BAD_OPTION = 4 << 5 | 2
    FORBIDDEN = 4 << 5 | 3
    NOT_FOUND = 4 << 5 | 4
    METHOD_NOT_ALLOWED = 4 << 5 | 5
    NOT_ACCEPTABLE = 4 << 5 | 6
    REQUEST_ENTITY_INCOMPLETE = 4 << 5 | 8
    CONFLICT = 4 << 5 | 9
    PRECONDITION_FAILED = 4 << 5 | 12
    REQUEST_ENTITY_TOO_LARGE = 4 << 5 | 13
    UNSUPPORTED_CONTENT_FORMAT = 4 << 5 | 15
    UNPROCESSABLE_ENTITY = 4 << 5 | 22

    INTERNAL_SERVER_ERROR = 5 << 5 | 0
    NOT_IMPLEMENTED = 5 << 5 | 1
    BAD_GATEWAY = 5 << 5 | 2
    SERVICE_UNAVAILABLE = 5 << 5 | 3
    GATEWAY_TIMEOUT = 5 << 5 | 4
    PROXY_NOT_SUPPORTED = 5 << 5 | 5


class CoapMessage:

    def load(self, dict_obj):
        for p in self.__dict__:
            if p == ‘block2‘ and dict_obj[p] is not None:
                bob = BlockOptionBean()
                bob.load(dict_obj[p])
                self.__dict__[p] = bob
            else:
                self.__dict__[p] = dict_obj[p]

    def __str__(self):
        attrs = ", ".join("{}={}".format(k, getattr(self, k)) for k in self.__dict__.keys())
        return "[{}: {}]".format(self.__class__.__name__, attrs)


class RequestBean(CoapMessage):

    def __init__(self):
        self.type = None
        self.code = None
        self.url = None
        self.contentFormat = None
        self.payload = None
        self.block2 = None


class ResponseBean(CoapMessage):

    def __init__(self):
        self.code = None
        self.contentFormat = None
        self.payload = None
        self.block2 = None


class BlockOptionBean(CoapMessage):

    def __init__(self):
        self.szx = None
        self.m = None
        self.num = None

class ParserTool:

    def __init__(self, scheme_file):
        self.scheme_file = scheme_file,
        self.codec = ‘uper‘
        self.asn1compiler = asn1tools.compile_files(self.scheme_file, self.codec)
        self.obj_names = {‘POST /s/2‘: (‘SyncRequest‘, ‘SyncResponse2‘)}

    def serialize(self, obj):
        if isinstance(obj, ResponseCode):
            return obj.name
        else:
            return obj.__dict__

    def asn1_encode(self, name, bean):
        return self.asn1compiler.encode(name, bean, check_constraints=True)

    def asn1_decode(self, name, blob):
        return self.asn1compiler.decode(name, blob, check_constraints=True)

    def get_resource_path(self, request_bean):
        url = request_bean.url
        if url.startswith(‘coaps‘):
            url = url[8:]
        i = url.index(‘/‘)
        return url[i:]

    def get_obj_path(self, resource_path):
        tmp = resource_path.split(‘/‘)
        if len(tmp) == 4:
            return ‘/%s/%s/_/%s‘ % (tmp[1], tmp[2], tmp[3])
        return resource_path

    def get_obj_name(self, code, path):
        key = ‘%s %s‘ % (code, path)
        return self.obj_names[key]

    def decode_request_payload(self, request_bean):
        if request_bean.payload is not None and len(request_bean.payload) > 0:
            path = self.get_resource_path(request_bean)
            path = self.get_obj_path(path)
            names = self.get_obj_name(request_bean.code, path)
            asnobj = self.asn1_decode(names[0], binascii.a2b_base64(request_bean.payload))
            logging.info(‘request asn1: %s‘ % asnobj)
            return asnobj

    def decode_request(self, mqtt_message):
        obj = json.loads(mqtt_message.payload.decode(‘utf-8‘))
        req = RequestBean()
        req.load(obj)
        return req

    def encode_response_payload(self, name, bean):
        logging.info(‘response asn1: %s‘ % bean)
        s = self.asn1_encode(name, bean)
        return s

    def encode_response(self, response_bean):
        data = json.dumps(response_bean, ensure_ascii=True, default=self.serialize)
        return bytes(data, ‘utf-8‘)

"ParserTool", "ResponseCode", "RequestBean", "ResponseBean", "BlockOptionBean"