智能电表

.
|-- LICENSE
|-- README.md
|-- code
|   |-- business.py
|   |-- constant.py
|   |-- demo.py
|   |-- dev.json
|   |-- protocol.py
|   `-- qframe
|       |-- __init__.py
|       |-- builtins
|       |   |-- __init__.py
|       |   |-- clients.py
|       |   |-- network.py
|       |   `-- uart.py
|       |-- collections.py
|       |-- core.py
|       |-- datetime.py
|       |-- globals.py
|       |-- led.py
|       |-- logging.py
|       |-- ota.py
|       |-- qsocket.py
|       |-- serial.py
|       `-- threading.py
`-- docs
    `-- media
        |-- UML.png
        |-- init.png
        |-- system.png
        `-- ...

def create_app(name='DTU', config_path='/usr/dev.json'):
    # init application instance
    _app = Application(name)
    # read settings from json file
    _app.config.from_json(config_path)
    #The code is omitted here
    return _app

app = create_app()

class Application(object):
    """Application Class"""

    def __init__(self, name):
        self.name = name
        self.config = LocalStorage()
        self.business_threads_pool = ThreadPoolExecutor(max_workers=4, enable_priority=True)
        self.submit = self.business_threads_pool.submit
        # init builtins dictionary and init common, we use OrderedDict to keep loading ordering
        self.extensions = OrderedDict()
        self.__append_builtin_extensions()
        # set global context variable
        CurrentApp.set(self)
        G.set(_AppCtxGlobals())

    #The code is omitted here

    def append_extension(self, extension):
        self.extensions[extension.name] = extension

    def mainloop(self):
        """load builtins"""
        for extension in self.extensions.values():
            if hasattr(extension, 'load'):
                extension.load()

class AppExtensionABC(object):
    """Abstract Application Extension Class"""

    def __init__(self, name, app=None):
        self.name = name  # extension name
        if app:
            self.init_app(app)

    def init_app(self, app):
        # register into app, then, you can use `app.{extesion.name}` to get current extension instance
        app.append_extesion(self)
        raise NotImplementedError

    def load(self):
        # loading extension functions, this method will be called in `app.mainloop`
        raise NotImplementedError

app = Application(__name__)
ext = ExtensionClass(app)

ext = ExtensionClass()
ext.init_app(app)

    rfc1662resolver.init_app(_app)
    uart.init_app(_app)
    client.init_app(_app)

import checkNet
from usr.qframe import Application
from usr.business import rfc1662resolver, client, uart

PROJECT_NAME = "QuecPython_Framework_DEMO"
PROJECT_VERSION = "1.0.0"
def poweron_print_once():
    checknet = checkNet.CheckNetwork(
        PROJECT_NAME,
        PROJECT_VERSION,
    )
    checknet.poweron_print_once()
def create_app(name='DTU', config_path='/code/dev.json'):
  # initialize Application
  _app = Application(name)
  # read settings from json file
  _app.config.from_json(config_path)

  # init rfc1662resolver extension
  rfc1662resolver.init_app(_app)
  # init uart extension
  uart.init_app(_app)
  # init tcp client extension
  client.init_app(_app)

  return _app

# create app with `create_app` factory function
app = create_app()
if __name__ == '__main__':
  poweron_print_once()
  # loading all extensions
  app.mainloop()

# we have inited a RFC1662ProtocolResolver object in `business.py` module
# import `rfc1662resolver`
from code.business import rfc1662resolver


# decorate with protocol 0x2100
@rfc1662resolver.register(0x2100)
def handle2100(msg):
  """when get a 0x2100 message，this function will be called"""
  pass

class TcpClient(object):
    # ...
    def recv_callback(self, data):
        raise NotImplementedError('you must implement this method to handle data received by tcp.')

    def send(self, data):
        # TODO: uplink data method
        pass

class BusinessClient(TcpClient):

    def recv_callback(self, data):
        # recv tcp data and send to uart
        data = RFC1662Protocol.build_rfc_0x2100(data)
        CurrentApp().uart.write(data)

class Uart(object):
    # ...
    def recv_callback(self, data):
        raise NotImplementedError('you must implement this method to handle data received from device.')

    def write(self, data):
        # TODO: write data to uart
        pass

class UartBusiness(Uart):

        def recv_callback(self, data):
            # parse 1662 protocol data
            pass

# >>>>>>>>>> handle rfc1662 message received from uart <<<<<<<<<<
@rfc1662resolver.register(0x2100)
def handle2100(msg):
  """post data received to cloud"""
  # message body bytes
  data = msg.info().request_data()
  if data:
    # post data to tcp server by `client` extension register in Application
    CurrentApp().client.send(data)