Spaces:
Running
Running
Download app.py from SucoCafe/CoffeeHealth-AI: direct link, hf CLI and curl.
- Browser
- Download file 15.4 kB
-
https://huggingface.co/spaces/SucoCafe/CoffeeHealth-AI/resolve/main/app.py
- Command line
-
hf download hf://spaces/SucoCafe/CoffeeHealth-AI/app.py
-
curl -L -o app.py https://huggingface.co/spaces/SucoCafe/CoffeeHealth-AI/resolve/main/app.py
15.4 kB
| import os | |
| import sys | |
| import torch | |
| import numpy as np | |
| import supervision as sv | |
| from inference import get_model | |
| from PIL import Image | |
| from typing import List, Dict | |
| from collections import Counter | |
| import shutil | |
| import tempfile | |
| from flask import Flask, request, jsonify, render_template | |
| from dotenv import load_dotenv | |
| # Carrega as variáveis do arquivo .env para o ambiente | |
| load_dotenv() | |
| # Adiciona o diretório 'Long-CLIP' ao path para encontrar a pasta 'model' | |
| long_clip_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "Long-CLIP") | |
| sys.path.append(long_clip_path) | |
| try: | |
| from model import longclip | |
| except ImportError: | |
| print(f"Erro: A pasta 'model' do LongCLIP não foi encontrada em: {long_clip_path}") | |
| # Suprime avisos de dependências ausentes da biblioteca inference (opcional) | |
| os.environ["CORE_MODEL_SAM_ENABLED"] = "False" | |
| os.environ["CORE_MODEL_SAM3_ENABLED"] = "False" | |
| os.environ["CORE_MODEL_GAZE_ENABLED"] = "False" | |
| os.environ["CORE_MODEL_YOLO_WORLD_ENABLED"] = "False" | |
| app = Flask(__name__) | |
| # Variável global para armazenar as análises em lote (Artigo) | |
| resultados_globais_artigo = [] | |
| # --- CONFIGURAÇÕES INICIAIS --- | |
| # A chave da API do Roboflow agora é carregada automaticamente do arquivo .env | |
| device = "cuda" if torch.cuda.is_available() else "cpu" | |
| output_dir = "static/outputs" | |
| app.config['UPLOAD_FOLDER'] = tempfile.gettempdir() | |
| # --- FUNÇÕES DO LONGCLIP --- | |
| def load_longclip_model(checkpoint_path, device="cpu"): | |
| checkpoint_path = os.path.normpath(checkpoint_path) | |
| print(f"Carregando LongCLIP de {checkpoint_path}...") | |
| if not os.path.exists(checkpoint_path): | |
| raise FileNotFoundError(f"Erro: O arquivo de checkpoint não foi encontrado em: {checkpoint_path}") | |
| model, preprocess = longclip.load(checkpoint_path, device=device) | |
| return model, preprocess | |
| def get_text_features(model, text_descriptions: Dict[str, str], device="cpu"): | |
| class_names = list(text_descriptions.keys()) | |
| descriptions = list(text_descriptions.values()) | |
| print(f"Tokenizando {len(descriptions)} descrições detalhadas...") | |
| tokens = longclip.tokenize(descriptions).to(device) | |
| with torch.no_grad(): | |
| features = model.encode_text(tokens) | |
| features /= features.norm(dim=-1, keepdim=True) | |
| return features, class_names | |
| # --- DICIONÁRIO DE DESCRIÇÕES OTIMIZADO PARA LONGCLIP --- | |
| text_descriptions_longclip = { | |
| "Mancha de Olho Pardo (Cercospora)": ( | |
| "A highly magnified view showing a distinct circular 'bullseye' or eye geometry. " | |
| "The absolute center must present a lighter, silvery, pale, or white core. " | |
| "This central pale area is obligatorily surrounded by prominent, dark concentric rings. " | |
| "Outside these dark rings lies a diffuse yellow halo." | |
| ), | |
| "Ferrugem do Cafeeiro": ( | |
| "A highly magnified view of bright, vivid cadmium-orange textures. " | |
| "The surface is densely covered in thick powdery granular dust or accumulated pollen-like spores. " | |
| "The bright orange fuzzy material forms soft, blurry, and diffuse shapes." | |
| ), | |
| "Bicho Mineiro do Café": ( | |
| "A highly magnified view of parched, light-tan or beige blisters. " | |
| "The texture consists of flat, withered material with a highly wrinkled, translucent, and papery appearance. " | |
| "It looks exactly like dry, flaky skin or torn parched parchment." | |
| ), | |
| "Mancha de Phoma": ( | |
| "A highly magnified view of a massive, solid, dark-brown or pitch-black necrotic mass. " | |
| "The affliction appears as a severe, thick dark rot or burn. " | |
| "It forms a continuous, opaque block of dark dead tissue, often causing the affected margins to become torn." | |
| ) | |
| } | |
| def home(): | |
| supabase_url = os.environ.get('SUPABASE_URL') | |
| supabase_key = os.environ.get('SUPABASE_ANON_KEY') | |
| return render_template('index.html', supabase_url=supabase_url, supabase_key=supabase_key) | |
| def redefinir_senha(): | |
| supabase_url = os.environ.get('SUPABASE_URL') | |
| supabase_key = os.environ.get('SUPABASE_ANON_KEY') | |
| return render_template('redefinir_senha.html', supabase_url=supabase_url, supabase_key=supabase_key) | |
| def process_leaf_image(image_path, model_roboflow, long_model, long_preprocess, text_features, class_names, device): | |
| # LÓGICA DE LIMPEZA DA PASTA | |
| if os.path.exists(output_dir): | |
| for filename in os.listdir(output_dir): | |
| file_path = os.path.join(output_dir, filename) | |
| try: | |
| if os.path.isfile(file_path) or os.path.islink(file_path): | |
| os.unlink(file_path) | |
| elif os.path.isdir(file_path): | |
| shutil.rmtree(file_path) | |
| except Exception as e: | |
| print(f'Falha ao deletar {file_path}. Motivo: {e}') | |
| else: | |
| os.makedirs(output_dir) | |
| # 1. DETECÇÃO (ROBOFLOW) | |
| image = Image.open(image_path) | |
| predictions = model_roboflow.infer(image, confidence=0.6)[0] | |
| detections = sv.Detections.from_inference(predictions) | |
| img_width, img_height = image.size | |
| padding_percentage = 0.20 | |
| min_dim_for_padding = 100 | |
| for i, xyxy in enumerate(detections.xyxy): | |
| x_min_orig, y_min_orig, x_max_orig, y_max_orig = xyxy | |
| bbox_width_orig = x_max_orig - x_min_orig | |
| bbox_height_orig = y_max_orig - y_min_orig | |
| final_x_min, final_y_min, final_x_max, final_y_max = x_min_orig, y_min_orig, x_max_orig, y_max_orig | |
| if bbox_width_orig < min_dim_for_padding or bbox_height_orig < min_dim_for_padding: | |
| pad_x = bbox_width_orig * padding_percentage | |
| pad_y = bbox_height_orig * padding_percentage | |
| final_x_min = max(0, x_min_orig - pad_x) | |
| final_y_min = max(0, y_min_orig - pad_y) | |
| final_x_max = min(img_width, x_max_orig + pad_x) | |
| final_y_max = min(img_height, y_max_orig + pad_y) | |
| final_x_min, final_y_min, final_x_max, final_y_max = int(final_x_min), int(final_y_min), int(final_x_max), int(final_y_max) | |
| cropped_image = image.crop((final_x_min, final_y_min, final_x_max, final_y_max)) | |
| cropped_image.save(os.path.join(output_dir, f"recorte_{i}.jpg")) | |
| # 2. CLASSIFICAÇÃO (LONGCLIP) | |
| results_list = [] | |
| image_urls = [] | |
| for file_name in os.listdir(output_dir): | |
| if file_name.startswith("recorte_") and file_name.endswith(".jpg"): | |
| img_path = os.path.join(output_dir, file_name) | |
| img_input = Image.open(img_path) | |
| image_input = long_preprocess(img_input).unsqueeze(0).to(device) | |
| with torch.no_grad(): | |
| image_features = long_model.encode_image(image_input) | |
| image_features /= image_features.norm(dim=-1, keepdim=True) | |
| logits = (100.0 * image_features @ text_features.T) | |
| probs = logits.softmax(dim=-1).cpu().numpy()[0] | |
| top_idx = np.argmax(probs) | |
| predicted_class = class_names[top_idx] | |
| top_prob = float(probs[top_idx]) | |
| results_list.append(predicted_class) | |
| image_urls.append({ | |
| "url": f"/static/outputs/{file_name}", | |
| "classe": predicted_class, | |
| "probabilidade": top_prob | |
| }) | |
| contagem = Counter(results_list) | |
| mais_frequente = contagem.most_common(1)[0][0] if results_list else "Nenhuma detecção encontrada" | |
| return { | |
| "contagem": dict(contagem), | |
| "mais_frequente": mais_frequente, | |
| "total": sum(contagem.values()), | |
| "imagens": image_urls | |
| } | |
| def predict(): | |
| if 'file' not in request.files: | |
| return jsonify({'error': 'Nenhum arquivo enviado'}), 400 | |
| file = request.files['file'] | |
| if file.filename == '': | |
| return jsonify({'error': 'Arquivo vazio selecionado'}), 400 | |
| try: | |
| image_path = os.path.join(app.config['UPLOAD_FOLDER'], "temp_leaf.jpg") | |
| file.save(image_path) | |
| model_roboflow = get_model("bracol-validado-region-detect-oupfv/1") | |
| checkpoint_abs_path = os.path.join(long_clip_path, "checkpoints", "longclip-L.pt") | |
| long_model, long_preprocess = load_longclip_model(checkpoint_abs_path, device) | |
| text_features, class_names = get_text_features(long_model, text_descriptions_longclip, device) | |
| result = process_leaf_image( | |
| image_path, model_roboflow, long_model, long_preprocess, | |
| text_features, class_names, device | |
| ) | |
| return jsonify(result) | |
| except Exception as e: | |
| return jsonify({'error': str(e)}), 500 | |
| def get_true_class_from_filename(filename): | |
| name_lower = filename.lower() | |
| if 'ferrugem' in name_lower: | |
| return "Ferrugem do Cafeeiro" | |
| elif 'cercospora' in name_lower or 'olho pardo' in name_lower: | |
| return "Mancha de Olho Pardo (Cercospora)" | |
| elif 'mineiro' in name_lower or 'bicho' in name_lower: | |
| return "Bicho Mineiro do Café" | |
| elif 'phoma' in name_lower: | |
| return "Mancha de Phoma" | |
| else: | |
| return None | |
| def predict_batch(): | |
| if 'files' not in request.files: | |
| return jsonify({'error': 'Nenhum arquivo enviado'}), 400 | |
| files = request.files.getlist('files') | |
| if not files or files[0].filename == '': | |
| return jsonify({'error': 'Nenhum arquivo selecionado'}), 400 | |
| try: | |
| # Carregar modelos uma única vez para o lote inteiro | |
| model_roboflow = get_model("bracol-validado-region-detect-oupfv/1") | |
| checkpoint_abs_path = os.path.join(long_clip_path, "checkpoints", "longclip-L.pt") | |
| long_model, long_preprocess = load_longclip_model(checkpoint_abs_path, device) | |
| text_features, class_names = get_text_features(long_model, text_descriptions_longclip, device) | |
| batch_results = [] | |
| y_true = [] | |
| y_pred = [] | |
| for idx, file in enumerate(files): | |
| image_path = os.path.join(app.config['UPLOAD_FOLDER'], f"temp_leaf_batch_{idx}.jpg") | |
| file.save(image_path) | |
| result = process_leaf_image( | |
| image_path, model_roboflow, long_model, long_preprocess, | |
| text_features, class_names, device | |
| ) | |
| predicted_class = result["mais_frequente"] | |
| batch_results.append({ | |
| "filename": file.filename, | |
| "mais_frequente": predicted_class | |
| }) | |
| # Extrair classe real: prioriza a seleção manual se existir, senão usa o nome do arquivo | |
| manual_true_class = request.form.get('true_class') | |
| if manual_true_class: | |
| true_class = manual_true_class | |
| else: | |
| true_class = get_true_class_from_filename(file.filename) | |
| if true_class and predicted_class != "Nenhuma detecção encontrada": | |
| y_true.append(true_class) | |
| y_pred.append(predicted_class) | |
| # Consolidar a contagem da doença principal de cada folha do lote | |
| doencas_principais = [res["mais_frequente"] for res in batch_results if res["mais_frequente"] != "Nenhuma detecção encontrada"] | |
| contagem_geral = Counter(doencas_principais) | |
| # Calcular porcentagens e garantir que todas as classes (doenças) apareçam no relatório | |
| total_validas = len(doencas_principais) | |
| estatisticas = {} | |
| for doenca in class_names: | |
| qtd = contagem_geral.get(doenca, 0) | |
| pct = (qtd / total_validas * 100) if total_validas > 0 else 0 | |
| estatisticas[doenca] = { | |
| "quantidade": qtd, | |
| "porcentagem": round(pct, 1) | |
| } | |
| # Calcular métricas de avaliação | |
| metricas_avaliacao = {} | |
| if len(y_true) > 0: | |
| acertos = sum(1 for yt, yp in zip(y_true, y_pred) if yt == yp) | |
| acuracia_geral = (acertos / len(y_true)) * 100 | |
| metricas_por_classe = {} | |
| for cls in class_names: | |
| tp = sum(1 for yt, yp in zip(y_true, y_pred) if yt == cls and yp == cls) | |
| fp = sum(1 for yt, yp in zip(y_true, y_pred) if yt != cls and yp == cls) | |
| fn = sum(1 for yt, yp in zip(y_true, y_pred) if yt == cls and yp != cls) | |
| precision = (tp / (tp + fp)) * 100 if (tp + fp) > 0 else 0 | |
| recall = (tp / (tp + fn)) * 100 if (tp + fn) > 0 else 0 | |
| f1_score = (2 * precision * recall) / (precision + recall) if (precision + recall) > 0 else 0 | |
| metricas_por_classe[cls] = { | |
| "precision": round(precision, 2), | |
| "recall": round(recall, 2), | |
| "f1_score": round(f1_score, 2), | |
| "suporte": sum(1 for yt in y_true if yt == cls) | |
| } | |
| matriz_confusao = {} | |
| for cls_real in class_names: | |
| matriz_confusao[cls_real] = {} | |
| for cls_pred in class_names: | |
| matriz_confusao[cls_real][cls_pred] = sum(1 for yt, yp in zip(y_true, y_pred) if yt == cls_real and yp == cls_pred) | |
| metricas_avaliacao = { | |
| "acuracia_geral": round(acuracia_geral, 2), | |
| "por_classe": metricas_por_classe, | |
| "matriz_confusao": matriz_confusao, | |
| "total_avaliado": len(y_true), | |
| "dados_brutos": { | |
| "y_true": y_true, | |
| "y_pred": y_pred | |
| } | |
| } | |
| resultado_final = { | |
| "estatisticas": estatisticas, | |
| "detalhes": batch_results, | |
| "total_folhas": len(files), | |
| "folhas_com_deteccao": total_validas, | |
| "metricas_avaliacao": metricas_avaliacao | |
| } | |
| import datetime | |
| timestamp_str = datetime.datetime.now().strftime("%d/%m/%Y %H:%M:%S") | |
| doencas_no_lote = list(set(y_true)) if len(y_true) > 0 else ["Desconhecida"] | |
| for doenca in doencas_no_lote: | |
| erros_confusao = {} | |
| suporte = len(files) | |
| if doenca != "Desconhecida" and "matriz_confusao" in metricas_avaliacao: | |
| erros_confusao = metricas_avaliacao["matriz_confusao"].get(doenca, {}) | |
| suporte = metricas_avaliacao.get("por_classe", {}).get(doenca, {}).get("suporte", len(files)) | |
| resultados_globais_artigo.append({ | |
| "timestamp": timestamp_str, | |
| "doenca_alvo": doenca, | |
| "total_folhas": suporte, | |
| "estatisticas": estatisticas, | |
| "acuracia_geral": metricas_avaliacao.get("acuracia_geral", 0), | |
| "metricas_classe": metricas_avaliacao.get("por_classe", {}).get(doenca, {}) if doenca != "Desconhecida" else {}, | |
| "erros_confusao": erros_confusao | |
| }) | |
| return jsonify(resultado_final) | |
| except Exception as e: | |
| return jsonify({'error': str(e)}), 500 | |
| def resultados_artigo(): | |
| return render_template('resultados_artigo.html', dados=resultados_globais_artigo) | |
| if __name__ == '__main__': | |
| app.run(debug=True, use_reloader=False) | |