Spaces:
Running
Running
File size: 4,708 Bytes
a31f556 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 | from fastapi import HTTPException, APIRouter, Response
from pydantic import BaseModel
router = APIRouter()
class SpeakPayload(BaseModel):
text: str
agent: str = "friday"
format: str = "wav" # always wav
class DevicePayload(BaseModel):
id: str
class SpaceAckPayload(BaseModel):
space: str # "dark_space" | "family_friendly"
persona: str = "friday"
@router.post("/speak")
async def speak(p: SpeakPayload):
"""Synthesise TTS via XTTS-v2 and return raw WAV bytes."""
try:
from backend.voice.tts import TTSPipeline
tts = TTSPipeline()
audio_bytes: bytes = await tts.synthesize(p.text, personality=p.agent)
return Response(content=audio_bytes, media_type="audio/wav")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("/space_ack")
async def space_ack(p: SpaceAckPayload):
"""
§2.1b — Space Context Voice Injection entry point.
Returns the persona-aware acknowledgement text. Android will take this text
and silently inject it into the local WAV recording without playing out loud.
"""
from backend.agent.personas import get_space_ack_prompt
prompt_text = get_space_ack_prompt(p.persona, p.space)
return {"status": "ok", "space": p.space, "persona": p.persona, "text": prompt_text}
@router.get("/devices")
async def get_devices():
return []
@router.post("/set_device")
async def set_device(p: DevicePayload):
return {"status": "ok"}
@router.get("/wake_word_status")
async def wake_status():
return {"status": "listening"}
from fastapi import UploadFile, File, BackgroundTasks
import os
import shutil
def process_offline_log_stt(file_path: str):
import asyncio
import logging
try:
from backend.voice.stt import STTPipeline
stt = STTPipeline()
with open(file_path, "rb") as f:
audio_bytes = f.read()
# Enqueue STT (wait for it as it's async)
text = asyncio.run(stt.transcribe(audio_bytes))
if text.strip():
logging.info(f"Offline Audio Parsed: {text}")
from backend.agent.react_agent import run_agent_pipeline
asyncio.run(run_agent_pipeline(text))
except Exception as e:
logging.error(f"Failed to process offline log STT: {e}")
@router.post("/upload_log")
async def upload_log(file: UploadFile = File(...), background_tasks: BackgroundTasks = BackgroundTasks()):
try:
sync_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "voice", "audio_logs", "phone_sync")
os.makedirs(sync_dir, exist_ok=True)
file_path = os.path.join(sync_dir, file.filename)
with open(file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
# Enqueue STT parsing for the offline log
background_tasks.add_task(process_offline_log_stt, file_path)
return {"status": "ok", "filename": file.filename, "message": "Successfully uploaded"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# ─── §0.4 JARVIS Voice Profile Preview & Test ────────────────────────────────
@router.get("/settings/preview/jarvis")
async def get_jarvis_voice_sample():
"""Returns current tuned JARVIS voice sample WAV for UI playback."""
import os
from fastapi.responses import FileResponse
app_data = os.environ.get("JARVIS_APP_DATA_DIR", ".")
sample_path = os.path.join(app_data, "voices", "jarvis.wav")
if os.path.exists(sample_path):
return FileResponse(sample_path, media_type="audio/wav")
raise HTTPException(status_code=404, detail="JARVIS voice sample not found. Place a jarvis.wav file in the voices directory.")
@router.post("/settings/test/jarvis")
async def test_jarvis_voice():
"""Synthesises a short test line through the full tuned §0.4 JARVIS pipeline."""
try:
from backend.voice.tts import TTSPipeline
tts = TTSPipeline()
# context="conversation" → routes through XTTS with full tuned profile
audio_bytes = await tts.synthesize(
"Good evening, sir. All systems are fully operational.",
personality="jarvis",
language="en",
context="conversation"
)
from fastapi.responses import Response
return Response(content=audio_bytes, media_type="audio/wav")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
|