Deepgram

pip install deepgram-sdk netra-sdk

import os
from netra import Netra

Netra.init(
    app_name="deepgram-stt-service",
    headers=f"x-api-key={os.environ.get('NETRA_API_KEY')}"
)

from deepgram import DeepgramClient, PrerecordedOptions, FileSource
from netra import task, workflow
import os

client = DeepgramClient(api_key=os.environ.get("DEEPGRAM_API_KEY"))

@task()
def transcribe_audio(audio_url: str) -> str:
    """Transcribe audio from URL using Deepgram."""
    options = PrerecordedOptions(
        model="nova-2",
        smart_format=True,
        punctuate=True,
        paragraphs=True
    )
    
    response = client.listen.prerecorded.transcribe_url(
        {"url": audio_url},
        options
    )
    
    return response.results.channels[0].alternatives[0].transcript

@task()
def transcribe_with_diarization(audio_url: str) -> list:
    """Transcribe with speaker diarization."""
    options = PrerecordedOptions(
        model="nova-2",
        smart_format=True,
        diarize=True,
        punctuate=True,
        utterances=True
    )
    
    response = client.listen.prerecorded.transcribe_url(
        {"url": audio_url},
        options
    )
    
    return response.results.utterances

@workflow()
def process_audio_file(audio_url: str) -> dict:
    """Process audio file with full transcription and diarization."""
    transcript = transcribe_audio(audio_url)
    speakers = transcribe_with_diarization(audio_url)
    
    return {
        "full_transcript": transcript,
        "speakers": speakers
    }

# Usage
result = process_audio_file("https://example.com/audio.mp3")

from deepgram import DeepgramClient, PrerecordedOptions, FileSource
from netra import SpanWrapper, ActionModel, UsageModel
import os
import time

client = DeepgramClient(api_key=os.environ.get("DEEPGRAM_API_KEY"))

def transcribe_with_tracking(audio_path: str) -> str:
    """Transcribe audio with detailed tracking."""
    span = SpanWrapper("deepgram-transcription")
    span.start()
    
    try:
        start_time = time.time_ns()
        
        # Read audio file
        with open(audio_path, "rb") as audio_file:
            audio_data = audio_file.read()
        
        audio_size_bytes = len(audio_data)
        span.set_attribute("audio_file", audio_path)
        span.set_attribute("audio_size_bytes", audio_size_bytes)
        span.set_attribute("model", "nova-2")
        
        # Transcribe
        payload = {"buffer": audio_data}
        options = PrerecordedOptions(
            model="nova-2",
            smart_format=True,
            punctuate=True,
            diarize=True
        )
        
        response = client.listen.prerecorded.transcribe_file(
            payload,
            options
        )
        
        transcript = response.results.channels[0].alternatives[0].transcript
        end_time = time.time_ns()
        duration_ms = (end_time - start_time) / 1_000_000
        
        # Extract metadata
        audio_duration = response.metadata.duration
        confidence = response.results.channels[0].alternatives[0].confidence
        word_count = len(response.results.channels[0].alternatives[0].words)
        
        # Track the STT API operation
        action = ActionModel(
            start_time=str(start_time),
            action="API",
            action_type="STT_TRANSCRIPTION",
            metadata={
                "provider": "deepgram",
                "model": "nova-2",
                "audio_size_bytes": str(audio_size_bytes),
                "audio_duration_seconds": str(audio_duration),
                "transcript_length": str(len(transcript)),
                "confidence": str(confidence),
                "duration_ms": str(duration_ms),
                "words_detected": str(word_count)
            },
            success=True
        )
        span.set_action([action])
        
        # Track usage
        usage = UsageModel(
            model="nova-2",
            usage_type="audio_seconds",
            units_used=audio_duration,
            cost_in_usd=audio_duration * 0.0043  # $0.0043 per second
        )
        span.set_usage([usage])
        
        span.set_attribute("transcript_length", len(transcript))
        span.set_attribute("confidence", confidence)
        span.set_status({"code": 1, "message": "Success"})
        span.end()
        
        return transcript
        
    except Exception as e:
        span.set_error(e)
        span.set_status({"code": 2, "message": "Error"})
        span.end()
        raise

# Usage
transcript = transcribe_with_tracking("./audio/sample.mp3")