开发者学堂课程【HaaS 物联网云端一体低代码开发课程：《 Python 轻应用方式打造播报音箱和温湿度监控报警系统》 】学习笔记，与课程紧密联系，让用户快速学习知识。
课程地址：https://developer.aliyun.com/learning/course/810/detail/13907
《Python 轻应用方式打造播报音箱和温湿度监控报警系统》
一、python 打造播报音箱和温湿度监控报警系统实例及操作
播报音响简介
“千里传音”服务，是阿里云 loT 针对带有语音播报能力的 a lot 设备，提供的一个云端一体的解决方案。
它为播报提醒类设备应用提供从播报语料合成，语料管理，语料推送到设备，播报设备管理等完善功能，配合集成了端侧播报能力的 HaaS 的设备，帮助用户高效完成播报类设备应用的开发和设备长期运行
”千里传音“，指的就是无论服务的使用者 (App) 和播放设备之间的物理距离有多远，都可以通过服务器，将自己想要传达的音频数据，传输给相关播报设备进行播报。
服务的使用者，可以通过千里传音服务提供的 SDK 和服务器进行通讯
目前千里传音服务提供了多种编程语言的 SDK ,包括 Java,JS,Python,PHP 等等，开发者可以选自自己熟悉的开发语言进行开发。调试阶段，可以使用在线调试工具进行调试。
服务器通过 MQTT 将播放资源和指令下发给播放设备进行播放
目前设备支持本地音频播放和在线音频播放，在线音频的播放需要通过物模型自定义服务，将音频的 url 发送给设备端，本地音频的播放，需要通过千里传音SpeechPost 服务先将音频链接发送给设备端，设备端将音频文件以 ID 命名并保存。当服务端需要播放的时候，将所有的音频文件按照 ID 组合起来，通过 SpeechBroadcast 服务下发给设备端，设备端将音频组合起来，进行播放。
千里传音服务提供的能力：
项目管理
客户通过项目形式管理不同应用场景中的设备和语料。
智能语料生成
通过人工智能算法帮助客户快速完成文字到固定播报语料的生成，支持 wav 和 mp3 格式输出。
语料组合播报
通过远程命令，告知特定设备将本地语料以特定顺序组合后播报，并支持加入动态数字内容。
动态语料合成
支持用户通过 API 生成动态播报语料，推送到端侧播报。此类语料设备端采用在线播放的形式，将不固化到设备中。
语料空中推送
为客户提供语料空中推送到单个和项目中全量设备的能力，实现设备端固化语料的更新。使设备播报语音内容变得可以运营。
云端 API
为客户提供平台能力对应API，以实现上述播报能力的云端控制。
1. 创建项目：以项目为维度管理一批设备和这批设备使用的语料
2. 生成语料：通过阿里云智能语音合成技术，为项目类设备生成播报语料
3. 创建设备：为项目创建设备，以使最终实现按设备指定播报
4.设备生产：打包下载生成的语料，设备身份信息在生成阶段预置到设备中
5.设备运行：设备投入实际运行，可以接收来自平台的组合播报命令，更新语料推送
5.1.远程组合播报命令：客户播报应用通过千里传音 API ，指定特定设备按特定语料标识组合在本地实现播报
5.2.动态语料播报：客户播报应用通过千里传音 API ，向设备推送动态播报命令，设备通过收到的 URL 进行单次播报
5.3.动态更新语料：客户可以根据具体的运营要求，通千里传音服务向设备空中推送更新语料
播报音响硬件连接
喇叭上面的蓝色按钮逆时针旋转到底（声音最大）
串口工具
mac 电脑推荐使用 picocom,windows 电脑推荐使用 putty , 务必保证设置串口波特率为 1500000
千里传音设备端 Python 轻应用代码：
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
'''
@File : haas_speech.py
@Description: file description
@Date : 2021/04/19 13:56:50
@Author : guoliang.wgl
@version : 1.0
'''
import iot
import utime
import http
import ujson as json
import math
AUDIO_ENABLED = True
if AUDIO_ENABLED:
from audio import Player,Snd
toneDir = "/sdcard/resource/"
tonenameSuffix = [".wav", ".mp3"]
tonenameNumb = ["SYS_TONE_0", "SYS_TONE_1", "SYS_TONE_2", "SYS_TONE_3", "SYS_TONE_4", "SYS_TONE_5", "SYS_TONE_6", "SYS_TONE_7", "SYS_TONE_8", "SYS_TONE_9"]
tonenameNumb1 = "SYS_TONE_yao"
tonenameDot = "SYS_TONE_dian"
tonenameUnit = ["SYS_TONE_MEASURE_WORD_ge", "SYS_TONE_MEASURE_WORD_shi", "SYS_TONE_MEASURE_WORD_bai", "SYS_TONE_MEASURE_WORD_qian"]
tonenameHunit = ["SYS_TONE_MEASURE_WORD_wan", "SYS_TONE_MEASURE_WORD_yi", "SYS_TONE_MEASURE_WORD_sw", "SYS_TONE_MEASURE_WORD_bw", "SYS_TONE_MEASURE_WORD_qw"]
tonepathConnected = "fs:/sdcard/resource/connected.wav"
tonepathPowerOn = "fs:/sdcard/resource/poweron.wav"
on_request = False
on_play = False
if AUDIO_ENABLED:
Snd.install_codec_driver()
Snd.init()
player = Player()
player.create()
player.set_volume(10)
def play(path):
global player
player.set_source(path
player.start()
player.wait_complete()
def playlist(pathlist):
for path in pathlist:
play('fs:'+path)
print('********end playlist*******')
def add_amount(num_str, toneList, formatFlag):
global tonenameNumb
global tonenameNumb1
global tonenameDot
global tonenameUnit
global tonenameHunit
global toneDir
global tonenameSuffix
num_f = float(num_str)
numb = int(num_f)
deci = num_f - numb
target = numb
subTarget = 0
subNumber = None
slot = 0
factor = 0
count = 0
prevSlotZero = False
hundredMillionExist = False
tenThousandExist = False
if (numb < 0 or numb >= 1000000000000):
print('amount overrange')
return toneIndex
if (deci < 0.0001 and deci > 0.0):
deci = 0.0001
i = 2
while(i >= 0):
factor = math.pow(10000,i)
if target < factor:
i = i -1
continue
subTarget = int(target / factor)
target %= factor
if (subTarget == 0):
i = i -1
continue
if (i == 2):
hundredMillionExist = True
elif (i == 1):
tenThousandExist = True
subNumber = subTarget
prevSlotZero = False
depth = 3
while(depth >= 0):
if(subNumber == 0):
break
factor = math.pow(10, depth)
if ((hundredMillionExist == True or tenThousandExist == True) and i == 0):
pass
elif (hundredMillionExist == True and tenThousandExist == True and depth > 0 and subTarget < factor):
pass
elif (subTarget < factor):
depth = depth - 1
continue
slot = int(subNumber / factor)
subNumber %= factor
if (slot == 0 and depth == 0):
depth = depth - 1
continue
if ((subTarget < 20 and depth == 1) or (slot == 0 and prevSlotZero) or (slot == 0 and depth == 0)):
pass
else:
toneList.append(toneDir + tonenameNumb[slot] + tonenameSuffix[formatFlag])
count += 1
if (slot == 0 and prevSlotZero == False):
prevSlotZero = True
elif (prevSlotZero == True and slot != 0):
prevSlotZero = False
if (slot > 0 and depth > 0) :
toneList.append(toneDir + tonenameUnit[depth] + tonenameSuffix[formatFlag])
count += 1
depth = depth - 1
if (i > 0):
toneList.append(toneDir + tonenameHunit[i - 1] + tonenameSuffix[formatFlag])
count += 1
i = i - 1
if (count == 0 and numb == 0):
toneList.append(toneDir + tonenameNumb[0] + tonenameSuffix[formatFlag])
if (deci >= 0.0001) :
toneList.append(toneDir + tonenameDot + tonenameSuffix[formatFlag])
deci ="{:.4f}".format(deci)
deci_tmp = str(deci).strip().rstrip('0')
deci_str = ''
got_dot = False
for j in range(len(deci_tmp)):
if(got_dot):
deci_str = deci_str + deci_tmp[j]
elif deci_tmp[j] == '.':
got_dot = True
deciArray = deci_str
for item in deciArray:
if (item >= '0' and item <= '9'):
toneList.append(toneDir + tonenameNumb[int(item)] + tonenameSuffix[formatFlag])
return toneList
def do_voice_test():
"""
@description :
---------
@param :
-------
@Returns :
-------
"""
global on_request
global add_amount
global playlist
global on_play
# 请替换物联网平台申请到的产品和设备信息,可以参考文章: https://blog.csdn.net/HaaSTech/article/details/114360517
productSecret = "xxxxxxxx"
productKey = "xxxxxxx"
deviceName = "xxxxxxx"
deviceSecret = "xxxxxxxx"
# 初始化linkkit sdk
key_info = {
'region' : 'cn-shanghai' ,
'productKey': productKey ,
'deviceName': deviceName ,
'deviceSecret': deviceSecret ,
'productSecret': productSecret
}
device = iot.Device(key_info)
# 物联网平台连接成功的回调函数
def on_connect():
if AUDIO_ENABLED:
global player
global tonepathConnected
print('linkkit is connected ')
if AUDIO_ENABLED:
play(tonepathConnected)
device.on('connect',on_connect)
# 设置 service 事件接收函数（本案例是千里传音）
def on_service(serviceid,request):
global on_request
global on_play
print("serviceid is :" + serviceid)
print("request is :" + request)
data = json.loads(request)
if serviceid == "SpeechPost":
on_request = data
print('**********SpeechPost data *******' )
print(on_request)
elif serviceid == "SpeechBroadcast":
on_play = data
print('**********SpeechPost data *******' )
print(on_play)
else:
pass
print("on_service end")
device.on('service',on_service)
# 连接物联网平台
device.connect()
def download_resource_file(on_request):
client=http.client()
client.set_header("Accept: */*\r\n")
client.get(on_request['url'])
response = json.loads(client.get_response())
#print(response)
format = response['format']
size = response['size']
format = response['format']
audio = response['audios'][0]
id = audio['id']
path = '/sdcard/resource/'+id+'.'+format
print('************ begin to download: ' + path)
ret = client.download(audio['url'],path)
if ret == 0 :
print('************ download: ' + path + ' succeed')
else:
print('************ download: ' + path + ' failed')
def play_voice(data):
global toneDir
global tonenameSuffix
global playlist
format = data['format']
audioResFormat = 0
if (format == 'mp3'):
audioResFormat = 1
speechs = data['speechs']
print(type(speechs))
print(speechs)
toneList = []
for speech in speechs:
print(speech)
# length = len(speech)
if speech.endswith('}') and speech.startswith('{') and (speech[1] == '$'):
speech_num = speech.strip('{').strip('$').strip('}')
toneList = add_amount(speech_num,toneList,audioResFormat)
else:
toneList.append(toneDir + speech + tonenameSuffix[audioResFormat])
print(toneList)
if AUDIO_ENABLED:
playlist(toneList)
# 触发linkit sdk持续处理server端信息
while(True):
device.do_yield(1000)
if on_request:
print('***************************')
print(on_request)
download_resource_file(on_request)
on_request = False
elif on_play:
print('***************************')
print(on_play)
play_voice(on_play)
on_play = False
# 断开连接
device.close()
if AUDIO_ENABLED:
player.stop()
player.release()
if __name__ == '__main__':
if AUDIO_ENABLED:
play(tonepathPowerOn)
do_voice_test()
SDK 调试
安装 SDK 核心库
pip install aliyun-python-sdk-core
修改如下 Python 代码中的 KeyID 和 Secret ,并保存为 SpeechByCombination.py
#coding=utf-8
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest
from aliyun_key import *
client = AcsClient(accessKeyId, accessSecret, 'cn-shanghai')
request = CommonRequest()
request.set_accept_format('json')
request.set_domain('iot.cn-shanghai.aliyuncs.com')
request.set_method('POST')
request.set_protocol_type('https') # https | http
request.set_version('2018-01-20')
request.set_action_name('SpeechByCombination')
request.add_query_param('RegionId', "cn-shanghai")
request.add_query_param('CombinationList.1', "welcome")
# request.add_query_param('CombinationList.1', "zfbGet")
# request.add_query_param('CombinationList.2', "{$10000.056}")
# request.add_query_param('CombinationList.3', "yuan")
request.add_query_param('ProductKey', "a1Ba4rCO9iM")
request.add_query_param('DeviceName', "py_voice_01")
response = client.do_action(request)
# python2: print(response)
print(str(response, encoding = 'utf-8'))
创建语料
点击管理服务按钮，进入语料页面
点击创建语料按钮，进行语料创建
语料标识：将作为生成语料的文件名，以及后续组合播报时的语料组合索引。建议按照播报内容的拼音首字母编写，如“银行到账”可将语料标识定义为 “YHDZ”.
语料内容：需要播报的语料内容
语料场景：千里传音服务为用户提供了5个场景下的45种不同语音合成能力，用户可以根据场景需求，选择对应的语音进行语料合成
推送代码到设备端
1. 选择/设置 HaaS EDU 要连接的路由器信息- SSID
2.设置 HaaS EDU 要连接的路由器 Password
3.选择 HaaS EDU 连接到电脑上面的串口
4.设置 UART 的波特率，选择目标路径
语料需要在执行前推送到设备，推送方法是点击推送到设备。
云端 API 调用
在线API调试地址，修改其中低调 ProductKey 和 DeviceName 以及需要发送的语料完成参数配置后，点击发起调用，即可将组合播报命令发送至设备端。
实例
1.实时显示设备温湿度
2.控制设备 RGB-LED 亮灭
3.当设备超过设定阈值时，触发告警
4.同时支持 Android 和 IOS
云端：
1. 云平台：
千里传音服务
设备属性
报警事件
2. loT Studio
应用界面
触发语音报警信息
设备端：
1. 云平台：
播报音箱
温度上报湿度上报
设定温度阈值
温度阈值检测报警
loT Studio 创建项目
https：//studio.iot.aliyun.com/
