cleanup

app.py

@@ -1,13 +1,6 @@
 import gradio as gr
-from transformers import (
-    pipeline,
-    BlipProcessor, BlipForConditionalGeneration, BlipForImageTextRetrieval,
-    CLIPProcessor, CLIPModel, ViltProcessor, ViltForQuestionAnswering,
-    AutoProcessor
-)
+from transformers import pipeline
 import torch
-from torch.nn.functional import cosine_similarity
-import torch.nn.functional as F
 from PIL import Image, ImageDraw, ImageFont
 import numpy as np
 import functools
@@ -305,149 +298,29 @@ def audio_classifier(audio, model_name):
 def audio_zero_shot_classifier(audio, candidate_labels, model_name):
     """Zero-shot классификация аудио"""
     try:
-        import soundfile as sf
-        import numpy as np
-        
         labels = [label.strip() for label in candidate_labels.split(",")]
+        classifier = get_pipeline("zero-shot-audio-classification", model_name)
+        result = classifier(audio, candidate_labels=labels)
         
-        # Загружаем аудио файл, если передан путь
-        if isinstance(audio, str):
-            # audio - это путь к файлу
-            audio_data, sample_rate = sf.read(audio)
-        elif isinstance(audio, tuple):
-            # audio - это кортеж (sample_rate, audio_data) от Gradio
-            sample_rate, audio_data = audio
-        else:
-            # audio уже является массивом numpy
-            audio_data = audio
-            sample_rate = None
-        
-        # Преобразуем в numpy array, если нужно
-        if not isinstance(audio_data, np.ndarray):
-            audio_data = np.array(audio_data)
-        
-        # Если аудио моно, убеждаемся что это 1D массив
-        if len(audio_data.shape) > 1:
-            audio_data = audio_data[:, 0] if audio_data.shape[1] > 0 else audio_data.flatten()
+        # Обрабатываем результат
+        if isinstance(result, list):
+            result = result[0] if result else {}
         
-        # Проверяем тип модели
-        if "clap" in model_name.lower():
-            # Используем CLAP для zero-shot классификации аудио
-            from transformers import ClapProcessor, ClapModel
-            
-            cache_key = f"audio_zero_shot_{model_name}"
-            cached = model_cache.get(cache_key)
-            if cached is None:
-                processor = ClapProcessor.from_pretrained(model_name)
-                model = ClapModel.from_pretrained(model_name)
-                cached = (processor, model)
-                model_cache.put(cache_key, cached)
-            
-            processor, model = cached
-            inputs = processor(text=labels, audios=audio_data, return_tensors="pt", padding=True)
-            with torch.no_grad():
-                outputs = model(**inputs)
-                logits_per_audio = outputs.logits_per_audio
-                probs = logits_per_audio.softmax(dim=1)
-            
-            output = "Результаты классификации:\n"
-            for label, prob in zip(labels, probs[0]):
-                output += f"{label}: {prob.item():.4f}\n"
-            return output
-        elif "wav2vec2" in model_name.lower() or "hubert" in model_name.lower():
-            # Используем подход с audio embeddings + text embeddings
-            # Получаем аудио эмбеддинги через audio model и текстовые эмбеддинги через text model
-            from transformers import AutoProcessor, AutoModel
-            from sentence_transformers import SentenceTransformer
-            import librosa
-            
-            cache_key = f"audio_zero_shot_{model_name}"
-            cached = model_cache.get(cache_key)
-            if cached is None:
-                # Загружаем модель для аудио эмбеддингов
-                audio_processor = AutoProcessor.from_pretrained(model_name)
-                audio_model = AutoModel.from_pretrained(model_name)
-                # Загружаем модель для текстовых эмбеддингов с размерностью 768
-                # Используем модель с размерностью 768 для совместимости с Wav2Vec2
-                text_model = SentenceTransformer('all-mpnet-base-v2')
-                cached = (audio_processor, audio_model, text_model)
-                model_cache.put(cache_key, cached)
-            
-            audio_processor, audio_model, text_model = cached
-            
-            # Получаем аудио эмбеддинги
-            # Ресемплируем до 16000 Гц если нужно (Wav2Vec2 требует 16000 Гц)
-            target_sample_rate = 16000
-            if sample_rate is None:
-                sample_rate = target_sample_rate
-            elif sample_rate != target_sample_rate:
-                # Ресемплируем аудио до нужной частоты дискретизации
-                audio_data = librosa.resample(audio_data, orig_sr=sample_rate, target_sr=target_sample_rate)
-                sample_rate = target_sample_rate
-            
-            inputs = audio_processor(audio_data, sampling_rate=sample_rate, return_tensors="pt")
-            with torch.no_grad():
-                audio_outputs = audio_model(**inputs)
-                # Используем последний скрытый слой как эмбеддинг
-                if hasattr(audio_outputs, 'last_hidden_state'):
-                    audio_embedding = audio_outputs.last_hidden_state.mean(dim=1)  # Усредняем по временной оси
-                else:
-                    audio_embedding = audio_outputs[0].mean(dim=1)
-                audio_embedding = audio_embedding / audio_embedding.norm(dim=1, keepdim=True)
-            
-            # Получаем текстовые эмбеддинги
-            text_embeddings = text_model.encode(labels, convert_to_tensor=True)
-            text_embeddings = text_embeddings / text_embeddings.norm(dim=1, keepdim=True)
-            
-            # Проверяем размерности и проецируем если нужно
-            audio_dim = audio_embedding.shape[1]
-            text_dim = text_embeddings.shape[1]
-            
-            if audio_dim != text_dim:
-                # Если размерности не совпадают, проецируем меньший эмбеддинг в большее пространство
-                if audio_dim > text_dim:
-                    # Проецируем текстовые эмбеддинги в пространство аудио
-                    projection = torch.nn.Linear(text_dim, audio_dim).to(text_embeddings.device)
-                    text_embeddings = projection(text_embeddings)
-                    text_embeddings = text_embeddings / text_embeddings.norm(dim=1, keepdim=True)
-                else:
-                    # Проецируем аудио эмбеддинги в пространство текста
-                    projection = torch.nn.Linear(audio_dim, text_dim).to(audio_embedding.device)
-                    audio_embedding = projection(audio_embedding)
-                    audio_embedding = audio_embedding / audio_embedding.norm(dim=1, keepdim=True)
-            
-            # Вычисляем косинусное сходство
-            similarities = cosine_similarity(audio_embedding, text_embeddings).squeeze(0)
-            # Применяем softmax для получения вероятностей
-            probs = torch.softmax(similarities * 10, dim=0)  # Масштабируем для лучшей точности
-            
-            output = "Результаты классификации (через audio + text embeddings):\n"
-            for label, prob in zip(labels, probs):
-                output += f"{label}: {prob.item():.4f}\n"
-            return output
+        output = "Результаты классификации:\n"
+        if isinstance(result, dict) and 'scores' in result:
+            for label, score in zip(result.get('labels', labels), result['scores']):
+                output += f"{label}: {score:.4f}\n"
+        elif isinstance(result, list):
+            for item in result:
+                if isinstance(item, dict):
+                    label = item.get('label', '')
+                    score = item.get('score', 0.0)
+                    output += f"{label}: {score:.4f}\n"
         else:
-            # Для других моделей используем CLAP по умолчанию
-            from transformers import ClapProcessor, ClapModel
-            
-            cache_key = f"audio_zero_shot_{model_name}"
-            cached = model_cache.get(cache_key)
-            if cached is None:
-                processor = ClapProcessor.from_pretrained(model_name)
-                model = ClapModel.from_pretrained(model_name)
-                cached = (processor, model)
-                model_cache.put(cache_key, cached)
-            
-            processor, model = cached
-            inputs = processor(text=labels, audios=audio_data, return_tensors="pt", padding=True)
-            with torch.no_grad():
-                outputs = model(**inputs)
-                logits_per_audio = outputs.logits_per_audio
-                probs = logits_per_audio.softmax(dim=1)
-            
-            output = "Результаты классификации:\n"
-            for label, prob in zip(labels, probs[0]):
-                output += f"{label}: {prob.item():.4f}\n"
-            return output
+            # Если формат неожиданный, пытаемся извлечь информацию
+            output += str(result)
+        
+        return output
     except Exception as e:
         return f"Ошибка: {str(e)}"
 
@@ -470,153 +343,70 @@ def speech_synthesis(text, model_name):
         import numpy as np
         import torch
         
-        if "speecht5" in model_name.lower():
-            from transformers import SpeechT5Processor, SpeechT5ForTextToSpeech, SpeechT5HifiGan
+        # Проверяем, что текст не пустой
+        if not text or not text.strip():
+            raise ValueError("Текст для синтеза не может быть пустым")
+        
+        # Используем стандартный pipeline
+        tts = get_pipeline("text-to-speech", model_name)
+        result = tts(text)
+        
+        # Pipeline может возвращать словарь или кортеж
+        if isinstance(result, dict):
+            # Стандартный формат: {"audio": array, "sampling_rate": int}
+            audio_data = result.get("audio", result.get("raw", None))
+            sample_rate = result.get("sampling_rate", result.get("sample_rate", 22050))
             
-            cache_key = f"tts_{model_name}"
-            cached = model_cache.get(cache_key)
-            if cached is None:
-                processor = SpeechT5Processor.from_pretrained(model_name)
-                model = SpeechT5ForTextToSpeech.from_pretrained(model_name)
-                vocoder = SpeechT5HifiGan.from_pretrained("microsoft/speecht5_hifigan")
-                
-                # Генерируем speaker embeddings используя модель напрямую
-                # Используем размерность speaker embeddings из конфигурации модели
-                speaker_embedding_dim = model.config.speaker_embedding_dim
-                # Создаем случайный speaker embedding (можно заменить на предобученный)
-                # Для более стабильного результата используем нормализованный случайный вектор
-                speaker_embeddings = torch.randn(1, speaker_embedding_dim)
-                speaker_embeddings = speaker_embeddings / torch.norm(speaker_embeddings, dim=1, keepdim=True)
-                
-                cached = (processor, model, vocoder, speaker_embeddings)
-                model_cache.put(cache_key, cached)
+            if audio_data is None:
+                raise ValueError("Не удалось извлечь аудио данные из результата pipeline")
             
-            processor, model, vocoder, speaker_embeddings = cached
-            inputs = processor(text=text, return_tensors="pt")
-            with torch.no_grad():
-                speech = model.generate_speech(inputs["input_ids"], speaker_embeddings, vocoder=vocoder)
+            # Конвертируем в numpy array если нужно
+            if isinstance(audio_data, torch.Tensor):
+                audio_data = audio_data.numpy()
+            elif not isinstance(audio_data, np.ndarray):
+                audio_data = np.array(audio_data)
             
-            # Конвертируем в numpy и нормализуем
-            audio_data = speech.numpy()
             # Убеждаемся, что это 1D массив
             if len(audio_data.shape) > 1:
                 audio_data = audio_data.flatten()
-            # Нормализуем в диапазон [-1, 1] если нужно
+            
+            # Нормализуем в float32
             if audio_data.dtype != np.float32:
                 audio_data = audio_data.astype(np.float32)
+            
             # Нормализуем если значения выходят за пределы [-1, 1]
             max_val = np.abs(audio_data).max()
             if max_val > 1.0:
                 audio_data = audio_data / max_val
             
-            sample_rate = 16000
             return (sample_rate, audio_data)
-        else:
-            # Проверяем, является ли модель неподдерживаемой
-            unsupported_models = {
-                "facebook/fastspeech2-en-ljspeech": "Эта модель требует библиотеку Fairseq, а не transformers. Используйте модель microsoft/speecht5_tts вместо неё.",
-                "espnet/kan-bayashi_ljspeech_vits": "Эта модель требует библиотеку ESPnet, а не transformers. Используйте модель microsoft/speecht5_tts вместо неё."
-            }
+        elif isinstance(result, tuple) and len(result) == 2:
+            # Уже в правильном формате (sample_rate, audio_data)
+            sample_rate, audio_data = result
             
-            if model_name in unsupported_models:
-                raise ValueError(unsupported_models[model_name])
+            # Конвертируем в numpy если нужно
+            if isinstance(audio_data, torch.Tensor):
+                audio_data = audio_data.numpy()
+            elif not isinstance(audio_data, np.ndarray):
+                audio_data = np.array(audio_data)
             
-            # Для других моделей используем стандартный pipeline
-            try:
-                # Проверяем, что текст не пустой
-                if not text or not text.strip():
-                    raise ValueError("Текст для синтеза не может быть пустым")
-                
-                # Для MMS TTS моделей проверяем язык
-                if "mms-tts" in model_name.lower():
-                    # MMS TTS модели обычно поддерживают только один язык
-                    # eng - английский, rus - русский и т.д.
-                    if "mms-tts-eng" in model_name.lower():
-                        # Проверяем, что текст на английском (простая проверка)
-                        # Если текст содержит кириллицу, это может быть проблемой
-                        has_cyrillic = any('\u0400' <= char <= '\u04FF' for char in text)
-                        if has_cyrillic:
-                            raise ValueError(
-                                f"Модель '{model_name}' поддерживает только английский язык. "
-                                f"Для русского текста используйте модель 'facebook/mms-tts-rus' или 'microsoft/speecht5_tts'."
-                            )
-                
-                tts = get_pipeline("text-to-speech", model_name)
-                result = tts(text)
-            except Exception as e:
-                error_msg = str(e)
-                if "does not appear to have a file named" in error_msg or "Unrecognized model" in error_msg:
-                    raise ValueError(
-                        f"Модель '{model_name}' не поддерживается библиотекой transformers. "
-                        f"Эта модель может требовать другую библиотеку (например, Fairseq или ESPnet). "
-                        f"Попробуйте использовать модель microsoft/speecht5_tts, которая полностью поддерживается."
-                    ) from e
-                elif "negative output size" in error_msg.lower() or "input size 0" in error_msg.lower():
-                    raise ValueError(
-                        f"Ошибка обработки текста моделью '{model_name}'. "
-                        f"Возможные причины: неподдерживаемый язык, пустой текст после обработки, или проблема с токенизацией. "
-                        f"Попробуйте использовать другую модель или проверьте язык текста."
-                    ) from e
-                raise
+            # Убеждаемся, что это 1D массив
+            if len(audio_data.shape) > 1:
+                audio_data = audio_data.flatten()
             
-            # Pipeline может возвращать словарь или кортеж
-            if isinstance(result, dict):
-                # Стандартный формат: {"audio": array, "sampling_rate": int}
-                audio_data = result.get("audio", result.get("raw", None))
-                sample_rate = result.get("sampling_rate", result.get("sample_rate", 22050))
-                
-                if audio_data is None:
-                    raise ValueError("Не удалось извлечь аудио данные из результата pipeline")
-                
-                # Конвертируем в numpy array если нужно
-                if isinstance(audio_data, torch.Tensor):
-                    audio_data = audio_data.numpy()
-                elif not isinstance(audio_data, np.ndarray):
-                    audio_data = np.array(audio_data)
-                
-                # Убеждаемся, что это 1D массив
-                if len(audio_data.shape) > 1:
-                    audio_data = audio_data.flatten()
-                
-                # Нормализуем в float32
-                if audio_data.dtype != np.float32:
-                    audio_data = audio_data.astype(np.float32)
-                
-                # Нормализуем если значения выходят за пределы [-1, 1]
-                max_val = np.abs(audio_data).max()
-                if max_val > 1.0:
-                    audio_data = audio_data / max_val
-                
-                return (sample_rate, audio_data)
-            elif isinstance(result, tuple) and len(result) == 2:
-                # Уже в правильном формате (sample_rate, audio_data)
-                sample_rate, audio_data = result
-                
-                # Конвертируем в numpy если нужно
-                if isinstance(audio_data, torch.Tensor):
-                    audio_data = audio_data.numpy()
-                elif not isinstance(audio_data, np.ndarray):
-                    audio_data = np.array(audio_data)
-                
-                # Убеждаемся, что это 1D массив
-                if len(audio_data.shape) > 1:
-                    audio_data = audio_data.flatten()
-                
-                # Нормализуем в float32
-                if audio_data.dtype != np.float32:
-                    audio_data = audio_data.astype(np.float32)
-                
-                # Нормализуем если значения выходят за пределы [-1, 1]
-                max_val = np.abs(audio_data).max()
-                if max_val > 1.0:
-                    audio_data = audio_data / max_val
-                
-                return (sample_rate, audio_data)
-            else:
-                raise ValueError(f"Неожиданный формат результата от pipeline: {type(result)}")
+            # Нормализуем в float32
+            if audio_data.dtype != np.float32:
+                audio_data = audio_data.astype(np.float32)
+            
+            # Нормализуем если значения выходят за пределы [-1, 1]
+            max_val = np.abs(audio_data).max()
+            if max_val > 1.0:
+                audio_data = audio_data / max_val
+            
+            return (sample_rate, audio_data)
+        else:
+            raise ValueError(f"Неожиданный формат результата от pipeline: {type(result)}")
     except Exception as e:
-        # В случае ошибки возвращаем None, чтобы Gradio мог обработать это
-        # Но также логируем ошибку в историю через декоратор
         raise Exception(f"Ошибка синтеза речи: {str(e)}")
 
 # ==================== ЗАДАЧИ С ИЗОБРАЖЕНИЯМИ ====================
@@ -898,26 +688,11 @@ def image_segmentation(image, model_name):
 def image_captioning(image, model_name):
     """Описание изображения"""
     try:
-        if "blip" in model_name.lower():
-            cache_key = f"caption_blip_{model_name}"
-            cached = model_cache.get(cache_key)
-            if cached is None:
-                processor = BlipProcessor.from_pretrained(model_name)
-                model = BlipForConditionalGeneration.from_pretrained(model_name)
-                cached = (processor, model)
-                model_cache.put(cache_key, cached)
-            
-            processor, model = cached
-            inputs = processor(image, return_tensors="pt")
-            out = model.generate(**inputs, max_length=50)
-            caption = processor.decode(out[0], skip_special_tokens=True)
-            return caption
-        else:
-            captioner = get_pipeline("image-to-text", model_name)
-            result = captioner(image)
-            if isinstance(result, list):
-                result = result[0]
-            return result['generated_text']
+        captioner = get_pipeline("image-to-text", model_name)
+        result = captioner(image)
+        if isinstance(result, list):
+            result = result[0]
+        return result['generated_text']
     except Exception as e:
         return f"Ошибка: {str(e)}"
 
@@ -925,28 +700,11 @@ def image_captioning(image, model_name):
 def visual_qa(image, question, model_name):
     """Визуальный вопрос-ответ"""
     try:
-        if "vilt" in model_name.lower():
-            cache_key = f"vqa_vilt_{model_name}"
-            cached = model_cache.get(cache_key)
-            if cached is None:
-                processor = ViltProcessor.from_pretrained(model_name)
-                model = ViltForQuestionAnswering.from_pretrained(model_name)
-                cached = (processor, model)
-                model_cache.put(cache_key, cached)
-            
-            processor, model = cached
-            inputs = processor(image, question, return_tensors="pt")
-            outputs = model(**inputs)
-            logits = outputs.logits
-            idx = logits.argmax(-1).item()
-            answer = model.config.id2label[idx]
-            return f"Ответ: {answer}"
-        else:
-            vqa = get_pipeline("visual-question-answering", model_name)
-            result = vqa(image=image, question=question)
-            if isinstance(result, list):
-                result = result[0]
-            return f"Ответ: {result['answer']}"
+        vqa = get_pipeline("visual-question-answering", model_name)
+        result = vqa(image=image, question=question)
+        if isinstance(result, list):
+            result = result[0]
+        return f"Ответ: {result['answer']}"
     except Exception as e:
         return f"Ошибка: {str(e)}"
 
@@ -954,89 +712,25 @@ def visual_qa(image, question, model_name):
 def image_zero_shot_classification(image, candidate_labels, model_name):
     """Zero-shot классификация изображений"""
     try:
-        cache_key = f"clip_zs_{model_name}"
-        cached = model_cache.get(cache_key)
-        if cached is None:
-            # Проверяем, является ли модель из sentence-transformers
-            if "sentence-transformers" in model_name:
-                from sentence_transformers import SentenceTransformer
-                model = SentenceTransformer(model_name)
-                cached = ("sentence_transformers", model)
-            # Проверяем, является ли модель LAION (требует OpenCLIP)
-            elif "laion/" in model_name.lower() or "laion5b" in model_name.lower():
-                import open_clip
-                # Определяем имя модели и веса для OpenCLIP
-                if "xlm-roberta-base-ViT-B-32" in model_name or "xlm-roberta-base" in model_name:
-                    clip_model_name = "xlm-roberta-base-ViT-B-32"
-                    pretrained = "laion5b_s13b_b90k"
-                else:
-                    # Пытаемся извлечь информацию из имени модели
-                    clip_model_name = "xlm-roberta-base-ViT-B-32"
-                    pretrained = "laion5b_s13b_b90k"
-                
-                model, _, preprocess = open_clip.create_model_and_transforms(
-                    clip_model_name, 
-                    pretrained=pretrained
-                )
-                tokenizer = open_clip.get_tokenizer(clip_model_name)
-                model.eval()
-                cached = ("openclip", model, preprocess, tokenizer)
-            else:
-                processor = CLIPProcessor.from_pretrained(model_name)
-                model = CLIPModel.from_pretrained(model_name)
-                cached = ("transformers", processor, model)
-            model_cache.put(cache_key, cached)
-        
         labels = [label.strip() for label in candidate_labels.split(",")]
+        classifier = get_pipeline("zero-shot-image-classification", model_name)
+        result = classifier(image, candidate_labels=labels)
         
-        if cached[0] == "sentence_transformers":
-            # Используем sentence-transformers
-            model = cached[1]
-            # Вычисляем эмбеддинги изображения и текстов
-            image_embedding = model.encode(image, convert_to_tensor=True)
-            text_embeddings = model.encode(labels, convert_to_tensor=True)
-            # Вычисляем косинусное сходство
-            similarities = cosine_similarity(image_embedding.unsqueeze(0), text_embeddings).squeeze(0)
-            # Нормализуем в диапазон [0, 1] и применяем softmax для вероятностей
-            similarities = (similarities + 1) / 2
-            probs = torch.softmax(similarities, dim=0)
-            
-            output = "Результаты классификации:\n"
-            for label, prob in zip(labels, probs):
-                output += f"{label}: {prob.item():.4f}\n"
-        elif cached[0] == "openclip":
-            # Используем OpenCLIP
-            model, preprocess, tokenizer = cached[1], cached[2], cached[3]
-            # Обрабатываем изображение и тексты
-            image_tensor = preprocess(image).unsqueeze(0)
-            text_tokens = tokenizer(labels)
-            
-            with torch.no_grad():
-                image_features = model.encode_image(image_tensor)
-                text_features = model.encode_text(text_tokens)
-                # Нормализуем признаки
-                image_features = image_features / image_features.norm(dim=-1, keepdim=True)
-                text_features = text_features / text_features.norm(dim=-1, keepdim=True)
-                # Вычисляем косинусное сходство (логиты)
-                logits_per_image = (image_features @ text_features.T) * 100  # Масштабируем для лучшей точности
-                probs = logits_per_image.softmax(dim=1)
-            
-            output = "Результаты классификации:\n"
-            for label, prob in zip(labels, probs[0]):
-                output += f"{label}: {prob.item():.4f}\n"
+        # Обрабатываем результат
+        if isinstance(result, list):
+            # Сортируем по score если нужно
+            result = sorted(result, key=lambda x: x.get('score', 0), reverse=True)
         else:
-            # Используем стандартный CLIP из transformers
-            processor, model = cached[1], cached[2]
-            inputs = processor(text=labels, images=image, return_tensors="pt", padding=True)
-            
-            with torch.no_grad():
-                outputs = model(**inputs)
-                logits_per_image = outputs.logits_per_image
-                probs = logits_per_image.softmax(dim=1)
-            
-            output = "Результаты классификации:\n"
-            for label, prob in zip(labels, probs[0]):
-                output += f"{label}: {prob.item():.4f}\n"
+            result = [result] if result else []
+        
+        output = "Результаты классификации:\n"
+        for item in result:
+            if isinstance(item, dict):
+                label = item.get('label', '')
+                score = item.get('score', 0.0)
+                output += f"{label}: {score:.4f}\n"
+            else:
+                output += f"{item}\n"
         
         return output
     except Exception as e: