准备工作
1. aku刷机前置教程
安装步骤
步骤一：安装系统依赖
apt-get update
apt-get install -y python3-pip python3-venv python3-pyaudio sox
步骤二：创建安装目录
mkdir -p /opt/ha
cd /opt/ha
步骤三：下载并安装 Wyoming Satellite
使用打包文件
将 ha.tar.gz 上传到设备并解压：
cd /opt/ha
tar -xzf ha.tar.gz
步骤四：创建 systemd 服务
创建 HA 语音服务文件：
cat > /usr/lib/systemd/system/ha.service << 'EOF'
[Unit]
Description=HA Voice Service
After=network.target display.service
Wants=display.service
[Service]
Type=simple
User=root
WorkingDirectory=/opt/ha/wyoming-satellite
ExecStart=/opt/ha/wyoming-satellite/.venv/bin/python3 -m wyoming_satellite \
--uri 'tcp://0.0.0.0:10700' \
--name 'aku57' \
--mic-command 'arecord -r 16000 -c 1 -f S16_LE -t raw' \
--snd-command 'aplay -r 22050 -c 1 -f S16_LE -t raw' \
--awake-wav 'sounds/awake.wav' \
--done-wav 'sounds/done.wav' \
--timer-finished-wav 'sounds/timer_finished.wav' \
--event-uri 'tcp://127.0.0.1:10500'
Restart=always
RestartSec=5
[Install]
WantedBy=multi-user.target
EOF
参数说明：
| 参数 |
说明 |
--uri |
卫星监听地址，HA 通过此端口连接 |
--name |
设备名称，会显示在 HA 中 |
--mic-command |
录音命令，16kHz 单声道 |
--snd-command |
播放命令，22050Hz 单声道 |
--event-uri |
事件发送地址，用于屏幕显示 |
步骤六：创建屏幕显示服务
此步骤为 AKU 设备专用，利用设备自带的屏幕显示对话内容。
创建显示处理脚本：
cat > /opt/ha/display_handler.py << 'EOF'
#!/usr/bin/env python3
import argparse, asyncio, logging, subprocess
from functools import partial
from wyoming.event import Event
from wyoming.server import AsyncEventHandler, AsyncServer
SHOW_TEXT = "/opt/aku/web/show_text"
PLAY_BMP = "/opt/aku/web/play_bmp_sequence"
EMOTIONS_DIR = "/opt/aku/web/emotions/emotion7"
WHITE, GREEN, YELLOW, CYAN, RED = "0xFFFF", "0x07E0", "0xFFE0", "0x07FF", "0xF800"
CHARS_PER_LINE = {10: 14, 12: 11, 14: 9}
LINES_PER_PAGE = {10: 8, 12: 6, 14: 5}
TIMEOUT_SECONDS = 30
_LOGGER = logging.getLogger()
def wrap_text(text, cpl):
lines = []
while text:
lines.append(text[:cpl])
text = text[cpl:]
return lines
class Display:
def __init__(self):
self.emoji_process = None
self.idle = True
self.loop = None
self._resume_handle = None
self._scroll_task = None
self._scrolling = False
self._timeout_handle = None
def set_loop(self, loop):
self.loop = loop
def stop_emoji(self):
subprocess.run(["killall", "-q", "play_bmp_sequence"], capture_output=True)
self.emoji_process = None
def start_emoji(self):
if self._scrolling:
return
self.stop_emoji()
self._cancel_timeout()
try:
self.emoji_process = subprocess.Popen([PLAY_BMP, "-d", "100", EMOTIONS_DIR], cwd="/opt/aku/web", stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
_LOGGER.info("Emoji started")
except Exception as e:
_LOGGER.error(f"Emoji error: {e}")
def _show_raw(self, text, size, color):
try:
subprocess.run([SHOW_TEXT, text, str(size), color, "1", "1"], timeout=3, capture_output=True)
except Exception as e:
_LOGGER.error(f"Show error: {e}")
def _cancel_timeout(self):
if self._timeout_handle:
self._timeout_handle.cancel()
self._timeout_handle = None
def _start_timeout(self):
self._cancel_timeout()
if self.loop:
self._timeout_handle = self.loop.call_later(TIMEOUT_SECONDS, self._on_timeout)
def _on_timeout(self):
_LOGGER.warning("Timeout! Forcing emoji restore")
self._timeout_handle = None
self.idle = True
self._scrolling = False
self.start_emoji()
def show(self, text, color=WHITE, size=12):
self.stop_emoji()
self._cancel_timeout()
if self._resume_handle:
self._resume_handle.cancel()
self._resume_handle = None
if self._scroll_task and not self._scroll_task.done():
self._scroll_task.cancel()
self._scrolling = False
cpl = CHARS_PER_LINE.get(size, 11)
lpp = LINES_PER_PAGE.get(size, 6)
lines = wrap_text(text, cpl)
if len(lines) <= lpp:
self._show_raw("\n".join(lines), size, color)
_LOGGER.info(f"Display({len(lines)}lines): {text[:30]}...")
else:
self._scrolling = True
self._scroll_task = asyncio.ensure_future(self._scroll_pages(lines, color, size, lpp))
async def _scroll_pages(self, lines, color, size, lpp):
try:
pages = []
for i in range(0, len(lines), lpp):
pages.append("\n".join(lines[i:i + lpp]))
_LOGGER.info(f"Scrolling {len(pages)} pages")
for i, page in enumerate(pages):
self._show_raw(page, size, color)
_LOGGER.info(f"Page[{i+1}/{len(pages)}]")
if i < len(pages) - 1:
await asyncio.sleep(3.0)
await asyncio.sleep(2.0)
except asyncio.CancelledError:
pass
except Exception as e:
_LOGGER.error(f"Scroll error: {e}")
finally:
self._scrolling = False
if self.idle:
self.start_emoji()
def set_idle(self, val):
self.idle = val
self._cancel_timeout()
if val and not self._scrolling:
self.resume_emoji_later(1.0)
def resume_emoji_later(self, delay=2.0):
if self._resume_handle:
self._resume_handle.cancel()
if self.loop:
self._resume_handle = self.loop.call_later(delay, self._maybe_resume)
def _maybe_resume(self):
self._resume_handle = None
if self.idle and not self._scrolling:
self.start_emoji()
display = Display()
class Handler(AsyncEventHandler):
def __init__(self, cli, *a, **k):
super().__init__(*a, **k)
self.conv = False
async def handle_event(self, ev):
t = ev.type
d = ev.data if hasattr(ev, 'data') else {}
_LOGGER.info(f"Event: {t}")
if t in ("detection", "wake-word-end"):
self.conv = True
display.idle = False
display._start_timeout()
display.show("在听...", CYAN, 20)
elif t == "transcript":
display.idle = False
display._start_timeout()
txt = d.get("text", "") if isinstance(d, dict) else ""
if txt:
display.show(f"你: {txt}", WHITE, 12)
elif t == "synthesize":
display.idle = False
display._start_timeout()
txt = ""
if hasattr(ev, 'data') and isinstance(ev.data, dict):
txt = ev.data.get("text", "")
display.show(f"AI: {txt}" if txt else "回复中...", GREEN if txt else YELLOW, 12 if txt else 16)
elif t == "played":
if self.conv:
self.conv = False
display.set_idle(True)
elif t == "error":
display.show("出错了", RED, 18)
self.conv = False
display.set_idle(True)
elif t == "satellite-connected":
display.idle = True
display.start_emoji()
return True
async def main():
p = argparse.ArgumentParser()
p.add_argument("--uri", default="tcp://127.0.0.1:10500")
p.add_argument("--debug", action="store_true")
a = p.parse_args()
logging.basicConfig(level=logging.DEBUG if a.debug else logging.INFO, format='%(asctime)s %(levelname)s: %(message)s')
display.set_loop(asyncio.get_event_loop())
_LOGGER.info("Starting...")
display.start_emoji()
srv = AsyncServer.from_uri(a.uri)
_LOGGER.info(f"Listening {a.uri}")
await srv.run(partial(Handler, a))
if __name__ == "__main__":
asyncio.run(main())
EOF
创建显示服务：
cat > /usr/lib/systemd/system/display.service << 'EOF'
[Unit]
Description=Wyoming Display Handler
After=network.target
Before=ha.service
[Service]
Type=simple
User=root
WorkingDirectory=/opt/ha
ExecStart=/opt/ha/wyoming-satellite/.venv/bin/python3 /opt/ha/display_handler.py --uri tcp://127.0.0.1:10500 --debug
Restart=always
RestartSec=3
[Install]
WantedBy=multi-user.target
EOF
禁用系统自带的开机动画（防止冲突）：
systemctl stop sysboot
systemctl disable sysboot
步骤七：启用并启动服务
# 重新加载 systemd
systemctl daemon-reload
# 启用服务（开机自启）
systemctl enable ha
systemctl enable display
# 启动服务
systemctl start display
systemctl start ha
# 查看状态
systemctl status ha display
在 Home Assistant 中添加设备
- 打开 Home Assistant
- 进入 设置 → 设备与服务 → 添加集成
- 搜索 Wyoming Protocol
- 输入设备 IP 和端口（默认 10700）
- 完成添加
设备会自动出现在语音助手配置中。
相关链接
免责声明
⚠️ 重要提示：
- 本教程仅在作者的 AKU R8-AkuBox 设备上进行了测试，未经过多设备验证和长时间稳定性测试
- 安装过程涉及系统服务修改和软件安装，操作前请务必备份重要数据
- 不同批次的 AKU 设备可能存在硬件或系统差异，安装结果可能不同
- 作者不对因使用本教程导致的任何设备故障、数据丢失或其他问题承担责任
- 建议在操作前充分了解 Linux 系统和 systemd 服务管理知识
- 如遇到问题，请先查看日志排查，必要时可恢复出厂设置
发表于