内容介绍

注：以上视频中文字为可商用，没有版权问题

基于 i.MX93 的智能家居语音中控终端设计

一、项目介绍与创意说明

随着智能家居的快速发展，家庭中的设备逐渐从“单点控制”向“集中控制”和“智能交互”演进。其中，语音作为最自然的人机交互方式之一，正在成为智能家居系统的重要入口。然而，目前市场上的智能家居中控设备多依赖云端语音服务，存在隐私风险、延迟较高以及网络依赖性强等问题。

本项目以“智能家居中控”为设计方向，基于 NXP i.MX93 处理器平台，结合高性能音频编解码芯片 AK4619，设计并实现了一套低延迟、本地化处理能力强的智能语音通信终端。该终端具备实时音频通信能力，可作为家庭中的语音交互节点，实现家庭成员之间的远程对讲、紧急呼叫以及语音交互入口功能。

项目的核心创意在于：

将“实时音频通信”能力引入智能家居中控系统
构建一个低延迟、本地优先的语音交互节点
结合嵌入式 Linux + baresip 音频框架，实现灵活的音频管理
提供可扩展接口，未来可接入语音识别、AI分析等功能

该系统不仅可以作为家庭语音中控终端，还可以扩展为家庭安全系统中的语音监测节点，具有较强的应用前景。

二、硬件系统介绍

硬件设计方案，以NXP iMX93 FRDM开发板为核心，设计了音频扩展板。

i.MX93 FRDM开发板通过I2S总线与AK4619音频编解码芯片进行数字音频数据交互，AK4619完成模拟音频与数字音频之间的转换。输入部分由两路麦克风构成，用于采集现场语音信号。输出部分由两片HT6872功放芯片组成双声道立体声放大电路，驱动扬声器输出。另外通过HDMI接口实现系统运行状态和通话界面的图形化显示。硬件架构图如下：

1. 主控平台：i.MX93 FRDM

i.MX93 是 NXP 推出的一款面向边缘计算与低功耗应用的处理器，具备以下特点：

ARM Cortex-A55 内核
支持 Linux 系统（Yocto）
丰富的外设接口（I2S、MIPI、USB等）
适合音视频及边缘智能应用

该平台为整个系统提供计算能力和系统调度能力。

2. 音频子系统：AK4619

扩展板上集成了 AK4619 高性能音频编解码芯片，主要功能包括：

多通道音频输入输出
高信噪比音频采集
支持 I2S 接口与 i.MX93 连接

外围设计包括：

麦克风输入电路（用于语音采集）
功放电路（驱动外接扬声器）

该部分是实现实时语音通信的核心硬件基础。

3. 显示子系统：HDMI 显示屏

系统通过 HDMI 接口扩展连接一块 1280*720 分辨率的 LCD 显示屏，主要用于：

显示来电/去电信息
展示系统状态

4. 自研扩展板设计

使用 Cadence 设计工具完成扩展板设计，主要功能模块包括：

音频编解码电路（AK4619）

麦克风与功放接口（LMV321A）

使用Allegro绘制的PCB

TOP视图

BOTTOM视图

焊接好的扩展板

三、系统方案框图与设计思路

1. 系统总体架构

系统整体架构如下：

2. 设计思路

系统设计遵循以下原则：

（1）模块化设计

音频处理、通信、UI、控制逻辑相互解耦
使用 baresip 模块化机制统一管理音频流

（2）低延迟优先

使用 Baresip 实现 SIP 通信
启用 WebRTC AEC 实现回声消除
优化音频 buffer 参数

（3）本地优先处理

音频采集与处理在本地完成
降低对云服务依赖

（4）可扩展性

预留 AI 语音识别接口
可扩展为智能家居控制中心

四、软件流程与关键实现

1. 软件架构

系统软件主要由以下几个部分组成：

Baresip：负责 SIP 通信
ALSA：音频路由与管理
LVGL UI：显示界面
Python 主程序：业务控制逻辑

2. 软件流程

整体流程如下：

自定义模块lvlg_ui

基于baresip定制的模块lvlg_ui截图

3. 关键技术点

（1）回声消除（AEC）

通过 baresip 集成 WebRTC AEC，实现：

消除扬声器回放声音被麦克风拾取的问题
提高通话清晰度

baresip相关配置如下：

audio_player            alsa,default
audio_source            alsa,default
audio_alert             alsa,default
# webrtc aec module
module                  webrtc_aec.so

（2）UI模块扩展

在 Baresip 中扩展 LVGL UI 模块，实现：

来电界面显示

呼叫状态显示
基本交互功能

baresip相关配置如下

video_display           lvgl_ui
module_app              lvgl_ui.so
# HDMI card0
lvgl_ui_device          /dev/dri/card0

（3）Python 控制逻辑

Python 程序负责：

启动/管理各服务
按键检测
控制呼叫流程
与 UI 交互

主控制程序main.py如下

import queue
import subprocess
import time
import os
import json

from app_db_config import AppDbConfig
from key_monitor import KeyMonitor
from baresip_client import BaresipClient
from mq import MsgQueue
from shell_executor import ShellExecutor
from sys_settings import SysSettings
from udp_cmd import UdpCommandServer
from app_utils import AppUtils
from multicast_sender import client as multicast_client


# Global event queue for inter-thread communication
event_queue = queue.Queue()


class MainApp:
    def __init__(self):
        self.baresip = None
        self.key_monitor = None
        self.user_cmd = None
        self.call_in_progress = False
        self.incoming_call_waiting = False
        self.ss = SysSettings()
        self.executor = ShellExecutor()

    def reload_config(self):
        """
        Reload the application configuration from the database.
        """
        c = AppDbConfig()
        c.init_db_config()

        # Update system settings based on new config
        self.ss.set_mic_level(c.micfil_volume)
        self.ss.set_ak4619_level(c.ak4619_volume)
        # self.ss.pw_update_volume(c.pw_source_volume, c.pw_sink_volume)

        return "OK"

    def pw_reload_links(self):
        self.ss.setup_pipewire_links()

    def play_wav_file(self, wav_path):
        if not os.path.exists(wav_path):
            print(f"File {wav_path} does not exist.")
            return False
        # alsa_output.platform-sound.stereo-fallback
        self.executor.start(f"aplay {wav_path}")
        return True

    def call_test(self):
        # multicast test: find PCUV device IP
        byteList = bytearray(16)
        byteList[0] = 0xFF
        byteList[7] = 0x01
        byteList[15] = 0xFE
        multicast_client.send(byteList)
        replies = multicast_client.find_device_ip_by_type("FRDM")
        print("replies", replies)
        if len(replies) > 0:
            frdm_ip = replies[0]
            print(f"Found FRDM device at {frdm_ip}, making SIP call...")
            client = f"sip:passenger_pei@{frdm_ip}"
            self.call_in_progress = True
            self.baresip.make_call(client, with_video=True)
        else:
            print("No FRDM device found.")

    def handle_app_cmd(self, cmd: str, data: dict):
        # {'cmd': 'sip-start', 'data': {'client': 'sip:passenger_pei@192.168.123.77'}}
        print(f"app-cmd: {data}")
        try:
            c = data['cmd']
            d = data['data']
            if c == "play":
                wavFile = d.get('file', '1.wav')
                if self.executor.get_pending_count() < 1:
                    # target = "alsa_output.platform-sound.stereo-fallback"
                    target = '"Echo Cancellation Sink"'
                    wav = f"./assets/{wavFile}"
                    if not os.path.exists(wav):
                        return json.dumps({"resp": c, "message": "fail", "info": f"{wavFile} not exist"})
                    # looping audio playback
                    # self.executor.start(f"while true; do pw-cat --playback --target={target} {wav}; done")
                    self.executor.start(f"while true; do aplay {wav}; done")
            elif c == "stop":
                self.executor.stop()
            elif c == "sip-start":
                client = d['client']
                with_video = d.get('video', False)
                self.baresip.make_call(client, with_video=with_video)
            elif c == "sip-stop":
                self.baresip.hangup_call()
            elif c == "sip-mute":
                self.baresip.call_mute()
            elif c == "key-press":
                # simulate key press event
                key = 'KEY_1'
                if d:
                    key = d
                msg = {'event': key, 'state': 'click'}
                event_queue.put(msg)
            else:
                print(f"unhandle app-cmd {cmd} -> {data}")
                return json.dumps({"resp": c, "message": "fail", "info": f"unhandle {cmd}"})

            return json.dumps({"resp": c, "message": "OK"})
        except Exception as e:
            print(f"handle_app_cmd err: {e}")

        return json.dumps({"message": "fail"})

    def start(self):
        try:
            AppUtils.sync_db_if_newer('./assets/app_data.db', '/mnt/param/app_data.db')
            AppUtils.sync_device_db('./assets/device.db', '/mnt/param/device.db')
            try:
                c = AppDbConfig('/mnt/param/app_data.db')
            except Exception as e:
                AppUtils.sync_db_if_newer('./assets/app_data.db', '/mnt/param/app_data.db', force_copy=True)
                c = AppDbConfig('/mnt/param/app_data.db')

            print(f"DB VERSION: {c.version}")

            # system configuration settings
            self.reload_config()
            # self.pw_reload_links()

            self.play_wav_file("./assets/booting.wav")

            # check network online
            print(f"[MAIN] Checking network on interface {c.netinterface}...")
            while True:
                ip = AppUtils.get_ip_address(c.netinterface)
                if ip and not ip.startswith("169.254"):
                    print(f"[MAIN] Network is online: {ip}")
                    break
                time.sleep(1)

            self.user_cmd = UdpCommandServer()
            self.user_cmd.start()

            self.mq = MsgQueue()
            self.mq.start()

            self.baresip = BaresipClient(event_queue)
            self.baresip.start()

            self.key_monitor = KeyMonitor(c.keypath, event_queue)
            self.key_monitor.start()

            # add json command for database
            self.user_cmd.register_json_command("update-sip", c.command)
            self.user_cmd.register_json_command("get-config", c.command)
            self.user_cmd.register_json_command("reload-config", lambda obj, addr: self.reload_config())
            self.mq.register_command("reload-config", lambda cmd, data: self.reload_config())
            self.mq.register_command("app-cmd", self.handle_app_cmd)

            print("[MAIN] FRDM Running ...")

            while True:
                e = event_queue.get()
                event = e.get('event', None)

                if event == 'BTN_1':            # BTN_1
                    print("[MAIN] BTN_1 pressed")
                    self.call_in_progress = False
                    self.incoming_call_waiting = False
                    self.baresip.hangup_call()
                elif event == 'BTN_2':          # BTN_2
                    print("[MAIN] BTN_2 pressed")
                    if self.incoming_call_waiting:
                        print("[MAIN] Answer incoming SIP call")
                        self.incoming_call_waiting = False
                        self.call_in_progress = True
                        self.baresip.answer_call(with_video=True)
                    elif self.call_in_progress:
                        print("[MAIN] Hangup current SIP call")
                        self.baresip.hangup_call()
                    else:
                        self.call_test()
                elif event == 'sipcall':        # MQTT new sip call
                    user = e.get('user', None)
                    if user:
                        print(f"[MAIN] SIP call to {user}")
                        self.baresip.make_call(user, with_video=True)
                elif event == 'sip':            # SIP MESSAGE
                    evtype = e.get('evtype', None)
                    if evtype == 'CALL_CLOSED':
                        self.call_in_progress = False
                        self.incoming_call_waiting = False
                    elif evtype == 'CALL_ESTABLISHED':
                        self.call_in_progress = True
                    elif evtype == 'CALL_INCOMING':
                        self.incoming_call_waiting = True
                        print("[MAIN] Incoming SIP call, press BTN_2 to answer")
                else:
                    print(f"[MAIN] Unknown event: {e}")

        except KeyboardInterrupt:
            self.mq.stop()
            self.baresip.stop()
            self.key_monitor.stop()
            self.user_cmd.stop()
            self.mq.join()
            self.key_monitor.join()
            self.user_cmd.join()


if __name__ == "__main__":
    app = MainApp()
    app.start()