valenceai 1.0.0

Python SDK for Valence AI Emotion Detection API - Real-time, Async, and Streaming Support

Valence SDK for Emotion Detection

valenceai is a Python client library for interacting with the Valence AI Pulse API for emotion detection. It provides a convenient interface to upload audio files, stream real-time audio, and retrieve detected emotional states.

Features

• Discrete audio processing - Real-time analysis for short audio clips (4-10s)
• Async audio processing - Multipart parallel upload for long files with timeline data
• Streaming API - Real-time WebSocket streaming for live audio
• Rate limiting - Monitor API usage and limits
• Model selection - Choose between 4emotions and 7emotions models
• Timeline analysis - Get emotion changes over time with timestamps
• Environment configuration - Built-in support for environment variables
• Enhanced logging - Configurable log levels
• 100% tested - Comprehensive test suite

The emotional classification model used in our APIs is optimized for North American English conversational data.

Emotion Models

The SDK supports two emotion detection models:

• 4emotions (default): angry, happy, neutral, sad
• 7emotions: happy, sad, angry, neutral, surprised, disgusted, calm

The number of emotions, emotional buckets, and language support can be customized. If you are interested in a custom model, please contact us.

API Overview

API	Best For	Input	Output	Response Time
Discrete	Real-time analysis	Short audio (4-10s)	Single emotion prediction	100-500ms
Async	Pre-recorded files	Long audio (up to 1GB)	Timeline with emotion changes	Depends on file size
Streaming	Live audio streams	Audio chunks via WebSocket	Real-time emotion updates	Near real-time

Async API Processing Workflow

The Async API uses a multi-step process to handle long audio files. Understanding this workflow is crucial for proper implementation:

1. Upload Phase (Client-Side)

When you call client.asynch.upload(file_path):

• SDK splits your file into parts (5MB chunks by default)
• Uploads parts to S3 in parallel using presigned URLs
• Returns a request_id - This is a tracking identifier, NOT a completion signal
• At this point: File is uploaded to S3, but NOT processed yet

2. Background Processing (Server-Side)

After upload completes, the server automatically:

• Background processor checks for new uploads every 10 seconds
• Downloads audio from S3 when detected
• Splits audio into 5-second segments
• Extracts audio features (MFCC) from each segment
• Invokes machine learning model for emotion detection
• Stores results in database
• Updates status to completed

Processing Time: Typically 1-2 minutes for a 60-minute audio file. The exact time depends on file length and current server load.

3. Results Retrieval (Client-Side)

When you call client.asynch.emotions(request_id):

• Polls the status endpoint at regular intervals
• Waits for status progression:
  • initiated → Upload started
  • upload_completed → File uploaded to S3 (processing not started)
  • processing → Background processing in progress
  • completed → Results ready
• Returns emotion timeline when status is completed

Status Values

Status	Meaning	What's Happening
initiated	Upload started	SDK is uploading file parts to S3
upload_completed	Upload finished	File is in S3, waiting for background processor
processing	Processing active	Server is analyzing audio with ML model
completed	Results ready	Emotion timeline is available

Important Notes

• The request_id is NOT a completion indicator - It's just a tracking ID
• upload() completing does NOT mean results are ready - It only means the file is in S3
• Background processing takes time - Plan for 1-2 minutes per hour of audio
• You can check status anytime - The request_id remains valid for retrieving results

Installation

pip install valenceai

Quick Start

from valenceai import ValenceClient

# Initialize client (uses VALENCE_API_KEY environment variable)
client = ValenceClient(api_key="your_api_key")

# Discrete API - Quick emotion detection
result = client.discrete.emotions(file_path="short_audio.wav", model="4emotions")
print(f"Emotion: {result['dominant_emotion']}")

# Async API - Long audio with timeline
# Step 1: Upload file to S3 (returns tracking ID, NOT results)
request_id = client.asynch.upload("long_audio.wav")
# Step 2: Wait for server processing and get results (polls until complete)
result = client.asynch.emotions(request_id, max_attempts=30, interval_seconds=10)
# Step 3: Access timeline and dominant emotion from results
timeline = client.asynch.get_timeline(request_id)
dominant = client.asynch.get_dominant_emotion(request_id)

# Streaming API - Real-time audio
stream = client.streaming.connect(model="4emotions")
stream.on_prediction(lambda data: print(data['main_emotion']))
stream.connect()
stream.send_audio(audio_bytes)
stream.disconnect()

# Rate Limit API - Monitor usage
status = client.rate_limit.get_status()
health = client.rate_limit.get_health()

Configuration

Environment Variables

export VALENCE_API_KEY="your_api_key"                    # Required
export VALENCE_API_BASE_URL="https://api.getvalenceai.com"  # Optional
export VALENCE_WEBSOCKET_URL="wss://api.getvalenceai.com"   # Optional
export VALENCE_LOG_LEVEL="INFO"                         # Optional: DEBUG, INFO, ERROR

Client Configuration

client = ValenceClient(
    api_key="your_api_key",           # API key (required)
    base_url="https://custom.api",    # Custom API endpoint (optional)
    websocket_url="wss://custom.api", # Custom WebSocket endpoint (optional)
    part_size=5*1024*1024,            # Upload chunk size (default: 5MB)
    show_progress=True,               # Show upload progress (default: True)
    max_threads=3                     # Concurrent upload threads (default: 3)
)

Usage Examples

Example 1: Analyze a Short Audio Clip

Use the Discrete API for quick emotion detection on short audio files (4-10 seconds).

from valenceai import ValenceClient

# Initialize the client
client = ValenceClient(api_key="your_api_key")

# Analyze a short customer service call snippet
result = client.discrete.emotions(
    file_path="customer_greeting.wav",
    model="4emotions"
)

# Access the results
print(f"Dominant emotion: {result['dominant_emotion']}")
print(f"Confidence scores:")
for emotion, score in result['emotions'].items():
    print(f"  {emotion}: {score:.2%}")

# Example output:
# Dominant emotion: happy
# Confidence scores:
#   happy: 78.00%
#   sad: 12.00%
#   angry: 5.00%
#   neutral: 5.00%

Example 2: Analyze a Long Recording with Timeline

Use the Async API to process longer audio files and get emotion changes over time.

from valenceai import ValenceClient

# Initialize the client with progress bar
client = ValenceClient(api_key="your_api_key", show_progress=True)

# STEP 1: Upload Phase - File is uploaded to S3 in parts
print("STEP 1: Uploading file to S3...")
request_id = client.asynch.upload("team_meeting_1hour.wav")
print(f"✓ Upload to S3 complete! Request ID: {request_id}")
print("Note: File is uploaded but NOT processed yet. Server will process in background.\n")

# STEP 2: Background Processing - Server automatically processes the audio
# The background processor will:
# - Detect the uploaded file (checks every 10 seconds)
# - Download from S3
# - Split into 5-second chunks
# - Run ML model on each chunk
# - Store results in database
print("STEP 2: Waiting for server to process audio...")
print("Status will progress: initiated → upload_completed → processing → completed")
print("This typically takes 1-2 minutes per hour of audio.\n")

# STEP 3: Poll for Results - Wait until processing is complete
result = client.asynch.emotions(
    request_id,
    max_attempts=50,        # Allow more attempts for long files
    interval_seconds=10     # Check every 10 seconds
)
print(f"✓ Processing complete! Status: {result['status']}\n")

# Access the emotion timeline
timeline = client.asynch.get_timeline(request_id)
print(f"Emotion timeline ({len(timeline)} data points):")
for point in timeline[:5]:  # Show first 5 points
    print(f"  {point['timestamp']}s: {point['emotion']} ({point['confidence']:.2%})")

# Find the dominant emotion across the entire meeting
dominant = client.asynch.get_dominant_emotion(request_id)
print(f"\nOverall meeting sentiment: {dominant}")

# Check emotion at specific moments
emotion_at_30min = client.asynch.get_emotion_at_time(request_id, timestamp=1800.0)
print(f"Emotion at 30 minutes: {emotion_at_30min['emotion']}")

Example 3: Real-Time Streaming Analysis

Use the Streaming API for live audio analysis via WebSocket.

from valenceai import ValenceClient
import pyaudio
import time

# Initialize the client
client = ValenceClient(api_key="your_api_key")

# Create a streaming connection
stream = client.streaming.connect(model="4emotions")

# Define callback functions
def on_prediction(data):
    """Called when an emotion prediction is received"""
    emotion = data['main_emotion']
    confidence = data['confidence']
    print(f"Detected: {emotion} (confidence: {confidence:.2%})")

    # Take action based on emotion
    if emotion == "angry" and confidence > 0.7:
        print("  Alert: High anger detected!")

def on_error(error):
    """Called when an error occurs"""
    print(f"Error: {error['error']}")

def on_connected(info):
    """Called when successfully connected"""
    print(f"Connected! Session ID: {info['session_id']}")

# Register the callbacks
stream.on_prediction(on_prediction)
stream.on_error(on_error)
stream.on_connected(on_connected)

# Connect to the WebSocket
stream.connect()

# Set up audio capture (using pyaudio as example)
audio = pyaudio.PyAudio()
audio_stream = audio.open(
    format=pyaudio.paInt16,
    channels=1,              # Mono
    rate=44100,              # 44.1kHz
    input=True,
    frames_per_buffer=4096
)

print("Listening... (Press Ctrl+C to stop)")

try:
    # Capture and stream audio in real-time
    while True:
        # Read audio chunk from microphone
        audio_chunk = audio_stream.read(4096, exception_on_overflow=False)

        # Send to Valence API for analysis
        stream.send_audio(audio_chunk)

        # Small delay to prevent overwhelming the API
        time.sleep(0.1)

except KeyboardInterrupt:
    print("\nStopping...")

finally:
    # Clean up
    audio_stream.stop_stream()
    audio_stream.close()
    audio.terminate()
    stream.disconnect()
    print("Disconnected")

Example 4: Monitor API Usage

Use the Rate Limit API to track your usage and ensure you stay within limits.

from valenceai import ValenceClient

# Initialize the client
client = ValenceClient(api_key="your_api_key")

# Check current rate limit status
status = client.rate_limit.get_status()

# Display current usage
print("API Usage Status:")
print(f"  Per second: {status['current_usage']['second']}/{status['limits']['second']['limit']}")
print(f"  Per minute: {status['current_usage']['minute']}/{status['limits']['minute']['limit']}")
print(f"  Per hour: {status['current_usage']['hour']}/{status['limits']['hour']['limit']}")
print(f"  Per day: {status['current_usage']['day']}/{status['limits']['day']['limit']}")

# Check remaining capacity
remaining = status['limits']['hour']['remaining']
if remaining < 100:
    print(f"\nWarning: Only {remaining} requests remaining this hour!")

# Check API health
health = client.rate_limit.get_health()
if health['status'] == 'healthy':
    print("\nAPI Status: All systems operational")
else:
    print(f"\nAPI Status: {health['status']}")

Example 5: Batch Processing Multiple Files

Process multiple audio files efficiently using the Async API.

from valenceai import ValenceClient
import os
import json

# Initialize the client
client = ValenceClient(api_key="your_api_key", show_progress=True)

# Directory containing audio files
audio_dir = "customer_calls/"
audio_files = [f for f in os.listdir(audio_dir) if f.endswith('.wav')]

print(f"Processing {len(audio_files)} audio files...")

# PHASE 1: Upload all files to S3 (fast, parallel uploads)
print("\nPHASE 1: Uploading files to S3...")
uploads = {}
for filename in audio_files:
    filepath = os.path.join(audio_dir, filename)
    print(f"  Uploading: {filename}")
    request_id = client.asynch.upload(filepath)  # Returns request_id after S3 upload
    uploads[filename] = request_id

print(f"✓ All {len(uploads)} files uploaded to S3")
print("Note: Files are uploaded but NOT processed yet.\n")

# PHASE 2: Server background processing (automatic, happens server-side)
print("PHASE 2: Server is processing files in background...")
print("Background processor will detect uploads and analyze each file.")
print("This happens automatically - no action needed.\n")

# PHASE 3: Poll for results (wait for each file to complete)
print("PHASE 3: Retrieving results as they become available...")
results = {}
for filename, request_id in uploads.items():
    print(f"  Waiting for: {filename}")

    # Poll until this specific file is processed
    result = client.asynch.emotions(request_id, max_attempts=50, interval_seconds=10)

    # Get the dominant emotion
    dominant = client.asynch.get_dominant_emotion(request_id)

    # Store results
    results[filename] = {
        'dominant_emotion': dominant,
        'timeline': result['emotions'],
        'status': result['status']
    }
    print(f"  ✓ {filename}: {dominant}")

# Generate summary report
print("\n" + "="*50)
print("BATCH PROCESSING SUMMARY")
print("="*50)

emotion_counts = {}
for filename, data in results.items():
    emotion = data['dominant_emotion']
    emotion_counts[emotion] = emotion_counts.get(emotion, 0) + 1
    print(f"{filename}: {emotion}")

print("\nEmotion Distribution:")
for emotion, count in sorted(emotion_counts.items(), key=lambda x: x[1], reverse=True):
    percentage = (count / len(results)) * 100
    print(f"  {emotion}: {count} files ({percentage:.1f}%)")

# Save detailed results to JSON
with open('batch_results.json', 'w') as f:
    json.dump(results, f, indent=2)
print("\nDetailed results saved to batch_results.json")

Example 6: Using In-Memory Audio Data

Process audio data that's already in memory (e.g., from an API or database).

from valenceai import ValenceClient
import numpy as np

# Initialize the client
client = ValenceClient(api_key="your_api_key")

# Example: Audio data from another source (as numpy array)
# This could come from a database, another API, or generated programmatically
audio_array = np.random.randn(44100).tolist()  # 1 second of audio at 44.1kHz

# Analyze without saving to disk
result = client.discrete.emotions(
    audio_array=audio_array,
    model="4emotions"
)

print(f"Emotion detected: {result['dominant_emotion']}")
print(f"Confidence: {result['emotions'][result['dominant_emotion']]:.2%}")

Example 7: Custom Error Handling

Implement robust error handling for production applications.

from valenceai import ValenceClient
from valenceai.exceptions import ValenceSDKException, UploadError, PredictionError
import logging

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Initialize the client
client = ValenceClient(api_key="your_api_key")

def process_audio_file(file_path, max_retries=3):
    """
    Process an audio file with retry logic and comprehensive error handling.

    Args:
        file_path: Path to the audio file
        max_retries: Maximum number of retry attempts

    Returns:
        dict: Processing results or None if failed
    """
    for attempt in range(max_retries):
        try:
            logger.info(f"Processing {file_path} (attempt {attempt + 1}/{max_retries})")

            # Upload the file
            request_id = client.asynch.upload(file_path)
            logger.info(f"Upload successful. Request ID: {request_id}")

            # Get emotions
            result = client.asynch.emotions(
                request_id,
                max_attempts=30,
                interval_seconds=5
            )

            logger.info(f"Processing complete for {file_path}")
            return result

        except FileNotFoundError:
            logger.error(f"File not found: {file_path}")
            return None  # Don't retry for missing files

        except UploadError as e:
            logger.error(f"Upload failed: {e}")
            if attempt < max_retries - 1:
                logger.info("Retrying upload...")
                continue
            else:
                logger.error("Max retries reached for upload")
                return None

        except PredictionError as e:
            logger.error(f"Prediction failed: {e}")
            if attempt < max_retries - 1:
                logger.info("Retrying prediction...")
                continue
            else:
                logger.error("Max retries reached for prediction")
                return None

        except ValenceSDKException as e:
            logger.error(f"SDK error: {e}")
            return None  # Don't retry for SDK errors

        except Exception as e:
            logger.error(f"Unexpected error: {e}")
            if attempt < max_retries - 1:
                logger.info("Retrying due to unexpected error...")
                continue
            else:
                logger.error("Max retries reached")
                return None

    return None

# Use the function
result = process_audio_file("important_meeting.wav")
if result:
    print(f"Success! Dominant emotion: {client.asynch.get_dominant_emotion(result)}")
else:
    print("Processing failed after all retry attempts")

API Reference

Discrete API

For short audio files requiring immediate emotion detection.

# File upload
result = client.discrete.emotions(
    file_path="audio.wav",
    model="4emotions"  # or "7emotions"
)

# In-memory audio array
result = client.discrete.emotions(
    audio_array=[0.1, 0.2, 0.3, ...],
    model="4emotions"
)

Response:

{
    "emotions": {
        "happy": 0.78,
        "sad": 0.12,
        "angry": 0.05,
        "neutral": 0.05
    },
    "dominant_emotion": "happy"
}

Async API

For long audio files with timeline analysis.

Workflow: The Async API uses a 3-step process:

1. Upload (upload()) - Multipart upload to S3, returns request_id (tracking ID)
2. Background Processing (automatic) - Server processes audio in 5-second chunks
3. Results Retrieval (emotions()) - Polls status endpoint until processing completes

Processing Time: Typically 1-2 minutes per hour of audio.

Status Progression: initiated → upload_completed → processing → completed

Upload Audio

# Upload file to S3 (multipart upload)
request_id = client.asynch.upload(file_path="long_audio.wav")
# Returns: request_id (tracking ID, NOT completion signal)
# File is uploaded to S3 but NOT processed yet

Get Emotion Results

# Poll for results until processing completes
result = client.asynch.emotions(
    request_id="abc-123",
    max_attempts=20,        # Max polling attempts (default: 20, range: 1-100)
    interval_seconds=5      # Polling interval (default: 5, range: 1-60)
)
# This method waits for server processing to complete
# Returns when status is 'completed'

Response:

{
    "emotions": [
        {
            "timestamp": 0.5,
            "start_time": 0.0,
            "end_time": 1.0,
            "emotion": "happy",
            "confidence": 0.9,
            "all_predictions": {"happy": 0.9, "sad": 0.1, ...}
        },
        {
            "timestamp": 1.5,
            "start_time": 1.0,
            "end_time": 2.0,
            "emotion": "neutral",
            "confidence": 0.85,
            "all_predictions": {"neutral": 0.85, "happy": 0.15, ...}
        }
    ],
    "status": "completed"
}

Timeline Analysis

# Get full timeline
timeline = client.asynch.get_timeline(request_id)

# Get emotion at specific time
emotion = client.asynch.get_emotion_at_time(request_id, timestamp=5.2)

# Get dominant emotion across entire audio
dominant = client.asynch.get_dominant_emotion(request_id)

Streaming API

For real-time emotion detection on live audio streams.

# Create streaming connection
stream = client.streaming.connect(model="4emotions")

# Register callbacks
stream.on_prediction(lambda data: print(f"Emotion: {data['main_emotion']}"))
stream.on_error(lambda error: print(f"Error: {error}"))
stream.on_connected(lambda info: print(f"Connected: {info['session_id']}"))

# Connect to WebSocket
stream.connect()

# Send audio chunks (PCM bytes)
stream.send_audio(audio_chunk_bytes)

# Check connection status
if stream.connected:
    print("Streaming active")

# Wait for events (blocking)
stream.wait()

# Disconnect
stream.disconnect()

Prediction Event:

{
    "main_emotion": "happy",
    "confidence": 0.87,
    "all_predictions": {
        "happy": 0.87,
        "sad": 0.05,
        "angry": 0.03,
        "neutral": 0.05
    },
    "timestamp": 1234567890
}

Rate Limit API

Monitor your API usage and limits.

# Get rate limit status
status = client.rate_limit.get_status()
print(status)
# {
#     "limits": {
#         "second": {"limit": 10, "remaining": 8, "reset": 1234567890},
#         "minute": {"limit": 100, "remaining": 95, "reset": 1234567890},
#         "hour": {"limit": 1000, "remaining": 950, "reset": 1234567890},
#         "day": {"limit": 10000, "remaining": 9500, "reset": 1234567890}
#     },
#     "current_usage": {
#         "second": 2,
#         "minute": 5,
#         "hour": 50,
#         "day": 500
#     }
# }

# Check API health
health = client.rate_limit.get_health()
print(health)
# {"status": "healthy", "timestamp": 1234567890}

Audio Input Requirements

Format Specifications

• Format: WAV (mono)
• Recommended sampling rate: 44.1 kHz (44100 Hz)
• Minimum sampling rate: 8 kHz
• Channel: Mono (single channel)

API-Specific Requirements

• Discrete API: 4-10 seconds per file
• Async API: Minimum 5 seconds, maximum 1 GB
• Streaming API: Real-time audio chunks (PCM bytes)

For custom microphone specifications or stereo/multi-channel support, please contact us.

Additional Examples

Example scripts are available in the examples/ directory:

# Run discrete audio example
python examples/upload_short_audio.py

# Run async audio example
python examples/upload_long_audio.py

# Run streaming example
python examples/streaming_audio.py

# Run batch processing example
python examples/batch_processing.py

Development

Testing

# Install development dependencies
pip install -e ".[dev]"

# Run all tests
pytest tests/

# Run tests with coverage
pytest tests/ --cov=valenceai --cov-report=html

# Run specific test file
pytest tests/test_discrete_api.py -v

Building

# Build distribution
python -m build

# Install locally
pip install .

# Install in editable mode for development
pip install -e .

Migration from v0.x

Key Changes in v1.0.0

1. Environment Variable: VALENCE_API_KEY is now the standard (consistent naming)
2. Streaming API: New WebSocket-based real-time emotion detection
3. Rate Limiting: New API for monitoring usage
4. Timeline Data: Async API now returns detailed timestamp information
5. Model Selection: Explicit model parameter for 4emotions or 7emotions

Updating Your Code

# Old (v0.x)
client = ValenceClient()
result = client.discrete.emotions("file.wav")

# New (v1.0.0) - mostly compatible, with new features
client = ValenceClient(api_key="your_key")
result = client.discrete.emotions(
    file_path="file.wav",
    model="4emotions"  # explicit model selection
)

# New streaming capability
stream = client.streaming.connect(model="4emotions")
stream.on_prediction(callback)
stream.connect()

Support

• Documentation: API Documentation
• Issues: GitHub Issues
• Contact: Valence AI Support

License

Private License © 2025 Valence Vibrations, Inc, a Delaware public benefit corporation.