""" KAIROS ML ENGINE v2.0 Sistema de Aprendizado Contínuo para Recomendações Agronômicas Integrado com: Embrapa + Supabase + TensorFlow AUTOR: Sistema AgroPlan MAPS DATA: 2026-02-03 """ import numpy as np import pandas as pd import requests import json from datetime import datetime, timedelta from sklearn.ensemble import RandomForestRegressor, GradientBoostingClassifier from sklearn.preprocessing import StandardScaler from sklearn.model_selection import train_test_split import joblib import warnings warnings.filterwarnings('ignore') # --- CONFIGURAÇÕES --- SUPABASE_URL = "https://mhbappxhnlpurfqhxpwa.supabase.co" SUPABASE_KEY = "sb_publishable_XgU3v47C8GjZO7E3RURXYA_abPUYDzD" class KairosMLEngine: """ Motor de IA que aprende com: 1. Dados históricos do solo (Supabase) 2. Recomendações da Embrapa (API) 3. Feedback do produtor (acerto/erro das sugestões) """ def __init__(self): self.modelo_producao = None # Prediz produtividade (sc/ha) self.modelo_correcao = None # Classifica necessidade de correção self.scaler = StandardScaler() self.historico_treino = [] self.acuracia_atual = 0.0 # ========================================== # 1. COLETA DE DADOS HISTÓRICOS # ========================================== def coletar_dados_supabase(self, limite=1000): """ Busca histórico de leituras do Supabase """ try: url = f"{SUPABASE_URL}/rest/v1/leituras?order=created_at.desc&limit={limite}" headers = { "apikey": SUPABASE_KEY, "Authorization": f"Bearer {SUPABASE_KEY}" } response = requests.get(url, headers=headers, timeout=10) if response.status_code == 200: dados = response.json() print(f"✅ Coletados {len(dados)} registros históricos") return pd.DataFrame(dados) else: print(f"⚠️ Erro ao coletar dados: {response.status_code}") return pd.DataFrame() except Exception as e: print(f"❌ Falha na coleta: {e}") return pd.DataFrame() def buscar_recomendacao_embrapa(self, cultura, estado, solo_tipo): """ Integração com API da Embrapa (exemplo com dados públicos) Na prática, você usaria o endpoint real da Embrapa """ # EMBRAPA tem várias APIs: # - https://www.embrapa.br/agencia-de-informacao-tecnologica # - Sistema de Recomendação de Calagem e Adubação # SIMULAÇÃO (substitua pelo endpoint real) recomendacoes_base = { "soja": { "ph_ideal": 6.0, "n_ideal": 65, "p_ideal": 20, "k_ideal": 140, "calagem_ton_ha": 2.5 }, "milho": { "ph_ideal": 5.8, "n_ideal": 120, "p_ideal": 80, "k_ideal": 80, "calagem_ton_ha": 3.0 } } return recomendacoes_base.get(cultura.lower(), recomendacoes_base["soja"]) # ========================================== # 2. PREPARAÇÃO DOS DADOS PARA ML # ========================================== def preparar_dataset(self, df_historico): """ Transforma dados brutos em features para o modelo """ if df_historico.empty: print("⚠️ Dataset vazio, usando dados sintéticos para demonstração") return self._gerar_dados_sinteticos() # Features (X): características do solo e clima features = df_historico[[ 'ph_solo', 'n_solo', 'p_solo', 'k_solo', 'temperatura_solo', 'umidade_solo' ]].copy() # Adiciona features derivadas features['deficit_ph'] = 6.0 - features['ph_solo'] # Quanto falta para pH ideal features['balanco_npk'] = features['n_solo'] + features['p_solo'] + features['k_solo'] features['estresse_termico'] = (features['temperatura_solo'] > 28).astype(int) # Target (y): produtividade real (você precisará coletar isso do produtor) # Por enquanto, vamos simular baseado na qualidade do solo target_producao = self._calcular_producao_estimada(features) target_correcao = (features['ph_solo'] < 5.5).astype(int) # 1 = precisa correção return features, target_producao, target_correcao def _calcular_producao_estimada(self, features): """ Estima produtividade baseada em fórmulas agronômicas (Substitua por dados reais quando disponíveis) """ # Fórmula simplificada: produção depende de pH, NPK e temperatura base = 40 # sacas/ha base bonus_ph = np.where(features['ph_solo'] >= 5.5, 10, -5) bonus_npk = (features['balanco_npk'] - 150) / 20 bonus_temp = np.where(features['temperatura_solo'].between(22, 28), 5, -3) producao = base + bonus_ph + bonus_npk + bonus_temp producao = np.clip(producao, 20, 70) # Limita entre 20-70 sc/ha return producao def _gerar_dados_sinteticos(self, n_amostras=500): """ Gera dados sintéticos para treinar o modelo inicial """ np.random.seed(42) features = pd.DataFrame({ 'ph_solo': np.random.uniform(4.5, 6.8, n_amostras), 'n_solo': np.random.randint(20, 80, n_amostras), 'p_solo': np.random.randint(10, 30, n_amostras), 'k_solo': np.random.randint(80, 180, n_amostras), 'temperatura_solo': np.random.uniform(20, 32, n_amostras), 'umidade_solo': np.random.uniform(25, 65, n_amostras) }) features['deficit_ph'] = 6.0 - features['ph_solo'] features['balanco_npk'] = features['n_solo'] + features['p_solo'] + features['k_solo'] features['estresse_termico'] = (features['temperatura_solo'] > 28).astype(int) target_producao = self._calcular_producao_estimada(features) target_correcao = (features['ph_solo'] < 5.5).astype(int) return features, target_producao, target_correcao # ========================================== # 3. TREINAMENTO DOS MODELOS # ========================================== def treinar_modelos(self, features, target_producao, target_correcao): """ Treina dois modelos: 1. Random Forest para prever produtividade 2. Gradient Boosting para classificar necessidade de correção """ print("\n🔄 Iniciando treinamento dos modelos...") # Normaliza features X = self.scaler.fit_transform(features) # Divide em treino e teste (80/20) X_train, X_test, y_prod_train, y_prod_test = train_test_split( X, target_producao, test_size=0.2, random_state=42 ) _, _, y_corr_train, y_corr_test = train_test_split( X, target_correcao, test_size=0.2, random_state=42 ) # MODELO 1: Predição de Produtividade self.modelo_producao = RandomForestRegressor( n_estimators=100, max_depth=10, min_samples_split=5, random_state=42, n_jobs=-1 ) self.modelo_producao.fit(X_train, y_prod_train) score_prod = self.modelo_producao.score(X_test, y_prod_test) # MODELO 2: Classificação de Necessidade de Correção self.modelo_correcao = GradientBoostingClassifier( n_estimators=100, learning_rate=0.1, max_depth=5, random_state=42 ) self.modelo_correcao.fit(X_train, y_corr_train) score_corr = self.modelo_correcao.score(X_test, y_corr_test) self.acuracia_atual = (score_prod + score_corr) / 2 print(f"✅ Modelo de Produtividade - R² Score: {score_prod:.2%}") print(f"✅ Modelo de Correção - Acurácia: {score_corr:.2%}") print(f"📊 Acurácia Média do Sistema: {self.acuracia_atual:.2%}") # Salva modelos self.salvar_modelos() return score_prod, score_corr def salvar_modelos(self): """ Salva modelos treinados em disco """ try: joblib.dump(self.modelo_producao, 'modelo_producao.pkl') joblib.dump(self.modelo_correcao, 'modelo_correcao.pkl') joblib.dump(self.scaler, 'scaler.pkl') print("💾 Modelos salvos com sucesso!") except Exception as e: print(f"⚠️ Erro ao salvar modelos: {e}") def carregar_modelos(self): """ Carrega modelos previamente treinados """ try: self.modelo_producao = joblib.load('modelo_producao.pkl') self.modelo_correcao = joblib.load('modelo_correcao.pkl') self.scaler = joblib.load('scaler.pkl') print("✅ Modelos carregados com sucesso!") return True except: print("⚠️ Nenhum modelo encontrado. Será necessário treinar.") return False # ========================================== # 4. PREDIÇÕES E RECOMENDAÇÕES # ========================================== def analisar_solo(self, dados_solo): """ Analisa condições do solo e retorna recomendações inteligentes Parâmetros: ----------- dados_solo : dict { 'ph_solo': 5.2, 'n_solo': 30, 'p_solo': 15, 'k_solo': 100, 'temperatura_solo': 26.5, 'umidade_solo': 45.0, 'chuva_prevista': 12.0 # mm nos próximos 3 dias } """ if self.modelo_producao is None: return {"erro": "Modelo não treinado. Execute treinar_modelos() primeiro."} # Prepara features features_dict = { 'ph_solo': dados_solo['ph_solo'], 'n_solo': dados_solo['n_solo'], 'p_solo': dados_solo['p_solo'], 'k_solo': dados_solo['k_solo'], 'temperatura_solo': dados_solo['temperatura_solo'], 'umidade_solo': dados_solo['umidade_solo'], 'deficit_ph': 6.0 - dados_solo['ph_solo'], 'balanco_npk': dados_solo['n_solo'] + dados_solo['p_solo'] + dados_solo['k_solo'], 'estresse_termico': 1 if dados_solo['temperatura_solo'] > 28 else 0 } features_df = pd.DataFrame([features_dict]) X = self.scaler.transform(features_df) # Predições producao_estimada = self.modelo_producao.predict(X)[0] prob_correcao = self.modelo_correcao.predict_proba(X)[0][1] # Probabilidade de precisar correção # Recomendação Embrapa embrapa_rec = self.buscar_recomendacao_embrapa("soja", "GO", "latossolo") # Lógica de Decisão Inteligente recomendacao = self._gerar_recomendacao_final( dados_solo, producao_estimada, prob_correcao, embrapa_rec ) return recomendacao def _gerar_recomendacao_final(self, solo, prod_est, prob_corr, embrapa): """ Combina ML + Embrapa + Regras Climáticas """ chuva = solo.get('chuva_prevista', 0) # Decisão baseada em ML if prob_corr > 0.7: # Alta probabilidade de precisar correção if chuva >= 20: acao = "⛔ AGUARDAR APLICAÇÃO" motivo = f"Risco de lixiviação (chuva: {chuva}mm)" cor = "#e74c3c" else: acao = "✅ APLICAR CALCÁRIO AGORA" dose_calc = round((embrapa['ph_ideal'] - solo['ph_solo']) * 2.5, 1) motivo = f"Dose sugerida: {dose_calc} ton/ha (Embrapa + ML)" cor = "#28a745" else: acao = "🟢 MONITORAR" motivo = f"Solo em condições aceitáveis (pH: {solo['ph_solo']})" cor = "#3498db" # Recomendações de NPK npk_rec = [] if solo['n_solo'] < embrapa['n_ideal'] * 0.7: npk_rec.append(f"Nitrogênio: aplicar {embrapa['n_ideal'] - solo['n_solo']} kg/ha") if solo['p_solo'] < embrapa['p_ideal'] * 0.7: npk_rec.append(f"Fósforo: aplicar {embrapa['p_ideal'] - solo['p_solo']} kg/ha") if solo['k_solo'] < embrapa['k_ideal'] * 0.7: npk_rec.append(f"Potássio: aplicar {embrapa['k_ideal'] - solo['k_solo']} kg/ha") return { "acao_principal": acao, "motivo": motivo, "cor": cor, "producao_estimada_sc_ha": round(prod_est, 1), "probabilidade_correcao": round(prob_corr * 100, 1), "confianca_ia": round(self.acuracia_atual * 100, 1), "recomendacoes_npk": npk_rec, "dados_embrapa": embrapa, "timestamp": datetime.now().isoformat() } # ========================================== # 5. APRENDIZADO CONTÍNUO (FEEDBACK LOOP) # ========================================== def registrar_feedback(self, dados_solo, recomendacao_dada, resultado_real): """ Registra o resultado real da recomendação para retreinamento futuro Parâmetros: ----------- dados_solo : dict Dados originais do solo recomendacao_dada : dict Recomendação que foi feita resultado_real : dict { 'producao_real_sc_ha': 58.5, # Produtividade real colhida 'correcao_efetiva': True, # Se a correção funcionou 'satisfacao_produtor': 5 # Nota de 1-5 } """ # Salva no histórico para retreinamento registro = { 'timestamp': datetime.now().isoformat(), 'dados_solo': dados_solo, 'recomendacao': recomendacao_dada, 'resultado': resultado_real } self.historico_treino.append(registro) # Salva no Supabase (tabela de feedback) self._salvar_feedback_supabase(registro) # Se acumulou 50+ feedbacks, retreina if len(self.historico_treino) >= 50: print("\n🔄 50 feedbacks acumulados! Iniciando retreinamento automático...") self.retreinar_com_feedback() def _salvar_feedback_supabase(self, registro): """ Salva feedback no banco para análise futura """ try: url = f"{SUPABASE_URL}/rest/v1/feedback_ia" headers = { "apikey": SUPABASE_KEY, "Authorization": f"Bearer {SUPABASE_KEY}", "Content-Type": "application/json", "Prefer": "return=minimal" } payload = { "created_at": registro['timestamp'], "dados_entrada": json.dumps(registro['dados_solo']), "recomendacao_ia": json.dumps(registro['recomendacao']), "resultado_real": json.dumps(registro['resultado']) } response = requests.post(url, json=payload, headers=headers, timeout=5) if response.status_code in [200, 201]: print("✅ Feedback salvo no Supabase") except Exception as e: print(f"⚠️ Erro ao salvar feedback: {e}") def retreinar_com_feedback(self): """ Retreina o modelo usando dados reais coletados """ if len(self.historico_treino) < 10: print("⚠️ Feedbacks insuficientes para retreinamento (mínimo: 10)") return # Extrai dados reais do feedback dados_reais = [] for fb in self.historico_treino: solo = fb['dados_solo'] resultado = fb['resultado'] dados_reais.append({ **solo, 'producao_real': resultado.get('producao_real_sc_ha', 0), 'correcao_efetiva': int(resultado.get('correcao_efetiva', False)) }) df_real = pd.DataFrame(dados_reais) # Prepara dataset features = df_real[[ 'ph_solo', 'n_solo', 'p_solo', 'k_solo', 'temperatura_solo', 'umidade_solo' ]] features['deficit_ph'] = 6.0 - features['ph_solo'] features['balanco_npk'] = features['n_solo'] + features['p_solo'] + features['k_solo'] features['estresse_termico'] = (features['temperatura_solo'] > 28).astype(int) target_producao = df_real['producao_real'] target_correcao = df_real['correcao_efetiva'] # Retreina print(f"\n🔄 Retreinando com {len(df_real)} registros reais...") self.treinar_modelos(features, target_producao, target_correcao) # Limpa histórico self.historico_treino = [] print("🎯 Retreinamento concluído! Modelo atualizado com dados reais.") # ========================================== # EXEMPLO DE USO # ========================================== if __name__ == "__main__": print("=" * 60) print("KAIROS ML ENGINE v2.0 - INICIALIZANDO") print("=" * 60) # Instancia o motor engine = KairosMLEngine() # Tenta carregar modelos existentes if not engine.carregar_modelos(): # Se não existe, treina novo modelo print("\n📚 Coletando dados históricos...") df_historico = engine.coletar_dados_supabase(limite=500) print("\n🔧 Preparando dataset...") features, target_prod, target_corr = engine.preparar_dataset(df_historico) print("\n🎓 Treinando modelos de Machine Learning...") engine.treinar_modelos(features, target_prod, target_corr) # Teste de análise print("\n" + "=" * 60) print("TESTE DE ANÁLISE DE SOLO") print("=" * 60) solo_teste = { 'ph_solo': 5.2, 'n_solo': 32, 'p_solo': 14, 'k_solo': 105, 'temperatura_solo': 26.5, 'umidade_solo': 42.0, 'chuva_prevista': 8.0 } resultado = engine.analisar_solo(solo_teste) print(f"\n🎯 RECOMENDAÇÃO KAIROS ML:") print(f" Ação: {resultado['acao_principal']}") print(f" Motivo: {resultado['motivo']}") print(f" Produção Estimada: {resultado['producao_estimada_sc_ha']} sc/ha") print(f" Confiança da IA: {resultado['confianca_ia']}%") print(f" Probabilidade de Correção: {resultado['probabilidade_correcao']}%") if resultado['recomendacoes_npk']: print(f"\n💊 Recomendações NPK:") for rec in resultado['recomendacoes_npk']: print(f" • {rec}") print("\n✅ Sistema pronto para uso em produção!")