CoffeeHealth-AI / app.py
Gabriel Henrique
merge: resolver conflito e incorporar atualizacoes do Hugging Face
7983a21
Raw History Blame Contribute Delete
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."
)
}
@app.route('/')
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)
@app.route('/redefinir-senha')
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
}
@app.route('/predict', methods=['POST'])
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
@app.route('/predict_batch', methods=['POST'])
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
@app.route('/resultados_artigo')
def resultados_artigo():
return render_template('resultados_artigo.html', dados=resultados_globais_artigo)
if __name__ == '__main__':
app.run(debug=True, use_reloader=False)