来源 | HaaS技术社区

1、前言

Python 语言的是一款非常容易使用的解释性语言，它的主要特点如下：
● 易学易用
● 可扩展可移植
● 丰富的三方库
正是由于以上这些特点，使得Python在一些逻辑简单，功能复杂的场景应用非常广泛。近几年，随着人工智能的飞速发展，Python 的优点得到的充分的体现，因此成为了AI开发的首选语言。

Python轻应用主要是基于micropython实现，在保留了python语言主要特性的基础上，他还对嵌入式系统的底层做了非常不错的封装，它的特点如下：
● 沿用python语法和基本数据结构，保持易学易用性
● 提供基于嵌入式系统的硬件功能库封装，让嵌入式开发简单便捷

今天要介绍的是如何使用Python 轻应用，连接阿里云的千里传音服务，打造一个现象级的云喇叭产品。

2、方案介绍

2.1、什么是千里传音？

“千里传音”服务，是阿里云IoT针对带有语音播报能力的AIoT设备，提供的一个云端一体的解决方案。为播报提醒类设备应用提供从播报语料合成，语料管理，语料推送到设备，播报设备管理等完善功能，配合集成了端侧播报能力的HaaS设备，帮助用户高效完成播报类设备应的开发和长期运行。

“千里传音”服务以项目为单位来帮助客户组织应用和管理设备，以便客户面向不同的用户来管理设备语料更新，以及批量或单个设备语料推送。同时，“千里传音”服务为客户应用提供云端API，通过传入语料组合逻辑，及设备ID，就可以完成对端设备播报的调用，简单省事。借助阿里云IoT平台提供的高并发设备通信能力，帮助客户无忧完成大规模设备部署和长期高可用运行。

它主要有三个部分组成：App,服务器，和播报设备

”千里传音“，指的就是无论服务的使用者(App)和播放设备之间的物理距离有多远，都可以通过服务器，将自己想要传达的音频数据，传输给相关播报设备进行播报。

服务的使用者，可以通过千里传音服务提供的SDK和服务器进行通讯

目前千里传音服务提供了多种编程语言的SDK,包括Java,JS,Python,PHP 等等，开发者可以选自自己熟悉的开发语言进行开发。调试阶段，可以使用在线调试工具进行调试。

服务器通过MQTT将播放资源和指令下发给播放设备进行播放

目前设备支持本地音频播放和在线音频播放，在线音频的播放需要通过物模型自定义服务，将音频的url发送给设备端，本地音频的播放，需要通过千里传音SpeechPost服务先将音频链接发送给设备端，设备端将音频文件以ID 命名并保存。当服务端需要播放的时候，将所有的音频文件按照ID 组合起来，通过SpeechBroadcast服务下发给设备端，设备端将音频组合起来，进行播放。

千里传音服务提供的能力如下:

● 项目管理
客户通过项目形式管理不同应用场景中的设备和语料。
● 智能语料生成
通过人工智能算法帮助客户快速完成文字到固定播报语料的生成，支持wav和mp3格式输出。
● 语料组合播报
通过远程命令，告知特定设备将本地语料以特定顺序组合后播报，并支持加入动态数字内容。
● 动态语料合成
支持用户通过API生成动态播报语料，推送到端侧播报。此类语料设备端采用在线播放的形式，将不固化到设备中。
● 语料空中推送
为客户提供语料空中推送到单个和项目中全量设备的能力，实现设备端固化语料的更新。使设备播报语音内容变得可以运营。
● 云端API
为客户提供平台能力对应API，以实现上述播报能力的云端控制。

2.2、如何使用千里传音？

本方案中，播报设备指的是阿里云的物联网硬件设备HaaS100，App的使用的是在线调试工具或者PC端的Python 应用。
具体的交互流程如下:

以上代码逻辑，均在Python轻应用中实现，具体细节轻参考物品清单章节中的Python轻应用固件软件部分

3、物品清单

3.1、硬件

HaaS100 开发板一块
电源一个
micro usb 一个
sdcard 一个
有源喇叭一个购买链接
杜邦线一根

注意事项:

HaaS100支持外接微型SD卡（Micro SD），最大支持64GB数据的存储,SD卡槽位于开发板背,可以外接微型SD存储卡
sdcard 格式支持: FAT16/FAT32，本案例中暂时不支持exFAT 格式
喇叭的链接方法如下:

Column 1	Column 2	Column 3
喇叭丝印接口	HaaS100 GPIO NUM	图中连接线颜色
IN	28	黄色
VCC	2	红色
GND	4	绿色

3.2、软件

3.2.1、千里传音产品和设备创建

● 创建项目
登录阿里云官网，选择产品阿里云IoT平台->企业物联网平台->控制台->增值服务->IoT云端一体服务，进入千里传音服务后台

在项目创建过程中，需要输入项目名称，并对项目进行简单描述，以便后期维护。

在完成项目创建后，系统将帮助用户创建一个与项目名相同的产品，以便后续加入设备。同时，如果用户希望为设备增加千里传音以外的能力，也可以直接到物联网平台的设备管理界面中，为设备添加物模型能力。
在项目列表中，点击需要配置的项目，可以对项目内容进行配置管理。

在项目配置界面，有两个选项，分别是服务配置和设备管理，其中服务配置是用来配置千里传音服务的的，主要是语料管理，设备配置是用来管理项目中设备列表的。
● 生成语料
如下图所示，在项目配置中，选中服务配置->管理服务，

进去千里传音服务配置界面以后，选中语料配置中的新增语料

新增自定义语料界面如下:

其中语料标识和语料内容非常重要,语料标识必须是项目级别唯一，它是作为语料组合调用时候的唯一标签，同时也作为设备端保存语料的文件名。语料内容指的就是我们要播报的文本.
除此之外，我们还可以自定义语料的场景，包括方言场景，童声场景，客服场景以及通用场景；自定义语料的播放速度；自定义语料的播报音量等。

● 创建设备
在完成语料构建和管理后，需要在项目中创建设备，以便最终用户的播报应用可以将命令发送到设备端，完成整个播报链路。点击“设备管理”标签，将进入设备管理界面。

设备创建可以有单个创建和批量创建两种方式。
点击“创建设备”会打开创建单个设备弹窗，并要求用户输入设备相关信息。

DeviceName：英文字符组成的设备名称，设备名称在项目中不可重复。
备注名称：为了便于用户区分设备，给设备赋予的别名。
点击“批量添加”会打开批量创建设备弹窗。

添加方式
○ 自动生成：指系统将为用户自动生成DeviceName。
○ 批量上传：需要用户通过.csv文件上传自定义的DeviceName。
设备数量：需要批量添加的设备个数。

在创建完成设备以后，设备管理界面，可以显示设备列表，点击设备对应的鉴权信息，可以查看设备的三元组信息

● 语料推送

在创建完成设备以后，我们就可以回到语料配置界面，将新增的语料推送到相应的设备上面进行测试了。
如下图所示，点击需要推送的语料，操作栏中的推送到设备按钮(下图中我们推送的语料id是 yuan)

进入语料推送界面，如下图所示，默认支持单个设备推送，分组推送和全部设备推送。

单个设备和分组模式，需要选择自己的设备名或者分组名，选中以后点击确定即可完成推送。

3.2.2、HaaS100 Python轻应用固件

可以通过HaaS100 AliOS Things 3.3 开源代码编译或者Python轻应用官方固件直接下载(2021.5月份以后)，相关下载方法都在Python轻应用快速上手中有详细介绍: Python轻应用快速上手

3.2.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()

以上是Python轻应用的主体代码，它的主要功能如下：

连接物联网平台
监听SpeechPost 服务，下载服务端推送过来的音频文件到 /sdcard/resource/
监听SpeechBroadcast服务，播放指定的音频

注意事项：

下载常用本地音频资源文件到/sdcard/resource目录资源文件下载
请务必将代码中的

PRODUCT_KEY = "xxxxxx"
PRODUCT_SECRET = "xxxxxxx"
DEVICE_NAME = "xxxxx"
DEVICE_SECRET = "xxxxxx"
替换成千里传音产品中相关信息

4、如何实现

4.1、固件烧录

请参考Python轻应用快速上手中烧录相关章节。

4.2、轻应用运行

将物品软件清单中Python轻应用的代码保存为haas_speech.py,并拷贝到sdcard中，设备插入sdcard后重启
● 执行联网操作

  python 
  /data/python-apps
  /wifi/main.py your_ssid. your_password
 

● 执行haas_speech.py

  python 
  /sdcard/haas_speech.py
 

4.3、调试

4.3.1在线调试

打开在线调试并登录
如下如所示，选中SpeechByCombination,选择右侧服务器华东(上海)，设置我们需要发送的语音序列信息，数字请使用{$xxx} 进行表示，其它语音请以此按照千里传音后台的语料ID 进行配置，最后设置需要调试的设备DeviceName 和 ProductKey。
点击发起调试即可

4.3.2、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'))

执行python SpeechByCombination.py

技术交流

Python轻应用继承了Python易学易用的特点，同时提供了基于嵌入式硬件的基础库封装，让开发者可以很方便的通过交互式的环境，实时进行嵌入式开发，让嵌入式开发也变得简单方便。

欢迎使用钉钉扫码加入Python轻应用SVIP群：

HaaS轻应用(Python): 使用“千里传音“服务打造您的专属云喇叭