AEP渐进均分性和数据压缩的关系

Posted 2023-04-05 Asandstar

渐进均分性AEP是和大数定律类似的概念，是弱大数定律的结果。而弱大数定律是指实验的重复次数不断增大时，某件事发生的频率越来越稳定，趋向于这件事发生的概率。

将有序集合划分成两个集合：①典型集，样本熵近似等于真实熵 ②非典型集，其他序列

关于典型序列的任何性质，都以很大概率成立，且将决定大样本的平均行为。

AEP的定义：

如果$X_1, X_2, …, X_n$是独立同分布（i.i.d.）序列，且服从$p(x)$，

则有$-\\frac1nlogp(X_1,X_2,…, X_n)→H(x)$也就是依概率收敛

个人理解：

因为典型序列的任何性质都大概率成立，所以我们可以在传递信息时，压缩发送典型序列所需要的信息，这样就起到了压缩信息的目的。

也就是出现多的、概率大的，用更少的信息表示，出现少的、概率小的，用更多的信息表示，这样可以在总体上减小需要发送的信息量。

 

参考资料：如何对AEP(渐近等分性)做一个很直白的解释？

BeyondSelf的回答 - 知乎 https://www.zhihu.com/question/351779388/answer/1633347140

归帆去棹残阳里的回答 - 知乎 https://www.zhihu.com/question/351779388/answer/1566973494

大漠孤盐的回答 - 知乎 https://www.zhihu.com/question/351779388/answer/870207447

ctwing电信AEP平台对接涂鸦智能云平台python程序源码

项目描述：
将无线倾角传感器、无线电流传感器数据对接到涂鸦云平台。
源代码：

 """
    www.daq-iot.com
    数采物联网版权所有，非授权禁止商业目的使用
 """
import base64
import hashlib
import hmac
import traceback
import ssl
import paho.mqtt.client as mqtt
import json
import os
import ssl
import time
import openpyxl
import pandas as pd
from flask import Flask
from flask import request


app = Flask(__name__)
def logintime():
    return int(time.time())
def datatime():
    return int(round(time.time() * 1000))
def getdeviceid(m):
    file_name = ".\\\\涂鸦云对接目录.xlsx"
    file_result = os.path.join(file_name)
    wb = openpyxl.load_workbook(file_result)  # 首先获取excel文件生成对象wb 使用openpyxl.load_workbook 方法
    lie = pd.DataFrame(pd.read_excel(file_result))
    sheet = wb.active
    for i in range(1, len(lie) + 2):
        cell = sheet.cell(row=i, column=1)
        cv = cell.value
        if cv ==  m:
            dev = sheet.cell(row=i, column=2)
            dv = dev.value
            return dv
def getdeviceselect(m):
    file_name = ".\\\\涂鸦云对接目录.xlsx"
    file_result = os.path.join(file_name)
    wb = openpyxl.load_workbook(file_result)  # 首先获取excel文件生成对象wb 使用openpyxl.load_workbook 方法
    lie = pd.DataFrame(pd.read_excel(file_result))
    sheet = wb.active
    for i in range(1, len(lie) + 2):
        cell = sheet.cell(row=i, column=1)
        cv = cell.value
        if cv == m:
            dev = sheet.cell(row=i, column=3)
            dv = dev.value
            return dv
def getdevtype(m):
    file_name = ".\\\\涂鸦云对接目录.xlsx"
    file_result = os.path.join(file_name)
    wb = openpyxl.load_workbook(file_result)  # 首先获取excel文件生成对象wb 使用openpyxl.load_workbook 方法
    lie = pd.DataFrame(pd.read_excel(file_result))
    sheet = wb.active
    for i in range(1, len(lie) + 2):
        cell = sheet.cell(row=i, column=1)
        cv = cell.value
        if cv == m:
            dev = sheet.cell(row=i, column=4)
            dv = dev.value
            return dv

def hmac_sha256(key, value):
    """
    hmacsha256加密
    :param key:
    :param value: 加密字符串
    :return: 加密结果转换为16进制字符串，并大写
    """
    message = value.encode("utf-8")
    return hmac.new(key.encode("utf-8"), message, digestmod=hashlib.sha256).hexdigest()

@app.route("/he/data/push/global", methods=['POST'])
def he_data_push_global():
    try:
        print('**** Receive Post Data *****')

        data = request.get_data()

        data= data.decode("utf-8")
        data=data.replace("'",'"')
        json_data = json.loads(data)
        print(json_data)
        IMEI=json_data["IMEI"]
        deviceid = getdeviceid(IMEI)
        deviceselect = getdeviceselect(IMEI)
        if getdevtype(IMEI)=='倾斜':
            time1=logintime()
            time2=datatime()
            # print(json_data)
            # data=data[2:-1]
            data=json_data["payload"]
            # print(data)
            data = str(data).replace("'", '"')
            data=json.loads(data)
            # print(data)
            data=data["APPdata"]
            # print(str(data))

            debs64=base64.b64decode(data)
            debs64=str(debs64)[11:-1]
            debs64=debs64.replace("",'"')
            debs64 = debs64.replace(",", '","')
            debs64 = debs64.replace(":", '":"')
            debs64 = debs64.replace("", '"')
            debs64=json.loads(debs64)


            jiaodu=int(debs64["angle"])/10
            dianliang=debs64["BT"]



            realdata=
            "msgId":"45lkj3551234001",
            "time":time2,
            "data":
                "jiaodu":
                    "value":int(jiaodu*10),
                    "time": time2
                ,
                "dianliang":
                      "value":int(float(dianliang)*10),
                      'time': time2
                
            
            
        elif getdevtype(IMEI) == '电流':
             time1 = logintime()
             time2 = datatime()
             payload = json_data['payload']['serviceData']
             dianliu=payload["IntData2"]
             dianlaing=payload["VoltageCur"]
             realdata = 
                "msgId": "45lkj3551234001",
                "time": time2,
                "data": 
                    "dianliu": 
                        "value": dianliu,
                        "time": time2
                    ,
                    "dianliang": 
                        "value": dianlaing,
                        'time': time2
                    
                
            

        print(realdata)
        TASK_TOPIC = 'tylink/%s/thing/property/report'%deviceid  # 客户端发布消息主题
        # print(TASK_TOPIC)

        client_id = "tuyalink_%s"%deviceid
        username="%s|signMethod=hmacSha256,timestamp=%s,secureMode=1,accessType=1"%(deviceid,time1)
        value="deviceId=,timestamp=,secureMode=1,accessType=1".format(deviceid,time1)
        password=hmac_sha256(deviceselect,value)
        # print(value)
        # print(client_id,username,password)
        """
        client_id是连接到代理。如果client_id的长度为零或为零，则行为为由使用的协议版本定义。如果使用MQTT v3.1.1，
        那么一个零长度的客户机id将被发送到代理，代理将被发送为客户端生成一个随机变量。如果使用MQTT v3.1，那么id将是
        随机生成的。在这两种情况下，clean_session都必须为True。如果这在这种情况下不会产生ValueError。
        注意：一般情况下如果客户端服务端启用两个监听那么客户端client_id 不能与服务器相同，如这里用时间"20190222142358"作为它的id，
        如果与服务器id相同，则无法接收到消息
        """
        client = mqtt.Client(client_id, transport='tcp')
        client.username_pw_set(username, password)
        client.tls_set(ca_certs=None, certfile=None, keyfile=None, cert_reqs=ssl.CERT_REQUIRED,
               tls_version=ssl.PROTOCOL_TLSv1_2, ciphers=None)
        client.connect("m1.tuyacn.com", 8883, 60)  # 此处端口默认为1883，通信端口期keepalive默认60


        # client.username_pw_set("", "")
        # client.connect("127.0.0.1", 1883, 5)
        client.loop_start()

        """
        客户端发布消息
        :param message: 消息主体
        :return:
        """
        # publish(主题：Topic; 消息内容)
        client.publish(TASK_TOPIC, json.dumps(realdata, ensure_ascii=False))
        print(realdata)
        print("Successful send message!")

        return 'POST SUCCESS'
    except:
        print(traceback.format_exc())
        return repr(Exception)
        pass

if __name__ == '__main__':
    app.run(host='0.0.0.0', debug=True, port=8002)