pipeline_antispoofing.py

from transformers import Pipeline
import torch
from .feature_extraction_antispoofing import AntispoofingFeatureExtractor
class AntispoofingPipeline(Pipeline):
    def __init__(self, model, **kwargs):
        super().__init__(model=model, **kwargs)
        self.feature_extractor = AntispoofingFeatureExtractor()

    def _sanitize_parameters(self, **kwargs):
        preprocess_kwargs = {}
        postprocess_kwargs = {}
        
        if "sampling_rate" in kwargs:
            preprocess_kwargs["sampling_rate"] = kwargs["sampling_rate"]
        
        return preprocess_kwargs, {}, postprocess_kwargs
    
    def preprocess(self, audio, sampling_rate=16000):
        audio = self.feature_extractor(audio)['input_values']
        inputs = {"input_values": audio}
        
        return inputs
    
    def _forward(self, model_inputs):
        outputs = self.model(**model_inputs)
        return outputs
    
    def postprocess(self, model_outputs):
        logits = model_outputs['logits']
        probs = torch.nn.functional.softmax(logits, dim=-1)
        predicted_class = torch.argmax(probs, dim=-1).item()
        confidence = probs[0][predicted_class].item()
        
        return {
            "label": self.model.config.id2label[predicted_class],
            "logits": logits.tolist(),
            "score": confidence,
            "all_scores": {
                self.model.config.id2label[i]: probs[0][i].item()
                for i in range(len(probs[0]))
            }
        }