fubuki-iot 0.2

An accessable Iot Terminal implementated by Python

Fubuki Iot —— 物联网智能终端

简介

Fubuki Iot是一款开源的物联网智能终端，类似于市面上的天猫精灵，小爱同学等。它可以监听智能家居的消息，也可以根据用户语音向智能家居 发送消息，从而实现家居智能化。与市面上各种终端相比，它具有以下优点：

• 定制化程度更高：用户可以自己实现对家居的控制，甚至对原有的家居电器改造
• 安全性更好：由于可以部署在本地局域网上，因此个人信息不会被上传到外网
• 效率更高：通过MQTT协议完成设备之间的交流，不需要复杂的流程

经过测试，本框架可以很好的支持Windows系统和树莓派（Linux）系统。

快速上手

安装

方案一：通过pip安装

（待完善）

方案二：下载安装

首先在终端执行以下命令：

git clone [e-mail]:littlebutt/fubuki-iot.git

然后进入根目录，执行以下命令安装fubuki-iot包

python setup.py install

方案三：通过docker安装

同样像方案三中下载项目到本地，然后执行以下命令

docker build -t fubuki-iot:latest .
docker run -it fubuki-iot:latest /bin/bash

就可以进入到容器内部了

启动

本项目内置了 百度云 的语音合成和语音识别的功能，因此使用之前需要免费申请百度云账号， 点击 这里 申请。申请之后开通"产品"-"人工智能"-"语音技术"的服务。 当然，您也可以用开源方案自己训练模型替代内置的语音功能，具体见 进阶功能 。

1.创建资源目录 首先创建一个Python工程 demo ，并在根目录下创建一个目录命名为resources

2.创建.env文件

在新建的工程下创建一个.env配置文件，其内容参考如下：

ENVIRONMENT=dev
RESOURCE_PATH=刚才创建的resources目录，结尾不要加“/”
BAIDU_ACCESS_TOKEN=百度云API的token，可以留空
BAIDU_API_KEY=申请的百度云的AK
BAIDU_SECRET_KEY=申请的百度云的密钥

3.创建主程序

在根目录下创建文件 app.py，并写入以下内容

from iot import Terminal


Terminal.run()

运行即可启动

使用

目前主流的物联网信息交换都采用 MQTT协议 ，因此要实现控制智能设备需要部署MQTT 服务器。本项目建议使用 EMQX 服务器，具体安装和部署方法点击 这里 。

若要完整的实现对硬件设备的控制可以参考 相关资料 。

内置功能

1.对话

运行智能终端后，按下键盘上的f可以进行录音。对着麦克风说出“在吗”、“你好”后，智能终端会回应“在的”。

2.控制开关和电灯

运行智能终端，按下键盘上的f后对着麦克风说出“打开开关”，然后它会向MQTT服务器的default/switch Topic发送以下信息：

{
  "switch": "on"
}

同样，对着麦克风说出“关闭开关”后，它会向MQTT服务器的default/switch Topic发送以下信息：

{
  "switch": "off"
}

具体效果需要由订阅了 default/switch Topic的智能设备实现。

此外，运行智能终端，按下键盘上的f后对着麦克风说出“打开卧室/客厅/餐厅灯”也会向MQTT服务器的default/light Topic发送以下信息：

{
  "position": "bedroom/livingroom/dinningroom" 
}

3.接受按钮信息

运行终端，当由设备向MQTT服务器的 self/button 发送如下消息后，终端会说“有人按下了按钮”。

{
  "topic":"self/button",
  "device":"button",
  "verbose":"false",
  "message":"有人按下了按钮"
}

自定义功能

1.自定义语音功能

语音功能可以理解为用户和智能终端进行对话，类似于机器人的对话功能。这种功能一般不涉及硬件。在天猫精灵中， 就内置了提醒助手、墨迹天气等语音功能。

首先在自己创建的Python项目的根目录中创建一个包（package）命名为 models ，在这个包中创建一个python文件 acoustics.py， 在文件中定义一个语义模型：

from iot import SemanticsGroup, SemanticsModel, SemanticsRedirectEnum, SemanticsFromEnum, SemanticsFunc

@SemanticsGroup.add_model
class MySemanticsModel(SemanticsModel):

    code = 'hello'                                  # 语义模型的标识，自定义

    frm = SemanticsFromEnum.USER                    # 语义模型的来源，这里是接受用户的语音命令，所以是USER

    topic = ''                                      # 由于不涉及发布消息，所以这个字段用不到，留空就行

    regex = "(.*)后提醒我(.*)"                       # 匹配用户语音命令的正则表达式，比如这里是一个有关提醒的命令

    regex_num = 3                                   # 上述表达式匹配后的分组（group）的数量，第一个为用户命令全量文本，第二个是“后”前面的文本，第三个是“我”后面的文本

    redirect = SemanticsRedirectEnum.ACOUSTICS      # 语义处理好后的重定向，由于不需要发送消息等后续操作，所以这里是直接语音返回

    func: SemanticsFunc = my_semantics_model_func   # 处理用户命令的回调函数

在上面的语义模型中，最后一个字段是一个 SemanticsFunc 实例，它是一个返回 FunctionDeviceModel 或者 UniverseNoticeModel 的方法 因此，需要这样定义：

from typing import Union
from iot import UniverseNoticeModel, FunctionDeviceModel

def my_semantics_model_func(*args) -> Union[FunctionDeviceModel, UniverseNoticeModel]:

    time = args[1]                                      # 获取时间

    content = args[2]                                   # 获取提醒内容

    # 处理提醒命令，可以借助其他API实现

    return FunctionDeviceModel(                         # 最后返回一个功能设备模型

        smt_code='hello',                               # 对应的语义模型标识

        is_raw=True,                                    # 是否为纯文本

        acoustics=f"好的，我会在{time}后提醒你{content}", # 返回给用户的语音内容

        data=""                                         # 由于是纯文本，所以这个字段用不到

    )

定义好以后需要在 app.py 中加入一行：

from iot import Terminal


Terminal.load_models('demo.models')
Terminal.run()

2.自定义设备功能

智能终端最大的优势就是可以通过语音控制智能家居。同样，需要定义一个语义模型实现这个功能：

from typing import Union

from iot import SemanticsGroup, SemanticsModel, SemanticsRedirectEnum, SemanticsFromEnum, SemanticsFunc, UniverseNoticeModel, FunctionDeviceModel


def curtain_semantics_model_func(*args) -> Union[FunctionDeviceModel, UniverseNoticeModel]:
    return FunctionDeviceModel(

        smt_code="hi",

        topic="default/curtain",                        # 发送的Topic，其实后续会被语义模型的Topic覆盖

        is_raw=False,                                   # 不再是纯文本返回了

        acoustics="好的，正在为你打开窗帘",                # 返回给用户的提示信息

        data={                                          # 发送的数据
            'state': 'on'
        }
    )


@SemanticsGroup.add_model
class SwitchOnSemanticsModel(SemanticsModel):

    code = "hi"

    frm = SemanticsFromEnum.USER

    topic = 'default/curtain'

    regex = "打开窗帘"

    regex_num = 1

    redirect = SemanticsRedirectEnum.MESSAGE # 重定向给消息，因为需要发送MQTT消息

    func: SemanticsFunc = curtain_semantics_model_func

具体怎么消费这个MQTT消息，即硬件设备如何处理则需要改造硬件，具体参考 相关资料 。

3.自定义消息推送

和之前一样，也需要定义一个语义模型：

from typing import Optional

from iot import SemanticsGroup, SemanticsModel, SemanticsRedirectEnum, SemanticsFromEnum, SemanticsFunc, UniverseNoticeModel


def button_semantics_model_func(model) -> UniverseNoticeModel:

    #处理设备推送的统一推送模型

    return UniverseNoticeModel(                 # 这次返回的是统一推送模型

        smt_code='hei', 

        topic='self/weather',                   # topic，被用来检索语义模型的

        device='remote_server',                 # 设备来源

        verbose=False,                          # 是否多语，这里只需要通知以下用户所以选择False

        message="天气播报：短期将有大量降雨"       # 返回给用户的信息
    )


@SemanticsGroup.add_model
class ButtonSemanticsModel(SemanticsModel):

    code = "hei"

    frm = SemanticsFromEnum.DEVICE                  # 来自设备

    topic = 'self/weather'

    regex: Optional[str] = None

    regex_num: Optional[str] = None

    redirect = SemanticsRedirectEnum.ACOUSTICS      #直接返回

    func: SemanticsFunc = button_semantics_model_func

至此，可以实现一个简单的物联网终端！

进阶功能

如果您对上述基本功能还不满足，可以试一下进阶功能。

1.语音唤醒功能

像主流的智能终端一样，本项目也可已开启语音唤醒功能。该功能是通过内置的 Pocketsphinx 实现的，因此需要安装其依赖，包括swig，C语言环境等，具体可以查看相关文档。

成功安装好依赖后将 .env 文件添加以下两行：

TERMINAL_MODE=0
DEVICE_REC=PocketsphinxRecorder

再次启动程序可以通过对它说hello或者hi唤醒。

2.生命周期和钩子函数

本智能终端在运行时分为以下几个阶段，在不同的阶段可以调用不同的钩子函数实现流程定制化：

                                                             ___________________________________循环______________________________________
                                                             |                                                                           |
    |加载用户模型| ->  |加载上下文| -> |执行启动钩子| -> |监听用户/设备请求| -> |执行前置语义处理钩子| -> |处理请求| -> |执行后置语义处理钩子| ->|转发请求| -> |执行卸载钩子| -> |卸载|
                                                             |                                                                           |
                                                             |——————————————————————————————————循环——————————————————————————————————————|

从上图可以看出，一共有四个钩子函数，分别是 OnStartUpHook、 OnModelPreprocessHook 、 OnModelPostprocessHook 和 OnTearDown。可以通过以下方法编写钩子函数：

from iot import HooksGroup

@HooksGroup.on_start_up
def start_up(context, semantics_group):
    ...


@HooksGroup.on_tear_down
def tear_down(context, semantics_group):
    ...


@HooksGroup.on_model_preprocess
def model_preprocess(context, function_device_model):
    ...


@HooksGroup.on_model_postprocess
def model_postprocess(context, function_device_model):
    ...

钩子函数可以获取到执行阶段的上下文，包括各种处理器信息和配置信息。此外，启动钩子 和 卸载钩子 可以获取语义处理模型的集合而 前置语义处理钩子 和 后置语义处理钩子 可以获取到语义模型。

3.自定义设备和语音处理器

本智能终端的设备（麦克风和扬声器）都是用的Windows默认的，如果要用在树莓派或者其他环境则需要自定义设备，包括麦克风（Recorder）和扬声器（Player）。

首先实现对应的类：

from iot import RecorderFactory, Recorder

@RecorderFactory.set
class MyRecorder(Recorder):      # 继承Recorder类，并加上注解
    
    def awake(self) -> bool:          # 实现awake方法，这个方法必须是个阻塞的方法，返回True则开始录音，返回False则推出程序
        ...

    def record(self, time: int) -> str: # 录音，time为录音时长，返回录音后保存的路径
        ...

然后在 .env 文件中修改默认的麦克风设备:

DEVICE_REC=MyRecorder

同样，扬声器也是这样的步骤：

from iot import PlayerFactory, Player

@PlayerFactory.set
class MyPlayer(Player):

    def play(self, path: str) -> None:    # path为存储语音文本的txt文件路径
        ...

然后更改 .env 文件

DEVICE_PLY=MyPlayer

您也可以修改默认的语音处理器，包括语音识别（AsrProcessor）和语音合成（TtsProcessor），方法也是一样的。

from iot import AsrProcessorFactory, AsrProcessor
from typing import Optional


@AsrProcessorFactory.set
class MyAsrProcessor(AsrProcessor):

    def asr(self, path: str) -> Optional[str]: # path为音频文件（一般为wav）的路径，返回语音文字，如果为None则说明处理失败
        ...

.env 文件

ASR_PROCESSOR=MyAsrProcessor

语音合成可以这样修改：

from iot import TtsProcessorFactory, TtsProcessor


@TtsProcessorFactory.set
class MyTtsProcessor(TtsProcessor):

    def tts(self, text: str) -> str:  # text为需要被合成的文字，返回合成后的音频文件路径
        ...

.env 文件

TTS_PROCESSOR=MyTtsProcessor

相关资料

1. 如何实现智能设备

2. 智能家居设备交流消息的抓取

待实现功能

1.语音追问功能

2.音频等流媒体的播放

3.语义模型的order