File size: 1,926 Bytes
cdc1bf3 |
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 |
import os
import io
import torch
import torchaudio
from typing import Any, Dict
from transformers import AutoConfig, AutoProcessor
from modeling_upstream_finetune import UpstreamFinetune
class EndpointHandler():
def __init__(self, model_dir: str, **kwargs: Any) -> None:
# Load config and model with trust_remote_code
device = 'cuda'
self.emotions = ['neutral','happy','sad','angry','surprise','contempt']
self.model = UpstreamFinetune.from_pretrained(
model_dir,
device=device,
)
self.model.eval()
def __call__(self, data: Dict[str, Any]) -> Dict[str, Any]:
# Expect raw audio bytes or a base64 string in `data["inputs"]`
audio = data["inputs"]
sampling_rate = data.get("sampling_rate", 16000)
# Decode MP3/WAV bytes → waveform tensor
waveform, sr = torchaudio.load(io.BytesIO(audio))
if sr != sampling_rate:
waveform = torchaudio.functional.resample(waveform, sr, sampling_rate)
# Forward pass
with torch.no_grad():
cat_logits, reg_outputs = self.model(
waveform,
sampling_rate
)
# Convert logits to probabilities using softmax
emotion_probs = torch.nn.functional.softmax(cat_logits, dim=1)
# Create emotion predictions
emotion_predictions = []
for i, emotion in enumerate(self.emotions):
emotion_predictions.append({
"label": emotion,
"score": float(emotion_probs[0, i]) # Convert tensor to float
})
# Add arousal and valence predictions
result = emotion_predictions + [
{"label": "arousal", "score": float(reg_outputs[0, 0])},
{"label": "valence", "score": float(reg_outputs[0, 1])}
]
return result |