BL0939 是上海贝岭股份有限公司开发的一款用于智能家居领域进行电能测量的专用芯片，支持两路测量，可同时进行计量和漏电故障检测，漏电检测电流可设，响应时间快，具有体积小，外围电路简单，成本低廉的优点。
链接: 上海贝岭官网文档-BL0939 .
链接: 文字转语音.
本案例中涉及到音频播报，所以需要若干.mp3文件。音频的内容，用户可以根据需要，自行定义。
本案例使用BL0939采集电力数据（2路数据，测试代码中只使用了其中1路数据），使用光电开关检测用电器件是否接入，若用电器件接入则打开继电器，给用电器件供电。由于使用的是M601核心板，所以需要自行打板测试。测试代码仅供参考，其中核心部分是”app_elec.py“中的电力采集、数据转换。
（1）新建产品
（2）打开动态注册、设置自定义功能
添加自定义功能
（3）发布物模型
（4）添加设备
（5）获取productKey和productSecret
进入产品页面，获取动态注册所需要的productKey和productsecret
import utime as time
import app_audio
import app_relay
import app_charge_port
import app_led
import app_elec
import app_aliyunserver
# 1端口的当前和过去状态
cur_port1_state=0
pre_port1_state=0
#电压，电流，功率，电能，温度
elecdata_port = [[0,0,0,0,0],[0,0,0,0,0]]
#每个端口的最大充电功率
max_power = [0,0]
def key_detect():
global cur_port1_state
global pre_port1_state
#获取端口1当前状态
cur_port1_state=app_charge_port.get_port_state(1)
if cur_port1_state!=pre_port1_state:
if cur_port1_state==1:
app_relay.relay_ctl(1,1)
app_audio.audio_play('in1')
else:
app_relay.relay_ctl(1,0)
app_audio.audio_play('out1')
pre_port1_state = cur_port1_state
# 初始化检测端口
def init_detect_port():
global cur_port1_state
global pre_port1_state
cur_port1_state = app_charge_port.get_port_state(1)
pre_port1_state = cur_port1_state
#获取电力数据，有两路数据
def get_elc_data():
global max_power
global elecdata_port
ret = 1
i=0
if i == 0:
ret,tempelecdata = app_elec.ask_uart2_data(i) #电压，电流1，电流2，功率1，功率2，电能1，电能2，温度
if ret == 0:
elecdata_port[2*i][0] = tempelecdata[0]
elecdata_port[2*i+1][0] = tempelecdata[0] #电压
elecdata_port[2*i][1] = tempelecdata[1]
elecdata_port[2*i+1][1] = tempelecdata[2] #电流
elecdata_port[2*i][2] = tempelecdata[3]
elecdata_port[2*i+1][2] = tempelecdata[4] #功率
elecdata_port[2*i][3] = tempelecdata[5]
elecdata_port[2*i+1][3] = tempelecdata[6]
elecdata_port[2*i][4] = tempelecdata[7]
elecdata_port[2*i+1][4] = tempelecdata[7] #温度
if elecdata_port[2*i][2] > max_power[2*i]:
max_power[2*i] = int(elecdata_port[2*i][2]) #单口最大充电功率
else:
pass
if elecdata_port[2*i+1][2] > max_power[2*i+1]:
max_power[2*i+1] = int(elecdata_port[2*i+1][2]) #单口最大充电功率
else:
pass
else:
print('------------elec data read failed i={}'.format(i))
else:
return 1
return 0
def main():
# 灯初始化
app_led.led_settings(0)
#开机关闭继电器
app_relay.relay_ctl(1,0)
#开机播放开机语音
app_audio.audio_init()
app_audio.audio_play('sbkj')
#清除电能计数器
app_elec.clearcft(0)
# 开机检测端口状态
init_detect_port()
#上云
app_aliyunserver.mqtt_main()
while True:
time.sleep_ms(500)
key_detect()
get_elc_data()
print("elecdata_port:",elecdata_port)
print("max_power:",max_power)
# 上报数据：电压，电流，功率，电能，温度
app_aliyunserver.upload_elc_data(elecdata_port)
#上报继电器状态
relay1_state=app_relay.get_relay_state(1)
app_aliyunserver.upload_relay_data(relay1_state)
if __name__=="__main__":
main()
对继电器操作时，最好开启过零点检测。当前测试代码没有开启，用户根据需要，开启或者关闭过零点检测。
import utime as time
from driver import GPIO
#创建继电器实例
JK1=GPIO()
JK1.open("JK1_CTRL")
#global relay_state
#global relay_port
# #AW9523_ZX1是过零点中断脚
# AW9523_ZX1 = GPIO()
# AW9523_ZX1.open('AW9523_ZX1')
# def zx1irq_cb(args): #2600us
# global AW9523_ZX1 # 电力采集芯片 自带的过零点中断脚
# global relay_state
# global relay_port
# AW9523_ZX1.disableIrq()
# time.sleep_us(8000)
# time.sleep_us(30)
# relay_ctl(relay_port,relay_state)
# # AW9523_ZX1.clearIrq()
# def control_raley(port,state):
# global relay_port
# relay_port = port
# global relay_state
# relay_state = state
# AW9523_ZX1.enableIrq(zx1irq_cb)
#继电器,打开/关闭指定的端口
def relay_ctl(port,state):
if port ==1:
JK1.write(state)
print('relay port1,state={}'.format(state))
elif port == 2:
pass
else:
pass
def get_relay_state(port):
if port==1:
return JK1.read()
elif port==2:
pass
from driver import GPIO
'''
三色灯 红绿黄,每一个灯由两个分量控制
即:
(LED1_1,LED1_2)=(0,0) 熄灯
(LED1_1,LED1_2)=(0,1) 绿灯
(LED1_1,LED1_2)=(1,0) 红灯
(LED1_1,LED1_2)=(1,1) 黄灯
'''
LED1_1 = GPIO()
LED1_2 = GPIO()
LED1_1.open('LED11')
LED1_2.open('LED12')
def led_settings(mode):
#亮红灯
if mode==0:
LED1_1.write(1)
LED1_2.write(0)
#亮绿灯
elif mode==1:
LED1_1.write(0)
LED1_2.write(1)
#亮黄灯
elif mode==2:
LED1_1.write(1)
LED1_2.write(1)
#熄灯
elif mode==3:
LED1_1.write(0)
LED1_2.write(0)
#闪烁灯
elif mode==5:
pass
根据芯片手册，获取到电能数据，并对该数据进行相应的转换。
'''
@note:电能检测处理
'''
from driver import UART
import utime as time
global ask_elecdata
global ask_writebuf
ask_writebuf = bytearray(2)
ask_elecdata = [0,0,0,0,0,0,0,0] #电压，电流1，电流2，功率1，功率2，电能1，电能2，温度
#校验和
def SumCheck(src):
sum = 0
for i in range(21):
sum += src[i]
if sum == src[21]:
return 0
else:
return 1
#清除电能计数器
def clearcft(channel):
if channel < 2 and channel >= 0:
chip = int(channel/2)
channel_ab = int(channel%2)
txdata0 = [0x55,0x00,0x00]
BLWriteData(chip,0x1A,txdata0)#打开写保护寄存器
time.sleep_ms(50)
txdata1 = [0x00,0x04,0x00]
BLWriteData(chip,0x18,txdata1)
time.sleep_ms(50)
if channel_ab == 0:
BLReadData(chip,0x0A)
else:
BLReadData(chip,0x0B)
txdata2 = [0x00,0x00,0x00]
BLWriteData(chip,0x18,txdata2)
time.sleep_ms(50)
txdata3 = [0x00,0x00,0x00]
BLWriteData(chip,0x1A,txdata3)#关闭写保护寄存器
print('----------clear cft channel {} success'.format(channel))
else:
print('----------clear cft channel {} failed'.format(channel))
# 读数据
def BLReadData(chip, Addr):
uart2 = UART()
uart2.open('serial2')
readbuf2 = bytearray(4)
txbuff = bytearray(2)
BL0939_w = 0x50
txbuff[0] = BL0939_w + chip
txbuff[1] = Addr
uart2.write(txbuff)
time.sleep_ms(10)
recvSize = uart2.read(readbuf2)
uart2.close()
temp = readbuf2[0]
temp <<= 8
temp += readbuf2[1]
temp <<= 8
temp += readbuf2[2]
return temp
#写数据
def BLWriteData(chip, Addr, uByte):
uart2 = UART()
uart2.open('serial2')
writebuf = bytearray(6)
BL0939_w = 0xA0
writebuf[0] = BL0939_w + chip
writebuf[1] = Addr
writebuf[2] = uByte[0]
writebuf[3] = uByte[1]
writebuf[4] = uByte[2]
check_sum = bytearray(1)
check_temp = 0
for i in range(5):
check_temp = check_temp + writebuf[i]
check_sum[0] = (check_temp) & 0xff
check_sum[0] = ~check_sum[0]
writebuf[5] = check_sum[0]
uart2.write(writebuf)
# print("BLWriteData writebuff={}".format(writebuf))
uart2.close()
#温度转换
def convertTemperature(temp):
result = (170*1.0/448)*(temp*1.0/2-32)-45
return int(result)
#电压转换 单位v
def convertVoltage(voltage):
result = (voltage*1.218*(1950+0.51))/(79931*510)
return int(result)
#电流转换
def convertCurrent(current, scale):
current = 1218.0*current/(324004*3)
if current < 15:
current = 0
else :
current -= 15 #15ma
return current
#功率转换 单位毫瓦
def convertPower(power, scale):
if (power & 0x800000) != 0:
power = 0x1000000 - power
power = -((1.218 *1.218) * (1950 + 0.51) / (4046 * 0.51*3)) * power
else :
power = (1.218*1.218*(1950 + 0.51) / (4046 * 0.51*3)) * power
if power < 200 and power > -200:
power = 0
else:
power -= 200 #200
return abs(int(power))
#电能转换 单位w.h
def convertenergy(cfa_calue):
if (cfa_calue & 0x800000) != 0:
cfa_calue = 0x1000000 - cfa_calue
result = -cfa_calue*1638.4*256*1.218*1.218*(1950+0.51)/(3600*4046*3*510)
else :
result = cfa_calue*1638.4*256*1.218*1.218*(1950+0.51)/(3600*4046*3*510)
if result < 1 and result > -1:
result = 0
else:
pass
return abs(int(result))
#获取bl0939的电力数据,该芯片有两路数据
def ask_uart2_data(chipid):
global ask_elecdata
global ask_writebuf
ret = 1
uart2 = UART()
uart2.open("serial2")
time.sleep_ms(10)
ask_elecdata = [0,0,0,0,0,0,0,0] #电压，电流1，电流2，功率1，功率2，电能1，电能2，温度
ask_writebuf[0] = (0x50 + chipid)
ask_writebuf[1] = 0xAA
uart2.write(ask_writebuf)
time.sleep_ms(10)
readBuf = bytearray(35)
recvSize = uart2.read(readBuf)
time.sleep_ms(5)
if recvSize == 35 and readBuf[0] == 0x55: #校验字节数35和包头0x55
temp1 = readBuf[12]
temp1 <<= 8
temp1 += readBuf[11]
temp1 <<= 8
temp1 += readBuf[10]
#电压
voltage = convertVoltage(temp1)
# print('Voltage-temp1={},voltage={}'.format(temp1,voltage))
ask_elecdata[0] = voltage
temp9 = readBuf[6]
temp9 <<= 8
temp9 += readBuf[5]
temp9 <<= 8
temp9 += readBuf[4]
#电流
current1 = convertCurrent(temp9,1)
# print('Current-temp9={},current1={}'.format(temp9,current1))
ask_elecdata[1] = current1
temp8 = readBuf[9]
temp8 <<= 8
temp8 += readBuf[8]
temp8 <<= 8
temp8 += readBuf[7]
#电流
current2 = convertCurrent(temp8,1)
# print('Current-temp8={},current2={}'.format(temp8,current2))
ask_elecdata[2] = current2
temp2 = readBuf[18]
temp2 <<= 8
temp2 += readBuf[17]
temp2 <<= 8
temp2 += readBuf[16]
#功率1
power1 = convertPower(temp2,1) #4.09
# print('Power-temp2={},power1={}'.format(temp2,power1))
ask_elecdata[3] = power1/1000 #mw
temp3 = readBuf[21]
temp3 <<= 8
temp3 += readBuf[20]
temp3 <<= 8
temp3 += readBuf[19]
#功率2
power2= convertPower(temp3,1)
# print('Power-temp3={},power2={}'.format(temp3,power2))
ask_elecdata[4] = power2/1000 #mw
temp4 = readBuf[24]
temp4 <<= 8
temp4 += readBuf[23]
temp4 <<= 8
temp4 += readBuf[22]
#电能脉冲计数1
energy1= convertenergy(temp4)
# print('energy1= {}-convertenergy1= {}'.format(temp4,energy1))
ask_elecdata[5] = int(energy1*1000) #mwh
temp5 = readBuf[27]
temp5 <<= 8
temp5 += readBuf[26]
temp5 <<= 8
temp5 += readBuf[25]
#电能脉冲计数2
energy2= convertenergy(temp5)
# print('energy2= {}-convertenergy2= {}'.format(temp5,energy2))
ask_elecdata[6] = int(energy2*1000) #mwh
temp6 = readBuf[30]
temp6 <<= 8
temp6 += readBuf[29]
temp6 <<= 8
temp6 += readBuf[28]
temperature= convertTemperature(temp6) #内部温度
# print('temperature= {}'.format(temperature))
ask_elecdata[7] = temperature
else:
pass
uart2.close()
check_sum = bytearray(1)
check_temp = 0
for i in range(34):
check_temp = check_temp + readBuf[i]
check_sum[0] = (check_temp + 0x50 + chipid) & 0xff
check_sum[0] = ~check_sum[0]
if check_sum[0] == readBuf[34]:
ret = 0
return ret,ask_elecdata
else:
return ret,ask_elecdata
'''
@note:播放MP3文件
'''
import audio
from driver import GPIO
#创建一个音频播放实例
aud = audio.Audio(2)
#喇叭播放使能
AMP_EN_PIN = GPIO()
AMP_EN_PIN.open('AMP_EN')
#当前工程文件夹下的路径
filepath = '/data/pyamp/'
def audio_init():
global filepath
enable_spkaker()
aud.setVolume(10)
def audio_play(types):
global filepath
#停止播放
aud.stop()
mp3path = filepath + types +'.mp3'
#开始播放
aud.play(mp3path)
def enable_spkaker():
#启用喇叭
AMP_EN_PIN.write(1)
'''
关电开关检测
'''
from driver import GPIO
#光电开关1
IR_LEFT=GPIO()
IR_LEFT.open('IR_LEFT')
#获取充电端口状态
def get_port_state(port):
global port1_state
port_state=0
if port==1:
# 获取光电开关的状态
port_state=IR_LEFT.read()
elif port ==2:
pass
return port_state
将productKey和productSecret复制到代码的相应的位置
# coding=utf-8
from driver import GPIO
import network
import ujson
import utime as time
import modem
from linksdk import Device
from driver import KV
import app_led
import app_relay
global g_connect_status,net,device,deviceSecret,deviceName,productKey,productSecret,device_dyn_resigter_succed
g_connect_status= False
net= None
device = None
deviceSecret = None
deviceName = None
productKey = "a1omBrxKuDa"
productSecret = "7Mwe9wqwL7wL52r1"
device_dyn_resigter_succed = False
#当iot设备连接到物联网平台的时候触发'connect' 事件
def on_connect(data):
print('***** connect lp succeed****')
app_led.led_settings(1)
#当iot云端下发属性设置时，触发'props'事件
def on_props(request):
print('clound req data is {}'.format(request))
# # #得到云平台下滑的消息
# clound req data is {'params_len': 17, 'params': '{"PORT1_RELAY":1}', 'msg_id': 1615129182}
# # #获取消息中的params数据
get_params=request['params']
print('get_params:',get_params) #'{"PORT1_RELAY":1}'
print('type of get_params:',type(get_params)) #type of get_params: <class 'str'>
# #去除字符串的''，得到字典数据
get_params_2=eval(get_params)
print('get_params2:',get_params_2) #{"PORT1_RELAY":1}
print('type of get_params_2:',type(get_params_2)) # type of get_params_2: <class 'dict'>
# #得到数据中 关于PORT1_RELAY的设置参数
get_relay_state=get_params_2['PORT1_RELAY']
print('get_relay_state:',get_relay_state) #get_realy_state: 1
# #根据云平台下发的指令，如 点亮开/关继电器
if get_relay_state==1:
#打开继电器
app_relay.relay_ctl(1,1)
elif get_relay_state==0:
#关闭继电器
app_relay.relay_ctl(1,0)
print('----------------------------')
#网络连接的回调函数
def on_4g_cb(args):
global g_connect_status
pdp = args[0]
netwk_sta = args[1]
if netwk_sta == 1:
g_connect_status = True
else:
g_connect_status = False
#网络连接
def connect_network():
global net,on_4g_cb,g_connect_status
#NetWorkClient该类是一个单例类，实现网络管理相关的功能，包括初始化，联网，状态信息等.
net = network.NetWorkClient()
g_register_network = False
if net._stagecode is not None and net._stagecode == 3 and net._subcode == 1:
g_register_network = True
else:
g_register_network = False
if g_register_network:
#注册网络连接的回调函数on(self,id,func); 1代表连接，func 回调函数 ；return 0 成功
net.on(1,on_4g_cb)
net.connect(None)
else:
print('网络注册失败')
while True:
if g_connect_status:
print('网络连接成功')
break
time.sleep_ms(20)
#动态注册回调函数
def on_dynreg_cb(data):
global deviceSecret,device_dyn_resigter_succed
deviceSecret = data
device_dyn_resigter_succed = True
# 连接物联网平台
def dyn_register_device(productKey,productSecret,deviceName):
global on_dynreg_cb,device,deviceSecret
kv = KV()
key = '_amp_customer_devicesecret'
deviceSecretdict = kv.getStorageSync(key)
if deviceSecretdict is not None:
deviceSecret = deviceSecretdict[key]
if deviceSecretdict is None or deviceSecret is None:
key_info = {
'productKey': productKey ,
'productSecret': productSecret ,
'deviceName': deviceName
}
# 动态注册一个设备，获取设备的deviceSecret
device.register(key_info,on_dynreg_cb)
def mqtt_main():
global g_connect_status,net,device,deviceSecret,deviceName,productKey,productSecret,device_dyn_resigter_succed,cloud_led,realy,flag_close_lock_request
flag_close_lock_request=0
#连接网络
connect_network()
#获取设备的IMEI 作为deviceName 进行动态注册
deviceName = modem.getDevImei()
#初始化物联网平台Device类，获取device实例
device = Device()
if deviceName is not None and len(deviceName) > 0 :
#动态注册一个设备
dyn_register_device(productKey,productSecret,deviceName)
else:
print("获取设备IMEI失败，无法进行动态注册")
while deviceSecret is None:
time.sleep(0.2)
print('动态注册成功:' + deviceSecret)
key_info = {
'region' : 'cn-shanghai' ,
'productKey': productKey ,
'deviceName': deviceName ,
'deviceSecret': deviceSecret ,
'keepaliveSec': 60
}
#打印设备信息
print(key_info)
#device.ON_CONNECT 是事件，on_connect是事件处理函数/回调函数
device.on(device.ON_CONNECT,on_connect)
device.on(device.ON_PROPS,on_props)
device.connect(key_info)
# 将IMEI数据放入字典
imei={}
imei["IMEI"]=deviceName
#将字典转换为字符串
imei_str=ujson.dumps(imei)
data0={
'param':imei_str
}
time.sleep_ms(1000)
device.postProps(data0)
def upload_elc_data(elcdata):
#上传端口一的电力数据elcdata[0]
port1_data={}
port1_data['PORT1']=elcdata[0]
port1_data_str=ujson.dumps(port1_data)
data1={
'param':port1_data_str
}
time.sleep_ms(500)
device.postProps(data1)
def upload_relay_data(relay1_state):
#上传端口1所在的继电器状态
relay1_data_str=ujson.dumps({"PORT1_RELAY":relay1_state})
data3={
'param':relay1_data_str
}
time.sleep_ms(500)
device.postProps(data3)
#上传端口2所在的继电器状态
#pass
{
"version": "1.0.0",
"io": {
"LED11": {
"type": "GPIO",
"port": 1,
"dir": "output",
"pull": "pullup"
},
"LED12": {
"type": "GPIO",
"port": 2,
"dir": "output",
"pull": "pullup"
},
"JK1_CTRL": {
"type": "GPIO",
"port": 10,
"dir": "output",
"pull": "pullup"
},
"JK2_CTRL": {
"type": "GPIO",
"port": 34,
"dir": "output",
"pull": "pullup"
},
"AMP_EN": {
"type": "GPIO",
"port": 4,
"dir": "output",
"pull": "pullup"
},
"GPIO_RI": {
"type": "GPIO",
"port": 46,
"dir": "output",
"pull": "pullup"
},
"IR_LEFT": {
"type": "GPIO",
"port": 11,
"dir": "input",
"pull": "pullup"
},
"IR_RIGHT": {
"type": "GPIO",
"port": 22,
"dir": "input",
"pull": "pullup"
},
"NFC_IRQ": {
"type": "GPIO",
"port": 5,
"dir": "irq",
"pull": "pullup",
"intMode": "rising"
},
"AW9523_ZX1": {
"type": "GPIO",
"port": 6,
"dir": "irq",
"pull": "pullup",
"intMode": "rising"
},
"SPI0": {
"type": "SPI",
"port": 0,
"mode": "master",
"freq": 2000000
},
"serial1":{
"type":"UART",
"port":0,
"dataWidth":8,
"baudRate":9600,
"stopBits":1,
"flowControl":"disable",
"parity":"none"
},
"serial2":{
"type":"UART",
"port":1,
"dataWidth":8,
"baudRate":4800,
"stopBits":1,
"flowControl":"disable",
"parity":"none"
},
"serial3":{
"type":"UART",
"port":2,
"dataWidth":8,
"baudRate":9600,
"stopBits":1,
"flowControl":"disable",
"parity":"none"
}
},
"debugLevel": "ERROR",
"repl":"disable"
}
电力数据是一个数组，数组中的元素分别为：电压，电流，功率，电能，温度
接入220V电源、用电器件，获取到“电压，电流，功率，电能，温度”数据
