Integra reSpeaker Clip en tu servicio

Esta guía muestra cómo añadir reSpeaker Clip como una fuente de audio gestionada a un servicio de Python existente. Se centra en el límite de integración: un adaptador de dispositivo, una conexión de larga duración, una pequeña superficie de API y una propiedad de estado clara.

reSpeaker Clip es una fuente de entrada. ClipClient gestiona la comunicación con el dispositivo, ClipService la adapta a tu aplicación y tu servicio mantiene la propiedad del procesamiento y la lógica de negocio.

Para la configuración del transporte, la API completa de ClipClient, los comandos AT y el protocolo de transferencia de archivos, consulta la Guía básica del SDK de reSpeaker Clip. Esta página no vuelve a explicar BLE/GATT, formatos de trama, manejo de CRC ni herramientas de CLI.

Enfoque de integración

Mantén el dispositivo detrás de un adaptador estrecho y envía las grabaciones descargadas al punto de entrada de procesamiento que tu servicio ya utiliza:

1. Crea un ClipService para cada dispositivo físico.
2. Conéctalo durante el arranque de la aplicación y desconéctalo durante el apagado.
3. Expón solo las operaciones de grabación que tu servicio necesite.
4. Pasa los archivos de audio completados a la canalización de procesamiento existente.

Antes de empezar

Esta guía asume que ya puedes comunicarte con un Clip en tu propia máquina. Antes de continuar, asegúrate de poder:

• instalar el SDK de Python de reSpeaker Clip y conectarte a un Clip;
• iniciar y detener una grabación;
• listar sesiones completadas;
• descargar una grabación correctamente.

Si cualquiera de estos pasos falla, detente aquí y completa primero la Guía básica del SDK de reSpeaker Clip. La Guía básica del SDK es la fuente de referencia para la configuración del transporte, los comandos del dispositivo y los detalles de la transferencia de archivos; esta guía se enlaza a ella en lugar de repetirla.

Elige la guía adecuada de reSpeaker Clip

La documentación de reSpeaker Clip está organizada por capa de desarrollo. Comienza con Getting Started para la configuración del producto y los flujos de trabajo normales. Usa el Basic SDK cuando una aplicación host necesite controlar el dispositivo o descargar grabaciones. Continúa con Service Integration cuando un servicio de larga ejecución necesite poseer la conexión del dispositivo y exponer APIs. Usa el Firmware SDK solo cuando necesites cambiar el comportamiento del lado del dispositivo, los protocolos o el procesamiento de audio.

Getting Started

Descripción general del producto, configuración de hardware y flujos de primera utilización.

Basic SDK

Controla el dispositivo, gestiona las grabaciones y descarga audio sin cambiar el firmware.

Service Integration

Añade reSpeaker Clip a un servicio de Python de larga ejecución mediante un adaptador y rutas de API.

Firmware SDK

Compila, flashea, depura y personaliza el comportamiento del lado del dispositivo.

Añade reSpeaker Clip como fuente de audio a tu servicio existente

Sigue los pasos a continuación para conectar el adaptador de dispositivo a un servicio FastAPI existente.

Paso 1: Preparar la canalización

Supón que tu servicio ya tiene una función que acepta un archivo de audio:

async def process_audio(path: Path) -> dict:
    # your existing processing pipeline
    ...

Una ruta de carga existente ya llama a esta función:

@app.post("/audio")
async def upload_audio(file: UploadFile):
    path = await save_upload(file)
    return await process_audio(path)

La integración de Clip debe generar un archivo de audio local estándar y llamar a la misma función process_audio(). El procesamiento debe seguir siendo independiente del dispositivo de captura.

Encapsula esta función en el objeto de servicio utilizado por la ruta de ingesta de Clip:

# audio_service.py
from pathlib import Path


class AudioService:
    def __init__(self, recordings_dir: Path):
        self.recordings_dir = recordings_dir
        self.recordings_dir.mkdir(parents=True, exist_ok=True)

    async def process(self, path: Path) -> dict:
        return await process_audio(path)

Paso 2: Crear el adaptador

Encapsula únicamente las operaciones que necesita esta integración: conectar, desconectar, iniciar, detener y descargar.

# clip_service.py
from __future__ import annotations
from pathlib import Path

from clip import ClipClient, ClipError


class ClipUnavailable(Exception):
    """The Clip device is not connected or could not perform an operation."""


class ClipStateError(Exception):
    """The device is in the wrong state for this operation."""


class ClipService:
    """Application-level adapter over a single long-lived ClipClient."""


        self._client = ClipClient(transport, command_timeout=command_timeout)
        self._connected = False
        self._active_session: str | None = None

    async def connect(self) -> None:
        await self._client.connect()
        self._connected = True


        if self._connected:
            await self._client.disconnect()
            self._connected = False

    async def start_recording(self, mode: str | None = None) -> str:
        if not self._connected:
            raise ClipUnavailable("Clip is not connected")
        if self._active_session is not None:
            raise ClipStateError("a recording is already in progress")
        try:
            session_id = await self._client.start_recording(mode)
        except ClipError as exc:
            raise ClipUnavailable("Clip could not start recording") from exc

            raise ClipStateError("Clip did not return a session id")
        self._active_session = session_id
        return session_id

    async def stop_recording(self) -> str | None:
        if not self._connected:
            raise ClipUnavailable("Clip is not connected")
        if self._active_session is None:
            raise ClipStateError("no recording in progress")
        session_id = self._active_session
        try:
            await self._client.stop_recording()

            raise ClipUnavailable("Clip could not stop recording") from exc
        self._active_session = None
        return session_id

    async def download_session(self, session_id: str, destination: str | Path):
        if not self._connected:
            raise ClipUnavailable("Clip is not connected")
        try:
            return await self._client.download_session(session_id, destination)
        except ClipError as exc:
            raise ClipUnavailable("Clip could not download the recording") from exc

Paso 3: Vincular el ciclo de vida

Crea el adaptador durante el arranque de FastAPI, guárdalo en el estado de la aplicación y desconéctalo durante el apagado.

Vincula la conexión al ciclo de vida de FastAPI para que se abra una vez al inicio y se cierre una vez al detenerse:

# main.py
from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI

from clip import BleTransport  # or UdpTransport after the Wi-Fi AP is on
from clip_service import ClipService
from audio_service import AudioService


@asynccontextmanager
async def lifespan(app: FastAPI):
    transport = BleTransport(name="Clip")
    clip = ClipService(transport)
    await clip.connect()
    app.state.clip = clip
    app.state.audio = AudioService(recordings_dir=Path("recordings"))
    try:
        yield
    finally:
        await clip.disconnect()


app = FastAPI(lifespan=lifespan)

Paso 4: Añadir rutas

Crea endpoints para iniciar, detener e ingerir una grabación completada:

POST /clip/recordings/start
POST /clip/recordings/stop
POST /clip/sessions/{session_id}/ingest

# routes.py
from pathlib import Path

from fastapi import APIRouter, HTTPException, Request

from audio_service import AudioService
from clip_service import ClipService, ClipStateError, ClipUnavailable

router = APIRouter(prefix="/clip")


@router.post("/recordings/start")
async def start_recording(request: Request, mode: str | None = None):
    clip: ClipService = request.app.state.clip
    try:
        session_id = await clip.start_recording(mode)
    except ClipStateError as exc:
        raise HTTPException(status_code=409, detail=str(exc))
    except ClipUnavailable as exc:
        raise HTTPException(status_code=503, detail=str(exc))
    return {"session_id": session_id}


@router.post("/recordings/stop")
async def stop_recording(request: Request):
    clip: ClipService = request.app.state.clip
    try:
        session_id = await clip.stop_recording()
    except ClipStateError as exc:
        raise HTTPException(status_code=409, detail=str(exc))
    except ClipUnavailable as exc:
        raise HTTPException(status_code=503, detail=str(exc))
    return {"session_id": session_id, "status": "stopped"}


@router.post("/sessions/{session_id}/ingest")
async def ingest_session(session_id: str, request: Request):
    clip: ClipService = request.app.state.clip
    audio: AudioService = request.app.state.audio
    try:
        result = await clip.download_session(session_id, audio.recordings_dir)
    except ClipUnavailable as exc:
        raise HTTPException(status_code=502, detail=str(exc))
    ingested = [await audio.process(Path(file.path)) for file in result.files]
    return {"session_id": session_id, "files": ingested}

Paso 5: Registrar el router

Importa el router en main.py y regístralo después de crear la aplicación FastAPI:

from routes import router

app = FastAPI(lifespan=lifespan)
app.include_router(router)

La integración ahora consta de main.py, routes.py, clip_service.py y audio_service.py en el paquete de tu aplicación.

Paso 6: Verificar la API

Inicia el servicio y luego llama a las rutas en orden:

curl -X POST "http://localhost:8000/clip/recordings/start"
curl -X POST "http://localhost:8000/clip/recordings/stop"
curl -X POST "http://localhost:8000/clip/sessions/<session_id>/ingest"

Las respuestas de inicio y parada deben contener el mismo session_id. La respuesta de ingestión debe contener los archivos devueltos por tu canalización AudioService existente.

Consideraciones de producción para un servicio de larga ejecución

Mantén explícitos el estado de conexión, de error y de flujo de trabajo cuando la integración se ejecute como un servicio.

Límite del adaptador

Mantén todas las llamadas a ClipClient dentro de ClipService. El adaptador es propietario del acceso a la conexión, los comandos de grabación, las descargas y la traducción de errores del SDK; no es propietario del procesamiento, el almacenamiento u otra lógica de negocio.

El código por encima del adaptador debe usar conceptos de la aplicación como session_id y excepciones de la aplicación en lugar de detalles de transporte, comandos AT o transferencia de archivos.

Reutilización del cliente

Asocia un Clip físico a un ClipClient de larga vida y a un ClipService. Debido a que el firmware no tiene ID de solicitud, ClipClient serializa los comandos. Para varios dispositivos, indexa las instancias de ClipService por ID de dispositivo en un ClipManager.

Duración de la conexión

Un error común es abrir el dispositivo dentro del manejador de la solicitud:

# WRONG — do not do this in a service
@app.post("/clip/start")
async def start():
    async with ClipClient(BleTransport(name="Clip")) as clip:
        await clip.start_recording()

Esto vuelve a conectar en cada llamada y pierde la sesión activa entre solicitudes. El ciclo de vida de la aplicación es el propietario de la conexión; las solicitudes la reutilizan.

Límite de errores

Traduce las excepciones del SDK en excepciones de la aplicación dentro de ClipService, y luego asígnalas a respuestas HTTP en el límite de la ruta.

Un ejemplo sencillo de mapeo (ilustrativo, no un mandato del SDK):

Condición de la aplicación	Estado HTTP
Clip no disponible	503
Estado de dispositivo no válido	409
Fallo en el comando del dispositivo	502
Solicitud incorrecta del cliente	400

Los códigos exactos dependen de la aplicación. El código fuera del adaptador debe depender de ClipUnavailable y ClipStateError, no de excepciones del SDK.

Reconexión

Ante un error del dispositivo, márcalo como no disponible y vuelve a conectar o recrea el transporte. Un tiempo de espera de comando requiere reconexión antes del siguiente comando porque el transporte ya no puede correlacionar respuestas. Expón el fallo como ClipUnavailable y luego vuelve a conectar desde una tarea en segundo plano o desde la siguiente solicitud.

Idempotencia

Las reintentos HTTP, los reintentos de workers y los reinicios pueden enviar el mismo session_id más de una vez. Comprueba el estado persistente antes de procesarlo.

Haz un seguimiento de cada sesión en tu base de datos:

session_id	downloaded	processed	result_id	created_at
Sesión de dispositivo única	Estado de descarga	Estado de la canalización	Referencia al resultado	Momento de la primera ingestión

async def ingest_clip_session(session_id: str):
    if repository.is_processed(session_id):
        return repository.get_result(session_id)

    result = await clip_service.download_session(session_id, Path("recordings"))
    for file in result.files:
        await process_audio(Path(file.path))

    repository.mark_processed(session_id)
    return repository.get_result(session_id)

Esto evita el procesamiento duplicado y los efectos secundarios posteriores.

Propiedad del estado

Mantén el estado del dispositivo separado del estado del flujo de trabajo:

Estado	Propietario	Valores de ejemplo
Estado del Clip	Capa de integración del dispositivo (ClipService)	conectado, grabando, sesión disponible, descargable
Estado del flujo de trabajo	Capa de aplicación/trabajo	ingiriendo, procesando, completado, fallido

No colapses ambas máquinas de estados en un solo campo. Una sesión puede descargarse mientras su flujo de trabajo de aplicación aún se está ejecutando.

Próximos pasos

Después de que el adaptador esté funcionando, mantén la integración del dispositivo separada de cualquier procesamiento específico de la aplicación. Páginas relacionadas:

• reSpeaker Clip Basic SDK Guide — protocolo del dispositivo, ClipClient API, transporte BLE/UDP, transferencia de archivos
• Getting Started with the reSpeaker Clip Firmware SDK — cuando necesites modificar el firmware del lado del dispositivo en lugar de integrar el SDK

Soporte técnico y debate sobre el producto

Gracias por elegir nuestros productos. Estamos aquí para ofrecerte diferentes tipos de soporte y garantizar que tu experiencia con nuestros productos sea lo más fluida posible. Ofrecemos varios canales de comunicación para adaptarnos a distintas preferencias y necesidades.