Deepgram

pip install deepgram-sdk netra-sdk

npm 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')}"
)

import Netra from 'netra-sdk';

await Netra.init({
  appName: 'deepgram-stt-service',
  headers: `x-api-key=${process.env.NETRA_API_KEY}`
});

from deepgram import DeepgramClient, PrerecordedOptions, FileSource
from netra.decorators 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")

import { createClient } from '@deepgram/sdk';
import { task, workflow } from 'netra-sdk';

const deepgram = createClient(process.env.DEEPGRAM_API_KEY);

class TranscriptionService {
  @task()
  async transcribeAudio(audioUrl: string): Promise<string> {
    const { result } = await deepgram.listen.prerecorded.transcribeUrl(
      { url: audioUrl },
      {
        model: 'nova-2',
        smart_format: true,
        punctuate: true,
        paragraphs: true
      }
    );

    return result.results.channels[0].alternatives[0].transcript;
  }

  @task()
  async transcribeWithDiarization(audioUrl: string): Promise<any> {
    const { result } = await deepgram.listen.prerecorded.transcribeUrl(
      { url: audioUrl },
      {
        model: 'nova-2',
        smart_format: true,
        diarize: true,
        punctuate: true,
        utterances: true
      }
    );

    return result.results.utterances;
  }

  @workflow()
  async processAudioFile(audioUrl: string): Promise<any> {
    const transcript = await this.transcribeAudio(audioUrl);
    const speakers = await this.transcribeWithDiarization(audioUrl);
    
    return {
      full_transcript: transcript,
      speakers: speakers
    };
  }
}

// Usage
const service = new TranscriptionService();
const result = await service.processAudioFile(
  '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")

import { createClient } from '@deepgram/sdk';
import { SpanWrapper, ActionModel } from 'netra-sdk';
import * as fs from 'fs';

const deepgram = createClient(process.env.DEEPGRAM_API_KEY);

async function transcribeWithTracking(audioPath: string): Promise<string> {
  const span = new SpanWrapper('deepgram-transcription');
  span.start();

  try {
    const startTime = Date.now();
    const audioBuffer = fs.readFileSync(audioPath);
    const audioSizeBytes = audioBuffer.length;
    
    span.setAttribute('audio_file', audioPath);
    span.setAttribute('audio_size_bytes', audioSizeBytes);
    span.setAttribute('model', 'nova-2');

    const { result } = await deepgram.listen.prerecorded.transcribeFile(
      audioBuffer,
      {
        model: 'nova-2',
        smart_format: true,
        punctuate: true,
        diarize: true
      }
    );

    const transcript = result.results.channels[0].alternatives[0].transcript;
    const duration = Date.now() - startTime;
    const audioDuration = result.metadata.duration;
    const confidence = result.results.channels[0].alternatives[0].confidence;

    // Track the STT API operation
    const action: ActionModel = {
      start_time: (startTime * 1000000).toString(),
      action: 'API',
      action_type: 'STT_TRANSCRIPTION',
      metadata: {
        provider: 'deepgram',
        model: 'nova-2',
        audio_size_bytes: audioSizeBytes.toString(),
        audio_duration_seconds: audioDuration.toString(),
        transcript_length: transcript.length.toString(),
        confidence: confidence.toString(),
        duration_ms: duration.toString(),
        words_detected: result.results.channels[0].alternatives[0].words.length.toString()
      },
      success: true
    };
    span.setAction([action]);

    span.setUsage({
      model: 'nova-2',
      usage_type: 'audio_seconds',
      units_used: audioDuration,
      cost_in_usd: audioDuration * 0.0043
    });

    span.setAttribute('transcript_length', transcript.length);
    span.setAttribute('confidence', confidence);
    span.setStatus({ code: 1, message: 'Success' });
    span.end();

    return transcript;
  } catch (error) {
    span.setError(error as Error);
    span.setStatus({ code: 2, message: 'Error' });
    span.end();
    throw error;
  }
}

// Usage
const transcript = await transcribeWithTracking('./audio/sample.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")