commit 2d7d27fac3f94ccd3a355de98ae147f289309251 Author: chexit Date: Tue Nov 10 15:21:59 2026 +0300 Build streaming voice assistant for Mini 2 diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..d991cac --- /dev/null +++ b/.dockerignore @@ -0,0 +1,5 @@ +* +!package.json +!package-lock.json +!server/ +!server/** diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..a183203 --- /dev/null +++ b/.env.example @@ -0,0 +1,7 @@ +# Provide the actual values privately in .env. +DEVICE_TOKEN= +ELEVENLABS_API_KEY= +OPENROUTER_API_KEY= +TELEGRAM_BOT_TOKEN= +TELEGRAM_OWNER_ID= +ADMIN_HOST=miniapp.pepega.systems diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bc65ecf --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +.DS_Store +/keys/ +.env +.env.* +!.env.example +/node_modules/ +/data/ +/dist/ diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..977f6cb --- /dev/null +++ b/Dockerfile @@ -0,0 +1,9 @@ +FROM node:24-alpine +WORKDIR /app +ENV NODE_ENV=production HOST=0.0.0.0 PORT=3000 DATA_DIR=/data +COPY package.json package-lock.json ./ +RUN npm ci --omit=dev --ignore-scripts +COPY server/ ./server/ +USER node +EXPOSE 3000 +CMD ["node", "server/index.mjs"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..8e0d4f6 --- /dev/null +++ b/README.md @@ -0,0 +1,45 @@ +# Алиса для Станции Мини 2 + +Собственный голосовой клиент для рутированной Mini 2, сервер на Node.js и пульт в Telegram. На колонке работает экспериментальный детектор, обученный локально на нескольких произнесениях «Алиса». После обнаружения он подаёт сигнал. Команду надо произнести после сигнала. + +```text +Mini 2 → WSS с PCM16 / 16 кГц → Scribe v2 Realtime + ← WSS с PCM16 / 24 кГц ← Eleven v4 Turbo ← OpenRouter / Haiku 5.5 +``` + +Scribe выдаёт промежуточный текст, а запрос к LLM начинается после подтверждения конца команды. Текстовые фрагменты из OpenRouter сразу идут в ElevenLabs, звуковые фрагменты сразу уходят на колонку. Запрос и ответ в текстовом виде сохраняются после успешного хода. Сервер держит 50 последних ходов и использует до 10 из них как контекст. Аудио сервер не пишет на диск. Облачные провайдеры обрабатывают отправленные им аудио и текст. + +## Запуск и проверка + +```sh +npm ci +npm test +cp .env.example .env +# Заполнить .env локально +node --env-file=.env server/index.mjs +``` + +Клиент, сборка и пробный запуск описаны в [station/README.md](station/README.md). На колонку передаётся отдельный `DEVICE_TOKEN`, ключи провайдеров остаются на сервере. Для интеграционного теста с реальными STT/TTS и без OpenRouter можно отдельно запустить сервер с `LLM_MODE=echo`. + +## Развёртывание + +На VPS код находится в `~/projects/alice-backend`. В корне проекта нужны `.env` с правами `0600` и каталог `data/` для настроек и истории, доступный на запись UID 1000. Пример конфигурации для уже существующей Docker-сети `web` и Caddy лежит в [deploy/](deploy/). + +```sh +docker compose -f deploy/compose.yaml up -d --build +``` + +Caddy проксирует голосовое соединение и `/healthz` на `alice.pepega.systems`, а пульт и API на `miniapp.pepega.systems`. История и редактирование настроек требуют Telegram `initData` с проверенной подписью и ID владельца из `TELEGRAM_OWNER_ID`. В пульте можно выбрать модель OpenRouter, голос и модель синтеза, изменить системный промпт и очистить историю. В текущем API распознавания в реальном времени доступна одна модель. + +## Протокол колонки + +`GET /v1/voice` с WebSocket Upgrade и `Authorization: Bearer `. Один ход на соединение. + +1. Сервер посылает `{"type":"ready"}`. +2. Клиент посылает `{"type":"start"}`, затем бинарные блоки mono PCM16LE 16 кГц и `{"type":"end"}`. +3. Сервер присылает JSON `partial` и `transcript` с полем `text`, затем бинарные блоки mono PCM16LE 24 кГц. +4. Сервер присылает `{"type":"done"}` или `{"type":"error","code":"…"}` и закрывает соединение. + +## Музыка + +Spotify Connect и управление музыкой пока не подключены. Для ручного Connect Spotify предлагает [Soloist](https://developer.spotify.com/documentation/soloist), которому нужны собственный ключ Spotify и PipeWire или PulseAudio на Linux. На Mini 2 штатный тракт ALSA. [Правила Spotify для разработчиков](https://developer.spotify.com/compliance-tips) запрещают использовать их инструменты для собственного голосового управления. diff --git a/deploy/Caddyfile b/deploy/Caddyfile new file mode 100644 index 0000000..9a2a752 --- /dev/null +++ b/deploy/Caddyfile @@ -0,0 +1,20 @@ +# alice-backend +alice.pepega.systems { + @voice path /v1/voice /healthz + handle @voice { + reverse_proxy alice-backend:3000 + } + handle { + respond 404 + } +} + +miniapp.pepega.systems { + @miniapp path / /style.css /app.js /api/admin /api/admin/* + handle @miniapp { + reverse_proxy alice-backend:3000 + } + handle { + respond 404 + } +} diff --git a/deploy/compose.yaml b/deploy/compose.yaml new file mode 100644 index 0000000..017a540 --- /dev/null +++ b/deploy/compose.yaml @@ -0,0 +1,29 @@ +services: + backend: + build: + context: .. + env_file: ../.env + volumes: + - ../data:/data + networks: + web: + aliases: [alice-backend] + restart: unless-stopped + init: true + read_only: true + cap_drop: [ALL] + security_opt: [no-new-privileges:true] + mem_limit: 192m + cpus: 0.75 + pids_limit: 64 + labels: + com.centurylinklabs.watchtower.enable: "false" + logging: + driver: json-file + options: + max-size: "2m" + max-file: "2" + +networks: + web: + external: true diff --git a/package-lock.json b/package-lock.json new file mode 100644 index 0000000..d54207c --- /dev/null +++ b/package-lock.json @@ -0,0 +1,39 @@ +{ + "name": "alice-backend", + "version": "0.1.0", + "lockfileVersion": 3, + "requires": true, + "packages": { + "": { + "name": "alice-backend", + "version": "0.1.0", + "dependencies": { + "ws": "8.21.0" + }, + "engines": { + "node": ">=22" + } + }, + "node_modules/ws": { + "version": "8.21.0", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.0.tgz", + "integrity": "sha512-Vsp28b7DRcimFQvrqu2Wek3z1iYxDCWqHYB8Qsnk/S4RfaCQzPGPyBNuVjJV3cd6UiKtUtp6sNM77gWvzcCH+g==", + "license": "MIT", + "engines": { + "node": ">=10.0.0" + }, + "peerDependencies": { + "bufferutil": "^4.0.1", + "utf-8-validate": ">=5.0.2" + }, + "peerDependenciesMeta": { + "bufferutil": { + "optional": true + }, + "utf-8-validate": { + "optional": true + } + } + } + } +} diff --git a/package.json b/package.json new file mode 100644 index 0000000..bd37495 --- /dev/null +++ b/package.json @@ -0,0 +1,12 @@ +{ + "name": "alice-backend", + "version": "0.1.0", + "private": true, + "type": "module", + "engines": {"node": ">=22"}, + "scripts": { + "start": "node server/index.mjs", + "test": "node --test" + }, + "dependencies": {"ws": "8.21.0"} +} diff --git a/server/admin.mjs b/server/admin.mjs new file mode 100644 index 0000000..ce4c009 --- /dev/null +++ b/server/admin.mjs @@ -0,0 +1,105 @@ +import { createHmac, timingSafeEqual } from 'node:crypto'; +import { readFileSync } from 'node:fs'; +import { validateSettings } from './store.mjs'; + +const assets = new Map([ + ['/', ['index.html', 'text/html; charset=utf-8']], + ['/app.js', ['app.js', 'text/javascript; charset=utf-8']], + ['/style.css', ['style.css', 'text/css; charset=utf-8']], +].map(([route, [file, type]]) => [route, [readFileSync(new URL('./public/' + file, import.meta.url)), type]])); + +const securityHeaders = { + 'cache-control': 'no-store', + 'x-content-type-options': 'nosniff', + 'referrer-policy': 'no-referrer', + 'content-security-policy': "default-src 'none'; script-src 'self' https://telegram.org; style-src 'self'; connect-src 'self'; img-src 'self'; base-uri 'none'; form-action 'none'; frame-ancestors 'self' https://web.telegram.org https://*.telegram.org", +}; + +export function verifyTelegram(initData, botToken, owner, now = Date.now()) { + if (!initData || initData.length > 8192 || !botToken || !owner) return false; + const params = new URLSearchParams(initData); + const entries = [...params.entries()]; + if (new Set(params.keys()).size !== entries.length) return false; + const hash = params.get('hash') ?? ''; + if (!/^[a-f0-9]{64}$/.test(hash)) return false; + const date = Number(params.get('auth_date')); + const age = now / 1000 - date; + if (!Number.isSafeInteger(date) || age < -60 || age > 3600) return false; + + const secret = createHmac('sha256', 'WebAppData').update(botToken).digest(); + const check = entries.filter(([key]) => key !== 'hash').sort(([a], [b]) => a < b ? -1 : a > b ? 1 : 0) + .map(([key, value]) => key + '=' + value).join('\n'); + const expected = createHmac('sha256', secret).update(check).digest(); + if (!timingSafeEqual(expected, Buffer.from(hash, 'hex'))) return false; + try { + const { id } = JSON.parse(params.get('user')); + return Number.isSafeInteger(id) && String(id) === owner; + } catch { + return false; + } +} + +function json(res, status, data) { + res.writeHead(status, { ...securityHeaders, 'content-type': 'application/json; charset=utf-8' }); + res.end(JSON.stringify(data)); +} + +async function readSettings(req) { + const chunks = []; + let bytes = 0; + for await (const chunk of req.iterator({ destroyOnReturn: false })) { + bytes += chunk.length; + if (bytes > 20_000) { + req.resume(); + return null; + } + chunks.push(chunk); + } + try { + return validateSettings(JSON.parse(Buffer.concat(chunks).toString())); + } catch { + return null; + } +} + +export async function handleAdmin(req, res, store) { + const route = req.url?.split('?')[0]; + const domain = process.env.ADMIN_HOST; + if (domain && req.headers.host !== domain) return false; + + const asset = assets.get(route); + if (asset && req.method === 'GET') { + const [content, type] = asset; + res.writeHead(200, { ...securityHeaders, 'content-type': type }); + res.end(content); + return true; + } + + if (route !== '/api/admin' && route !== '/api/admin/settings' && route !== '/api/admin/history') return false; + const auth = req.headers.authorization ?? ''; + if (!auth.startsWith('tma ') || !verifyTelegram(auth.slice(4), process.env.TELEGRAM_BOT_TOKEN, process.env.TELEGRAM_OWNER_ID)) { + json(res, 401, { error: 'telegram_auth' }); + return true; + } + + if (route === '/api/admin' && req.method === 'GET') { + json(res, 200, { + settings: store.getSettings(), + history: store.getHistory().reverse(), + }); + } else if (route === '/api/admin/settings' && req.method === 'PUT') { + const settings = await readSettings(req); + if (settings) { + store.updateSettings(settings); + json(res, 200, { settings }); + } else { + json(res, 400, { error: 'invalid_settings' }); + } + } else if (route === '/api/admin/history' && req.method === 'DELETE') { + store.clearHistory(); + json(res, 200, { history: [] }); + } else { + json(res, 405, { error: 'method_not_allowed' }); + } + return true; +} diff --git a/server/index.mjs b/server/index.mjs new file mode 100644 index 0000000..ed6efb1 --- /dev/null +++ b/server/index.mjs @@ -0,0 +1,33 @@ +import http from 'node:http'; +import { timingSafeEqual } from 'node:crypto'; +import { WebSocketServer } from 'ws'; +import { openStore } from './store.mjs'; +import { handleAdmin } from './admin.mjs'; +import { handleTurn } from './turns.mjs'; + +const { DEVICE_TOKEN, ELEVENLABS_API_KEY, LLM_MODE = 'openrouter', OPENROUTER_API_KEY, PORT = '3000', HOST = '127.0.0.1' } = process.env; +if (!DEVICE_TOKEN || !ELEVENLABS_API_KEY || !['openrouter', 'echo'].includes(LLM_MODE) || (LLM_MODE === 'openrouter' && !OPENROUTER_API_KEY)) { + console.error('Missing required configuration or invalid LLM_MODE'); process.exit(1); +} +const store = openStore(); +const server = http.createServer(async (req, res) => { + try { + if (req.method === 'GET' && req.url === '/healthz') { res.writeHead(200, { 'content-type': 'application/json', 'cache-control': 'no-store' }); res.end('{"ok":true}'); return; } + if (await handleAdmin(req, res, store)) return; + res.writeHead(404); res.end(); + } catch { if (!res.headersSent) res.writeHead(500); res.end(); } +}); +const sockets = new WebSocketServer({ noServer: true, maxPayload: 128 * 1024, perMessageDeflate: false }); +let active = false; +server.on('upgrade', (req, socket, head) => { + const token = req.headers.authorization?.startsWith('Bearer ') ? req.headers.authorization.slice(7) : ''; + const a = Buffer.from(token), b = Buffer.from(DEVICE_TOKEN); + if (req.url !== '/v1/voice' || a.length !== b.length || !timingSafeEqual(a, b)) { socket.end('HTTP/1.1 401 Unauthorized\r\nConnection: close\r\n\r\n'); return; } + if (active) { socket.end('HTTP/1.1 503 Service Unavailable\r\nConnection: close\r\n\r\n'); return; } + sockets.handleUpgrade(req, socket, head, ws => { + active = true; + ws.once('close', () => { active = false; }); + handleTurn(ws, store); + }); +}); +server.listen(Number(PORT), HOST, () => console.log(`Voice server listening on ${HOST}:${PORT}`)); diff --git a/server/providers.mjs b/server/providers.mjs new file mode 100644 index 0000000..c767f81 --- /dev/null +++ b/server/providers.mjs @@ -0,0 +1,94 @@ +import WebSocket from 'ws'; + +const env = process.env; +const sttUrl = env.ELEVEN_STT_URL || 'wss://api.elevenlabs.io/v1/speech-to-text/realtime'; +const ttsUrl = env.ELEVEN_TTS_URL || 'wss://api.elevenlabs.io/v1/text-to-dialogue/stream-input'; +const flashUrl = env.ELEVEN_FLASH_URL || 'wss://api.elevenlabs.io/v1/text-to-speech/{voice_id}/stream-input'; +const llmUrl = env.OPENROUTER_URL || 'https://openrouter.ai/api/v1/chat/completions'; + +function endpoint(base, params) { + const url = new URL(base); + for (const [key, value] of Object.entries(params)) url.searchParams.set(key, value); + return url.toString(); +} +function connect(url, signal) { + return new Promise((resolve, reject) => { + if (signal.aborted) return reject(new Error('aborted')); + const ws = new WebSocket(url, { headers: { 'xi-api-key': env.ELEVENLABS_API_KEY }, handshakeTimeout: 10000, maxPayload: 2 * 1024 * 1024 }); + const abort = () => ws.terminate(); + signal.addEventListener('abort', abort, { once: true }); + ws.once('close', () => signal.removeEventListener('abort', abort)); + ws.once('open', () => resolve(ws)); + ws.once('error', reject); + }); +} + +export async function openStt(settings, signal, onPartial, onFinal, onFailure) { + const ws = await connect(endpoint(sttUrl, { model_id: settings.sttModel, audio_format: 'pcm_16000', language_code: 'ru', commit_strategy: 'manual' }), signal); + let final = false; + ws.on('message', data => { + try { + const msg = JSON.parse(data.toString()); + if (msg.message_type === 'partial_transcript') onPartial(msg.text || ''); + else if (msg.message_type === 'committed_transcript') { final = true; onFinal(msg.text || ''); } + else if (msg.error || /(?:error|rate_limited|auth_error|quota_exceeded)/.test(msg.message_type || '')) onFailure(); + } catch { onFailure(); } + }); + ws.on('error', onFailure); + ws.on('close', () => { if (!final && !signal.aborted) onFailure(); }); + return ws; +} + +export async function openTts(settings, signal, onAudio) { + const flash = settings.ttsModel === 'eleven_flash_v2_5'; + const url = flash ? flashUrl.replace('{voice_id}', encodeURIComponent(settings.voiceId)) : ttsUrl; + const ws = await connect(endpoint(url, { model_id: settings.ttsModel, output_format: 'pcm_24000', ...(flash ? { auto_mode: 'true' } : { language_code: 'ru' }) }), signal); + let settled = false; + let resolveDone, rejectDone; + const done = new Promise((resolve, reject) => { resolveDone = resolve; rejectDone = reject; }); + // The caller may still be generating text when the provider fails. + done.catch(() => {}); + ws.on('message', data => { + try { + const msg = JSON.parse(data.toString()); + if (msg.error || msg.type === 'error') throw new Error('tts_error'); + if (msg.audio) onAudio(Buffer.from(msg.audio, 'base64')); + if (msg.is_final || msg.isFinal) { settled = true; resolveDone(); } + } catch { settled = true; rejectDone(new Error('tts_error')); } + }); + ws.on('error', () => { if (!settled) { settled = true; rejectDone(new Error('tts_error')); } }); + ws.on('close', () => { if (!settled) { settled = true; rejectDone(new Error('tts_closed')); } }); + ws.send(JSON.stringify(flash ? { text: ' ' } : { voices: [settings.voiceId] })); + const keepalive = setInterval(() => { if (ws.readyState === WebSocket.OPEN) ws.send(JSON.stringify(flash ? { text: ' ' } : { keep_alive: true })); }, 10000); + ws.once('close', () => clearInterval(keepalive)); + return { ws, done, text(text) { if (ws.readyState !== WebSocket.OPEN) throw new Error('tts_closed'); ws.send(JSON.stringify(flash ? { text } : { inputs: [{ text, voice_id: settings.voiceId }] })); }, finish() { if (ws.readyState !== WebSocket.OPEN) throw new Error('tts_closed'); ws.send(JSON.stringify(flash ? { text: '' } : { close_socket: true })); } }; +} + +export async function streamAnswer(settings, history, userText, signal, onText) { + if (env.LLM_MODE === 'echo') { onText(userText); return userText; } + if (!env.OPENROUTER_API_KEY) throw new Error('llm_config'); + const messages = [{ role: 'system', content: settings.systemPrompt }, ...history.slice(-10).flatMap(t => [{ role: 'user', content: t.user }, { role: 'assistant', content: t.assistant }]), { role: 'user', content: userText }]; + const response = await fetch(llmUrl, { method: 'POST', signal, headers: { Authorization: `Bearer ${env.OPENROUTER_API_KEY}`, 'Content-Type': 'application/json' }, body: JSON.stringify({ model: settings.llmModel, messages, stream: true, max_tokens: 512 }) }); + if (!response.ok || !response.body) throw new Error('llm_error'); + let buffer = '', answer = '', finished = false; + const decoder = new TextDecoder(); + const consume = event => { + const payload = event.split(/\r?\n/).filter(line => line.startsWith('data:')).map(line => line.slice(5).trimStart()).join('\n'); + if (!payload) return; + if (payload === '[DONE]') { finished = true; return; } + const message = JSON.parse(payload); + if (message.error) throw new Error('llm_error'); + const content = message.choices?.[0]?.delta?.content; + if (typeof content === 'string' && content) { answer += content; if (answer.length > 12000) throw new Error('llm_limit'); onText(content); } + }; + for await (const chunk of response.body) { + buffer += decoder.decode(chunk, { stream: true }); + if (buffer.length > 65536) throw new Error('llm_limit'); + let match; + while ((match = /\r?\n\r?\n/.exec(buffer))) { consume(buffer.slice(0, match.index)); buffer = buffer.slice(match.index + match[0].length); } + } + buffer += decoder.decode(); + if (buffer.trim()) consume(buffer); + if (!finished || !answer.trim()) throw new Error('llm_incomplete'); + return answer; +} diff --git a/server/public/app.js b/server/public/app.js new file mode 100644 index 0000000..f5e04d2 --- /dev/null +++ b/server/public/app.js @@ -0,0 +1,109 @@ +const telegram = window.Telegram?.WebApp; +const status = document.querySelector('#status'); +const admin = document.querySelector('#admin'); +const history = document.querySelector('#history'); +const settings = document.querySelector('#settings'); +const showHistory = document.querySelector('#show-history'); +const showSettings = document.querySelector('#show-settings'); +const clear = document.querySelector('#clear'); +const form = document.querySelector('#form'); +const turns = document.querySelector('#turns'); + +const dateFormat = new Intl.DateTimeFormat('ru', { dateStyle: 'medium', timeStyle: 'short' }); + +async function api(path = '', options = {}) { + const response = await fetch('/api/admin' + path, { + ...options, + headers: { Authorization: 'tma ' + telegram.initData, 'Content-Type': 'application/json' }, + }); + if (response.status === 401) throw new Error('Доступ открыт владельцу. Если сессия истекла, заново открой приложение из меню бота.'); + if (response.status === 400) throw new Error('Проверь ID модели, голос и системный промпт.'); + if (!response.ok) throw new Error('Сервер недоступен. Попробуй ещё раз.'); + return response.json(); +} + +function renderTurns(items) { + turns.replaceChildren(); + clear.disabled = items.length === 0; + if (!items.length) { + const empty = document.createElement('p'); + empty.className = 'empty'; + empty.textContent = 'Здесь появятся последние ответы станции.'; + turns.append(empty); + } + for (const turn of items) { + const article = document.createElement('article'); + const date = document.createElement('time'); + date.dateTime = turn.createdAt; + date.textContent = dateFormat.format(new Date(turn.createdAt)); + const question = document.createElement('p'); + question.className = 'question'; + question.textContent = turn.user; + const answer = document.createElement('p'); + answer.textContent = turn.assistant; + article.append(date, question, answer); + turns.append(article); + } +} + +function selectTab(isHistory) { + showHistory.setAttribute('aria-pressed', String(isHistory)); + showSettings.setAttribute('aria-pressed', String(!isHistory)); + history.hidden = !isHistory; + settings.hidden = isHistory; + status.textContent = ''; +} +showHistory.onclick = async () => { + selectTab(true); + try { + renderTurns((await api()).history); + } catch (error) { + status.textContent = error.message; + } +}; +showSettings.onclick = () => selectTab(false); + +form.onsubmit = async (event) => { + event.preventDefault(); + const button = form.querySelector('button'); + button.disabled = true; + try { + const body = JSON.stringify(Object.fromEntries(new FormData(form))); + await api('/settings', { method: 'PUT', body }); + status.textContent = 'Сохранено. Настройки вступят в силу со следующего запроса.'; + } catch (error) { + status.textContent = error.message; + } finally { + button.disabled = false; + } +}; + +clear.onclick = () => { + const erase = async (confirmed) => { + if (!confirmed) return; + clear.disabled = true; + try { + renderTurns((await api('/history', { method: 'DELETE' })).history); + } catch (error) { + status.textContent = error.message; + clear.disabled = false; + } + }; + if (telegram.isVersionAtLeast('6.2')) telegram.showConfirm('Очистить историю разговоров?', erase); + else erase(window.confirm('Очистить историю разговоров?')); +}; + +telegram?.ready(); +if (!telegram?.initData) { + status.textContent = 'Открой пульт из меню Telegram-бота.'; +} else { + try { + const data = await api(); + for (const [name, value] of Object.entries(data.settings)) form.elements.namedItem(name).value = value; + renderTurns(data.history); + status.textContent = ''; + admin.hidden = false; + } catch (error) { + status.textContent = error.message; + } +} diff --git a/server/public/index.html b/server/public/index.html new file mode 100644 index 0000000..26b54f0 --- /dev/null +++ b/server/public/index.html @@ -0,0 +1,46 @@ + + + + + + + Алиса · пульт + + + + + +
+
+ +

СТАНЦИЯ МИНИ 2

Пульт Алисы

+
+

Подключаемся…

+ +
+ + diff --git a/server/public/style.css b/server/public/style.css new file mode 100644 index 0000000..57d9c32 --- /dev/null +++ b/server/public/style.css @@ -0,0 +1,41 @@ +:root { + color-scheme: light dark; + font: 16px/1.45 system-ui, sans-serif; + color: var(--tg-theme-text-color, light-dark(#181724, #f4f1ff)); + background: var(--tg-theme-bg-color, light-dark(#faf8ff, #171421)); + --muted: var(--tg-theme-hint-color, light-dark(#645e76, #afa7c0)); + --surface: var(--tg-theme-secondary-bg-color, light-dark(#eeeaf7, #242031)); + --accent: var(--tg-theme-button-color, light-dark(#6636d2, #8b61ef)); + --accent-text: var(--tg-theme-button-text-color, white); +} +* { box-sizing: border-box; } +body { margin: 0; } +main { max-width: 640px; margin: auto; padding: max(26px, env(safe-area-inset-top)) 20px max(36px, env(safe-area-inset-bottom)); } +header { display: flex; gap: 16px; align-items: center; margin-bottom: 20px; } +.badge { width: 54px; height: 54px; border-radius: 18px; background: var(--accent); display: flex; align-items: center; justify-content: center; gap: 3px; } +.badge i { width: 3px; height: 11px; background: var(--accent-text); border-radius: 4px; } +.badge i:nth-child(even) { height: 19px; } +.badge i:nth-child(3) { height: 27px; } +h1, h2, p { margin: 0; } +h1 { font-size: 27px; letter-spacing: -.04em; } +h2 { font-size: 20px; } +.eyebrow { font-size: 10px; font-weight: 800; letter-spacing: .19em; color: var(--muted); } +#status { color: var(--muted); margin-bottom: 14px; } +#status:empty { display: none; } +nav { display: flex; gap: 4px; border-radius: 15px; padding: 4px; background: var(--surface); margin-bottom: 22px; } +button, input, select, textarea { font: inherit; color: inherit; } +button { min-height: 42px; padding: 8px 13px; border: 0; border-radius: 11px; background: transparent; cursor: pointer; } +button:disabled { opacity: .55; cursor: default; } +nav button { flex: 1; } +nav button[aria-pressed="true"], .primary { background: var(--accent); color: var(--accent-text); font-weight: 650; } +.section-title { display: flex; align-items: center; justify-content: space-between; margin-bottom: 10px; } +.quiet, time { color: var(--muted); font-size: 13px; } +article { margin: 10px 0; padding: 14px; border-radius: 15px; background: var(--surface); } +article p { white-space: pre-wrap; overflow-wrap: anywhere; margin-top: 7px; } +article .question { font-weight: 650; } +.empty { color: var(--muted); } +form { display: grid; gap: 14px; margin-top: 15px; } +label { display: grid; gap: 5px; font-size: 13px; font-weight: 650; } +input, select, textarea { width: 100%; background: var(--surface); border: 1px solid transparent; border-radius: 11px; padding: 10px; font-size: 16px; font-weight: 400; } +textarea { resize: vertical; } +:focus-visible { outline: 2px solid var(--accent); outline-offset: 2px; } diff --git a/server/store.mjs b/server/store.mjs new file mode 100644 index 0000000..7656e3b --- /dev/null +++ b/server/store.mjs @@ -0,0 +1,62 @@ +import { mkdirSync, readFileSync, renameSync, writeFileSync } from 'node:fs'; +import { join } from 'node:path'; + +export const defaults = Object.freeze({ + llmModel: 'anthropic/claude-haiku-5.5', + sttModel: 'scribe_v2_realtime', + ttsModel: 'eleven_v4_turbo', + voiceId: '21m00Tcm4TlvDq8ikWAM', + systemPrompt: 'Ты голосовой помощник Алиса. Отвечай по-русски, коротко и по делу. Пиши так, чтобы ответ хорошо звучал вслух: обычный текст, без Markdown. Если не знаешь ответа, скажи об этом прямо.', +}); + +export const ttsModels = ['eleven_v4_turbo', 'eleven_v3_conversational', 'eleven_flash_v2_5']; + +export function validateSettings(input) { + if (typeof input !== 'object' || input === null) return null; + const { llmModel, sttModel, ttsModel, voiceId, systemPrompt } = input; + if (typeof llmModel !== 'string' || !/^[\w.~:-]+\/[\w.:~-]+$/.test(llmModel) || llmModel.length > 120) return null; + if (sttModel !== defaults.sttModel || !ttsModels.includes(ttsModel)) return null; + if (typeof voiceId !== 'string' || !/^[a-zA-Z0-9]{20}$/.test(voiceId)) return null; + if (typeof systemPrompt !== 'string' || !systemPrompt.trim() || systemPrompt.length > 4000) return null; + return { llmModel, sttModel, ttsModel, voiceId, systemPrompt: systemPrompt.trim() }; +} + +export function openStore(directory = process.env.DATA_DIR ?? './data') { + mkdirSync(directory, { recursive: true, mode: 0o700 }); + const path = join(directory, 'assistant.json'); + let state = { settings: { ...defaults }, history: [] }; + try { + const saved = JSON.parse(readFileSync(path, 'utf8')); + const settings = validateSettings({ ...defaults, ...saved.settings }); + if (!settings || !Array.isArray(saved.history)) throw new Error('Invalid assistant state'); + state = { settings, history: saved.history }; + } catch (error) { + if (error.code !== 'ENOENT') throw error; + } + + function commit(next) { + const temporary = path + '.tmp'; + writeFileSync(temporary, JSON.stringify(next) + '\n', { mode: 0o600 }); + renameSync(temporary, path); + state = next; + } + + return { + getSettings: () => ({ ...state.settings }), + getHistory: () => state.history.map(turn => ({ ...turn })), + updateSettings(settings) { + commit({ ...state, settings }); + }, + recordTurn(user, assistant) { + const turn = { + user: user.slice(0, 6000), + assistant: assistant.slice(0, 6000), + createdAt: new Date().toISOString(), + }; + commit({ ...state, history: [...state.history.slice(-49), turn] }); + }, + clearHistory() { + commit({ ...state, history: [] }); + }, + }; +} diff --git a/server/turns.mjs b/server/turns.mjs new file mode 100644 index 0000000..e8a5dea --- /dev/null +++ b/server/turns.mjs @@ -0,0 +1,81 @@ +import WebSocket from 'ws'; +import { openStt, openTts, streamAnswer } from './providers.mjs'; + +const MAX_AUDIO = 60 * 16000 * 2; +const MAX_BUFFERED = 2 * 1024 * 1024; + +export function handleTurn(client, store) { + const controller = new AbortController(); + const { signal } = controller; + let state = 'waiting', stt, tts, pending, bytes = 0, received = 0, messages = 0, ended = false, terminal = false; + const timeout = setTimeout(() => fail('timeout'), 120000); + const send = obj => { if (client.readyState === WebSocket.OPEN) client.send(JSON.stringify(obj)); }; + function cleanup() { + clearTimeout(timeout); + controller.abort(); + if (stt && stt.readyState < WebSocket.CLOSING) stt.terminate(); + if (tts?.ws && tts.ws.readyState < WebSocket.CLOSING) tts.ws.terminate(); + } + function fail(code) { + if (terminal) return; + terminal = true; send({ type: 'error', code }); cleanup(); + if (client.readyState === WebSocket.OPEN) client.close(); + } + function sendChunk(chunk, commit = false) { + if (!stt || stt.readyState !== WebSocket.OPEN || stt.bufferedAmount > MAX_BUFFERED) return fail('stt_unavailable'); + stt.send(JSON.stringify({ message_type: 'input_audio_chunk', audio_base_64: chunk.toString('base64'), ...(commit ? { commit: true } : {}) })); + } + async function respond(text, settings) { + if (state !== 'processing' || terminal) return; + text = text.trim(); + if (!text) return fail('empty_transcript'); + send({ type: 'transcript', text }); + try { + tts = await openTts(settings, signal, audio => { + if (terminal || client.readyState !== WebSocket.OPEN) return; + if (client.bufferedAmount > MAX_BUFFERED) return fail('client_too_slow'); + client.send(audio, { binary: true }); + }); + if (terminal) return; + const answer = await streamAnswer(settings, store.getHistory(), text, signal, part => tts.text(part)); + tts.finish(); + await tts.done; + if (terminal) return; + store.recordTurn(text, answer); + terminal = true; send({ type: 'done' }); cleanup(); + if (client.readyState === WebSocket.OPEN) client.close(1000); + } catch { fail('upstream_error'); } + } + let settings; + let chain = Promise.resolve(); + client.on('message', (data, binary) => { + if (++messages > 10000 || (received += data.length) > MAX_AUDIO + 65536) return fail('input_limit'); + chain = chain.then(async () => { + if (terminal) return; + if (binary) { + if (state !== 'recording') return fail('invalid_state'); + const audio = Buffer.from(data); + if (!audio.length || audio.length % 2 || (bytes += audio.length) > MAX_AUDIO) return fail('invalid_audio'); + if (pending) sendChunk(pending); + pending = audio; + return; + } + let msg; + try { msg = JSON.parse(data.toString()); } catch { return fail('invalid_message'); } + if (msg?.type === 'start' && state === 'waiting') { + state = 'connecting'; settings = store.getSettings(); + try { + stt = await openStt(settings, signal, text => { if (!terminal && state === 'recording') send({ type: 'partial', text }); }, text => { if (!ended || terminal) return; respond(text, settings); }, () => fail('stt_error')); + if (terminal) return; + state = 'recording'; + } catch { fail('stt_error'); } + } else if (msg?.type === 'end' && state === 'recording') { + if (!pending) return fail('empty_audio'); + ended = true; state = 'processing'; sendChunk(pending, true); pending = null; + } else fail('invalid_state'); + }).catch(() => fail('internal_error')); + }); + client.on('error', () => { if (!terminal) { terminal = true; cleanup(); } }); + client.on('close', () => { if (!terminal) { terminal = true; cleanup(); } }); + send({ type: 'ready' }); +} diff --git a/station/README.md b/station/README.md new file mode 100644 index 0000000..507ce2c --- /dev/null +++ b/station/README.md @@ -0,0 +1,44 @@ +# Клиент Mini 2 + +Клиент собирается в статический ARM64-бинарник для ядра Mini 2. Для захвата и вывода он запускает штатные `arecord` и `aplay`, которые на этой прошивке собраны для armhf. + +```sh +cd station +go test ./... +CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -trimpath -ldflags='-s -w' -o alice-station . +``` + +На тестовой колонке файлы лежат в `/data/opt/alice-backend/`. Помести туда бинарник, `with-mic.sh` и отдельный `DEVICE_TOKEN` в файле `device-token` с правами `0600`. Для WSS нужны доступ в сеть, корректное время и доверенный набор сертификатов. + +## Ручной тест с микрофоном + +На этой Mini 2 `hw:0,1` обычно занят штатным `platformd`. Скрипт `with-mic.sh` освобождает микрофон на время теста, даёт не больше 90 секунд и возвращает штатный процесс после выхода. Для остановки нажми Ctrl+C в SSH-сеансе с терминалом. + +```sh +cd /data/opt/alice-backend +./with-mic.sh ptt -voice-threshold 55 +``` + +Нажми Enter, дождись сигнала, затем скажи команду. Настройки `-voice-threshold`, `-silence` и `-max` регулируют порог голоса, паузу в конце и предел длительности. Значение 55 проверено на этой колонке вблизи микрофона. + +## Локальная активация по фразе + +```sh +./with-mic.sh enroll -samples 4 -voice-threshold 55 +./with-mic.sh wake -voice-threshold 55 -once +``` + +При обучении нажимай Enter перед каждым образцом и после сигнала произноси «Алиса», затем сделай паузу. Клиент сохраняет спектральные признаки в `wake-templates.json` с правами `0600`. На тестовой колонке эти образцы уже записаны. В режиме `wake` произнеси «Алиса», дождись сигнала и скажи команду. Флаг `-once` завершает сеанс после первого успешного ответа. Без флага можно провести несколько ходов за один пробный запуск. + +Клиент сопоставляет целый сегмент речи со спектральными шаблонами методом DTW. Из-за этого после «Алиса» требуется короткая пауза. Оценки пропусков и ложных срабатываний на расстоянии, под музыку и у разных людей пока нет. Клиент использует один канал микрофона и простой даунсемплинг, AEC и прерывание речи в ответе отсутствуют. + +До локального срабатывания соединение с сервером не открывается. После сигнала начало команды буферизуется в RAM, а слово активации остаётся на колонке. Проверенное состояние кнопки mute на этой Mini 2 читается из `/sys/class/gpio/gpio414/value`, где `1` означает разрешённый микрофон. В mute клиент отбрасывает вход, сбрасывает состояние детектора и прерывает текущую передачу. При недоступном GPIO применяется то же поведение. Состояние можно читать из другого узла через `-mute-gpio`, если на конкретной ревизии этот сигнал расположен иначе. + +## Проверка из файла + +```sh +export DEVICE_TOKEN="$(cat device-token)" +./alice-station -mode file -input /path/to/input.pcm +``` + +Этот режим отправляет тестовый файл (mono PCM16LE, 16 кГц, до 30 секунд) и выводит ответ через ALSA (mono PCM16LE, 24 кГц). Микрофон при этом не открывается. diff --git a/station/audio_linux.go b/station/audio_linux.go new file mode 100644 index 0000000..2ba6e91 --- /dev/null +++ b/station/audio_linux.go @@ -0,0 +1,10 @@ +package main + +import ( + "os/exec" + "syscall" +) + +func terminateWithClient(cmd *exec.Cmd) { + cmd.SysProcAttr = &syscall.SysProcAttr{Pdeathsig: syscall.SIGKILL} +} diff --git a/station/audio_other.go b/station/audio_other.go new file mode 100644 index 0000000..f47627e --- /dev/null +++ b/station/audio_other.go @@ -0,0 +1,7 @@ +//go:build !linux + +package main + +import "os/exec" + +func terminateWithClient(*exec.Cmd) {} diff --git a/station/go.mod b/station/go.mod new file mode 100644 index 0000000..17ca92c --- /dev/null +++ b/station/go.mod @@ -0,0 +1,5 @@ +module alice-back/station + +go 1.21 + +require github.com/gorilla/websocket v1.5.3 diff --git a/station/go.sum b/station/go.sum new file mode 100644 index 0000000..25a9fc4 --- /dev/null +++ b/station/go.sum @@ -0,0 +1,2 @@ +github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= +github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= diff --git a/station/main.go b/station/main.go new file mode 100644 index 0000000..a71e83d --- /dev/null +++ b/station/main.go @@ -0,0 +1,643 @@ +package main + +import ( + "bufio" + "bytes" + "context" + "crypto/tls" + "encoding/binary" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "log" + "math" + "net/http" + "os" + "os/exec" + "os/signal" + "strings" + "sync" + "syscall" + "time" + + "github.com/gorilla/websocket" +) + +const rate = 16000 +const frameSamples = 320 +const hop = 160 +const captureBlock = hop * 3 * 16 + +type template struct { + Frames [][]float64 `json:"frames"` +} +type message struct { + Type string `json:"type"` + Code string `json:"code,omitempty"` +} + +var hann [frameSamples]float64 +var coefficients [20]float64 + +func init() { + for i := range hann { + hann[i] = 0.5 - 0.5*math.Cos(2*math.Pi*float64(i)/float64(frameSamples-1)) + } + for b := range coefficients { + freq := 150 * math.Pow(7000.0/150.0, float64(b)/19) + coefficients[b] = 2 * math.Cos(2*math.Pi*freq/rate) + } +} + +// features calculates log spectral energy in coarse bands, normalized against overall level. +// It is a small experimental template matcher, not a production speech model. +func features(samples []int16) []float64 { + const bands = 20 + out := make([]float64, bands) + for b := 0; b < bands; b++ { + coeff := coefficients[b] + var a, c float64 + for i, s := range samples { + v := float64(s)*hann[i] + coeff*a - c + c = a + a = v + } + power := a*a + c*c - coeff*a*c + out[b] = math.Log(1 + math.Max(0, power)) + } + var mean float64 + for _, v := range out { + mean += v + } + mean /= bands + for i := range out { + out[i] -= mean + } + return out +} +func rms(s []int16) float64 { + var v float64 + for _, x := range s { + v += float64(x) * float64(x) + } + return math.Sqrt(v / float64(len(s))) +} +func distance(a, b [][]float64) float64 { + if len(a) == 0 || len(b) == 0 { + return math.Inf(1) + } + prev := make([]float64, len(b)+1) + for i := 1; i < len(prev); i++ { + prev[i] = math.Inf(1) + } + for _, x := range a { + cur := make([]float64, len(b)+1) + cur[0] = math.Inf(1) + for j, y := range b { + var cost float64 + for k := range x { + d := x[k] - y[k] + cost += d * d + } + cost = math.Sqrt(cost / float64(len(x))) + cur[j+1] = cost + math.Min(prev[j], math.Min(prev[j+1], cur[j])) + } + prev = cur + } + return prev[len(b)] / float64(len(a)+len(b)) +} +func loadTemplates(path string) ([]template, error) { + f, e := os.Open(path) + if e != nil { + return nil, e + } + defer f.Close() + b, e := io.ReadAll(io.LimitReader(f, 2<<20+1)) + if len(b) > 2<<20 { + return nil, errors.New("template file too large") + } + if e != nil { + return nil, e + } + var t []template + e = json.Unmarshal(b, &t) + if e != nil { + return nil, e + } + if len(t) == 0 || len(t) > 10 { + return nil, errors.New("no templates") + } + for _, x := range t { + if len(x.Frames) == 0 || len(x.Frames) > 400 { + return nil, errors.New("empty template") + } + for _, f := range x.Frames { + if len(f) != 20 { + return nil, errors.New("invalid feature dimensions") + } + } + } + return t, nil +} + +// Convert each 480-frame/8-channel block to 160 mono samples. Only physical +// mic channel 0 is used, never the odd channels (which may have other meanings). +func downsample(raw []byte) []int16 { + n := len(raw) / (8 * 2 * 3) + out := make([]int16, n) + for i := 0; i < n; i++ { + var sum int32 + for j := 0; j < 3; j++ { + off := (i*3 + j) * 16 + sum += int32(int16(binary.LittleEndian.Uint16(raw[off:]))) + } + out[i] = int16(sum / 3) + } + return out +} +func pcm(s []int16) []byte { + b := make([]byte, len(s)*2) + for i, x := range s { + binary.LittleEndian.PutUint16(b[i*2:], uint16(x)) + } + return b +} +func capture(device string) (*exec.Cmd, io.ReadCloser, error) { + c := exec.Command("arecord", "-q", "-D", device, "-t", "raw", "-f", "S16_LE", "-r", "48000", "-c", "8", "--period-time=10000", "--buffer-time=80000") + c.Stderr = os.Stderr + terminateWithClient(c) + p, e := c.StdoutPipe() + if e != nil { + return nil, nil, e + } + if e = c.Start(); e != nil { + return nil, nil, e + } + return c, p, nil +} + +// Continuously drain the microphone even when the network or user is slow. Drop +// old frames rather than sending stale audio captured before activation. +type liveReader struct { + chunks chan []byte + remainder []byte + err error +} + +func newLiveReader(r io.Reader) *liveReader { + l := &liveReader{chunks: make(chan []byte, 2)} + go func() { + for { + b := make([]byte, captureBlock) + if _, e := io.ReadFull(r, b); e != nil { + l.err = e + close(l.chunks) + return + } + select { + case l.chunks <- b: + default: + select { + case <-l.chunks: + default: + } + l.chunks <- b + } + } + }() + return l +} +func (l *liveReader) discardPending() { + l.remainder = nil + for { + select { + case _, ok := <-l.chunks: + if !ok { + return + } + default: + return + } + } +} +func (l *liveReader) Read(p []byte) (int, error) { + if len(l.remainder) == 0 { + b, ok := <-l.chunks + if !ok { + return 0, l.err + } + l.remainder = b + } + n := copy(p, l.remainder) + l.remainder = l.remainder[n:] + return n, nil +} + +var mutePath string +var errMuted = errors.New("microphone muted or mute state unavailable") + +func unmuted() bool { + b, e := os.ReadFile(mutePath) + return e == nil && strings.TrimSpace(string(b)) == "1" +} +func cue(device string) error { + samples := make([]int16, 2400) + for i := range samples { + samples[i] = int16(12000 * math.Sin(2*math.Pi*880*float64(i)/24000) * math.Sin(math.Pi*float64(i)/float64(len(samples)))) + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + c := exec.CommandContext(ctx, "aplay", "-q", "-D", device, "-t", "raw", "-f", "S16_LE", "-r", "24000", "-c", "1") + c.Stdin = bytes.NewReader(pcm(samples)) + c.Stderr = os.Stderr + terminateWithClient(c) + return c.Run() +} +func waitRecord(r io.Reader, max time.Duration, silence time.Duration, threshold float64) ([][]float64, error) { + var samples []int16 + var frames [][]float64 + start := time.Now() + last := start + heard := false + buf := make([]byte, hop*3*16) + for time.Since(start) < max { + if _, e := io.ReadFull(r, buf); e != nil { + return nil, e + } + if !unmuted() { + samples = nil + frames = nil + heard = false + last = time.Now() + start = time.Now() + continue + } + s := downsample(buf) + samples = append(samples, s...) + if len(samples) >= frameSamples { + frame := samples[:frameSamples] + if rms(frame) >= threshold { + heard = true + last = time.Now() + } + if heard { + frames = append(frames, features(frame)) + } + samples = samples[hop:] + } + if len(frames) >= 250 { + return frames, nil + } + if heard && time.Since(last) >= silence { + return frames, nil + } + } + if !heard { + return nil, errors.New("no speech detected") + } + return frames, nil +} +func enroll(r io.Reader, path string, count int, threshold float64, playback string) error { + var all []template + in := bufio.NewReader(os.Stdin) + for i := 0; i < count; i++ { + fmt.Printf("Sample %d/%d: press Enter, then say the wake phrase.\n", i+1, count) + if _, e := in.ReadString('\n'); e != nil { + return e + } + for !unmuted() { + time.Sleep(100 * time.Millisecond) + } + if e := cue(playback); e != nil { + return e + } + if live, ok := r.(*liveReader); ok { + live.discardPending() + } + f, e := waitRecord(r, 4*time.Second, 350*time.Millisecond, threshold) + if e != nil { + return e + } + all = append(all, template{f}) + fmt.Printf("Recorded %d frames.\n", len(f)) + } + b, e := json.Marshal(all) + if e != nil { + return e + } + return os.WriteFile(path, b, 0600) +} +func connect(url, token string) (*websocket.Conn, error) { + h := http.Header{} + h.Set("Authorization", "Bearer "+token) + d := websocket.Dialer{HandshakeTimeout: 15 * time.Second, TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12}} + c, resp, e := d.Dial(url, h) + if e != nil { + if resp != nil { + return nil, fmt.Errorf("websocket handshake HTTP %d: %w", resp.StatusCode, e) + } + return nil, e + } + c.SetReadLimit(4 << 20) + _ = c.SetReadDeadline(time.Now().Add(20 * time.Second)) + var m message + if e = c.ReadJSON(&m); e != nil { + c.Close() + return nil, e + } + if m.Type != "ready" { + c.Close() + return nil, fmt.Errorf("expected ready, got %q", m.Type) + } + _ = c.SetReadDeadline(time.Time{}) + return c, nil +} +func exchange(url, token string, live *liveReader, fileAudio []byte, threshold float64, max, silence time.Duration, playback string) error { + if fileAudio == nil && !unmuted() { + return errMuted + } + c, e := connect(url, token) + if e != nil { + return e + } + defer c.Close() + if fileAudio == nil && !unmuted() { + return errMuted + } + _ = c.SetWriteDeadline(time.Now().Add(15 * time.Second)) + if e = c.WriteJSON(message{Type: "start"}); e != nil { + return e + } + if fileAudio != nil { + for len(fileAudio) > 0 { + n := 3200 + if n > len(fileAudio) { + n = len(fileAudio) + } + _ = c.SetWriteDeadline(time.Now().Add(15 * time.Second)) + if e = c.WriteMessage(websocket.BinaryMessage, fileAudio[:n]); e != nil { + return e + } + fileAudio = fileAudio[n:] + } + } else { + if e = cue(playback); e != nil { + return e + } + live.discardPending() + // Keep the onset in a short local buffer. It contains only post-cue audio. + const prerollSamples = rate / 5 + pre := make([]int16, 0, prerollSamples+hop) + started := time.Now() + lastVoice := started + heard := false + var voiced int + buf := make([]byte, captureBlock) + for time.Since(started) < max { + if _, e = io.ReadFull(live, buf); e != nil { + return e + } + if !unmuted() { + return errMuted + } + s := downsample(buf) + if rms(s) >= threshold { + voiced++ + lastVoice = time.Now() + } else { + voiced = 0 + } + if !heard { + pre = append(pre, s...) + if len(pre) > prerollSamples { + copy(pre, pre[len(pre)-prerollSamples:]) + pre = pre[:prerollSamples] + } + if voiced < 4 { + continue + } + heard = true + s = pre + } + _ = c.SetWriteDeadline(time.Now().Add(15 * time.Second)) + if e = c.WriteMessage(websocket.BinaryMessage, pcm(s)); e != nil { + return e + } + if heard && time.Since(lastVoice) >= silence { + break + } + } + if !unmuted() { + return errMuted + } + if !heard { + return errors.New("no speech detected after cue") + } + } + _ = c.SetWriteDeadline(time.Now().Add(15 * time.Second)) + if e = c.WriteJSON(message{Type: "end"}); e != nil { + return e + } + player := exec.Command("aplay", "-q", "-D", playback, "-t", "raw", "-f", "S16_LE", "-r", "24000", "-c", "1") + player.Stderr = os.Stderr + terminateWithClient(player) + w, e := player.StdinPipe() + if e != nil { + return e + } + if e = player.Start(); e != nil { + return e + } + defer func() { + w.Close() + if player.Process != nil { + player.Process.Kill() + } + player.Wait() + }() + _ = c.SetReadDeadline(time.Now().Add(90 * time.Second)) + for { + typ, data, e := c.ReadMessage() + if e != nil { + return e + } + if typ == websocket.BinaryMessage { + if _, e = w.Write(data); e != nil { + return e + } + continue + } + if typ == websocket.TextMessage { + var m message + if e = json.Unmarshal(data, &m); e != nil { + return e + } + switch m.Type { + case "done": + if e = w.Close(); e != nil { + return e + } + e = player.Wait() + player.Process = nil + return e + case "error": + return fmt.Errorf("server error: %s", m.Code) + } + } + } +} +func main() { + mode := flag.String("mode", "ptt", "ptt, enroll, wake, or file") + muteGPIO := flag.String("mute-gpio", "/sys/class/gpio/gpio414/value", "physical microphone mute value, 1 = unmuted") + input := flag.String("input", "", "16 kHz mono PCM16LE file for file mode (max 30 seconds)") + device := flag.String("capture", "hw:0,1", "ALSA capture device") + playback := flag.String("playback", "default", "ALSA playback device") + url := flag.String("url", "wss://alice.pepega.systems/v1/voice", "server websocket URL") + path := flag.String("templates", "wake-templates.json", "local enrollment file") + th := flag.Float64("voice-threshold", 350, "speech RMS threshold") + match := flag.Float64("match-threshold", 1.3, "experimental DTW maximum distance") + oneTurn := flag.Bool("once", false, "exit wake mode after one successful response") + count := flag.Int("samples", 4, "enrollment samples") + max := flag.Duration("max", 12*time.Second, "maximum request duration") + silence := flag.Duration("silence", 1200*time.Millisecond, "end after speech silence") + flag.Parse() + mutePath = *muteGPIO + if *th <= 0 || *max <= 0 || *silence <= 0 || *count < 1 || *count > 10 { + log.Fatal("invalid threshold, duration, or sample count") + } + token := strings.TrimSpace(os.Getenv("DEVICE_TOKEN")) + if *mode != "enroll" && token == "" { + log.Fatal("DEVICE_TOKEN is required") + } + var ts []template + var e error + if *mode == "wake" { + ts, e = loadTemplates(*path) + if e != nil { + log.Fatal(e) + } + } else if *mode != "ptt" && *mode != "enroll" && *mode != "file" { + log.Fatal("unknown mode") + } + if *mode == "file" { + f, e := os.Open(*input) + if e != nil { + log.Fatal(e) + } + b, e := io.ReadAll(io.LimitReader(f, 30*rate*2+1)) + f.Close() + if e != nil { + log.Fatal(e) + } + if len(b) == 0 || len(b) > 30*rate*2 || len(b)%2 != 0 { + log.Fatal("input must contain at most 30 seconds of PCM16LE and have an even nonzero size") + } + if e = exchange(*url, token, nil, b, *th, *max, *silence, *playback); e != nil { + log.Fatal(e) + } + return + } + if *mode == "ptt" { + fmt.Println("Press Enter to start speaking.") + if _, e = bufio.NewReader(os.Stdin).ReadString('\n'); e != nil { + log.Fatal(e) + } + } + cmd, r, e := capture(*device) + if e != nil { + log.Fatal(e) + } + var once sync.Once + cleanup := func() { once.Do(func() { r.Close(); cmd.Process.Kill(); cmd.Wait() }) } + defer cleanup() + sig := make(chan os.Signal, 1) + signal.Notify(sig, os.Interrupt, syscall.SIGTERM) + go func() { <-sig; cleanup(); os.Exit(0) }() + live := newLiveReader(r) + if *mode == "enroll" { + if e = enroll(live, *path, *count, *th, *playback); e != nil { + cleanup() + log.Fatal(e) + } + return + } + if *mode == "ptt" { + for { + for !unmuted() { + time.Sleep(100 * time.Millisecond) + } + e = exchange(*url, token, live, nil, *th, *max, *silence, *playback) + if errors.Is(e, errMuted) { + log.Println("microphone muted; waiting for unmute") + continue + } + if e != nil { + cleanup() + log.Fatal(e) + } + return + } + } + + // Detect whole speech segments locally before opening a voice connection. + log.Println("Listening locally for enrolled phrase") + raw := make([]byte, hop*3*16) + var samples []int16 + var f [][]float64 + active := false + quiet := 0 + for { + if _, e = io.ReadFull(live, raw); e != nil { + cleanup() + log.Fatal(e) + } + if !unmuted() { + samples = nil + f = nil + active = false + quiet = 0 + continue + } + s := downsample(raw) + samples = append(samples, s...) + if len(samples) < frameSamples { + continue + } + v := rms(samples[:frameSamples]) + frame := samples[:frameSamples] + if v >= *th { + active = true + quiet = 0 + } + samples = samples[hop:] + if active { + f = append(f, features(frame)) + if v < *th { + quiet++ + } + if quiet >= 35 || len(f) >= 250 { + best := math.Inf(1) + for _, t := range ts { + if d := distance(f, t.Frames); d < best { + best = d + } + } + log.Printf("wake candidate distance=%.3f", best) + if best <= *match { + log.Println("wake detected") + if e = exchange(*url, token, live, nil, *th, *max, *silence, *playback); e != nil { + log.Printf("session: %v", e) + } else if *oneTurn { + return + } + } + f = nil + active = false + quiet = 0 + samples = nil + } + } + } +} diff --git a/station/main_test.go b/station/main_test.go new file mode 100644 index 0000000..8bdecbd --- /dev/null +++ b/station/main_test.go @@ -0,0 +1,54 @@ +package main + +import ( + "encoding/binary" + "io" + "math" + "strings" + "testing" +) + +func TestDownsampleUsesMicZero(t *testing.T) { + raw := make([]byte, 3*16) + for i, v := range []int16{300, 600, 900} { + binary.LittleEndian.PutUint16(raw[i*16:], uint16(v)) + binary.LittleEndian.PutUint16(raw[i*16+2:], 1234) + } + got := downsample(raw) + if len(got) != 1 || got[0] != 600 { + t.Fatalf("got %v", got) + } +} +func TestFeaturesAndDistance(t *testing.T) { + a := make([]int16, frameSamples) + for i := range a { + a[i] = int16(4000 * math.Sin(2*math.Pi*440*float64(i)/rate)) + } + f := features(a) + if len(f) != 20 { + t.Fatal(len(f)) + } + d := distance([][]float64{f, f}, [][]float64{f, f}) + if math.Abs(d) > 1e-9 { + t.Fatal(d) + } + if !math.IsInf(distance(nil, [][]float64{f}), 1) { + t.Fatal("empty distance") + } +} +func TestTemplateValidation(t *testing.T) { + p := t.TempDir() + "/template.json" + if _, e := loadTemplates(p); e == nil { + t.Fatal("missing template accepted") + } +} + +func TestLiveReaderEOFStaysEOF(t *testing.T) { + r := newLiveReader(strings.NewReader("")) + b := make([]byte, 1) + for i := 0; i < 2; i++ { + if _, err := r.Read(b); err != io.EOF { + t.Fatalf("read %d: %v", i, err) + } + } +} diff --git a/station/with-mic.sh b/station/with-mic.sh new file mode 100755 index 0000000..5a65c36 --- /dev/null +++ b/station/with-mic.sh @@ -0,0 +1,104 @@ +#!/bin/sh +# Run a supervised test while the stock microphone service is temporarily paused. +set -eu +cd "$(dirname "$0")" +mode=${1:-ptt} +case "$mode" in + ptt|enroll|wake) ;; + *) echo "Usage: $0 [ptt|enroll|wake] [client flags]" >&2; exit 2 ;; +esac +if [ "$#" -gt 0 ]; then shift; fi + +umask 077 +exec 9>/tmp/alice-backend-mic.lock +if ! flock -n 9; then + echo 'A station microphone test is already running.' >&2 + exit 1 +fi +exec 3<&0 +if [ "$mode" != enroll ]; then + DEVICE_TOKEN=$(cat device-token) + export DEVICE_TOKEN +fi +set -f +owners=$(fuser /dev/snd/pcmC0D1c 2>/dev/null || true) +launcher= +guard= +client= + +mic_owned_by_stock() { + mic_pid=$(fuser /dev/snd/pcmC0D1c 2>/dev/null | tr -d '[:space:]' || true) + case "$mic_pid" in + ''|*[!0-9]*) return 1 ;; + esac + [ "$(tr '\0' ' ' < "/proc/$mic_pid/cmdline" 2>/dev/null)" = '/system/vendor/quasar/maind --platformd ' ] +} + +restore() { + if [ -n "$guard" ]; then kill "$guard" 2>/dev/null || true; fi + if [ -n "$client" ]; then + kill -TERM "$client" 2>/dev/null || true + wait "$client" 2>/dev/null || true + fi + if [ -n "$launcher" ] && [ "$(readlink "/proc/$launcher/exe")" = /system/vendor/quasar/quasar_launcher2 ]; then + kill -CONT "$launcher" 2>/dev/null || true + for second in 1 2 3 4 5 6 7 8; do + if mic_owned_by_stock; then return; fi + sleep 1 + done + echo 'Restarting stock Quasar to restore microphone ownership.' >&2 + /system/vendor/quasar/q restart + fi +} +trap restore EXIT +trap 'exit 130' HUP INT TERM + +if [ -n "$owners" ]; then + platform=$(printf '%s' "$owners" | tr -d '[:space:]') + case "$platform" in + ''|*[!0-9]*) echo "Unexpected microphone owner" >&2; exit 1 ;; + esac + if [ "$(tr '\0' ' ' < "/proc/$platform/cmdline")" != '/system/vendor/quasar/maind --platformd ' ]; then + echo 'Microphone is busy.' >&2 + exit 1 + fi + launcher=$(awk '$1 == "PPid:" { print $2 }' "/proc/$platform/status") + if [ "$(readlink "/proc/$launcher/exe")" != /system/vendor/quasar/quasar_launcher2 ]; then + launcher= + echo 'Unexpected parent process.' >&2 + exit 1 + fi +fi + +# A finite test and detached fallback restore the stock launcher even after SSH loss. +lease=90 +owner=$$ +started=$(awk '{print $22}' "/proc/$owner/stat") +nohup sh -c ' + sleep "$1" + if [ "$(awk "{print \$22}" "/proc/$2/stat" 2>/dev/null)" = "$3" ]; then + kill -TERM "$2" 2>/dev/null || true + fi + if [ -n "$4" ] && [ "$(readlink "/proc/$4/exe")" = /system/vendor/quasar/quasar_launcher2 ]; then + kill -CONT "$4" 2>/dev/null || true + fi +' sh "$lease" "$owner" "$started" "$launcher" /dev/null 2>&1 9>&- & +guard=$! + +if [ -n "$launcher" ]; then + kill -STOP "$launcher" + kill -TERM "$platform" + for second in 1 2 3; do + if ! fuser /dev/snd/pcmC0D1c >/dev/null 2>&1; then break; fi + sleep 1 + done + if fuser /dev/snd/pcmC0D1c >/dev/null 2>&1; then + echo 'The stock service kept the microphone.' >&2 + exit 1 + fi +fi + +./alice-station -mode "$mode" "$@" <&3 9>&- & +client=$! +wait "$client" +client= diff --git a/test/admin.test.mjs b/test/admin.test.mjs new file mode 100644 index 0000000..7688fc3 --- /dev/null +++ b/test/admin.test.mjs @@ -0,0 +1,43 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { createHmac } from 'node:crypto'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { verifyTelegram } from '../server/admin.mjs'; +import { openStore, validateSettings } from '../server/store.mjs'; + +test('Telegram signatures expire and authorize only the configured owner', () => { + const now = 1794312000000; + const token = '1234:local-test-secret'; + const owner = '42'; + const params = new URLSearchParams({ + user: '{"id":42}', + signature: 'third-party-signature', + auth_date: String(now / 1000), + }); + const secret = createHmac('sha256', 'WebAppData').update(token).digest(); + const check = 'auth_date=' + now / 1000 + '\nsignature=third-party-signature\nuser={"id":42}'; + params.set('hash', createHmac('sha256', secret).update(check).digest('hex')); + + assert.equal(verifyTelegram(params.toString(), token, owner, now), true); + assert.equal(verifyTelegram(params.toString(), token, '43', now), false); + assert.equal(verifyTelegram(params.toString(), token, owner, now + 3601000), false); + params.set('user', '{"id":43}'); + assert.equal(verifyTelegram(params.toString(), token, '43', now), false); +}); + +test('settings and bounded history survive a restart', t => { + const directory = mkdtempSync(join(tmpdir(), 'alice-store-')); + t.after(() => rmSync(directory, { recursive: true })); + const store = openStore(directory); + const changed = validateSettings({ ...store.getSettings(), systemPrompt: ' Тест ' }); + store.updateSettings(changed); + for (let i = 0; i < 55; i++) store.recordTurn(String(i), 'Ответ'); + const reopened = openStore(directory); + assert.equal(reopened.getSettings().systemPrompt, 'Тест'); + assert.equal(reopened.getHistory().length, 50); + assert.equal(reopened.getHistory()[0].user, '5'); + reopened.clearHistory(); + assert.deepEqual(openStore(directory).getHistory(), []); +}); diff --git a/test/voice.test.mjs b/test/voice.test.mjs new file mode 100644 index 0000000..c705c88 --- /dev/null +++ b/test/voice.test.mjs @@ -0,0 +1,44 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { once } from 'node:events'; +import { WebSocketServer, WebSocket } from 'ws'; + +// Local upstreams exercise the real framing without provider credentials or network access. +test('one PCM turn streams through Scribe and dialogue TTS', async () => { + const stt = new WebSocketServer({ port: 0 }); + const tts = new WebSocketServer({ port: 0 }); + await Promise.all([once(stt, 'listening'), once(tts, 'listening')]); + process.env.ELEVENLABS_API_KEY = 'test'; + process.env.LLM_MODE = 'echo'; + process.env.ELEVEN_STT_URL = `ws://127.0.0.1:${stt.address().port}/stt`; + process.env.ELEVEN_TTS_URL = `ws://127.0.0.1:${tts.address().port}/tts`; + let input = Buffer.alloc(0), ttsText = '', recorded; + stt.on('connection', (ws, req) => { + assert.equal(new URL(req.url, 'http://x').searchParams.get('commit_strategy'), 'manual'); + ws.on('message', data => { + const msg = JSON.parse(data); + input = Buffer.concat([input, Buffer.from(msg.audio_base_64, 'base64')]); + if (msg.commit) ws.send(JSON.stringify({ message_type: 'committed_transcript', text: 'Привет' })); + }); + }); + tts.on('connection', ws => ws.on('message', data => { + const msg = JSON.parse(data); + if (msg.inputs) ttsText += msg.inputs[0].text; + if (msg.close_socket) { ws.send(JSON.stringify({ audio: Buffer.from([1, 2, 3, 4]).toString('base64') })); ws.send(JSON.stringify({ is_final: true })); } + })); + const { handleTurn } = await import('../server/turns.mjs'); + const local = new WebSocketServer({ port: 0 }); + await once(local, 'listening'); + const store = { getSettings: () => ({ sttModel: 'scribe_v2_realtime', ttsModel: 'eleven_v4_turbo', voiceId: 'voice', systemPrompt: '' }), getHistory: () => [], recordTurn: (...args) => { recorded = args; } }; + local.on('connection', ws => handleTurn(ws, store)); + const client = new WebSocket(`ws://127.0.0.1:${local.address().port}`); + const seen = [], audio = []; + client.on('message', (data, binary) => { if (binary) audio.push(...data); else { const msg = JSON.parse(data); seen.push(msg.type); if (msg.type === 'ready') { client.send(JSON.stringify({ type: 'start' })); client.send(Buffer.from([5, 6, 7, 8])); client.send(JSON.stringify({ type: 'end' })); } } }); + await once(client, 'close'); + assert.deepEqual(seen, ['ready', 'transcript', 'done']); + assert.deepEqual([...input], [5, 6, 7, 8]); + assert.deepEqual(audio, [1, 2, 3, 4]); + assert.equal(ttsText, 'Привет'); + assert.deepEqual(recorded, ['Привет', 'Привет']); + await Promise.all([new Promise(r => local.close(r)), new Promise(r => stt.close(r)), new Promise(r => tts.close(r))]); +});