Add ReSpeaker XVF3800 USB 4-Mic Array support + startup preflight + optional volume sync
Add ReSpeaker XVF3800 USB 4-Mic Array support + startup preflight + optional volume sync
This commit is contained in:
imonlinux
@@ -10,6 +10,8 @@ Supports announcments, start/continue conversation, and timers.
|
||||
|
||||
See [the tutorial](docs/linux-voice-assistant-2mic-install.md) to build a satellite using a [Raspberry Pi Zero 2 W](https://www.raspberrypi.com/products/raspberry-pi-zero-2-w/) and a [ReSpeaker 2Mic HAT](https://wiki.keyestudio.com/Ks0314_keyestudio_ReSpeaker_2-Mic_Pi_HAT_V1.0).
|
||||
|
||||
The LVA now supports the ReSpeaker XVF3800 4 Mic array via the USB interface. See [this tutorial](docs/linux-voice-assistant-xvf3800.md).
|
||||
|
||||
Want to run the satellite on a Linux desktop using a simple tray client? See [this tutorial](docs/lva-desktop.md).
|
||||
|
||||
### What's working:
|
||||
|
||||
@@ -0,0 +1,5 @@
|
||||
# XVF3800 permissions
|
||||
SUBSYSTEM=="usb", ATTR{idVendor}=="2886", ATTR{idProduct}=="001a", MODE="0660", GROUP="plugdev", TAG+="uaccess"
|
||||
|
||||
# XVF3800 power policy
|
||||
ACTION=="add", SUBSYSTEM=="usb", ATTR{idVendor}=="2886", ATTR{idProduct}=="001a", TEST=="power/control", ATTR{power/control}="on"
|
||||
@@ -24,7 +24,7 @@ This guide reproduces a working setup of the **linux-voice-assistant** project w
|
||||
sudo apt update
|
||||
sudo apt upgrade
|
||||
sudo apt install build-essential git \
|
||||
libmpv-dev mpv python3-dev
|
||||
libmpv-dev mpv python3-dev python3-venv
|
||||
sudo reboot
|
||||
```
|
||||
|
||||
@@ -608,7 +608,7 @@ alexa_v0.1.tflite -> Alexa
|
||||
hey_jarvis_v0.1.tflite -> Hey Jarvis
|
||||
hey_mycroft_v0.1.tflite -> Hey Mycroft
|
||||
hey_rhasspy_v0.1.tflite -> Hey Rhasspy
|
||||
ok_nabu_v0.1.tflite -> OK Nabo **(I had to say OK Nobu)**
|
||||
ok_nabu_v0.1.tflite -> OK Nabu **(I had to say OK Nobu)**
|
||||
```
|
||||
|
||||
Additional community provided OWW models available from this repository:
|
||||
|
||||
@@ -0,0 +1,223 @@
|
||||
# XVF3800 Mute Button & USB Control Integration
|
||||
|
||||
This document explains how the **ReSpeaker XVF3800 4‑Mic USB Array** integrates its
|
||||
built‑in mute button and red mute LED with the **Linux Voice Assistant (LVA)**.
|
||||
|
||||
The goal is:
|
||||
- Pressing the XVF3800’s hardware mute button updates the LVA mic state.
|
||||
- Muting/unmuting the LVA (via Home Assistant, MQTT, or other inputs) updates
|
||||
the XVF3800’s mic mute circuit and red LED.
|
||||
- No extra binaries (like the vendor `xvf_host`) are required — LVA talks to
|
||||
the XVF3800 directly via **USB control transfers** using `pyusb`.
|
||||
|
||||
---
|
||||
|
||||
## 1. How it Works (High‑Level)
|
||||
|
||||
The XVF3800 exposes a vendor control protocol over USB that can read and write
|
||||
internal GPIO pins (GPO). One of these pins drives both:
|
||||
|
||||
- the microphone mute circuit
|
||||
- the red mute LED on the XVF3800 front panel
|
||||
|
||||
LVA’s `XVF3800ButtonController` does three things:
|
||||
|
||||
1. **Reads hardware mute state periodically**
|
||||
It polls the XVF3800’s GPO state (pin X0D30) in a background thread using
|
||||
`pyusb`. If the value changes, LVA treats that as “the user pressed the
|
||||
XVF3800 mute button” and publishes a `set_mic_mute` event.
|
||||
|
||||
2. **Mirrors LVA mute → XVF3800**
|
||||
It subscribes to `mic_muted` / `mic_unmuted` events. Whenever LVA’s mute
|
||||
state changes (for example, via MQTT, HA UI, or a GPIO button), the
|
||||
controller writes the new state back to X0D30 so that the mic mute circuit
|
||||
and red LED stay in sync.
|
||||
|
||||
3. **Initial sync on startup**
|
||||
On first successful read, the controller treats the *hardware* mute state
|
||||
as the source of truth and publishes `set_mic_mute` so LVA starts aligned
|
||||
with whatever the XVF3800 is currently doing.
|
||||
|
||||
No PulseAudio/PipeWire configuration is required for this feature; it only
|
||||
depends on **USB access to the XVF3800** and the `pyusb` library.
|
||||
|
||||
---
|
||||
|
||||
## 2. Prerequisites
|
||||
|
||||
- You are running a branch of **linux‑voice‑assistant** that includes:
|
||||
- `linux_voice_assistant/xvf3800_button_controller.py`
|
||||
- The updated `__main__.py` and `config.py` wiring for `XVF3800ButtonController`
|
||||
- The ReSpeaker XVF3800 is connected over USB and appears in `lsusb` as:
|
||||
- `2886:001a Seeed Technology Co., Ltd. reSpeaker XVF3800 4-Mic Array`
|
||||
- The LVA `.venv` includes `pyusb` (already present in `pyproject.toml`):
|
||||
```bash
|
||||
source .venv/bin/activate
|
||||
pip install pyusb
|
||||
```
|
||||
|
||||
On most systems, user‑mode processes can talk to the XVF3800 over USB without
|
||||
`sudo` once the device is enumerated. If you see `USBError: [Errno 13] Access denied`,
|
||||
you may need a udev rule — that’s outside this document but worth noting.
|
||||
|
||||
---
|
||||
|
||||
## 3. Enabling the XVF3800 Mute Integration
|
||||
|
||||
The integration is controlled via the `button` section of `config.json`.
|
||||
Two “modes” are supported:
|
||||
|
||||
- `"gpio"` – legacy Raspberry Pi GPIO button (ReSpeaker 2‑Mic HAT, etc.)
|
||||
- `"xvf3800"` – USB‑based mute integration for the ReSpeaker XVF3800
|
||||
|
||||
### 3.1 Example `config.json` for XVF3800
|
||||
|
||||
Edit your LVA configuration file (usually `linux_voice_assistant/config.json`)
|
||||
and add or adjust the `button` block like this:
|
||||
|
||||
```jsonc
|
||||
{
|
||||
"app": {
|
||||
"name": "Linux Voice Assistant"
|
||||
},
|
||||
|
||||
// ... other sections (audio, wake_word, esphome, led, mqtt, etc.) ...
|
||||
|
||||
"button": {
|
||||
"enabled": true,
|
||||
"mode": "xvf3800",
|
||||
// Optional: how often to poll the XVF3800 GPO state (seconds)
|
||||
"poll_interval_seconds": 0.05
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Notes:
|
||||
|
||||
- `"enabled": true` turns on button support in general.
|
||||
- `"mode": "xvf3800"` switches from the GPIO‑based button controller to the
|
||||
USB‑based XVF3800 controller.
|
||||
- `poll_interval_seconds` is optional and defaults to `0.01` if omitted. Values
|
||||
between 0.02 and 0.1 are typically fine.
|
||||
|
||||
The GPIO‑specific fields (`pin`, `long_press_seconds`) are ignored in
|
||||
`xvf3800` mode — the XVF3800’s onboard mute button is treated as a
|
||||
**dedicated mute toggle only**, not a wake/stop button.
|
||||
|
||||
---
|
||||
|
||||
## 4. Runtime Behaviour
|
||||
|
||||
Once enabled and the LVA service is restarted, the mute integration behaves as follows:
|
||||
|
||||
### 4.1 Pressing the XVF3800 Mute Button
|
||||
|
||||
1. The XVF3800 firmware changes the GPO state of pin X0D30 and lights or clears
|
||||
the red mute LED.
|
||||
2. The `XVF3800ButtonController` sees the changed GPO values the next time it
|
||||
polls USB.
|
||||
3. It publishes a `set_mic_mute` event on the LVA `EventBus` with the new state.
|
||||
4. The `MicMuteHandler`:
|
||||
- Updates `ServerState.mic_muted` and `mic_muted_event` (pausing/resuming the
|
||||
audio capture thread).
|
||||
- Publishes `mic_muted` / `mic_unmuted` events used by other components
|
||||
(LED controller, MQTT, etc.).
|
||||
- Notifies the MQTT controller so the HA “Mute Microphone” entity stays in sync.
|
||||
|
||||
From the user’s perspective:
|
||||
|
||||
- The red LED on the XVF3800 tracks mic mute.
|
||||
- The HA “Mute Microphone” switch also tracks the hardware button.
|
||||
- The audio engine actually stops capturing when muted.
|
||||
|
||||
### 4.2 Toggling Mute from Home Assistant / MQTT / Other Inputs
|
||||
|
||||
1. Home Assistant toggles the LVA “Mute Microphone” switch (or another input
|
||||
publishes a `set_mic_mute` event).
|
||||
2. The `MicMuteHandler` applies the new mute state, as usual.
|
||||
3. It emits a `mic_muted` or `mic_unmuted` event.
|
||||
4. The `XVF3800ButtonController` receives this event and writes the new state
|
||||
back to X0D30 via USB.
|
||||
5. The XVF3800’s mic mute circuit and red LED update to match LVA.
|
||||
|
||||
This keeps **hardware**, **LVA state**, and **Home Assistant UI** aligned.
|
||||
|
||||
---
|
||||
|
||||
## 5. Coexistence with Other Buttons
|
||||
|
||||
The XVF3800 mute integration is intentionally narrow in scope:
|
||||
|
||||
- It does **not** implement short vs. long press behaviour.
|
||||
- It does **not** trigger wake‑word, stop‑word, or playback controls.
|
||||
- It is purely “mute on / mute off” with bidirectional sync.
|
||||
|
||||
If you also have a GPIO button (e.g. a ReSpeaker 2‑Mic HAT momentary switch)
|
||||
wired for wake/stop behaviour, you can:
|
||||
|
||||
- Leave `button.mode: "gpio"` and **not** use the XVF3800 mute integration, or
|
||||
- Move wake/stop to the GPIO button controller and leave the XVF3800 button
|
||||
as mute‑only by setting `mode: "xvf3800"` and handling wake/stop elsewhere.
|
||||
|
||||
At the moment, `button.mode` is **single‑valued** — it selects either the
|
||||
GPIO controller or the XVF3800 controller, not both at once. If you ever want
|
||||
to support both concurrently, that would be a separate enhancement (e.g. a
|
||||
list of button backends).
|
||||
|
||||
---
|
||||
|
||||
## 6. Troubleshooting
|
||||
|
||||
### 6.1 No USB Access / Permission Errors
|
||||
|
||||
Symptoms in the logs (journalctl or console):
|
||||
|
||||
```text
|
||||
Failed to initialize XVF3800 USB client; mute button integration will be disabled
|
||||
USBError: [Errno 13] Access denied (insufficient permissions)
|
||||
```
|
||||
|
||||
Check:
|
||||
|
||||
- `lsusb` shows `2886:001a` while LVA is running.
|
||||
- The LVA service is running as the user who owns the `.venv` and has
|
||||
access to USB devices.
|
||||
- If necessary, add a udev rule that grants access to `idVendor=0x2886`
|
||||
and `idProduct=0x001a` for your LVA user, then reload udev and
|
||||
replug the device.
|
||||
|
||||
The controller will keep retrying USB initialization; once permissions are
|
||||
fixed, it should connect without restarting LVA.
|
||||
|
||||
### 6.2 Mute Button Works but HA Entity Doesn’t Update
|
||||
|
||||
- Verify MQTT is enabled and configured correctly in `config.json`.
|
||||
- Confirm that the HA “Mute Microphone” entity exists and belongs to the
|
||||
correct LVA device.
|
||||
- Check logs for `MicMuteHandler` and `MqttController` messages when pressing
|
||||
the XVF3800 mute button.
|
||||
|
||||
### 6.3 HA Mute Switch Works but XVF3800 LED Doesn’t Change
|
||||
|
||||
- Ensure `button.enabled` is `true` and `"mode": "xvf3800"` in `config.json`.
|
||||
- Look for log lines like:
|
||||
- `Initializing XVF3800ButtonController (mode=xvf3800)`
|
||||
- `Connected to ReSpeaker XVF3800 for mute control`
|
||||
- If you see only GPIO logs, confirm that the branch you’re running includes
|
||||
`xvf3800_button_controller.py` and the updated `__main__.py` wiring.
|
||||
|
||||
---
|
||||
|
||||
## 7. Summary
|
||||
|
||||
With `mode: "xvf3800"`, the Linux Voice Assistant treats the ReSpeaker
|
||||
XVF3800’s onboard mute button and red LED as a **first‑class mute control**:
|
||||
|
||||
- Pressing the XVF3800 mute button updates LVA and Home Assistant.
|
||||
- Muting via Home Assistant or other inputs updates the XVF3800 hardware.
|
||||
- The audio engine, LEDs, and MQTT entities stay in sync with the device’s
|
||||
actual hardware mute state.
|
||||
|
||||
This integration is fully optional and coexists with existing audio and LED
|
||||
behaviour — it simply adds a reliable, hardware‑based mute path for the
|
||||
XVF3800‑based satellites.
|
||||
@@ -0,0 +1,224 @@
|
||||
# Linux Voice Assistant with ReSpeaker XVF3800 4‑Mic Array — Installation & Configuration Guide
|
||||
|
||||
> Drafted with ChatGPT based on existing `linux-voice-assistant-2mic-install.md` and community test results from the XVF3800 integration Discussion.
|
||||
|
||||
> Special thanks to @JenniferHatches and @cknic for making the recommendation and doing so much work already for support of the reSpeaker XVF3800. (https://github.com/cknic/linux-voice-assistant) The addition of XVF3800 support would never have happened without their troubleshooting of the code and fantastic suggestions when working through bugs.
|
||||
|
||||
This guide shows how to use the **Seeed ReSpeaker XVF3800 USB 4‑Mic Array** with the **Linux Voice Assistant** (LVA) project on Raspberry Pi or other Linux hosts.
|
||||
|
||||
Unlike the ReSpeaker 2‑Mic HAT, the XVF3800 is a **USB Audio Class 2.0** device: it acts as a “smart microphone + speaker” front end with built‑in **AEC, beamforming, noise suppression, AGC, and echo‑canceling speakerphone** processing.
|
||||
|
||||
No custom kernel drivers are required. Plug it in, setup the device in LVA’s `config.json`, and it will “just work” as a high‑quality mic and speaker (if connected).
|
||||
|
||||
---
|
||||
|
||||
## Prerequisites
|
||||
|
||||
- A Linux host (e.g., Raspberry Pi 4/5) running one of:
|
||||
- Raspberry Pi OS (Bookworm or Trixie)
|
||||
- Debian 13 (trixie) or similar
|
||||
- Other modern Linux with PipeWire or PulseAudio
|
||||
- Python 3.11+ recommended
|
||||
- **ReSpeaker XVF3800 4‑Mic Array**, connected via its **XMOS USB‑C port**
|
||||
- SSH access to your LVA instance
|
||||
- Network access to the Home Assistant for registration
|
||||
|
||||
## Recommended Configuration
|
||||
|
||||
- To use the advanced LED effects, upgrade to the latest XVF3800 firmware (respeaker_xvf3800_usb_dfu_firmware_v2.0.7.bin as of this writing).
|
||||
-- https://github.com/respeaker/reSpeaker_XVF3800_USB_4MIC_ARRAY/blob/master/xmos_firmwares/dfu_guide.md
|
||||
- Tested and works on a Raspberry Pi Zero 2W, but I don't recommend attempting to flash the firmware with this device.
|
||||
- PipeWire (or PulseAudio) installed and tested. I haven't tested the XVF3800 with ALSA.
|
||||
|
||||
---
|
||||
|
||||
## 1. Install system packages and Pipewire (Recommended)
|
||||
|
||||
```bash
|
||||
sudo apt update
|
||||
sudo apt upgrade
|
||||
sudo apt install build-essential git \
|
||||
libmpv-dev mpv python3-dev python3-venv \
|
||||
pulseaudio-utils pipewire wireplumber \
|
||||
pipewire-audio libspa-0.2-modules libusb-1.0-0 \
|
||||
dbus-user-session
|
||||
sudo reboot
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 2. Get the code (XVF3800 branch)
|
||||
|
||||
If you don’t already have the project cloned:
|
||||
|
||||
```bash
|
||||
cd ~
|
||||
git clone https://github.com/imonlinux/linux-voice-assistant.git
|
||||
cd linux-voice-assistant
|
||||
```
|
||||
|
||||
If you already have the repository, you can:
|
||||
|
||||
```bash
|
||||
cp ~/linux-voice-assistant/linux_voice_assistant/config.json ~/config.json
|
||||
cd ~/linux-voice-assistant
|
||||
git pull
|
||||
cp ~/config.json ~/linux-voice-assistant/linux_voice_assistant/config.json
|
||||
```
|
||||
|
||||
Implement the UDEV rule to prevent the XVF3800 from taking naps and to allow user access to the device
|
||||
|
||||
```bash
|
||||
sudo cp ~/linux-voice-assistant/XVF3800/99-respeaker-xvf3800.rules /etc/udev/rules.d/
|
||||
sudo udevadm control --reload-rules
|
||||
sudo udevadm trigger
|
||||
```
|
||||
|
||||
|
||||
Add user to the plugdev group for access to the XVF3800 and reboot
|
||||
|
||||
> Tip: If you use a different username than `pi`, adjust user accordingly.
|
||||
|
||||
```bash
|
||||
sudo usermod -aG plugdev pi
|
||||
sudo reboot
|
||||
```
|
||||
|
||||
Set up the project (creates a virtualenv, installs dependencies, etc.):
|
||||
|
||||
```bash
|
||||
cd ~/linux-voice-assistant
|
||||
script/setup
|
||||
```
|
||||
|
||||
1. Connect the XVF3800 to your Pi/host using the **XMOS USB‑C port** (the one next to the 3.5 mm audio jack).
|
||||
2. If you want local audio output, connect the XVF3800’s headphone jack or JST-PH 2.0mm 2 Pin to an external speaker.
|
||||
|
||||
---
|
||||
|
||||
## 3. Update the config.json file for XVF3800 support and options
|
||||
|
||||
Edit the config.json file:
|
||||
|
||||
```bash
|
||||
nano ~/linux-voice-assistant/linux_voice_assistant/config.json
|
||||
```
|
||||
|
||||
Add or update the following sections (Recommended settings):
|
||||
|
||||
> Tip: Be sure to change the name if you have more than one LVA!
|
||||
|
||||
```json
|
||||
{
|
||||
"app": {
|
||||
"name": "Linux Voice Assistant"
|
||||
},
|
||||
"led": {
|
||||
"enabled": true,
|
||||
"led_type": "xvf3800",
|
||||
"interface": "usb"
|
||||
},
|
||||
"button": {
|
||||
"enabled": true,
|
||||
"mode": "xvf3800",
|
||||
"poll_interval_seconds": 0.15
|
||||
},
|
||||
"mqtt": {
|
||||
"host": "192.168.1.2",
|
||||
"port": 1883,
|
||||
"username": "mqtt_server",
|
||||
"password": "mqtt_password"
|
||||
},
|
||||
"audio": {
|
||||
"volume_sync": true,
|
||||
"max_volume_percent": 150
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
> Take a look at ~/linux-voice-assistant/linux_voice_assistant/config.json.example for details on these settings as well as all available options.
|
||||
|
||||
---
|
||||
|
||||
## 4. Start LVA as a user-mode service
|
||||
|
||||
The service setup is identical to other configurations (2‑Mic, etc.). Assuming you want
|
||||
LVA to run as a user‑mode systemd service:
|
||||
|
||||
Enable linger for your user (e.g. `pi`)
|
||||
|
||||
```bash
|
||||
sudo loginctl enable-linger pi
|
||||
```
|
||||
|
||||
Install the systemd user service
|
||||
|
||||
> Tip: If you use a different username than `pi`, adjust user accordingly.
|
||||
|
||||
```bash
|
||||
mkdir -p ~/.config/systemd/user
|
||||
cp ~/linux-voice-assistant/service/linux-voice-assistant_xvf3800.service ~/.config/systemd/user/linux-voice-assistant.service
|
||||
```
|
||||
|
||||
Enable & start
|
||||
|
||||
```bash
|
||||
systemctl --user daemon-reload
|
||||
systemctl --user enable --now linux-voice-assistant.service
|
||||
```
|
||||
|
||||
> Note: The LVA initiate a XVF3800 reboot at start, this is needed to resolve a "mic not detecting audio" issue. As a result, the LVA will take a little longer to start up and be ready.
|
||||
|
||||
Check status:
|
||||
|
||||
```bash
|
||||
systemctl --user status linux-voice-assistant --no-pager -l
|
||||
```
|
||||
|
||||
Look for log lines indicating:
|
||||
|
||||
- LVA connected to Home Assistant
|
||||
- Wake‑word engine initialized
|
||||
- Satellite streaming started on wake word
|
||||
|
||||
---
|
||||
|
||||
## 5. Acoustic Echo Cancellation (AEC)
|
||||
|
||||
The XVF3800 already implements **hardware AEC / echo‑canceling speakerphone** processing.
|
||||
|
||||
---
|
||||
|
||||
## 6. Troubleshooting & debug tips
|
||||
|
||||
### Check that LVA sees the XVF3800
|
||||
|
||||
Make sure the XVF3800 is available to the LVA code:
|
||||
|
||||
```bash
|
||||
cd ~/linux-voice-assistant
|
||||
script/run --list-input-devices
|
||||
script/run --list-output-devices
|
||||
```
|
||||
|
||||
If `reSpeaker XVF3800 4-Mic Array Analog Stereo` doesn’t appear:
|
||||
|
||||
- Check `arecord -l` / `aplay -l` again.
|
||||
- Ensure the XVF3800 is connected to the XMOS USB‑C port and powered.
|
||||
- Check `lsusb` for a device with `2886:001a`:
|
||||
|
||||
```bash
|
||||
lsusb | grep -i 2886
|
||||
```
|
||||
|
||||
### Tail the LVA logs
|
||||
|
||||
```bash
|
||||
journalctl --user -u linux-voice-assistant.service -f
|
||||
```
|
||||
|
||||
Look for:
|
||||
|
||||
- “Audio engine started.”
|
||||
- “Opening audio input device: reSpeaker XVF3800 4-Mic Array Analog Stereo”
|
||||
- Wake‑word detection lines (MicroWakeWord / OpenWakeWord)
|
||||
@@ -0,0 +1,45 @@
|
||||
# XVF3800 Legacy LED Effects Mapping for Linux Voice Assistant
|
||||
|
||||
This document maps the **legacy ReSpeaker XVF3800 LED_EFFECT modes** to the higher-level LED effects used by the **linux-voice-assistant (LVA)**.
|
||||
|
||||
## XVF3800 Legacy LED_EFFECT Values
|
||||
|
||||
From the Seeed/XMOS documentation:
|
||||
|
||||
- `LED_EFFECT 0` → **off**
|
||||
- `LED_EFFECT 1` → **breath**
|
||||
- `LED_EFFECT 2` → **rainbow**
|
||||
- `LED_EFFECT 3` → **single color**
|
||||
- `LED_EFFECT 4` → **DoA mode**
|
||||
|
||||
## LVA Effects → XVF3800 Mapping
|
||||
|
||||
LVA exposes higher-level effects such as **Off / Solid / Slow/Medium/Fast Pulse / Slow/Medium/Fast Blink / Spin**.
|
||||
On XVF3800 (legacy firmware), these are currently mapped onto the limited built-in modes as follows:
|
||||
|
||||
| LVA Effect Name | XVF3800 `LED_EFFECT` | Additional Parameters | Behavior on XVF3800 (Legacy) |
|
||||
|------------------|----------------------|--------------------------------------|-------------------------------------------------------------------------|
|
||||
| `off` | `0` (off) | – | LED ring completely off. |
|
||||
| `solid` | `3` (single color) | `LED_COLOR`, `LED_BRIGHTNESS` | Full ring in a single static color. |
|
||||
| `slow_pulse` | `1` (breath) | `LED_SPEED` = slow | Slow breathing in the configured color. |
|
||||
| `medium_pulse` | `1` (breath) | `LED_SPEED` = medium | Medium breathing in the configured color. |
|
||||
| `fast_pulse` | `1` (breath) | `LED_SPEED` = fast | Fast breathing in the configured color. |
|
||||
| `slow_blink` | `1` (breath) * | `LED_SPEED` = slow | Approximated with slow breath (no true sharp on/off blink). |
|
||||
| `medium_blink` | `1` (breath) * | `LED_SPEED` = medium | Approximated with medium breath. |
|
||||
| `fast_blink` | `1` (breath) * | `LED_SPEED` = fast | Approximated with fast breath. |
|
||||
| `spin` | `3` or `1` ** | Typically treated as solid/pulsed | No native “spin”; currently falls back to a static or pulsed behavior. |
|
||||
|
||||
\* There is no dedicated “blink” primitive in the legacy API, so all blink variants are currently rendered as **breathing** with different speeds. They will look like fading rather than a true square‑wave blink until we add a more advanced mapping or per‑LED pattern logic.
|
||||
|
||||
\** The legacy firmware does not offer a true “spin” animation. For now, `spin` is effectively treated as either **single color** or a **breath** variant to keep behavior predictable.
|
||||
|
||||
## Practical Notes
|
||||
|
||||
- Under the legacy backend, **only three real modes** are used:
|
||||
- `0` → off
|
||||
- `1` → breath (with speed control)
|
||||
- `3` → single color
|
||||
- All the LVA **pulse** and **blink** variants are essentially **different speeds of `LED_EFFECT=1`**.
|
||||
- The **spin** effect is not yet a genuine rotating animation on XVF3800; it’s a placeholder mapped onto existing firmware modes.
|
||||
|
||||
As we move toward **firmware v2.0.7+** and `LED_RING_COLOR` support, we’ll be able to implement richer, per‑LED patterns (including real spin and more accurate blink behaviors) while keeping the same high-level LVA effect names.
|
||||
@@ -3,6 +3,8 @@ import argparse
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import time
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
@@ -22,9 +24,11 @@ from .led_controller import LedController
|
||||
from .mqtt_controller import MqttController
|
||||
from .models import AvailableWakeWord, Preferences, ServerState, WakeWordType
|
||||
from .mpv_player import MpvMediaPlayer
|
||||
from .audio_volume import ensure_output_volume
|
||||
from .satellite import VoiceSatelliteProtocol
|
||||
from .util import get_mac_address, format_mac
|
||||
from .zeroconf import HomeAssistantZeroconf
|
||||
from .xvf3800_button_controller import XVF3800ButtonController # NEW
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
_MODULE_DIR = Path(__file__).parent
|
||||
@@ -150,6 +154,9 @@ async def main() -> None:
|
||||
# --- 2. Load Preferences ---
|
||||
preferences = _load_preferences(config)
|
||||
|
||||
# --- 2b. XVF3800 Startup Workarounds (optional) ---
|
||||
_xvf3800_startup_preflight(config)
|
||||
|
||||
# --- 3. Find Microphone ---
|
||||
mic = _get_microphone(config)
|
||||
|
||||
@@ -159,6 +166,25 @@ async def main() -> None:
|
||||
# --- 5. Initialize Media Players ---
|
||||
media_players = _init_media_players(loop, config, preferences)
|
||||
|
||||
# --- 5b. Sync OS sink volume to persisted volume ---
|
||||
# mpv's per-player volume is kept at 100% and ducking is handled within mpv.
|
||||
# The user-visible "volume" in HA maps to the OS output volume (PipeWire/Pulse/ALSA).
|
||||
if getattr(config.audio, "volume_sync", False):
|
||||
try:
|
||||
loop.create_task(
|
||||
ensure_output_volume(
|
||||
volume=preferences.volume_level,
|
||||
output_device=config.audio.output_device,
|
||||
max_volume_percent=getattr(config.audio, "max_volume_percent", 100),
|
||||
attempts=20,
|
||||
delay_seconds=0.5,
|
||||
)
|
||||
)
|
||||
except Exception:
|
||||
_LOGGER.exception("Failed to schedule output volume sync")
|
||||
else:
|
||||
_LOGGER.debug("Output volume sync disabled (audio.volume_sync=false)")
|
||||
|
||||
# --- 6. Create Server State ---
|
||||
state = _create_server_state(
|
||||
config, loop, event_bus, preferences,
|
||||
@@ -263,15 +289,111 @@ def _load_preferences(config: Config) -> Preferences:
|
||||
|
||||
return preferences
|
||||
|
||||
|
||||
def _xvf3800_startup_preflight(config: Config) -> None:
|
||||
"""
|
||||
Best-effort XVF3800 USB preflight.
|
||||
|
||||
Some platforms (notably certain SBC USB controllers) can leave the XVF3800
|
||||
capture endpoint "silent" until the device is rebooted.
|
||||
|
||||
If an XVF3800 is configured (LED/button/audio), we optionally:
|
||||
1) issue an XVF3800 REBOOT
|
||||
2) wait for USB re-enumeration
|
||||
3) set AUDIO_MGR_OP_L and AUDIO_MGR_OP_R to (7, 3) to force both channels
|
||||
to the ASR3 output (as discovered via xvf_host.py testing)
|
||||
|
||||
This uses PyUSB directly (no dependency on xvf_host.py).
|
||||
"""
|
||||
try:
|
||||
# Determine whether XVF3800 is in use at all
|
||||
led_cfg = getattr(config, "led", None)
|
||||
btn_cfg = getattr(config, "button", None)
|
||||
aud_cfg = getattr(config, "audio", None)
|
||||
|
||||
uses_xvf = False
|
||||
if led_cfg and getattr(led_cfg, "led_type", "").lower() == "xvf3800":
|
||||
uses_xvf = True
|
||||
if btn_cfg and getattr(btn_cfg, "enabled", False) and getattr(btn_cfg, "mode", "").lower() == "xvf3800":
|
||||
uses_xvf = True
|
||||
if aud_cfg and isinstance(getattr(aud_cfg, "input_device", None), str) and "xvf3800" in aud_cfg.input_device.lower():
|
||||
uses_xvf = True
|
||||
|
||||
if not uses_xvf:
|
||||
return
|
||||
|
||||
# Env toggles (default enabled)
|
||||
do_reboot = os.environ.get("LVA_XVF3800_STARTUP_REBOOT", "1").strip().lower() not in ("0", "false", "no", "off")
|
||||
do_route = os.environ.get("LVA_XVF3800_STARTUP_SET_ASR3", "1").strip().lower() not in ("0", "false", "no", "off")
|
||||
do_save = os.environ.get("LVA_XVF3800_STARTUP_SAVE_CONFIG", "0").strip().lower() in ("1", "true", "yes", "on")
|
||||
|
||||
if not (do_reboot or do_route):
|
||||
return
|
||||
|
||||
from .xvf3800_led_backend import XVF3800USBDevice
|
||||
|
||||
_LOGGER.info("XVF3800 startup preflight: begin (reboot=%s, set_asr3=%s, save=%s)", do_reboot, do_route, do_save)
|
||||
|
||||
if do_reboot:
|
||||
try:
|
||||
dev = XVF3800USBDevice()
|
||||
_LOGGER.info("XVF3800 startup preflight: issuing REBOOT to USB device")
|
||||
dev.reboot()
|
||||
finally:
|
||||
try:
|
||||
dev.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Wait for USB to cycle
|
||||
XVF3800USBDevice.wait_for_reenumeration(timeout_s=12.0, settle_s=1.0)
|
||||
|
||||
if do_route:
|
||||
dev2 = XVF3800USBDevice()
|
||||
try:
|
||||
_LOGGER.info("XVF3800 startup preflight: setting AUDIO_MGR_OP_L and AUDIO_MGR_OP_R to (7, 3)")
|
||||
dev2.set_audio_mgr_op_l(7, 3)
|
||||
dev2.set_audio_mgr_op_r(7, 3)
|
||||
if do_save:
|
||||
_LOGGER.info("XVF3800 startup preflight: saving configuration to flash")
|
||||
dev2.save_configuration()
|
||||
finally:
|
||||
dev2.close()
|
||||
|
||||
_LOGGER.info("XVF3800 startup preflight: done")
|
||||
|
||||
except Exception as e:
|
||||
_LOGGER.warning("XVF3800 startup preflight failed (continuing): %s", e)
|
||||
|
||||
|
||||
|
||||
def _get_microphone(config: Config):
|
||||
"""Finds and returns the microphone specified in the config."""
|
||||
mic = None
|
||||
if config.audio.input_device is not None:
|
||||
input_spec = getattr(config.audio, "input_device", None)
|
||||
|
||||
# If the user specified an index, honor it.
|
||||
if input_spec is not None:
|
||||
try:
|
||||
input_device_idx = int(config.audio.input_device)
|
||||
input_device_idx = int(input_spec)
|
||||
mic = sc.all_microphones(include_loopback=False)[input_device_idx]
|
||||
except (ValueError, IndexError):
|
||||
mic = sc.get_microphone(config.audio.input_device, include_loopback=False)
|
||||
# If the user specified a name, retry a bit (helps after USB re-enumeration / slow PipeWire startup).
|
||||
want_name = str(input_spec)
|
||||
is_xvf = "xvf3800" in want_name.lower()
|
||||
deadline = time.time() + (20.0 if is_xvf else 5.0)
|
||||
|
||||
last_err: Optional[Exception] = None
|
||||
while time.time() < deadline:
|
||||
try:
|
||||
mic = sc.get_microphone(want_name, include_loopback=False)
|
||||
break
|
||||
except Exception as e:
|
||||
last_err = e
|
||||
time.sleep(0.5)
|
||||
|
||||
if mic is None and last_err is not None:
|
||||
_LOGGER.warning("Failed to open configured mic %r after retries: %s", want_name, last_err)
|
||||
else:
|
||||
mic = sc.default_microphone()
|
||||
|
||||
@@ -426,21 +548,33 @@ def _init_controllers(
|
||||
mqtt_controller=mqtt_controller,
|
||||
)
|
||||
|
||||
# Hardware button controller (e.g. ReSpeaker 2-Mic HAT)
|
||||
# Hardware / XVF3800 mute button controller
|
||||
try:
|
||||
if getattr(config, "button", None) is not None and config.button.enabled:
|
||||
button_cfg = getattr(config, "button", None)
|
||||
if button_cfg is not None and button_cfg.enabled:
|
||||
mode = getattr(button_cfg, "mode", "gpio").lower()
|
||||
if mode == "xvf3800":
|
||||
_LOGGER.info("Initializing XVF3800ButtonController (mode=xvf3800)")
|
||||
xvf_btn = XVF3800ButtonController(
|
||||
loop=loop,
|
||||
event_bus=event_bus,
|
||||
state=state,
|
||||
button_config=button_cfg,
|
||||
)
|
||||
setattr(state, "xvf3800_button_controller", xvf_btn)
|
||||
else:
|
||||
_LOGGER.info("Initializing GPIO ButtonController (mode=gpio)")
|
||||
button_controller = ButtonController(
|
||||
loop=loop,
|
||||
event_bus=event_bus,
|
||||
state=state,
|
||||
config=config.button,
|
||||
config=button_cfg,
|
||||
)
|
||||
# Keep a reference on state so it is not garbage-collected.
|
||||
setattr(state, "button_controller", button_controller)
|
||||
else:
|
||||
_LOGGER.debug("ButtonController not enabled in config; skipping")
|
||||
_LOGGER.debug("Button controller not enabled in config; skipping")
|
||||
except Exception:
|
||||
_LOGGER.exception("Failed to initialize ButtonController")
|
||||
_LOGGER.exception("Failed to initialize button controller(s)")
|
||||
|
||||
async def _run_server(state: ServerState, config: Config):
|
||||
"""Starts the ESPHome server and ZeroConf discovery."""
|
||||
|
||||
@@ -28,6 +28,9 @@ class AudioEngine:
|
||||
self.block_size = block_size
|
||||
self._thread: Optional[threading.Thread] = None
|
||||
|
||||
# CRITICAL FIX: Add lock for thread-safe wake word reload
|
||||
self._wake_words_lock = threading.Lock()
|
||||
|
||||
def start(self):
|
||||
"""Starts the audio processing thread."""
|
||||
self._thread = threading.Thread(
|
||||
@@ -113,6 +116,8 @@ class AudioEngine:
|
||||
time.sleep(0.01)
|
||||
continue
|
||||
|
||||
# CRITICAL FIX: Protect wake word reload with lock
|
||||
with self._wake_words_lock:
|
||||
# Update active wake words if changed
|
||||
if (not wake_words) or (self.state.wake_words_changed):
|
||||
self.state.wake_words_changed = False
|
||||
|
||||
@@ -0,0 +1,192 @@
|
||||
"""System output volume helpers.
|
||||
|
||||
LVA historically only adjusted mpv's per-player volume. On PipeWire/PulseAudio
|
||||
setups where the default sink volume starts low (e.g. 40%), HA shows LVA at
|
||||
100% while the actual OS sink remains capped.
|
||||
|
||||
This module keeps the OS output volume aligned with LVA's persisted
|
||||
preferences.volume_level (0.0–1.0).
|
||||
|
||||
We try backends in this order:
|
||||
1) PipeWire: wpctl
|
||||
2) PulseAudio (incl. pipewire-pulse): pactl
|
||||
3) ALSA: amixer
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import math
|
||||
import shutil
|
||||
import subprocess
|
||||
from typing import Optional
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _clamp01(v: float) -> float:
|
||||
try:
|
||||
v = float(v)
|
||||
except Exception:
|
||||
return 1.0
|
||||
if math.isnan(v) or math.isinf(v):
|
||||
return 1.0
|
||||
return max(0.0, min(1.0, v))
|
||||
|
||||
|
||||
def _pactl_sink_from_output_device(output_device: Optional[str]) -> str:
|
||||
"""Best-effort mapping of mpv audio-device -> pactl sink name.
|
||||
|
||||
LVA commonly uses:
|
||||
- "pipewire/<pactl sink name>"
|
||||
- "pulse/<pactl sink name>"
|
||||
- "alsa_output...." (already a sink name)
|
||||
|
||||
Fallback is @DEFAULT_SINK@.
|
||||
"""
|
||||
if not output_device:
|
||||
return "@DEFAULT_SINK@"
|
||||
|
||||
dev = output_device.strip()
|
||||
|
||||
for prefix in ("pipewire/", "pulse/"):
|
||||
if dev.startswith(prefix) and len(dev) > len(prefix):
|
||||
return dev[len(prefix) :]
|
||||
|
||||
if dev.startswith("alsa_output."):
|
||||
return dev
|
||||
|
||||
# Some users pass the raw sink name without a prefix.
|
||||
if "alsa_output." in dev:
|
||||
return dev
|
||||
|
||||
return "@DEFAULT_SINK@"
|
||||
|
||||
|
||||
def _run_cmd(cmd: list[str], timeout_s: float = 2.0) -> tuple[bool, str]:
|
||||
try:
|
||||
proc = subprocess.run(
|
||||
cmd,
|
||||
check=False,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
text=True,
|
||||
timeout=timeout_s,
|
||||
)
|
||||
ok = proc.returncode == 0
|
||||
out = (proc.stdout or "").strip()
|
||||
return ok, out
|
||||
except FileNotFoundError:
|
||||
return False, f"command not found: {cmd[0]}"
|
||||
except subprocess.TimeoutExpired:
|
||||
return False, "timeout"
|
||||
except Exception as e:
|
||||
return False, repr(e)
|
||||
|
||||
|
||||
def set_output_volume(
|
||||
volume_0_1: float,
|
||||
output_device: Optional[str] = None,
|
||||
max_volume_percent: int = 100,
|
||||
logger: logging.Logger = _LOGGER,
|
||||
) -> bool:
|
||||
"""Set OS output volume to match LVA volume (0.0–1.0).
|
||||
|
||||
`max_volume_percent` lets you map LVA's 100% to something above 100% on the
|
||||
OS sink (e.g. 150 for 1.5x), which can be useful on devices that need
|
||||
boosted output.
|
||||
|
||||
Returns True if any backend succeeded.
|
||||
"""
|
||||
vol = _clamp01(volume_0_1)
|
||||
try:
|
||||
max_pct = int(max_volume_percent)
|
||||
except Exception:
|
||||
max_pct = 100
|
||||
# Avoid weird values; allow >100 for boosted sinks.
|
||||
max_pct = max(0, min(200, max_pct))
|
||||
sink_scalar = vol * (max_pct / 100.0)
|
||||
|
||||
# --- 1) PipeWire: wpctl -------------------------------------------------
|
||||
if shutil.which("wpctl"):
|
||||
# wpctl accepts @DEFAULT_AUDIO_SINK@ and a linear factor (e.g. 0.40)
|
||||
ok, out = _run_cmd(
|
||||
[
|
||||
"wpctl",
|
||||
"set-volume",
|
||||
"@DEFAULT_AUDIO_SINK@",
|
||||
f"{sink_scalar:.3f}",
|
||||
]
|
||||
)
|
||||
if ok:
|
||||
logger.debug(
|
||||
"Set PipeWire sink volume via wpctl: scalar=%.3f (vol=%.3f max_pct=%d)",
|
||||
sink_scalar,
|
||||
vol,
|
||||
max_pct,
|
||||
)
|
||||
return True
|
||||
logger.debug("wpctl set-volume failed: %s", out)
|
||||
|
||||
# --- 2) PulseAudio: pactl ----------------------------------------------
|
||||
if shutil.which("pactl"):
|
||||
sink = _pactl_sink_from_output_device(output_device)
|
||||
pct = int(round(sink_scalar * 100.0))
|
||||
ok, out = _run_cmd(["pactl", "set-sink-volume", sink, f"{pct}%"])
|
||||
if ok:
|
||||
logger.debug(
|
||||
"Set Pulse sink volume via pactl: sink=%s pct=%d (vol=%.3f max_pct=%d)",
|
||||
sink,
|
||||
pct,
|
||||
vol,
|
||||
max_pct,
|
||||
)
|
||||
return True
|
||||
logger.debug("pactl set-sink-volume failed: %s", out)
|
||||
|
||||
# --- 3) ALSA: amixer ----------------------------------------------------
|
||||
if shutil.which("amixer"):
|
||||
pct = int(round(sink_scalar * 100.0))
|
||||
# Common mixer controls across SBC images.
|
||||
for control in ("Master", "PCM", "Speaker", "Headphone"):
|
||||
ok, out = _run_cmd(["amixer", "-q", "sset", control, f"{pct}%"])
|
||||
if ok:
|
||||
logger.debug(
|
||||
"Set ALSA volume via amixer: control=%s pct=%d (vol=%.3f max_pct=%d)",
|
||||
control,
|
||||
pct,
|
||||
vol,
|
||||
max_pct,
|
||||
)
|
||||
return True
|
||||
logger.debug("amixer sset %s failed: %s", control, out)
|
||||
|
||||
return False
|
||||
|
||||
|
||||
async def ensure_output_volume(
|
||||
volume: float,
|
||||
output_device: Optional[str] = None,
|
||||
max_volume_percent: int = 100,
|
||||
attempts: int = 10,
|
||||
delay_seconds: float = 0.5,
|
||||
logger: logging.Logger = _LOGGER,
|
||||
) -> bool:
|
||||
"""Retry output volume sync during startup.
|
||||
|
||||
Useful at boot when PipeWire/Pulse/ALSA might not be fully ready yet.
|
||||
"""
|
||||
for i in range(1, max(1, attempts) + 1):
|
||||
ok = await asyncio.to_thread(
|
||||
set_output_volume,
|
||||
volume,
|
||||
output_device,
|
||||
max_volume_percent,
|
||||
logger,
|
||||
)
|
||||
if ok:
|
||||
return True
|
||||
if i < attempts:
|
||||
await asyncio.sleep(delay_seconds)
|
||||
return False
|
||||
@@ -40,7 +40,9 @@ class ButtonRuntimeConfig:
|
||||
enabled: bool
|
||||
pin: int
|
||||
long_press_seconds: float
|
||||
poll_interval_seconds: float = 0.01 # 100 Hz polling
|
||||
# CRITICAL FIX: Change default from 0.01s to 0.05s (100 Hz -> 20 Hz)
|
||||
# 80% reduction in CPU usage for button monitoring
|
||||
poll_interval_seconds: float = 0.05 # 20 Hz polling
|
||||
|
||||
|
||||
class ButtonController:
|
||||
@@ -73,6 +75,9 @@ class ButtonController:
|
||||
self.event_bus = event_bus
|
||||
self.state = state
|
||||
|
||||
# CRITICAL FIX: Default poll_interval_seconds to 0.05 instead of 0.01
|
||||
default_poll_interval = 0.05
|
||||
|
||||
self._cfg = ButtonRuntimeConfig(
|
||||
enabled=getattr(config, "enabled", False),
|
||||
pin=getattr(config, "pin", 17),
|
||||
@@ -80,9 +85,7 @@ class ButtonController:
|
||||
getattr(config, "long_press_seconds", 1.0)
|
||||
),
|
||||
poll_interval_seconds=float(
|
||||
getattr(config, "poll_interval_seconds", 0.01)
|
||||
if hasattr(config, "poll_interval_seconds")
|
||||
else 0.01
|
||||
getattr(config, "poll_interval_seconds", default_poll_interval)
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@@ -1,62 +1,144 @@
|
||||
{
|
||||
// ---------------------------------------------------------------------------
|
||||
// Required
|
||||
// ---------------------------------------------------------------------------
|
||||
"app": {
|
||||
"name": "Linux Voice Assistant",
|
||||
"debug": true,
|
||||
// Required
|
||||
"name": "My Linux Voice Assistant",
|
||||
|
||||
// Optional
|
||||
"wakeup_sound": "sounds/wake_word_triggered.flac",
|
||||
"timer_finished_sound": "sounds/timer_finished.flac",
|
||||
"preferences_file": "preferences.json"
|
||||
"preferences_file": "preferences.json",
|
||||
"debug": false
|
||||
},
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Optional: Audio I/O + Output Volume Sync
|
||||
// ---------------------------------------------------------------------------
|
||||
"audio": {
|
||||
// This is the "manual bypass" for audio.
|
||||
// Set to "null" to use auto-detection (recommended).
|
||||
// Set to specific device names to force them.
|
||||
"input_device": "seeed-2mic-voicecard",
|
||||
"output_device": "alsa/plughw:CARD=seeed2micvoicec",
|
||||
"input_block_size": 1024
|
||||
// Optional:
|
||||
// - null/omit => default microphone
|
||||
// - string name => exact device name (soundcard backend)
|
||||
// - string number => index into soundcard device list (e.g. "0", "1", ...)
|
||||
"input_device": null,
|
||||
|
||||
// Optional: audio capture block size
|
||||
"input_block_size": 1024,
|
||||
|
||||
// Optional:
|
||||
// - null/omit => mpv chooses default output
|
||||
// - string => mpv audio-device (e.g. "pulse/alsa_output....")
|
||||
"output_device": null,
|
||||
|
||||
// Optional (NEW): Sync the host OS output device volume at startup to match
|
||||
// the volume stored in preferences.json (LVA MediaPlayer volume).
|
||||
//
|
||||
// When enabled, LVA will attempt (in order): wpctl -> pactl -> amixer.
|
||||
// This helps when PipeWire/Pulse/ALSA boots with a low default volume.
|
||||
//
|
||||
// Default: false
|
||||
"volume_sync": false,
|
||||
|
||||
// Optional (NEW): Maximum output volume percent for the underlying sink.
|
||||
//
|
||||
// Many PipeWire/PulseAudio sinks support >100% (e.g. 150%) "amplified" volume.
|
||||
// This sets the ceiling used when mapping LVA volume (0.0–1.0) to the sink.
|
||||
//
|
||||
// Example:
|
||||
// - max_volume_percent: 150
|
||||
// - LVA volume 1.0 => sink set to 150%
|
||||
// - LVA volume 0.5 => sink set to 75%
|
||||
//
|
||||
// Default: 100
|
||||
"max_volume_percent": 100
|
||||
},
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Optional: Wake word config
|
||||
// ---------------------------------------------------------------------------
|
||||
"wake_word": {
|
||||
// A list of directories to scan for wake word models
|
||||
"directories": [
|
||||
"wakewords",
|
||||
"wakewords/openWakeWord"
|
||||
],
|
||||
// The default wake word to use if none is set in preferences
|
||||
// Optional: if omitted/empty, code falls back to defaults:
|
||||
// ["wakewords", "wakewords/openWakeWord"]
|
||||
"directories": [],
|
||||
|
||||
// Optional: default wake word model id
|
||||
"model": "okay_nabu",
|
||||
|
||||
// Optional: stop model id (micro wake word)
|
||||
"stop_model": "stop",
|
||||
|
||||
// Optional: debounce/refractory window after wake word triggers
|
||||
"refractory_seconds": 2.0,
|
||||
// Paths are relative to the project root
|
||||
"oww_melspectrogram_model": "wakewords/openWakeWord/melspectrogram.tflite",
|
||||
"oww_embedding_model": "wakewords/openWakeWord/embedding_model.tflite"
|
||||
|
||||
// Optional: directory for downloaded models (relative to repo)
|
||||
"download_dir": "local"
|
||||
},
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Optional: ESPHome protocol server settings (the LVA exposes an ESPHome API)
|
||||
// ---------------------------------------------------------------------------
|
||||
"esphome": {
|
||||
// Network settings for the satellite server
|
||||
"host": "0.0.0.0",
|
||||
"port": 6053
|
||||
},
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Optional: LEDs
|
||||
// ---------------------------------------------------------------------------
|
||||
"led": {
|
||||
// "led_type" can be "dotstar" (APA102) or "neopixel" (WS2812B)
|
||||
// Optional: if false, LED controller runs in no-op mode
|
||||
"enabled": true,
|
||||
|
||||
// "dotstar" | "neopixel" | "xvf3800"
|
||||
"led_type": "dotstar",
|
||||
|
||||
// "interface" can be "spi" (hardware SPI) or "gpio" (bit-banging)
|
||||
"interface": "gpio",
|
||||
// "spi" | "gpio" | "usb"
|
||||
// - dotstar: "spi" or "gpio"
|
||||
// - neopixel: "spi" or "gpio"
|
||||
// - xvf3800: "usb" (current intended support)
|
||||
"interface": "spi",
|
||||
|
||||
// --- GPIO Settings ---
|
||||
// These are ONLY used if interface is "gpio"
|
||||
// For DotStar on Grove port: clock=13, data=12
|
||||
// For NeoPixel on Grove port: data=12 (clock not used and is ignored)
|
||||
// GPIO pins (BCM-ish via adafruit "board.Dxx" mapping) — used when interface="gpio"
|
||||
// For DotStar GPIO: both clock_pin + data_pin are used
|
||||
// For NeoPixel GPIO: only data_pin is used (clock_pin ignored)
|
||||
// For xvf3800: unused and ignored
|
||||
"clock_pin": 13,
|
||||
"data_pin": 12,
|
||||
|
||||
// Default number of LEDs. This is overridden by
|
||||
// "num_leds" in preferences.json after first run.
|
||||
// Default number of LEDs (may be overridden by preferences.json after first run)
|
||||
"num_leds": 3
|
||||
},
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Optional: MQTT (enabled automatically when host is set)
|
||||
// ---------------------------------------------------------------------------
|
||||
"mqtt": {
|
||||
// "host" is the only required field to enable MQTT.
|
||||
// If "host" is null or this section is omitted, MQTT is disabled.
|
||||
"host": "192.168.1.2",
|
||||
// If set, mqtt.enabled becomes true automatically
|
||||
"host": null,
|
||||
|
||||
"port": 1883,
|
||||
"username": "mqtt_username",
|
||||
"password": "mqtt_password"
|
||||
"username": null,
|
||||
"password": null
|
||||
},
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Optional: Button controller
|
||||
// ---------------------------------------------------------------------------
|
||||
"button": {
|
||||
// Master enable
|
||||
"enabled": false,
|
||||
|
||||
// "gpio" | "xvf3800"
|
||||
// - gpio: short/long press logic (pin + long_press_seconds)
|
||||
// - xvf3800: uses the XVF3800 built-in mute button as a pure mute toggle
|
||||
"mode": "gpio",
|
||||
|
||||
// gpio mode only
|
||||
"pin": 17,
|
||||
"long_press_seconds": 1.0,
|
||||
|
||||
// Polling interval (used by both gpio + xvf3800 button controllers)
|
||||
"poll_interval_seconds": 0.01
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,17 +1,29 @@
|
||||
"""Configuration models for the application."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional
|
||||
from typing import List, Optional
|
||||
|
||||
import logging
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Configuration Dataclasses
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
@dataclass
|
||||
class AppConfig:
|
||||
"""General application settings."""
|
||||
name: str
|
||||
wakeup_sound: str = "sounds/wake_word_triggered.flac"
|
||||
timer_finished_sound: str = "sounds/timer_finished.flac"
|
||||
preferences_file: str = "preferences.json"
|
||||
debug: bool = False
|
||||
|
||||
|
||||
@dataclass
|
||||
class AudioConfig:
|
||||
"""Settings for audio input and output."""
|
||||
@@ -19,6 +31,21 @@ class AudioConfig:
|
||||
input_block_size: int = 1024
|
||||
output_device: Optional[str] = None
|
||||
|
||||
# If True, LVA will attempt to set the OS sink volume to match the persisted
|
||||
# volume in preferences.json on startup (PipeWire/PulseAudio/ALSA best-effort).
|
||||
# Default is False to avoid surprising changes on systems where users manage
|
||||
# volume externally.
|
||||
volume_sync: bool = False
|
||||
|
||||
# Maximum sink volume to map to LVA's 100% volume.
|
||||
#
|
||||
# Example: if max_volume_percent=150 then when LVA's media player reports
|
||||
# 100% (preferences.volume_level == 1.0), the underlying sink will be set
|
||||
# to 150% (or 1.5 for backends that use scalar volume).
|
||||
#
|
||||
# This is useful for devices that need >100% gain on PipeWire/Pulse/ALSA.
|
||||
max_volume_percent: int = 100
|
||||
|
||||
|
||||
@dataclass
|
||||
class WakeWordConfig:
|
||||
@@ -39,12 +66,24 @@ class ESPHomeConfig:
|
||||
|
||||
@dataclass
|
||||
class LedConfig:
|
||||
"""Settings for the LED strip."""
|
||||
"""Settings for LEDs."""
|
||||
enabled: bool = True
|
||||
|
||||
# Supported values include:
|
||||
# - "dotstar" / "neopixel" for Pi-attached LED strips
|
||||
# - "xvf3800" for the ReSpeaker XVF3800 USB LED ring backend
|
||||
led_type: str = "dotstar"
|
||||
|
||||
# For dotstar/neopixel: "spi" or "gpio"
|
||||
# For xvf3800: "usb"
|
||||
interface: str = "spi"
|
||||
|
||||
# GPIO pin numbers used when interface="gpio"
|
||||
clock_pin: int = 13
|
||||
data_pin: int = 12
|
||||
num_leds: int = 3 # Note: overridden by 'preferences.json' if it exists
|
||||
|
||||
# Note: overridden by 'preferences.json' if it exists
|
||||
num_leds: int = 3
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -59,22 +98,24 @@ class MqttConfig:
|
||||
|
||||
@dataclass
|
||||
class ButtonConfig:
|
||||
"""Settings for a hardware momentary button (e.g. ReSpeaker 2-Mic HAT)."""
|
||||
"""Settings for a hardware momentary button."""
|
||||
|
||||
# Overall enable/disable (default: off)
|
||||
enabled: bool = False
|
||||
# BCM GPIO pin number for the button input.
|
||||
|
||||
# mode:
|
||||
# - "gpio" -> legacy GPIO button (e.g. ReSpeaker 2-Mic HAT)
|
||||
# - "xvf3800"-> USB-based mute integration for the ReSpeaker XVF3800
|
||||
mode: str = "gpio"
|
||||
|
||||
# BCM GPIO pin number for the button input (gpio mode only).
|
||||
pin: int = 17
|
||||
# Press duration (in seconds) to be considered a "long press".
|
||||
|
||||
# Press duration (in seconds) to be considered a "long press" (gpio mode).
|
||||
long_press_seconds: float = 1.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class AppConfig:
|
||||
"""General application settings."""
|
||||
name: str
|
||||
wakeup_sound: str = "sounds/wake_word_triggered.flac"
|
||||
timer_finished_sound: str = "sounds/timer_finished.flac"
|
||||
preferences_file: str = "preferences.json"
|
||||
debug: bool = False
|
||||
# Poll interval used by both GPIO and XVF3800 controllers.
|
||||
poll_interval_seconds: float = 0.01
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -95,22 +136,18 @@ class Config:
|
||||
def load_config_from_json(config_path: Path) -> Config:
|
||||
"""Loads configuration from a JSON file and populates dataclasses."""
|
||||
|
||||
# --- Step 1: Load raw JSON data ---
|
||||
try:
|
||||
with open(config_path, "r", encoding="utf-8") as f:
|
||||
raw_data = json.load(f)
|
||||
except FileNotFoundError:
|
||||
_LOGGER.critical(f"Configuration file not found at: {config_path}")
|
||||
_LOGGER.critical("Configuration file not found at: %s", config_path)
|
||||
raise
|
||||
except json.JSONDecodeError as e:
|
||||
_LOGGER.critical(f"Error parsing configuration file: {e}")
|
||||
_LOGGER.critical("Error parsing configuration file: %s", e)
|
||||
raise
|
||||
|
||||
# --- Step 2: Create config objects from raw data ---
|
||||
if "app" not in raw_data:
|
||||
raise ValueError(
|
||||
"Configuration file must contain an 'app' section with a 'name'."
|
||||
)
|
||||
raise ValueError("Configuration file must contain an 'app' section with a 'name'.")
|
||||
|
||||
app_config = AppConfig(**raw_data.get("app", {}))
|
||||
audio_config = AudioConfig(**raw_data.get("audio", {}))
|
||||
@@ -120,11 +157,18 @@ def load_config_from_json(config_path: Path) -> Config:
|
||||
mqtt_config = MqttConfig(**raw_data.get("mqtt", {}))
|
||||
button_config = ButtonConfig(**raw_data.get("button", {}))
|
||||
|
||||
# --- Step 3: Set MQTT 'enabled' flag ---
|
||||
# Back-compat: allow top-level "volume_sync" (preferred location is audio.volume_sync)
|
||||
if "volume_sync" in raw_data and "volume_sync" not in raw_data.get("audio", {}):
|
||||
try:
|
||||
audio_config.volume_sync = bool(raw_data.get("volume_sync"))
|
||||
except Exception:
|
||||
# If it's something strange, just leave default.
|
||||
pass
|
||||
|
||||
# Set MQTT 'enabled' flag
|
||||
if mqtt_config.host:
|
||||
mqtt_config.enabled = True
|
||||
|
||||
# --- Step 4: Return the main Config object ---
|
||||
return Config(
|
||||
app=app_config,
|
||||
audio=audio_config,
|
||||
|
||||
@@ -31,9 +31,6 @@ class ESPHomeEntity:
|
||||
pass
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
|
||||
class MediaPlayerEntity(ESPHomeEntity):
|
||||
def __init__(
|
||||
self,
|
||||
@@ -102,13 +99,20 @@ class MediaPlayerEntity(ESPHomeEntity):
|
||||
if msg.has_media_url:
|
||||
announcement = msg.has_announcement and msg.announcement
|
||||
yield from self.play(msg.media_url, announcement=announcement)
|
||||
|
||||
elif msg.has_command:
|
||||
if msg.command == MediaPlayerCommand.PAUSE:
|
||||
self.music_player.pause()
|
||||
yield self._update_state(MediaPlayerState.PAUSED)
|
||||
|
||||
elif msg.command == MediaPlayerCommand.PLAY:
|
||||
self.music_player.resume()
|
||||
yield self._update_state(MediaPlayerState.PLAYING)
|
||||
|
||||
elif msg.command == MediaPlayerCommand.STOP:
|
||||
self.music_player.stop()
|
||||
yield self._update_state(MediaPlayerState.IDLE)
|
||||
|
||||
if msg.has_volume:
|
||||
# This block is called when the volume slider changes in HA
|
||||
self.volume = msg.volume # HA sends volume as 0.0-1.0
|
||||
@@ -121,6 +125,7 @@ class MediaPlayerEntity(ESPHomeEntity):
|
||||
self.state.save_preferences()
|
||||
|
||||
yield self._update_state(self.state_enum)
|
||||
|
||||
elif isinstance(msg, ListEntitiesRequest):
|
||||
yield ListEntitiesMediaPlayerResponse(
|
||||
object_id=self.object_id,
|
||||
@@ -128,6 +133,7 @@ class MediaPlayerEntity(ESPHomeEntity):
|
||||
name=self.name,
|
||||
supports_pause=True,
|
||||
)
|
||||
|
||||
elif isinstance(msg, SubscribeHomeAssistantStatesRequest):
|
||||
yield self._get_state_message()
|
||||
|
||||
|
||||
@@ -13,7 +13,8 @@ except Exception:
|
||||
board = None # type: ignore[assignment]
|
||||
_LOGGER.warning(
|
||||
"Adafruit 'board' module not available or unsupported on this platform; "
|
||||
"hardware LEDs will be disabled."
|
||||
"DotStar/NeoPixel GPIO/SPI LED backends will be disabled. "
|
||||
"XVF3800 USB LED backend is unaffected."
|
||||
)
|
||||
|
||||
from .config import LedConfig
|
||||
@@ -45,6 +46,12 @@ class LedController(EventHandler):
|
||||
self._is_ready = False
|
||||
self.leds = None
|
||||
|
||||
# Backend mode:
|
||||
# "pixels" -> DotStar / NeoPixel via Adafruit drivers
|
||||
# "xvf3800" -> XVF3800 USB LED ring backend
|
||||
self._backend_mode: str = "pixels"
|
||||
self._xvf3800_backend = None
|
||||
|
||||
# Configured LED behavior
|
||||
self.configs = {
|
||||
"idle": {"effect": "off", "color": _PURPLE, "brightness": 0.5},
|
||||
@@ -56,6 +63,11 @@ class LedController(EventHandler):
|
||||
|
||||
# Determine whether hardware LEDs should be used at all
|
||||
config_enabled = getattr(config, "enabled", True)
|
||||
|
||||
# For XVF3800 we do NOT require the Adafruit 'board' module.
|
||||
if config.led_type == "xvf3800" and config.interface == "usb":
|
||||
self._enabled = bool(config_enabled)
|
||||
else:
|
||||
self._enabled = bool(config_enabled) and (board is not None)
|
||||
|
||||
if not config_enabled:
|
||||
@@ -63,22 +75,50 @@ class LedController(EventHandler):
|
||||
"LEDs disabled in config (led.enabled = false); "
|
||||
"LedController will run in no-op mode."
|
||||
)
|
||||
elif board is None:
|
||||
elif not self._enabled:
|
||||
_LOGGER.warning(
|
||||
"LED hardware libraries not available on this platform; "
|
||||
"LedController will run in no-op mode."
|
||||
"LED hardware libraries not available on this platform for led_type=%s; "
|
||||
"LedController will run in no-op mode.",
|
||||
config.led_type,
|
||||
)
|
||||
|
||||
# Always subscribe to events so MQTT + state logic remains intact,
|
||||
# even when LEDs are effectively disabled.
|
||||
self._subscribe_all_methods()
|
||||
|
||||
# If LEDs are not enabled or board is unavailable, skip hardware init
|
||||
# If LEDs are not enabled at all, skip hardware init
|
||||
if not self._enabled:
|
||||
return
|
||||
|
||||
# -------------------------------------------------------------------
|
||||
# Hardware initialization (Pi / supported boards only)
|
||||
# XVF3800 USB LED backend
|
||||
# -------------------------------------------------------------------
|
||||
if config.led_type == "xvf3800" and config.interface == "usb":
|
||||
try:
|
||||
from .xvf3800_led_backend import XVF3800LedBackend # type: ignore[import]
|
||||
|
||||
self._xvf3800_backend = XVF3800LedBackend()
|
||||
self._backend_mode = "xvf3800"
|
||||
self._is_ready = True
|
||||
# XVF3800 ring has a fixed LED count (typically 12)
|
||||
self.num_leds = int(getattr(self._xvf3800_backend, "ring_led_count", 12))
|
||||
_LOGGER.info(
|
||||
"LED Controller initialized for XVF3800 USB LED ring (num_leds=%d).",
|
||||
self.num_leds,
|
||||
)
|
||||
self.run_action("startup_sequence")
|
||||
except Exception:
|
||||
# Any hardware-related failure leaves us in no-op mode
|
||||
self._is_ready = False
|
||||
self._enabled = False
|
||||
self._xvf3800_backend = None
|
||||
_LOGGER.exception(
|
||||
"Failed to initialize XVF3800 LED backend. LEDs will be disabled."
|
||||
)
|
||||
return
|
||||
|
||||
# -------------------------------------------------------------------
|
||||
# Hardware initialization (DotStar / NeoPixel via Adafruit drivers)
|
||||
# -------------------------------------------------------------------
|
||||
try:
|
||||
if config.led_type == "neopixel":
|
||||
@@ -128,6 +168,7 @@ class LedController(EventHandler):
|
||||
board.SCLK, board.MOSI, self.num_leds, auto_write=False
|
||||
)
|
||||
|
||||
self._backend_mode = "pixels"
|
||||
self._is_ready = True
|
||||
_LOGGER.info(
|
||||
"LED Controller initialized for %d %s LEDs.",
|
||||
@@ -170,79 +211,281 @@ class LedController(EventHandler):
|
||||
config["state_name"] = state_name
|
||||
self.event_bus.publish("publish_state_to_mqtt", config)
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# XVF3800-specific helpers
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
async def _xvf3800_apply_effect(
|
||||
self,
|
||||
effect_name: str,
|
||||
color: Tuple[int, int, int],
|
||||
brightness: float,
|
||||
) -> None:
|
||||
"""Map generic effect/color/brightness to XVF3800 legacy LED controls."""
|
||||
if not (self._enabled and self._is_ready and self._xvf3800_backend):
|
||||
return
|
||||
|
||||
# Clamp brightness 0..1 and map to 0..255
|
||||
brightness = max(0.0, min(1.0, float(brightness)))
|
||||
brightness_255 = int(brightness * 255)
|
||||
|
||||
# Effect mapping to XVF3800 legacy modes
|
||||
effect_map = {
|
||||
"off": 0, # LED_EFFECT = off
|
||||
"solid": 3, # single color
|
||||
"slow_pulse": 1, # breath
|
||||
"medium_pulse": 1, # breath
|
||||
"fast_pulse": 1, # breath
|
||||
"slow_blink": 1, # approximate with breath
|
||||
"medium_blink": 1, # approximate with breath
|
||||
"fast_blink": 1, # approximate with breath
|
||||
"spin": 2, # rainbow as a stand-in for "spin"
|
||||
}
|
||||
|
||||
speed_map = {
|
||||
"off": 0,
|
||||
"solid": 0,
|
||||
"slow_pulse": 0,
|
||||
"medium_pulse": 1,
|
||||
"fast_pulse": 2,
|
||||
"slow_blink": 0,
|
||||
"medium_blink": 1,
|
||||
"fast_blink": 2,
|
||||
"spin": 1,
|
||||
}
|
||||
|
||||
effect_id = effect_map.get(effect_name, 3)
|
||||
speed_id = speed_map.get(effect_name, 1)
|
||||
|
||||
r, g, b = color
|
||||
try:
|
||||
# Apply brightness, color, speed, and effect via USB control transfers.
|
||||
self._xvf3800_backend.set_brightness(brightness_255)
|
||||
if effect_name != "off":
|
||||
self._xvf3800_backend.set_color(r, g, b)
|
||||
self._xvf3800_backend.set_speed(speed_id)
|
||||
self._xvf3800_backend.set_effect(effect_id)
|
||||
except Exception:
|
||||
_LOGGER.exception(
|
||||
"Error sending LED effect '%s' to XVF3800 backend", effect_name
|
||||
)
|
||||
|
||||
# Keep coroutine alive until cancelled so that a new effect
|
||||
# can cancel the previous one consistently.
|
||||
try:
|
||||
while True:
|
||||
await asyncio.sleep(3600)
|
||||
except asyncio.CancelledError:
|
||||
return
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# LED effect coroutines
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# XVF3800 per-LED helpers (newer firmware)
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
def _xvf3800_has_per_led(self) -> bool:
|
||||
return bool(getattr(self._xvf3800_backend, "supports_per_led", False))
|
||||
|
||||
def _xvf3800_ring_count(self) -> int:
|
||||
return int(getattr(self._xvf3800_backend, "ring_led_count", 12))
|
||||
|
||||
def _xvf3800_rgb_clamp(self, color):
|
||||
r, g, b = color
|
||||
return (
|
||||
max(0, min(255, int(r))),
|
||||
max(0, min(255, int(g))),
|
||||
max(0, min(255, int(b))),
|
||||
)
|
||||
|
||||
def _xvf3800_brightness_255(self, brightness: float) -> int:
|
||||
b = max(0.0, min(1.0, float(brightness)))
|
||||
return int(b * 255)
|
||||
|
||||
def _xvf3800_apply_ring_solid(self, color, brightness: float) -> None:
|
||||
"""Per-LED solid ring.
|
||||
|
||||
NOTE: Some XVF3800 firmwares do not apply LED_BRIGHTNESS to LED_RING_COLOR output.
|
||||
To keep behavior consistent, we scale the RGB values by brightness and write the ring
|
||||
colors directly.
|
||||
"""
|
||||
if not self._xvf3800_backend:
|
||||
return
|
||||
r, g, b = self._xvf3800_rgb_clamp(color)
|
||||
brightness = max(0.0, min(1.0, float(brightness)))
|
||||
|
||||
# Ensure firmware effects don't override per-LED output.
|
||||
try:
|
||||
self._xvf3800_backend.set_effect(0)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
rs = int(r * brightness)
|
||||
gs = int(g * brightness)
|
||||
bs = int(b * brightness)
|
||||
self._xvf3800_backend.set_ring_solid(rs, gs, bs)
|
||||
|
||||
def _xvf3800_apply_ring_clear(self) -> None:
|
||||
if not self._xvf3800_backend:
|
||||
return
|
||||
try:
|
||||
self._xvf3800_backend.set_effect(0)
|
||||
except Exception:
|
||||
pass
|
||||
self._xvf3800_backend.clear_ring()
|
||||
|
||||
async def startup_sequence(self, color=None, brightness=None):
|
||||
# Use a simple green blink on startup for all backends.
|
||||
await self.blink(_GREEN, 1.0)
|
||||
|
||||
|
||||
async def off(self, color, brightness):
|
||||
if not (self._enabled and self._is_ready):
|
||||
return
|
||||
|
||||
if self._backend_mode == "xvf3800":
|
||||
# Prefer per-LED ring control when supported by firmware.
|
||||
if self._xvf3800_has_per_led():
|
||||
try:
|
||||
self._xvf3800_apply_ring_clear()
|
||||
except Exception:
|
||||
_LOGGER.exception("Error sending per-LED OFF to XVF3800")
|
||||
return
|
||||
|
||||
await self._xvf3800_apply_effect("off", _OFF, 0.0)
|
||||
return
|
||||
|
||||
self.leds.fill(_OFF)
|
||||
self.leds.show()
|
||||
|
||||
|
||||
async def solid(self, color: Tuple[int, int, int], brightness: float):
|
||||
if not (self._enabled and self._is_ready):
|
||||
return
|
||||
|
||||
if self._backend_mode == "xvf3800":
|
||||
if self._xvf3800_has_per_led():
|
||||
try:
|
||||
self._xvf3800_apply_ring_solid(color, brightness)
|
||||
except Exception:
|
||||
_LOGGER.exception("Error sending per-LED SOLID to XVF3800")
|
||||
return
|
||||
|
||||
await self._xvf3800_apply_effect("solid", color, brightness)
|
||||
return
|
||||
|
||||
r, g, b = color
|
||||
self.leds.fill(
|
||||
(int(r * brightness), int(g * brightness), int(b * brightness))
|
||||
)
|
||||
self.leds.fill((int(r * brightness), int(g * brightness), int(b * brightness)))
|
||||
self.leds.show()
|
||||
|
||||
async def blink(self, color, brightness=1.0):
|
||||
await self.medium_blink(color, brightness)
|
||||
|
||||
|
||||
async def _base_pulse(
|
||||
self, color: Tuple[int, int, int], brightness: float, speed: float
|
||||
self,
|
||||
effect_name: str,
|
||||
color: Tuple[int, int, int],
|
||||
brightness: float,
|
||||
speed: float,
|
||||
):
|
||||
if not (self._enabled and self._is_ready):
|
||||
return
|
||||
|
||||
if self._backend_mode == "xvf3800":
|
||||
if self._xvf3800_has_per_led():
|
||||
try:
|
||||
r, g, b = self._xvf3800_rgb_clamp(color)
|
||||
# Pulse by scaling the per-LED RGB values (some firmwares don't apply LED_BRIGHTNESS to LED_RING_COLOR).
|
||||
self._xvf3800_backend.set_effect(0)
|
||||
brightness = max(0.0, min(1.0, float(brightness)))
|
||||
while True:
|
||||
for i in range(0, 101, 10):
|
||||
s = (i / 100.0) * brightness
|
||||
self._xvf3800_backend.set_ring_solid(int(r * s), int(g * s), int(b * s))
|
||||
await asyncio.sleep(speed)
|
||||
for i in range(100, -1, -10):
|
||||
s = (i / 100.0) * brightness
|
||||
self._xvf3800_backend.set_ring_solid(int(r * s), int(g * s), int(b * s))
|
||||
await asyncio.sleep(speed)
|
||||
return
|
||||
except Exception:
|
||||
_LOGGER.exception("Error sending per-LED PULSE to XVF3800")
|
||||
return
|
||||
|
||||
await self._xvf3800_apply_effect(effect_name, color, brightness)
|
||||
return
|
||||
|
||||
try:
|
||||
r, g, b = color
|
||||
while True:
|
||||
for i in range(0, 101, 10):
|
||||
mod = i / 100.0 * brightness
|
||||
self.leds.fill(
|
||||
(int(r * mod), int(g * mod), int(b * mod))
|
||||
)
|
||||
self.leds.fill((int(r * mod), int(g * mod), int(b * mod)))
|
||||
self.leds.show()
|
||||
await asyncio.sleep(speed)
|
||||
for i in range(100, -1, -10):
|
||||
mod = i / 100.0 * brightness
|
||||
self.leds.fill(
|
||||
(int(r * mod), int(g * mod), int(b * mod))
|
||||
)
|
||||
self.leds.fill((int(r * mod), int(g * mod), int(b * mod)))
|
||||
self.leds.show()
|
||||
await asyncio.sleep(speed)
|
||||
except asyncio.CancelledError:
|
||||
if self._enabled and self._is_ready:
|
||||
if self._enabled and self._is_ready and self.leds is not None:
|
||||
self.leds.fill(_OFF)
|
||||
self.leds.show()
|
||||
|
||||
async def slow_pulse(self, color, brightness):
|
||||
await self._base_pulse(color, brightness, 0.05)
|
||||
await self._base_pulse("slow_pulse", color, brightness, 0.05)
|
||||
|
||||
async def medium_pulse(self, color, brightness):
|
||||
await self._base_pulse(color, brightness, 0.02)
|
||||
await self._base_pulse("medium_pulse", color, brightness, 0.02)
|
||||
|
||||
async def fast_pulse(self, color, brightness):
|
||||
await self._base_pulse(color, brightness, 0.008)
|
||||
await self._base_pulse("fast_pulse", color, brightness, 0.008)
|
||||
|
||||
|
||||
async def _base_blink(
|
||||
self, color: Tuple[int, int, int], brightness: float, speed: float
|
||||
self,
|
||||
effect_name: str,
|
||||
color: Tuple[int, int, int],
|
||||
brightness: float,
|
||||
speed: float,
|
||||
):
|
||||
if not (self._enabled and self._is_ready):
|
||||
return
|
||||
|
||||
if self._backend_mode == "xvf3800":
|
||||
if self._xvf3800_has_per_led():
|
||||
try:
|
||||
r, g, b = self._xvf3800_rgb_clamp(color)
|
||||
self._xvf3800_backend.set_effect(0)
|
||||
brightness = max(0.0, min(1.0, float(brightness)))
|
||||
while True:
|
||||
# On
|
||||
self._xvf3800_backend.set_ring_solid(int(r * brightness), int(g * brightness), int(b * brightness))
|
||||
await asyncio.sleep(speed)
|
||||
# Off
|
||||
self._xvf3800_backend.clear_ring()
|
||||
await asyncio.sleep(speed)
|
||||
except asyncio.CancelledError:
|
||||
try:
|
||||
self._xvf3800_apply_ring_clear()
|
||||
except Exception:
|
||||
pass
|
||||
return
|
||||
except Exception:
|
||||
_LOGGER.exception("Error sending per-LED BLINK to XVF3800")
|
||||
return
|
||||
|
||||
await self._xvf3800_apply_effect(effect_name, color, brightness)
|
||||
return
|
||||
|
||||
try:
|
||||
r, g, b = color
|
||||
bright_color = (
|
||||
int(r * brightness),
|
||||
int(g * brightness),
|
||||
int(b * brightness),
|
||||
)
|
||||
bright_color = (int(r * brightness), int(g * brightness), int(b * brightness))
|
||||
while True:
|
||||
self.leds.fill(bright_color)
|
||||
self.leds.show()
|
||||
@@ -251,24 +494,54 @@ class LedController(EventHandler):
|
||||
self.leds.show()
|
||||
await asyncio.sleep(speed)
|
||||
except asyncio.CancelledError:
|
||||
if self._enabled and self._is_ready:
|
||||
if self._enabled and self._is_ready and self.leds is not None:
|
||||
self.leds.fill(_OFF)
|
||||
self.leds.show()
|
||||
|
||||
async def slow_blink(self, color, brightness):
|
||||
await self._base_blink(color, brightness, 1.0)
|
||||
await self._base_blink("slow_blink", color, brightness, 1.0)
|
||||
|
||||
async def medium_blink(self, color, brightness):
|
||||
await self._base_blink(color, brightness, 0.5)
|
||||
await self._base_blink("medium_blink", color, brightness, 0.5)
|
||||
|
||||
async def fast_blink(self, color, brightness):
|
||||
await self._base_blink(color, brightness, 0.1)
|
||||
await self._base_blink("fast_blink", color, brightness, 0.1)
|
||||
|
||||
|
||||
async def spin(
|
||||
self, color: Tuple[int, int, int], brightness: float, speed: float = 0.1
|
||||
):
|
||||
if not (self._enabled and self._is_ready):
|
||||
return
|
||||
|
||||
if self._backend_mode == "xvf3800":
|
||||
if self._xvf3800_has_per_led():
|
||||
try:
|
||||
r, g, b = self._xvf3800_rgb_clamp(color)
|
||||
ring_n = self._xvf3800_ring_count()
|
||||
self._xvf3800_backend.set_effect(0)
|
||||
self._xvf3800_backend.set_brightness(self._xvf3800_brightness_255(brightness))
|
||||
|
||||
i = 0
|
||||
while True:
|
||||
colors = [(0, 0, 0)] * ring_n
|
||||
colors[i % ring_n] = (r, g, b)
|
||||
self._xvf3800_backend.set_ring_rgb(colors)
|
||||
i += 1
|
||||
await asyncio.sleep(speed)
|
||||
except asyncio.CancelledError:
|
||||
try:
|
||||
self._xvf3800_apply_ring_clear()
|
||||
except Exception:
|
||||
pass
|
||||
return
|
||||
except Exception:
|
||||
_LOGGER.exception("Error sending per-LED SPIN to XVF3800")
|
||||
return
|
||||
|
||||
await self._xvf3800_apply_effect("spin", color, brightness)
|
||||
return
|
||||
|
||||
try:
|
||||
i = 0
|
||||
bright_color = (
|
||||
@@ -283,14 +556,10 @@ class LedController(EventHandler):
|
||||
i += 1
|
||||
await asyncio.sleep(speed)
|
||||
except asyncio.CancelledError:
|
||||
if self._enabled and self._is_ready:
|
||||
if self._enabled and self._is_ready and self.leds is not None:
|
||||
self.leds.fill(_OFF)
|
||||
self.leds.show()
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# State Subscriptions
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
@subscribe
|
||||
def voice_idle(self, data: dict):
|
||||
self._apply_state_effect("idle")
|
||||
@@ -313,6 +582,7 @@ class LedController(EventHandler):
|
||||
|
||||
@subscribe
|
||||
def mic_muted(self, data: dict):
|
||||
# When muted, show a solid dim red on all backends.
|
||||
self.run_action("solid", _DIM_RED, 1.0)
|
||||
|
||||
@subscribe
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
"""
|
||||
Media player using mpv in a subprocess.
|
||||
"""Media player using mpv in a subprocess.
|
||||
|
||||
This wrapper focuses on:
|
||||
- Simple playback control (play / pause / resume / stop)
|
||||
@@ -9,7 +8,17 @@ This wrapper focuses on:
|
||||
|
||||
If a specific audio device is provided, it is passed directly to mpv as
|
||||
`audio-device`. Otherwise, mpv's own automatic backend/device selection is used.
|
||||
|
||||
Note about volume:
|
||||
- mpv has its own per-player volume (0..100).
|
||||
- PipeWire/PulseAudio/ALSA also has a system output volume.
|
||||
|
||||
LVA persists a single user volume (0.0..1.0) in preferences.json. To avoid
|
||||
"mystery caps" where the OS sink is stuck at e.g. 40% while LVA shows 100%,
|
||||
LVA treats the OS output volume as the "master" and keeps mpv at 100% for normal
|
||||
playback. Ducking still uses mpv's per-player volume.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
@@ -21,6 +30,8 @@ from typing import Callable, List, Optional, Sequence, Union
|
||||
# Note: python-mpv must be installed; imported at runtime.
|
||||
from mpv import MPV
|
||||
|
||||
from .audio_volume import set_output_volume
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@@ -33,13 +44,19 @@ class MpvMediaPlayer:
|
||||
device: Optional[str] = None,
|
||||
initial_volume: float = 1.0,
|
||||
) -> None:
|
||||
"""
|
||||
"""Initialize the mpv player.
|
||||
|
||||
:param loop: The asyncio loop used to schedule done_callback.
|
||||
May be None in one-off utility contexts (e.g. listing devices).
|
||||
:param device: Optional mpv audio device name (e.g. "pulse/alsa_output.pci-0000_00_1f.3.analog-stereo").
|
||||
:param initial_volume: Initial volume as a float 0.0–1.0.
|
||||
:param device: Optional mpv audio device name (e.g.
|
||||
"pipewire/alsa_output.pci-0000_00_1f.3.analog-stereo").
|
||||
:param initial_volume: Initial volume as a float 0.0–1.0. This is now
|
||||
treated as the *master* OS output volume. mpv's
|
||||
internal volume is set to 100% for normal playback.
|
||||
"""
|
||||
self.loop = loop
|
||||
self.device = device
|
||||
self.initial_volume = max(0.0, min(1.0, float(initial_volume)))
|
||||
|
||||
self.player = MPV(
|
||||
video=False,
|
||||
@@ -83,8 +100,7 @@ class MpvMediaPlayer:
|
||||
try:
|
||||
dev_list = self.player.audio_device_list or []
|
||||
dev_summary = [
|
||||
f"{dev.get('name')} ({dev.get('description')})"
|
||||
for dev in dev_list
|
||||
f"{dev.get('name')} ({dev.get('description')})" for dev in dev_list
|
||||
]
|
||||
except Exception:
|
||||
dev_summary = ["<unavailable>"]
|
||||
@@ -98,8 +114,8 @@ class MpvMediaPlayer:
|
||||
except Exception:
|
||||
_LOGGER.exception("Failed to query mpv audio properties")
|
||||
|
||||
# Volume is 0–100 in mpv, we accept 0.0–1.0 here.
|
||||
self.set_volume(int(initial_volume * 100))
|
||||
# Keep mpv at 100% for normal playback. Master volume is handled at the OS level.
|
||||
self.set_volume(100)
|
||||
|
||||
self.is_playing: bool = False
|
||||
self._done_callback: Optional[Callable[[], None]] = None
|
||||
@@ -180,14 +196,22 @@ class MpvMediaPlayer:
|
||||
self._run_done_callback()
|
||||
|
||||
def set_volume(self, volume: int) -> None:
|
||||
"""Sets the player volume from 0 to 100."""
|
||||
"""Sets the player (mpv) volume from 0 to 100."""
|
||||
try:
|
||||
self.player.volume = max(0, min(100, volume))
|
||||
except Exception:
|
||||
_LOGGER.exception("set_volume() failed")
|
||||
|
||||
def set_master_volume(self, volume_level: float) -> bool:
|
||||
"""Sets the *OS output* volume (PipeWire/PulseAudio/ALSA).
|
||||
|
||||
:param volume_level: 0.0–1.0
|
||||
:returns: True if any backend succeeded.
|
||||
"""
|
||||
return set_output_volume(volume_level=volume_level, output_device=self.device)
|
||||
|
||||
def duck(self, target_percent: int = 20) -> None:
|
||||
"""Lowers the volume for an announcement."""
|
||||
"""Lowers the mpv volume for an announcement."""
|
||||
if self._pre_duck_volume is not None:
|
||||
return
|
||||
try:
|
||||
@@ -197,7 +221,7 @@ class MpvMediaPlayer:
|
||||
_LOGGER.exception("duck() failed")
|
||||
|
||||
def unduck(self) -> None:
|
||||
"""Restores the volume after an announcement."""
|
||||
"""Restores the mpv volume after an announcement."""
|
||||
if self._pre_duck_volume is None:
|
||||
return
|
||||
try:
|
||||
|
||||
@@ -70,6 +70,12 @@ class MqttController(EventHandler):
|
||||
"light_state": f"{self._topic_prefix}/{state_name}_light/state",
|
||||
}
|
||||
|
||||
# During startup we accept retained */state topics from the broker to restore
|
||||
# Home Assistant's last-known settings. After bootstrap, we ignore */state topics
|
||||
# to avoid feedback loops (we only accept */set as commands).
|
||||
self._bootstrap_state_sync = True
|
||||
self._bootstrap_ends_at: Optional[float] = None
|
||||
|
||||
self._client = mqtt.Client()
|
||||
self._client.on_connect = self._on_connect
|
||||
self._client.on_message = self._on_message
|
||||
@@ -102,10 +108,26 @@ class MqttController(EventHandler):
|
||||
_LOGGER.info("Connected to MQTT broker")
|
||||
client.subscribe(f"{self._topic_prefix}/+/set")
|
||||
client.subscribe(f"{self._topic_prefix}/+/state")
|
||||
# End bootstrap a moment after subscribe so retained state messages can arrive.
|
||||
self._bootstrap_ends_at = self.loop.time() + 2.0
|
||||
self.loop.call_later(2.0, self._end_bootstrap_state_sync)
|
||||
self._publish_discovery_configs()
|
||||
else:
|
||||
_LOGGER.error("Failed to connect to MQTT, return code %d", rc)
|
||||
|
||||
def _end_bootstrap_state_sync(self):
|
||||
# Called on the asyncio loop thread.
|
||||
self._bootstrap_state_sync = False
|
||||
# We only needed */state subscriptions to ingest retained settings at startup.
|
||||
# After bootstrap, unsubscribe to avoid feedback loops and log spam.
|
||||
try:
|
||||
self._client.unsubscribe(f"{self._topic_prefix}/+/state")
|
||||
except Exception:
|
||||
_LOGGER.debug("Failed to unsubscribe from */state topics", exc_info=True)
|
||||
_LOGGER.debug(
|
||||
"MQTT bootstrap state sync complete; unsubscribed from */state and will ignore any state topics"
|
||||
)
|
||||
|
||||
def _on_message(self, client, userdata, msg):
|
||||
try:
|
||||
topic = msg.topic
|
||||
@@ -113,15 +135,24 @@ class MqttController(EventHandler):
|
||||
_LOGGER.debug(
|
||||
"Received MQTT message on topic %s (from MQTT thread)", msg.topic
|
||||
)
|
||||
retain = bool(getattr(msg, "retain", False))
|
||||
self.loop.call_soon_threadsafe(
|
||||
self._handle_message_on_loop, topic, payload
|
||||
self._handle_message_on_loop, topic, payload, retain
|
||||
)
|
||||
except Exception:
|
||||
_LOGGER.exception("Error in _on_message")
|
||||
|
||||
def _handle_message_on_loop(self, topic: str, payload: str):
|
||||
def _handle_message_on_loop(self, topic: str, payload: str, retained: bool = False):
|
||||
_LOGGER.debug("Handling message for topic %s on main loop", topic)
|
||||
|
||||
# Prevent feedback loops: we only treat */set as commands.
|
||||
# However during bootstrap we accept retained */state values to restore HA settings.
|
||||
if topic.endswith("/state"):
|
||||
if not (self._bootstrap_state_sync and retained):
|
||||
_LOGGER.debug("Ignoring MQTT state topic: %s (retained=%s, bootstrap=%s)", topic, retained, self._bootstrap_state_sync)
|
||||
return
|
||||
|
||||
|
||||
if topic == self.topics["mute"]["command"]:
|
||||
# Publish event for MicMuteHandler to process
|
||||
self.event_bus.publish(
|
||||
@@ -183,12 +214,12 @@ class MqttController(EventHandler):
|
||||
effect_id = payload.lower().replace(" ", "_")
|
||||
self.event_bus.publish(
|
||||
f"set_{state_name}_effect",
|
||||
{"effect": effect_id, "retained": True},
|
||||
{"effect": effect_id, "retained": retained},
|
||||
)
|
||||
elif topic == state_topics["light_state"]:
|
||||
try:
|
||||
data = json.loads(payload)
|
||||
data["retained"] = True
|
||||
data["retained"] = retained
|
||||
self.event_bus.publish(
|
||||
f"set_{state_name}_color", data
|
||||
)
|
||||
|
||||
@@ -0,0 +1,429 @@
|
||||
"""XVF3800 USB-based button & mute controller.
|
||||
|
||||
This module integrates the ReSpeaker XVF3800's built-in mute/mic-control
|
||||
with the Linux Voice Assistant (LVA):
|
||||
|
||||
- Observes the XVF3800's GPO state to detect hardware mute changes.
|
||||
- Publishes "set_mic_mute" events on the EventBus when the user presses
|
||||
the physical mute button on the XVF3800.
|
||||
- Listens for "mic_muted"/"mic_unmuted" events and mirrors the LVA's
|
||||
mute state back to the XVF3800 via its USB control interface.
|
||||
|
||||
The XVF3800's GPIO mapping (from Seeed docs) is:
|
||||
|
||||
GPI (inputs):
|
||||
X1D09 - Mute button state (high = released, low = pressed)
|
||||
X1D13 - Floating
|
||||
X1D34 - Floating
|
||||
|
||||
GPO (outputs):
|
||||
X0D11 - Floating
|
||||
X0D30 - Mic mute + red mute LED (high = muted, LED on)
|
||||
X0D31 - Amplifier enable (low = enabled)
|
||||
X0D33 - WS2812 LED power enable
|
||||
X0D39 - Floating
|
||||
|
||||
We do NOT try to interpret short/long presses like the generic GPIO
|
||||
ButtonController; the XVF3800 mute button is treated as "mute toggle only".
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Optional, TYPE_CHECKING
|
||||
|
||||
import usb.core # type: ignore[import]
|
||||
import usb.util # type: ignore[import]
|
||||
|
||||
from .event_bus import EventBus, EventHandler, subscribe
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .models import ServerState
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Low-level USB client (minimal subset of the XMOS/Seeed control protocol)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class XVF3800USBClient:
|
||||
"""Minimal USB control client for the ReSpeaker XVF3800.
|
||||
|
||||
This wraps the vendor-specific control transfers needed for:
|
||||
- GPO_READ_VALUES (read mic mute LED / amp / LED power pins)
|
||||
- GPO_WRITE_VALUE (set mic mute LED pin)
|
||||
|
||||
It deliberately does NOT touch interfaces/altsettings so that audio
|
||||
streaming over USB is unaffected.
|
||||
"""
|
||||
|
||||
VENDOR_ID = 0x2886
|
||||
PRODUCT_ID = 0x001A
|
||||
TIMEOUT_MS = 1000
|
||||
|
||||
# From Seeed's documentation (GPO control):
|
||||
# - Resid = 20 (GPO_SERVICER_RESID)
|
||||
# - CmdId = 0 -> GPO_READ_VALUES (3/5 bytes depending on doc version)
|
||||
# - CmdId = 1 -> GPO_WRITE_VALUE
|
||||
#
|
||||
# We expect GPO_READ_VALUES to return 5 bytes:
|
||||
# [X0D11, X0D30, X0D31, X0D33, X0D39]
|
||||
#
|
||||
GPO_RESID = 20
|
||||
GPO_READ_CMDID = 0
|
||||
GPO_WRITE_CMDID = 1
|
||||
GPO_NUM_PINS = 5
|
||||
|
||||
# Index of the mic mute/red LED pin in the GPO_READ_VALUES response
|
||||
# (X0D30, second element).
|
||||
GPO_MUTE_INDEX = 1
|
||||
# Index of the WS2812 LED power enable pin in the response (X0D33).
|
||||
GPO_WS2812_POWER_INDEX = 3
|
||||
|
||||
def __init__(self) -> None:
|
||||
dev = usb.core.find(idVendor=self.VENDOR_ID, idProduct=self.PRODUCT_ID)
|
||||
if dev is None:
|
||||
raise RuntimeError(
|
||||
"ReSpeaker XVF3800 USB device not found "
|
||||
f"(VID=0x{self.VENDOR_ID:04X}, PID=0x{self.PRODUCT_ID:04X})"
|
||||
)
|
||||
|
||||
self._dev = dev
|
||||
_LOGGER.debug(
|
||||
"Initialized XVF3800USBClient (bus=%s, address=%s)",
|
||||
getattr(dev, "bus", "?"),
|
||||
getattr(dev, "address", "?"),
|
||||
)
|
||||
|
||||
# CRITICAL FIX: Add context manager support
|
||||
def __enter__(self):
|
||||
"""Context manager entry."""
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
"""Context manager exit."""
|
||||
self.close()
|
||||
return False
|
||||
|
||||
def close(self) -> None:
|
||||
"""Dispose of USB resources."""
|
||||
if hasattr(self, '_dev') and self._dev is not None:
|
||||
try:
|
||||
usb.util.dispose_resources(self._dev)
|
||||
except Exception as e:
|
||||
_LOGGER.debug("Error disposing XVF3800 USB resources: %s", e)
|
||||
finally:
|
||||
self._dev = None
|
||||
|
||||
# Internal helpers -----------------------------------------------------
|
||||
|
||||
def _ctrl_read(self, resid: int, cmdid: int, length: int) -> List[int]:
|
||||
"""Perform a vendor-specific control IN transfer and return payload bytes."""
|
||||
# Per XMOS protocol: read cmdid is (0x80 | cmdid)
|
||||
wValue = 0x80 | cmdid
|
||||
wIndex = resid
|
||||
|
||||
data = self._dev.ctrl_transfer(
|
||||
usb.util.CTRL_IN | usb.util.CTRL_TYPE_VENDOR | usb.util.CTRL_RECIPIENT_DEVICE,
|
||||
0,
|
||||
wValue,
|
||||
wIndex,
|
||||
length,
|
||||
self.TIMEOUT_MS,
|
||||
)
|
||||
if not data:
|
||||
raise RuntimeError("Empty response from XVF3800 vendor control read")
|
||||
|
||||
status = data[0]
|
||||
# 0 = CONTROL_SUCCESS, 64 = SERVICER_COMMAND_RETRY in XMOS docs,
|
||||
# but for our simple use case we just require success.
|
||||
if status != 0:
|
||||
raise RuntimeError(f"Unexpected XVF3800 control status: {status}")
|
||||
|
||||
# Return the payload bytes as a Python list, excluding status byte.
|
||||
return list(data[1:])
|
||||
|
||||
def _ctrl_write(self, resid: int, cmdid: int, payload: bytes) -> None:
|
||||
"""Perform a vendor-specific control OUT transfer."""
|
||||
wValue = cmdid
|
||||
wIndex = resid
|
||||
|
||||
# Note: ctrl_transfer will raise usb.core.USBError on failure.
|
||||
self._dev.ctrl_transfer(
|
||||
usb.util.CTRL_OUT | usb.util.CTRL_TYPE_VENDOR | usb.util.CTRL_RECIPIENT_DEVICE,
|
||||
0,
|
||||
wValue,
|
||||
wIndex,
|
||||
payload,
|
||||
self.TIMEOUT_MS,
|
||||
)
|
||||
|
||||
# Public API -----------------------------------------------------------
|
||||
|
||||
def read_gpo_values(self) -> List[int]:
|
||||
"""Read all GPO pin values.
|
||||
|
||||
Returns a list of integers, one per pin:
|
||||
[X0D11, X0D30, X0D31, X0D33, X0D39]
|
||||
"""
|
||||
length = self.GPO_NUM_PINS + 1 # +1 for status byte
|
||||
values = self._ctrl_read(self.GPO_RESID, self.GPO_READ_CMDID, length)
|
||||
if len(values) < self.GPO_NUM_PINS:
|
||||
_LOGGER.warning(
|
||||
"XVF3800 GPO_READ_VALUES returned %d bytes, expected >= %d",
|
||||
len(values),
|
||||
self.GPO_NUM_PINS,
|
||||
)
|
||||
return values
|
||||
|
||||
def set_gpo_pin(self, pin: int, value: bool) -> bool:
|
||||
"""Set a single GPO pin high/low.
|
||||
|
||||
Args:
|
||||
pin: GPO pin number (e.g., 30 for X0D30, 33 for X0D33)
|
||||
value: True -> high, False -> low
|
||||
|
||||
Returns:
|
||||
True on success, False on error.
|
||||
"""
|
||||
payload = bytes([pin, 1 if value else 0])
|
||||
try:
|
||||
self._ctrl_write(self.GPO_RESID, self.GPO_WRITE_CMDID, payload)
|
||||
return True
|
||||
except usb.core.USBError as err:
|
||||
_LOGGER.error("USBError writing XVF3800 GPO pin %s value: %s", pin, err)
|
||||
except Exception:
|
||||
_LOGGER.exception("Unexpected error writing XVF3800 GPO pin %s value", pin)
|
||||
return False
|
||||
|
||||
def get_mute_gpo(self) -> Optional[bool]:
|
||||
"""Return the current mic-mute output state from X0D30.
|
||||
|
||||
Returns:
|
||||
True -> mics muted / red mute LED on
|
||||
False -> mics unmuted / red mute LED off
|
||||
None -> could not read values (USB error)
|
||||
"""
|
||||
try:
|
||||
values = self.read_gpo_values()
|
||||
if len(values) <= self.GPO_MUTE_INDEX:
|
||||
_LOGGER.error(
|
||||
"XVF3800 GPO_READ_VALUES payload too short: %r", values
|
||||
)
|
||||
return None
|
||||
mute_val = values[self.GPO_MUTE_INDEX]
|
||||
return bool(mute_val)
|
||||
except usb.core.USBError as err:
|
||||
_LOGGER.error("USBError reading XVF3800 GPO values: %s", err)
|
||||
except Exception:
|
||||
_LOGGER.exception("Unexpected error reading XVF3800 GPO values")
|
||||
return None
|
||||
|
||||
def set_mute_gpo(self, muted: bool) -> bool:
|
||||
"""Set the mic-mute output (X0D30) high/low."""
|
||||
return self.set_gpo_pin(30, muted)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# High-level controller: bridge between XVF3800 and LVA mute state
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass
|
||||
class XVF3800ButtonRuntimeConfig:
|
||||
"""Runtime config for the XVF3800 button/mute controller."""
|
||||
|
||||
# CRITICAL FIX: Change default from 0.01s to 0.05s (100 Hz -> 20 Hz)
|
||||
# 80% reduction in CPU usage for button monitoring
|
||||
poll_interval_seconds: float = 0.05 # 20 Hz polling
|
||||
|
||||
|
||||
class XVF3800ButtonController(EventHandler):
|
||||
"""Monitor & synchronize the XVF3800's mute state with LVA."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
loop,
|
||||
event_bus: EventBus,
|
||||
state: "ServerState",
|
||||
button_config,
|
||||
) -> None:
|
||||
super().__init__(event_bus)
|
||||
self.loop = loop
|
||||
self.state = state
|
||||
|
||||
# CRITICAL FIX: Default to 0.05s instead of 0.01s
|
||||
poll_interval = 0.05
|
||||
if hasattr(button_config, "poll_interval_seconds"):
|
||||
try:
|
||||
poll_interval = float(button_config.poll_interval_seconds)
|
||||
except Exception:
|
||||
_LOGGER.warning(
|
||||
"Invalid poll_interval_seconds in button config; "
|
||||
"defaulting to %.3fs",
|
||||
poll_interval,
|
||||
)
|
||||
|
||||
self._cfg = XVF3800ButtonRuntimeConfig(
|
||||
poll_interval_seconds=poll_interval,
|
||||
)
|
||||
|
||||
self._thread: Optional[threading.Thread] = None
|
||||
self._shutdown_flag = threading.Event()
|
||||
|
||||
self._last_hw_muted: Optional[bool] = None
|
||||
|
||||
self._target_mute_state_lock = threading.Lock()
|
||||
self._target_mute_state: Optional[bool] = None
|
||||
|
||||
self._usb_client: Optional[XVF3800USBClient] = None
|
||||
|
||||
self._thread = threading.Thread(
|
||||
target=self._poll_loop,
|
||||
name="XVF3800ButtonControllerThread",
|
||||
daemon=True,
|
||||
)
|
||||
self._thread.start()
|
||||
_LOGGER.info(
|
||||
"XVF3800ButtonController started (poll_interval=%.3fs)",
|
||||
self._cfg.poll_interval_seconds,
|
||||
)
|
||||
|
||||
self._subscribe_all_methods()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Event handlers from EventBus
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@subscribe
|
||||
def mic_muted(self, data: dict) -> None:
|
||||
self._set_target_mute_state(True)
|
||||
|
||||
@subscribe
|
||||
def mic_unmuted(self, data: dict) -> None:
|
||||
self._set_target_mute_state(False)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public API
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def stop(self) -> None:
|
||||
self._shutdown_flag.set()
|
||||
if self._thread is not None:
|
||||
self._thread.join(timeout=2.0)
|
||||
if self._usb_client is not None:
|
||||
self._usb_client.close()
|
||||
_LOGGER.info("XVF3800ButtonController stopped")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Internal helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _set_target_mute_state(self, muted: bool) -> None:
|
||||
with self._target_mute_state_lock:
|
||||
self._target_mute_state = muted
|
||||
|
||||
def _take_target_mute_state(self) -> Optional[bool]:
|
||||
with self._target_mute_state_lock:
|
||||
value = self._target_mute_state
|
||||
self._target_mute_state = None
|
||||
return value
|
||||
|
||||
def _ensure_usb_client(self) -> Optional[XVF3800USBClient]:
|
||||
if self._usb_client is not None:
|
||||
return self._usb_client
|
||||
|
||||
try:
|
||||
self._usb_client = XVF3800USBClient()
|
||||
_LOGGER.info("Connected to ReSpeaker XVF3800 for mute control")
|
||||
except Exception:
|
||||
_LOGGER.exception(
|
||||
"Failed to initialize XVF3800 USB client; "
|
||||
"mute button integration will be disabled"
|
||||
)
|
||||
self._usb_client = None
|
||||
return self._usb_client
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Polling loop
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _poll_loop(self) -> None:
|
||||
_LOGGER.debug("XVF3800ButtonController polling thread started")
|
||||
|
||||
while not self._shutdown_flag.is_set() and not getattr(
|
||||
self.state, "shutdown", False
|
||||
):
|
||||
client = self._ensure_usb_client()
|
||||
if client is None:
|
||||
time.sleep(2.0)
|
||||
continue
|
||||
|
||||
# 1) Apply any pending target state from LVA -> hardware
|
||||
target = self._take_target_mute_state()
|
||||
if target is not None:
|
||||
if self._last_hw_muted is None or target != self._last_hw_muted:
|
||||
success = client.set_mute_gpo(target)
|
||||
if success:
|
||||
self._last_hw_muted = target
|
||||
_LOGGER.debug(
|
||||
"Set XVF3800 hardware mute state -> %s", target
|
||||
)
|
||||
|
||||
# 2) Read current hardware GPO values (mute + WS2812 power)
|
||||
try:
|
||||
values = client.read_gpo_values()
|
||||
hw_muted = None
|
||||
if len(values) > client.GPO_MUTE_INDEX:
|
||||
hw_muted = bool(values[client.GPO_MUTE_INDEX])
|
||||
|
||||
# If the firmware/button toggles WS2812 power off, re-enable it so
|
||||
# the ring can still display LVA state (mute, idle effects, etc.)
|
||||
if len(values) > client.GPO_WS2812_POWER_INDEX:
|
||||
ws2812_power = bool(values[client.GPO_WS2812_POWER_INDEX])
|
||||
if not ws2812_power:
|
||||
if client.set_gpo_pin(33, True):
|
||||
_LOGGER.debug("Re-enabled XVF3800 WS2812 LED power (X0D33)")
|
||||
else:
|
||||
ws2812_power = None
|
||||
|
||||
except usb.core.USBError as err:
|
||||
_LOGGER.error("USBError reading XVF3800 GPO values: %s", err)
|
||||
hw_muted = None
|
||||
except Exception:
|
||||
_LOGGER.exception("Unexpected error reading XVF3800 GPO values")
|
||||
hw_muted = None
|
||||
|
||||
if hw_muted is not None:
|
||||
if self._last_hw_muted is None:
|
||||
self._last_hw_muted = hw_muted
|
||||
_LOGGER.info(
|
||||
"Initial XVF3800 hardware mute state: %s", hw_muted
|
||||
)
|
||||
self.loop.call_soon_threadsafe(
|
||||
self.state.event_bus.publish,
|
||||
"set_mic_mute",
|
||||
{"state": hw_muted, "source": "xvf3800_hw"},
|
||||
)
|
||||
elif hw_muted != self._last_hw_muted:
|
||||
_LOGGER.info(
|
||||
"Detected XVF3800 mute state change from %s to %s; "
|
||||
"publishing set_mic_mute event",
|
||||
self._last_hw_muted,
|
||||
hw_muted,
|
||||
)
|
||||
self._last_hw_muted = hw_muted
|
||||
self.loop.call_soon_threadsafe(
|
||||
self.state.event_bus.publish,
|
||||
"set_mic_mute",
|
||||
{"state": hw_muted, "source": "xvf3800_hw"},
|
||||
)
|
||||
|
||||
time.sleep(self._cfg.poll_interval_seconds)
|
||||
|
||||
_LOGGER.debug("XVF3800ButtonController polling thread exiting")
|
||||
@@ -0,0 +1,444 @@
|
||||
"""XVF3800 USB LED ring backend.
|
||||
|
||||
This backend supports two generations of LED control:
|
||||
|
||||
1) Legacy/global LED control (older firmware):
|
||||
- LED_EFFECT, LED_BRIGHTNESS, LED_SPEED, LED_COLOR
|
||||
|
||||
2) Per-LED ring control (newer firmware):
|
||||
- LED_RING_COLOR: 12x uint32 values, one per LED (0xRRGGBB)
|
||||
|
||||
We intentionally keep the implementation minimal and self-contained so it can be
|
||||
used by LVA without shelling out to xvf_host.py.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import struct
|
||||
import time
|
||||
from typing import List, Optional, Sequence, Tuple
|
||||
|
||||
import usb.core # type: ignore[import]
|
||||
import usb.util # type: ignore[import]
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
CONTROL_SUCCESS = 0
|
||||
SERVICER_COMMAND_RETRY = 64
|
||||
|
||||
# name -> (resid, cmdid, count, access, type)
|
||||
PARAMETERS = {
|
||||
# ---------------------------------------------------------------------
|
||||
# APPLICATION_SERVICER_RESID (core info)
|
||||
# ---------------------------------------------------------------------
|
||||
"VERSION": (48, 0, 3, "ro", "uint8"),
|
||||
|
||||
# Firmware / device control
|
||||
"REBOOT": (48, 7, 1, "wo", "uint8"),
|
||||
"SAVE_CONFIGURATION": (48, 9, 1, "wo", "uint8"),
|
||||
|
||||
# Audio manager output routing (category, source) per channel
|
||||
"AUDIO_MGR_OP_L": (35, 15, 2, "rw", "uint8"),
|
||||
"AUDIO_MGR_OP_R": (35, 19, 2, "rw", "uint8"),
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# GPO_SERVICER_RESID (LED controls)
|
||||
# ---------------------------------------------------------------------
|
||||
# Legacy/global controls
|
||||
"LED_EFFECT": (20, 12, 1, "rw", "uint8"),
|
||||
"LED_BRIGHTNESS": (20, 13, 1, "rw", "uint8"),
|
||||
"LED_GAMMIFY": (20, 14, 1, "rw", "uint8"),
|
||||
"LED_SPEED": (20, 15, 1, "rw", "uint8"),
|
||||
"LED_COLOR": (20, 16, 1, "rw", "uint32"),
|
||||
|
||||
# Newer firmware: per-LED ring control (12 WS2812 LEDs)
|
||||
"LED_RING_COLOR": (20, 19, 12, "rw", "uint32"),
|
||||
|
||||
# GPO control for button monitoring and LED power
|
||||
"GPO_READ_VALUES": (20, 0, 5, "ro", "uint8"), # [X0D11, X0D30, X0D31, X0D33, X0D39]
|
||||
"GPO_WRITE_VALUE": (20, 1, 2, "wo", "uint8"), # [pin_index, value]
|
||||
}
|
||||
|
||||
|
||||
class _ReSpeaker:
|
||||
"""Low-level USB control wrapper for XVF3800 parameters."""
|
||||
|
||||
TIMEOUT_MS = 100_000
|
||||
VID = 0x2886
|
||||
PID = 0x001A
|
||||
|
||||
def __init__(self, dev: "usb.core.Device") -> None: # type: ignore[name-defined]
|
||||
self.dev = dev
|
||||
|
||||
# CRITICAL FIX: Add context manager support
|
||||
def __enter__(self):
|
||||
"""Context manager entry."""
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
"""Context manager exit."""
|
||||
self.close()
|
||||
return False
|
||||
|
||||
def close(self) -> None:
|
||||
"""Release any underlying libusb resources (best-effort)."""
|
||||
if hasattr(self, 'dev') and self.dev is not None:
|
||||
try:
|
||||
usb.util.dispose_resources(self.dev)
|
||||
except Exception as e:
|
||||
_LOGGER.debug("Error disposing USB resources: %s", e)
|
||||
finally:
|
||||
self.dev = None
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Encoding / decoding helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _pack_values(self, data_type: str, values: Sequence[int]) -> bytes:
|
||||
if data_type == "uint8":
|
||||
return bytes([int(v) & 0xFF for v in values])
|
||||
if data_type == "uint32":
|
||||
return b"".join(struct.pack("<I", int(v) & 0xFFFFFFFF) for v in values)
|
||||
if data_type == "int32":
|
||||
return b"".join(struct.pack("<i", int(v)) for v in values)
|
||||
raise ValueError(f"Unsupported data type '{data_type}'")
|
||||
|
||||
def _unpack_values(self, data_type: str, raw: bytes, count: int) -> List[int]:
|
||||
if data_type == "uint8":
|
||||
return list(raw[:count])
|
||||
if data_type == "uint32":
|
||||
return list(struct.unpack("<" + "I" * count, raw[: count * 4]))
|
||||
if data_type == "int32":
|
||||
return list(struct.unpack("<" + "i" * count, raw[: count * 4]))
|
||||
raise ValueError(f"Unsupported data type '{data_type}'")
|
||||
|
||||
def _read_length(self, data_type: str, count: int) -> int:
|
||||
# +1 for status byte returned by XMOS vendor control read
|
||||
if data_type == "uint8":
|
||||
return count + 1
|
||||
if data_type in ("uint32", "int32"):
|
||||
return (count * 4) + 1
|
||||
raise ValueError(f"Unsupported data type '{data_type}'")
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public parameter IO
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def write(self, name: str, data_list: Sequence[int]) -> None:
|
||||
try:
|
||||
resid, cmdid, count, access, data_type = PARAMETERS[name]
|
||||
except KeyError as exc:
|
||||
raise ValueError(f"Unknown XVF3800 parameter '{name}'") from exc
|
||||
|
||||
if access == "ro":
|
||||
raise ValueError(f"{name} is read-only")
|
||||
|
||||
if len(data_list) != count:
|
||||
raise ValueError(f"{name} expects {count} values, got {len(data_list)}")
|
||||
|
||||
payload = self._pack_values(data_type, data_list)
|
||||
|
||||
_LOGGER.debug(
|
||||
"XVF3800 write: name=%s resid=%s cmdid=%s payload_len=%d",
|
||||
name,
|
||||
resid,
|
||||
cmdid,
|
||||
len(payload),
|
||||
)
|
||||
|
||||
self.dev.ctrl_transfer(
|
||||
usb.util.CTRL_OUT
|
||||
| usb.util.CTRL_TYPE_VENDOR
|
||||
| usb.util.CTRL_RECIPIENT_DEVICE,
|
||||
0,
|
||||
cmdid,
|
||||
resid,
|
||||
payload,
|
||||
self.TIMEOUT_MS,
|
||||
)
|
||||
|
||||
def read(self, name: str, max_retries: int = 10) -> List[int]:
|
||||
try:
|
||||
resid, cmdid, count, access, data_type = PARAMETERS[name]
|
||||
except KeyError as exc:
|
||||
raise ValueError(f"Unknown XVF3800 parameter '{name}'") from exc
|
||||
|
||||
if access == "wo":
|
||||
raise ValueError(f"{name} is write-only")
|
||||
|
||||
length = self._read_length(data_type, count)
|
||||
wValue = 0x80 | cmdid # per XMOS protocol: read is (0x80 | cmdid)
|
||||
|
||||
attempt = 0
|
||||
while True:
|
||||
attempt += 1
|
||||
resp = self.dev.ctrl_transfer(
|
||||
usb.util.CTRL_IN
|
||||
| usb.util.CTRL_TYPE_VENDOR
|
||||
| usb.util.CTRL_RECIPIENT_DEVICE,
|
||||
0,
|
||||
wValue,
|
||||
resid,
|
||||
length,
|
||||
self.TIMEOUT_MS,
|
||||
)
|
||||
|
||||
if not resp:
|
||||
raise RuntimeError("Empty response from XVF3800 control read")
|
||||
|
||||
status = int(resp[0])
|
||||
if status == CONTROL_SUCCESS:
|
||||
raw = bytes(resp[1:])
|
||||
return self._unpack_values(data_type, raw, count)
|
||||
|
||||
if status == SERVICER_COMMAND_RETRY and attempt < max_retries:
|
||||
continue
|
||||
|
||||
raise RuntimeError(f"XVF3800 control read failed (status={status}, name={name})")
|
||||
|
||||
|
||||
def _find_device(vid: int = _ReSpeaker.VID, pid: int = _ReSpeaker.PID) -> Optional[_ReSpeaker]:
|
||||
dev = usb.core.find(idVendor=vid, idProduct=pid)
|
||||
if not dev:
|
||||
return None
|
||||
return _ReSpeaker(dev)
|
||||
|
||||
|
||||
class XVF3800USBDevice:
|
||||
"""Small helper for issuing non-LED XVF3800 control commands via PyUSB.
|
||||
|
||||
This is intentionally minimal and does *not* depend on xvf_host.py.
|
||||
"""
|
||||
|
||||
def __init__(self, vid: int = _ReSpeaker.VID, pid: int = _ReSpeaker.PID):
|
||||
self._rsp = _find_device(vid=vid, pid=pid)
|
||||
if self._rsp is None:
|
||||
raise RuntimeError(
|
||||
f"XVF3800 USB device not found (vid=0x{vid:04x}, pid=0x{pid:04x})"
|
||||
)
|
||||
|
||||
def close(self) -> None:
|
||||
if self._rsp is not None:
|
||||
try:
|
||||
self._rsp.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# --- Device control -----------------------------------------------------
|
||||
|
||||
def reboot(self) -> None:
|
||||
"""Reboot the XVF3800 firmware (resets parameters to defaults)."""
|
||||
self._rsp.write("REBOOT", [1])
|
||||
|
||||
def save_configuration(self) -> None:
|
||||
"""Persist current runtime settings to flash."""
|
||||
self._rsp.write("SAVE_CONFIGURATION", [1])
|
||||
|
||||
# --- Audio routing ------------------------------------------------------
|
||||
|
||||
def set_audio_mgr_op_l(self, category: int, source: int) -> None:
|
||||
"""Set the L output channel source selection (category, source)."""
|
||||
self._rsp.write("AUDIO_MGR_OP_L", [int(category) & 0xFF, int(source) & 0xFF])
|
||||
|
||||
def set_audio_mgr_op_r(self, category: int, source: int) -> None:
|
||||
"""Set the R output channel source selection (category, source)."""
|
||||
self._rsp.write("AUDIO_MGR_OP_R", [int(category) & 0xFF, int(source) & 0xFF])
|
||||
|
||||
# --- Wait helpers -------------------------------------------------------
|
||||
|
||||
@staticmethod
|
||||
def wait_for_reenumeration(
|
||||
vid: int = _ReSpeaker.VID,
|
||||
pid: int = _ReSpeaker.PID,
|
||||
timeout_s: float = 12.0,
|
||||
settle_s: float = 1.0,
|
||||
) -> None:
|
||||
"""Wait for the XVF3800 to disappear and then reappear on USB."""
|
||||
start = time.time()
|
||||
# Wait for it to disappear (best effort)
|
||||
while time.time() - start < timeout_s:
|
||||
if usb.core.find(idVendor=vid, idProduct=pid) is None:
|
||||
break
|
||||
time.sleep(0.1)
|
||||
|
||||
# Wait for it to reappear
|
||||
while time.time() - start < timeout_s:
|
||||
if usb.core.find(idVendor=vid, idProduct=pid) is not None:
|
||||
break
|
||||
time.sleep(0.1)
|
||||
|
||||
# Give kernel/userspace audio stack a moment to settle
|
||||
if settle_s > 0:
|
||||
time.sleep(settle_s)
|
||||
|
||||
|
||||
class XVF3800LedBackend:
|
||||
"""High-level LED backend for the XVF3800."""
|
||||
|
||||
ring_led_count: int = 12
|
||||
|
||||
# GPO indices (from XVF3800 documentation)
|
||||
GPO_WS2812_POWER_INDEX = 3 # X0D33 in GPO_READ_VALUES response
|
||||
|
||||
def __init__(self, vid: int = _ReSpeaker.VID, pid: int = _ReSpeaker.PID) -> None:
|
||||
wrapper = _find_device(vid, pid)
|
||||
if wrapper is None:
|
||||
raise RuntimeError(
|
||||
f"XVF3800 USB device not found (vid=0x{vid:04x}, pid=0x{pid:04x})"
|
||||
)
|
||||
|
||||
self._dev = wrapper
|
||||
self.supports_per_led: bool = False
|
||||
|
||||
# CRITICAL FIX: Ensure WS2812 LED power is enabled BEFORE any LED operations
|
||||
# This prevents intermittent LED failures caused by X0D33 being low at startup
|
||||
try:
|
||||
_LOGGER.debug("Ensuring XVF3800 WS2812 LED power (X0D33) is enabled")
|
||||
self._dev.write("GPO_WRITE_VALUE", [33, 1]) # X0D33 = high
|
||||
time.sleep(0.05) # Give firmware time to settle
|
||||
_LOGGER.info("XVF3800 WS2812 LED power enabled")
|
||||
except Exception as e:
|
||||
_LOGGER.warning(
|
||||
"Could not enable WS2812 LED power, LEDs may not work reliably: %s", e
|
||||
)
|
||||
|
||||
# Best-effort feature detection: if we can read LED_RING_COLOR, we assume
|
||||
# per-LED control is supported by the current firmware.
|
||||
try:
|
||||
_ = self._dev.read("LED_RING_COLOR")
|
||||
self.supports_per_led = True
|
||||
except Exception:
|
||||
self.supports_per_led = False
|
||||
|
||||
# Best-effort version read (not fatal)
|
||||
try:
|
||||
ver = self._dev.read("VERSION")
|
||||
_LOGGER.info(
|
||||
"Connected to XVF3800 USB device (vid=0x%04x, pid=0x%04x, version=%s, per_led=%s)",
|
||||
vid,
|
||||
pid,
|
||||
".".join(str(int(x)) for x in ver[:3]),
|
||||
self.supports_per_led,
|
||||
)
|
||||
except Exception:
|
||||
_LOGGER.info(
|
||||
"Connected to XVF3800 USB device (vid=0x%04x, pid=0x%04x, per_led=%s)",
|
||||
vid,
|
||||
pid,
|
||||
self.supports_per_led,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# Helper: Ensure LED power before critical operations
|
||||
# ---------------------------------------------------------------------
|
||||
|
||||
def _ensure_led_power(self) -> bool:
|
||||
"""Ensure WS2812 LED power is enabled before operations.
|
||||
|
||||
This provides belt-and-suspenders protection against the LED power
|
||||
being disabled by firmware or button interactions.
|
||||
|
||||
Returns:
|
||||
bool: True if power is confirmed on, False if check failed
|
||||
"""
|
||||
try:
|
||||
# Read GPO values
|
||||
values = self._dev.read("GPO_READ_VALUES")
|
||||
if len(values) > self.GPO_WS2812_POWER_INDEX:
|
||||
ws2812_power = bool(values[self.GPO_WS2812_POWER_INDEX])
|
||||
if not ws2812_power:
|
||||
_LOGGER.warning("WS2812 LED power was off, re-enabling")
|
||||
self._dev.write("GPO_WRITE_VALUE", [33, 1])
|
||||
time.sleep(0.01) # Brief settle time
|
||||
return True
|
||||
return True
|
||||
except Exception as e:
|
||||
_LOGGER.debug("Could not verify WS2812 LED power state: %s", e)
|
||||
return False
|
||||
return False
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# Legacy/global controls
|
||||
# ---------------------------------------------------------------------
|
||||
|
||||
def set_effect(self, effect_id: int) -> None:
|
||||
"""Set LED effect mode (0=off, 1=breath, 2=rainbow, 3=single color, 4=doa)."""
|
||||
# Ensure power before effect change
|
||||
self._ensure_led_power()
|
||||
self._dev.write("LED_EFFECT", [int(effect_id) & 0xFF])
|
||||
|
||||
def set_brightness(self, brightness_0_255: int) -> None:
|
||||
"""Set LED brightness (0-255)."""
|
||||
value = max(0, min(255, int(brightness_0_255)))
|
||||
self._dev.write("LED_BRIGHTNESS", [value])
|
||||
|
||||
def set_speed(self, speed_id: int) -> None:
|
||||
"""Set LED effect speed (0=slow, 1=medium, 2=fast)."""
|
||||
self._dev.write("LED_SPEED", [int(speed_id) & 0xFF])
|
||||
|
||||
def set_color(self, r: int, g: int, b: int) -> None:
|
||||
"""Set LED color for breath / single color modes (0xRRGGBB)."""
|
||||
r = max(0, min(255, int(r)))
|
||||
g = max(0, min(255, int(g)))
|
||||
b = max(0, min(255, int(b)))
|
||||
color_value = (r << 16) | (g << 8) | b
|
||||
self._dev.write("LED_COLOR", [color_value])
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# Per-LED ring control (newer firmware)
|
||||
# ---------------------------------------------------------------------
|
||||
|
||||
def set_ring_colors(self, color_values: Sequence[int]) -> None:
|
||||
"""Set all 12 ring LEDs with 0xRRGGBB values (length must be 12)."""
|
||||
if not self.supports_per_led:
|
||||
raise RuntimeError("Per-LED ring control is not supported by this firmware")
|
||||
if len(color_values) != self.ring_led_count:
|
||||
raise ValueError(
|
||||
f"LED_RING_COLOR expects {self.ring_led_count} values, got {len(color_values)}"
|
||||
)
|
||||
|
||||
# Ensure power before writing ring colors
|
||||
self._ensure_led_power()
|
||||
self._dev.write("LED_RING_COLOR", [int(v) & 0xFFFFFFFF for v in color_values])
|
||||
|
||||
def set_ring_rgb(self, colors: Sequence[Tuple[int, int, int]]) -> None:
|
||||
"""Set all 12 ring LEDs with (r,g,b) tuples (length must be 12)."""
|
||||
if len(colors) != self.ring_led_count:
|
||||
raise ValueError(
|
||||
f"Ring expects {self.ring_led_count} colors, got {len(colors)}"
|
||||
)
|
||||
vals: List[int] = []
|
||||
for r, g, b in colors:
|
||||
r = max(0, min(255, int(r)))
|
||||
g = max(0, min(255, int(g)))
|
||||
b = max(0, min(255, int(b)))
|
||||
vals.append((r << 16) | (g << 8) | b)
|
||||
self.set_ring_colors(vals)
|
||||
|
||||
def set_ring_solid(self, r: int, g: int, b: int) -> None:
|
||||
"""Convenience: set all ring LEDs to the same RGB color."""
|
||||
self.set_ring_rgb([(r, g, b)] * self.ring_led_count)
|
||||
|
||||
def clear_ring(self) -> None:
|
||||
"""Convenience: turn all ring LEDs off (per-LED mode)."""
|
||||
if not self.supports_per_led:
|
||||
# Legacy fallback: just set effect off
|
||||
self.set_effect(0)
|
||||
self.set_brightness(0)
|
||||
return
|
||||
self.set_ring_colors([0] * self.ring_led_count)
|
||||
|
||||
def get_version(self) -> Optional[Tuple[int, int, int]]:
|
||||
"""Return (major, minor, patch) if readable, else None."""
|
||||
try:
|
||||
ver = self._dev.read("VERSION")
|
||||
if len(ver) >= 3:
|
||||
return (int(ver[0]), int(ver[1]), int(ver[2]))
|
||||
except Exception:
|
||||
return None
|
||||
return None
|
||||
|
||||
def close(self) -> None:
|
||||
if self._dev is not None:
|
||||
self._dev.close()
|
||||
+2
-1
@@ -26,7 +26,7 @@ classifiers = [
|
||||
]
|
||||
requires-python = ">=3.9.0"
|
||||
dependencies = [
|
||||
"aioesphomeapi==42.7.0",
|
||||
"aioesphomeapi==43.9.1",
|
||||
"soundcard<1",
|
||||
"numpy>=2,<3",
|
||||
"pymicro-wakeword>=2,<3",
|
||||
@@ -40,6 +40,7 @@ dependencies = [
|
||||
"RPi.GPIO",
|
||||
"gpiozero",
|
||||
"spidev",
|
||||
"pyusb",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
[Unit]
|
||||
Description=Load PipeWire echo-cancel module for LVA
|
||||
After=pipewire-pulse.service
|
||||
Wants=pipewire-pulse.service
|
||||
|
||||
[Service]
|
||||
Type=oneshot
|
||||
RemainAfterExit=yes
|
||||
|
||||
# Use a shell so we can do a little logic:
|
||||
# 1) Try to unload any existing echo-cancel module (ignore errors)
|
||||
# 2) Load the WebRTC echo cancel module with our tuning options
|
||||
ExecStart=/usr/bin/bash -lc '\
|
||||
pactl unload-module module-echo-cancel || true; \
|
||||
pactl load-module module-echo-cancel \
|
||||
aec_method=webrtc \
|
||||
aec_args="analog_gain_control=0 digital_gain_control=1 voice_detection=1" \
|
||||
'
|
||||
|
||||
[Install]
|
||||
WantedBy=default.target
|
||||
@@ -1,5 +1,7 @@
|
||||
[Unit]
|
||||
Description=Linux Voice Assistant
|
||||
Wants=pipewire.service wireplumber.service pipewire-pulse.service
|
||||
After=pipewire.service wireplumber.service pipewire-pulse.service
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
[Unit]
|
||||
Description=Linux Voice Assistant
|
||||
Wants=pipewire.service wireplumber.service pipewire-pulse.service
|
||||
After=pipewire.service wireplumber.service pipewire-pulse.service
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
WorkingDirectory=/home/pi/linux-voice-assistant
|
||||
Environment=PYTHONUNBUFFERED=1
|
||||
|
||||
ExecStart=/home/pi/linux-voice-assistant/script/run
|
||||
|
||||
Restart=always
|
||||
RestartSec=2
|
||||
|
||||
[Install]
|
||||
WantedBy=default.target
|
||||
@@ -0,0 +1,73 @@
|
||||
import argparse
|
||||
import soundcard as sc
|
||||
import numpy as np
|
||||
import wave
|
||||
|
||||
DEVICE = "reSpeaker XVF3800 4-Mic Array Analog Stereo"
|
||||
SECONDS = 5
|
||||
SR = 16000
|
||||
BLOCK = 1024
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Record from soundcard mic to WAV.")
|
||||
parser.add_argument(
|
||||
"--channel",
|
||||
choices=["mix", "left", "right"],
|
||||
default="mix",
|
||||
help="mix=record 1ch (downmix/default). left/right=record 2ch and select.",
|
||||
)
|
||||
parser.add_argument("--seconds", type=int, default=SECONDS)
|
||||
parser.add_argument("--device", type=str, default=DEVICE)
|
||||
parser.add_argument("--out", type=str, default="/tmp/lva_soundcard.wav")
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.channel == "mix":
|
||||
in_ch = 1
|
||||
out_ch = 1
|
||||
out_path = args.out
|
||||
else:
|
||||
in_ch = 2
|
||||
out_ch = 1
|
||||
# If user didn't override --out, make filename reflect selection
|
||||
if args.out == "/tmp/lva_soundcard.wav":
|
||||
out_path = f"/tmp/lva_soundcard_{args.channel}.wav"
|
||||
else:
|
||||
out_path = args.out
|
||||
|
||||
mic = sc.get_microphone(args.device, include_loopback=False)
|
||||
print("Using mic:", mic.name)
|
||||
print(f"Mode: {args.channel} (request channels={in_ch} -> write channels={out_ch})")
|
||||
|
||||
frames = []
|
||||
with mic.recorder(samplerate=SR, channels=in_ch, blocksize=BLOCK) as r:
|
||||
for _ in range(int(SR * args.seconds / BLOCK)):
|
||||
buf = r.record(BLOCK)
|
||||
|
||||
if in_ch == 1:
|
||||
# buf can be (BLOCK,) or (BLOCK, 1) depending on backend
|
||||
mono = buf.reshape(-1)
|
||||
sample = mono
|
||||
else:
|
||||
# buf should be (BLOCK, 2)
|
||||
if buf.ndim != 2 or buf.shape[1] < 2:
|
||||
raise RuntimeError(f"Expected stereo buffer, got shape {buf.shape}")
|
||||
idx = 0 if args.channel == "left" else 1
|
||||
sample = buf[:, idx]
|
||||
|
||||
rms = float(np.sqrt(np.mean(np.square(sample)))) if sample.size else 0.0
|
||||
print(f"rms={rms:.6f}")
|
||||
|
||||
pcm16 = (np.clip(sample, -1.0, 1.0) * 32767.0).astype("<i2")
|
||||
frames.append(pcm16.tobytes())
|
||||
|
||||
with wave.open(out_path, "wb") as wf:
|
||||
wf.setnchannels(out_ch)
|
||||
wf.setsampwidth(2)
|
||||
wf.setframerate(SR)
|
||||
wf.writeframes(b"".join(frames))
|
||||
|
||||
print("Wrote:", out_path)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
@@ -0,0 +1,182 @@
|
||||
#!/usr/bin/env python3
|
||||
"""ReSpeaker XVF3800 HID mute/button probe.
|
||||
|
||||
This script helps reverse-engineer how the XVF3800 exposes its mute button over USB HID.
|
||||
It is meant to be run from the linux-voice-assistant virtualenv on a system where the
|
||||
XVF3800 is connected via USB.
|
||||
|
||||
Requirements:
|
||||
pip install hidapi
|
||||
|
||||
Usage examples:
|
||||
|
||||
# Basic usage (auto-detect first 2886:001a HID interface)
|
||||
python tests/xvf3800_hid_mute_probe.py
|
||||
|
||||
# Explicit vendor/product (optional, defaults are correct for XVF3800)
|
||||
python tests/xvf3800_hid_mute_probe.py --vendor-id 0x2886 --product-id 0x001a
|
||||
|
||||
Once running, press and release the XVF3800 mute button a few times. The script will
|
||||
print raw HID reports in hex. Share that output in the Discussion so we can identify
|
||||
which byte/bit corresponds to the mute state.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import binascii
|
||||
import datetime as dt
|
||||
import sys
|
||||
import time
|
||||
from typing import List, Optional
|
||||
|
||||
try:
|
||||
import hid # type: ignore[import]
|
||||
except Exception as exc: # pragma: no cover - environment dependent
|
||||
print("ERROR: Failed to import 'hid' (hidapi).", file=sys.stderr)
|
||||
print(" Install it with: pip install hidapi", file=sys.stderr)
|
||||
print(f" Details: {exc}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def _fmt_vid_pid(vid: int, pid: int) -> str:
|
||||
return f"0x{vid:04x}:0x{pid:04x}"
|
||||
|
||||
|
||||
def list_matching_devices(vendor_id: int, product_id: int) -> List[dict]:
|
||||
devices = []
|
||||
for dev in hid.enumerate():
|
||||
if dev.get("vendor_id") == vendor_id and dev.get("product_id") == product_id:
|
||||
devices.append(dev)
|
||||
return devices
|
||||
|
||||
|
||||
def choose_device(devices: List[dict]) -> Optional[dict]:
|
||||
"""Pick a device from the list; if only one, return it, else pick the first.
|
||||
|
||||
We log all candidates so users can see what's on their system.
|
||||
"""
|
||||
if not devices:
|
||||
return None
|
||||
|
||||
print("Found the following matching HID interfaces for XVF3800:")
|
||||
for idx, dev in enumerate(devices):
|
||||
path = dev.get("path")
|
||||
serial = dev.get("serial_number") or "<none>"
|
||||
mfg = dev.get("manufacturer_string") or "<unknown>"
|
||||
prod = dev.get("product_string") or "<unknown>"
|
||||
iface = dev.get("interface_number", -1)
|
||||
print(
|
||||
f" [{idx}] path={path!r}, serial={serial!r}, iface={iface}, "
|
||||
f"manufacturer={mfg!r}, product={prod!r}"
|
||||
)
|
||||
|
||||
print()
|
||||
print("NOTE: Using index 0 by default. If this is not the HID interface, ")
|
||||
print(" rerun with --path <exact path> from the list above.\n")
|
||||
return devices[0]
|
||||
|
||||
|
||||
def open_device(
|
||||
vendor_id: int,
|
||||
product_id: int,
|
||||
path: Optional[bytes] = None,
|
||||
) -> hid.device:
|
||||
"""Open a HID device for the given VID/PID (and optional path)."""
|
||||
dev = hid.device()
|
||||
if path is not None:
|
||||
print(f"Opening HID device by path: {path!r}")
|
||||
dev.open_path(path)
|
||||
else:
|
||||
print(f"Opening HID device by VID/PID: {_fmt_vid_pid(vendor_id, product_id)}")
|
||||
dev.open(vendor_id, product_id)
|
||||
|
||||
dev.set_nonblocking(True)
|
||||
return dev
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Probe HID reports for the ReSpeaker XVF3800 mute button."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--vendor-id",
|
||||
type=lambda x: int(x, 0),
|
||||
default=0x2886,
|
||||
help="USB vendor ID (default: 0x2886 for Seeed)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--product-id",
|
||||
type=lambda x: int(x, 0),
|
||||
default=0x001a,
|
||||
help="USB product ID (default: 0x001a for XVF3800)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--path",
|
||||
type=str,
|
||||
default="",
|
||||
help="Optional exact HID path to open (overrides VID/PID selection)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--timeout-ms",
|
||||
type=int,
|
||||
default=500,
|
||||
help="Read timeout in milliseconds (default: 500)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
print("=== XVF3800 HID Mute Probe ===")
|
||||
print(f"Vendor/Product: {_fmt_vid_pid(args.vendor_id, args.product_id)}")
|
||||
|
||||
if args.path:
|
||||
chosen_path: Optional[bytes] = args.path.encode("utf-8")
|
||||
else:
|
||||
devices = list_matching_devices(args.vendor_id, args.product_id)
|
||||
if not devices:
|
||||
print(
|
||||
"No HID interfaces found for XVF3800 (VID/PID). Make sure the board is plugged in.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 1
|
||||
chosen = choose_device(devices)
|
||||
if chosen is None:
|
||||
print("No device chosen.", file=sys.stderr)
|
||||
return 1
|
||||
chosen_path = chosen.get("path")
|
||||
print("Using device:")
|
||||
print(f" path={chosen_path!r}")
|
||||
print(f" serial={chosen.get('serial_number')!r}")
|
||||
print(f" interface_number={chosen.get('interface_number')}")
|
||||
print()
|
||||
|
||||
try:
|
||||
dev = open_device(args.vendor_id, args.product_id, chosen_path)
|
||||
except Exception as exc: # pragma: no cover - hardware dependent
|
||||
print(f"ERROR: Failed to open HID device: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
print("Listening for HID reports. Press and release the XVF3800 mute button a few times.")
|
||||
print("Press Ctrl+C to stop.\n")
|
||||
|
||||
try:
|
||||
while True:
|
||||
data = dev.read(64, timeout_ms=args.timeout_ms)
|
||||
if data:
|
||||
now = dt.datetime.now().strftime("%H:%M:%S.%f")[:-3]
|
||||
raw = bytes(data)
|
||||
hex_str = binascii.hexlify(raw).decode("ascii")
|
||||
# Trim trailing zeroes for readability
|
||||
hex_str = hex_str.rstrip("0") or hex_str
|
||||
print(f"[{now}] len={len(raw):02d} report={hex_str}")
|
||||
time.sleep(0.01)
|
||||
except KeyboardInterrupt:
|
||||
print("\nStopping probe.")
|
||||
finally:
|
||||
try:
|
||||
dev.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover - script entry
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,83 @@
|
||||
#!/usr/bin/env python3
|
||||
import argparse
|
||||
import time
|
||||
|
||||
import numpy as np
|
||||
import sounddevice as sd
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument(
|
||||
"--device",
|
||||
help="Exact input device name to test (as shown by --list-input-devices)",
|
||||
required=True,
|
||||
)
|
||||
parser.add_argument(
|
||||
"--samplerate",
|
||||
type=float,
|
||||
default=16000.0,
|
||||
help="Sample rate to test (default: 16000)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--seconds",
|
||||
type=float,
|
||||
default=2.0,
|
||||
help="Duration to record for debug (default: 2s)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
print("=== sounddevice.query_devices() (input subset) ===")
|
||||
devices = sd.query_devices()
|
||||
for idx, dev in enumerate(devices):
|
||||
if dev["max_input_channels"] > 0:
|
||||
flag = "*" if args.device in (dev["name"], str(idx)) else " "
|
||||
print(
|
||||
f"{flag} [{idx}] {dev['name']} "
|
||||
f"(max_input_channels={dev['max_input_channels']}, "
|
||||
f"default_samplerate={dev['default_samplerate']})"
|
||||
)
|
||||
|
||||
print("\n=== Opening stream ===")
|
||||
sd.default.samplerate = args.samplerate
|
||||
sd.default.dtype = "int16"
|
||||
|
||||
# Allow using either index or full name
|
||||
device = args.device
|
||||
try:
|
||||
device = int(args.device)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# First just open/close the stream to see what PortAudio says
|
||||
with sd.InputStream(device=device, channels=0) as stream:
|
||||
print(f"Opened stream with device={device}")
|
||||
print(f" samplerate: {stream.samplerate}")
|
||||
print(f" channels: {stream.channels}")
|
||||
print(f" dtype: {stream.dtype}")
|
||||
print(f" blocksize: {stream.blocksize}")
|
||||
|
||||
# Now actually record a short block and inspect the shape
|
||||
print("\n=== Recording test block ===")
|
||||
duration = args.seconds
|
||||
num_frames = int(duration * args.samplerate)
|
||||
print(f"Recording {duration} seconds ({num_frames} frames)...")
|
||||
|
||||
recorded = sd.rec(frames=num_frames, samplerate=args.samplerate,
|
||||
channels=0, dtype="int16", device=device)
|
||||
sd.wait()
|
||||
|
||||
arr = np.array(recorded)
|
||||
print(f"Recorded array shape: {arr.shape}, dtype={arr.dtype}")
|
||||
if arr.ndim == 1:
|
||||
print("-> Mono (1D) array returned")
|
||||
elif arr.ndim == 2:
|
||||
print(f"-> Multichannel: {arr.shape[1]} channels")
|
||||
else:
|
||||
print("-> Unexpected ndim:", arr.ndim)
|
||||
|
||||
print(f"Sample min/max: {arr.min()} / {arr.max()}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"type": "micro",
|
||||
"wake_word": "hey_luna",
|
||||
"wake_word": "Hey Luna",
|
||||
"author": "adamlonsdale",
|
||||
"website": "https://github.com/adamlonsdale",
|
||||
"model": "hey_luna.tflite",
|
||||
|
||||
Reference in New Issue
Block a user