import gradio as gr import pandas as pd import numpy as np from PIL import Image import os import tensorflow as tf import glob import requests # ── HACK PRO STARÉ MODELY Z TEACHABLE MACHINE ───────────────── original_init = tf.keras.layers.DepthwiseConv2D.__init__ def patched_init(self, *args, **kwargs): if 'groups' in kwargs: kwargs.pop('groups') original_init(self, *args, **kwargs) tf.keras.layers.DepthwiseConv2D.__init__ = patched_init # ────────────────────────────────────────────────────────────── # Načtení API klíčů z prostředí Hugging Face api_key_openrouter = os.environ.get("OPENROUTER_API_KEY") api_key_gemini = os.environ.get("GEMINI_API_KEY") # ── Funkce pro načtení všech CSV dat ── def nacti_vsechna_data(): vsechna_data_text = "" soubory = glob.glob("*.csv") if not soubory: return "Pozor: V repozitáři nebyla nalezena žádná data (.csv)." for soubor in soubory: try: df = pd.read_csv(soubor, sep=None, engine='python') vsechna_data_text += f"\n--- OBSAH SOUBORU (Vzorek 15 řádků): {soubor} ---\n" vsechna_data_text += df.head(15).to_string() + "\n" except Exception as e: vsechna_data_text += f"\nChyba při čtení {soubor}: {str(e)}\n" return vsechna_data_text # ── Funkce pro model Teachable Machine ──────────── def nacti_model(): if not os.path.exists("keras_model.h5") or not os.path.exists("labels.txt"): return None, "Chybí soubory modelu (keras_model.h5, labels.txt)." model = tf.keras.models.load_model("keras_model.h5", compile=False) labels = open("labels.txt", "r", encoding="utf-8").read().strip().split("\n") return model, labels GLOBAL_MODEL, GLOBAL_LABELS = nacti_model() # ── Funkce pro vyhodnocení střechy ───────────── def vyhodnot_strechu(obrazek): if obrazek is None: return "Prosím, nahrajte snímek střechy." model, labels = GLOBAL_MODEL, GLOBAL_LABELS if model is None: return labels img = obrazek.resize((224, 224)) arr = np.array(img, dtype=np.float32) / 255.0 arr = np.expand_dims(arr, axis=0) predikce = model.predict(arr)[0] index = np.argmax(predikce) skore = float(predikce[index]) * 100 label = labels[index] vysledek = f"Analýza: {label} ({skore:.1f}%)\n\n" label_lower = label.lower() if skore < 70: vysledek += "⚠️ Závěr: Výsledek je nejednoznačný (spolehlivost je nižší než 70 %).\n" vysledek += "💡 Doporučení: Umělá inteligence si v tomto případě není dostatečně jistá. Pro přesné posouzení doporučujeme fyzickou obhlídku střechy technikem." elif "nevhod" in label_lower or "2" in label_lower: vysledek += "❌ Závěr: Střecha NENÍ vhodná pro FVE" elif "vhod" in label_lower or "0" in label_lower: vysledek += "✅ Závěr: Střecha JE vhodná pro FVE" else: vysledek += "❌ Závěr: Střecha NENÍ vhodná pro FVE (Neurčitý výsledek)" return vysledek # ── JÁDRO S PODROBNÝM TRASOVÁNÍM HISTORIE VÝBĚRU (ROUTING LOG) ── def proved_api_volani(otazka, historie): log_list = [] log_list.append("⏱️ **[00:00]** Inicializace požadavku uživatele.") data_text = nacti_vsechna_data() log_list.append(f"📄 **[Data]** Úspěšně načteny lokální CSV soubory ({len(data_text)} znaků v kontextu).") prompt = ( "Jsi špičkový datový analytik a expert na fotovoltaiku pro firmu DAMR58H v Ústeckém kraji. " "Odpovídej vždy věcně, logicky a VÝHRADNĚ PERFEKTNÍ GRAMATICKOU ČEŠTINOU.\n\n" "Pravidla pro analýzu:\n" "1. Pokud se uživatel ptá na data ze souborů CSV, NEKOPÍRUJ suše strukturu sloupců. " "Místo toho se podívej na hodnoty, hledej trendy, porovnávej obce a dělej reálné energetické závěry.\n" "2. Pravidlo o 70% jistotě a fyzické obhlídce technikem aplikuj POUZE tehdy, když vyhodnocuješ konkrétní skóre z analýzy FOTOGRAFIE střechy. Nemíchej to do tabulek.\n\n" f"Tady jsou data z našich souborů k analýze:\n{data_text}\n\n" f"Otázka: {otazka}" ) historie.append([otazka, "⏳ Spouštím vyhledávání volného serveru..."]) yield historie, "### 🤖 Aktivní model: Vyhledávám...", "\n".join(log_list) # KROK 1: Pokus o OpenRouter servery if api_key_openrouter: log_list.append("🔗 **[OpenRouter]** API klíč detekován. Sestavuji prioritní frontu free modelů.") url_or = "https://openrouter.ai/api/v1/chat/completions" headers_or = { "Authorization": f"Bearer {api_key_openrouter}", "Content-Type": "application/json", "HTTP-Referer": "https://huggingface.co", "X-Title": "Solar Advisor" } modely_or = [ "meta-llama/llama-3.3-70b-instruct:free", "meta-llama/llama-3.2-3b-instruct:free", "openrouter/free" ] for model in modely_or: model_short = model.split('/')[-1] log_list.append(f"📡 **[Požadavek]** Odesílám dotaz na endpoint: `{model}`") yield historie, f"### 🤖 OpenRouter: Zkouším `{model_short}`...", "\n".join(log_list) payload = { "model": model, "messages": [{"role": "user", "content": prompt}], "temperature": 0.5 } try: res = requests.post(url_or, headers=headers_or, json=payload, timeout=18) if res.status_code == 200: data = res.json() realny_model = data.get("model", model) odpoved = data["choices"][0]["message"]["content"] log_list.append(f"✅ **[Odezva]** Server odpověděl úspěšně (Kód 200).") log_list.append(f"🏷️ **[Identifikace]** OpenRouter přidělil reálnou instanci: `{realny_model}`") # FILTR PROTI "MRDKÁM": Pokud router podhodil slabý model if model == "openrouter/free": nevhodne_prazdne_modely = ["llama-3-8b", "pygmalion", "gemma-7b"] if any(x in realny_model.lower() for x in nevhodne_prazdne_modely): log_list.append(f"❌ **[Filtr]** Odmítnuto! Přidělený model `{realny_model}` nesplňuje kvalitativní kritéria pro češtinu. Zahazuji spojení.") yield historie, "### ⚠️ OpenRouter podstrčil slabý model. Přepínám...", "\n".join(log_list) break # Vyskočí z OpenRouteru a padá do KROKU 2 (Gemini) historie[-1][1] = odpoved log_list.append("🏁 **[Dokončeno]** Data úspěšně vygenerována and odeslána do chatu.") yield historie, f"### 🤖 Aktivní model: `{realny_model.split('/')[-1]}` (OpenRouter)", "\n".join(log_list) return else: log_list.append(f"⚠️ **[Chyba]** Model `{model_short}` vrátil status kód {res.status_code}. Zkouším další v pořadí.") except Exception as e: log_list.append(f"💥 **[Selhání]** Spojení s `{model_short}` spadlo na timeout nebo síťovou chybu.") continue else: log_list.append("⚠️ **[OpenRouter]** Klíč OPENROUTER_API_KEY chybí. Přeskakuji celou OpenRouter sadu.") # KROK 2: Totální záloha přes Google Gemini API if api_key_gemini: log_list.append("🚀 **[Fallback]** Inicializuji nouzový most na Google Cloud (Gemini API)...") yield historie, "### 🚀 Nouzový režim: Připojuji se na Google Gemini API...", "\n".join(log_list) url_gemini = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash:generateContent?key={api_key_gemini}" payload_gemini = { "contents": [{"parts": [{"text": prompt}]}], "generationConfig": {"temperature": 0.4} } try: res = requests.post(url_gemini, json=payload_gemini, timeout=20) if res.status_code == 200: data = res.json() odpoved = data['candidates'][0]['content']['parts'][0]['text'] log_list.append("✅ **[Odezva]** Google Gemini 1.5 Flash zareagoval okamžitě s kódem 200.") log_list.append("🏁 **[Dokončeno]** Výstup vygenerován záložní infrastrukturou Google.") historie[-1][1] = odpoved yield historie, "### ⚡ Aktivní model: `Gemini 1.5 Flash` (Google Cloud)", "\n".join(log_list) return else: log_list.append(f"❌ **[Kritická chyba]** Google API odmítlo požadavek se statusem {res.status_code}.") status_g = f"Chyba {res.status_code}" except Exception as e: log_list.append(f"💥 **[Kritická chyba]** Google API je nedostupné nebo vypršel časový limit.") status_g = "Network/Timeout" else: log_list.append("❌ **[Kritická chyba]** Záložní klíč GEMINI_API_KEY není v systému definován.") status_g = "Klíč chybí" # Pokud selhalo úplně všechno log_list.append("🚨 **[FATAL]** Systém vyčerpal všechny dostupné servery i nouzové zálohy. Komunikace přerusená.") historie[-1][1] = f"❌ Selhal OpenRouter i záložní Google API. (Detaily: Gemini status: {status_g}). Zkus to poslat znovu za minutku." yield historie, "### 🤖 Aktivní model: Žádný (Všechny sítě přetíženy)", "\n".join(log_list) # ── OPRAVENÁ LOGIKA OVLÁDACÍCH TLAČÍTEK (OPRAVENO 'v' -> 'in') ── def chatbot_odpoved_manual(otazka, historie): if not otazka.strip(): yield "", historie, "### 🤖 Aktivní model: Bez zátěže", "Žádný dotaz ke zpracování." return historie = historie or [] for nova_hist, status, log in proved_api_volani(otazka, historie): yield "", nova_hist, status, log def undo_posledni(historie): if historie: historie.pop() return historie, "### 🤖 Aktivní model: Poslední zpráva smazána", "🗑️ Poslední log vyčištěn smazáním zprávy." def upravit_posledni(historie): if not historie: return "", historie, "### 🤖 Aktivní model: Žrádná historie k úpravě", "Žádný log k dispozici." posledni_dotaz = historie[-1][0] historie.pop() return posledni_dotaz, historie, "### 🤖 Aktivní model: Upravujete zprávu...", f"✏️ Upravuje se dotaz: '{posledni_dotaz}'" def resend_funkce(historie): if not historie: yield historie, "### 🤖 Aktivní model: Žádná historie k odeslání", "Není co posílat znovu." return posledni_otazka = historie[-1][0] historie.pop() for nova_hist, status, log in proved_api_volani(posledni_otazka, historie): yield nova_hist, status, log # ── UI Rozhraní (Gradio) ────────────────────────── with gr.Blocks(title="Solární poradce") as demo: gr.Markdown("# Solární poradce pro Ústecký kraj") with gr.Tabs(): with gr.TabItem("Analýza střechy"): gr.Markdown("### Nahrajte snímek střechy") with gr.Row(): with gr.Column(): vstup_img = gr.Image(type="pil", label="Vyberte obrázek") analyzovat_btn = gr.Button("Vyhodnotit", variant="primary") with gr.Column(): vystup_text = gr.Textbox(label="Výsledek", lines=4) analyzovat_btn.click(fn=vyhodnot_strechu, inputs=vstup_img, outputs=vystup_text) with gr.TabItem("Chatbot s daty"): status_model = gr.Markdown("### 🤖 Aktivní model: Bez zátěže") chatbot = gr.Chatbot(label="Odpovědi solárního asistenta", show_copy_button=True) vstup_text = gr.Textbox(label="Váš dotaz", placeholder="Napište otázku a stiskněte Enter...") with gr.Row(): btn_undo = gr.Button("↩️ Zpět (Undo)", variant="secondary", size="sm") btn_edit = gr.Button("✏️ Upravit (Edit)", variant="secondary", size="sm") btn_resend = gr.Button("🔄 Znovu poslat (Resend)", variant="primary", size="sm") with gr.Accordion("🔍 Zobrazit historii hledání modelu (Routing Log)", open=False): log_vystup = gr.Markdown("Zatím neproběhlo žádné vyhledávání modelu.") vstup_text.submit( fn=chatbot_odpoved_manual, inputs=[vstup_text, chatbot], outputs=[vstup_text, chatbot, status_model, log_vystup] ) btn_undo.click(fn=undo_posledni, inputs=[chatbot], outputs=[chatbot, status_model, log_vystup]) btn_edit.click(fn=upravit_posledni, inputs=[chatbot], outputs=[vstup_text, chatbot, status_model, log_vystup]) btn_resend.click(fn=resend_funkce, inputs=[chatbot], outputs=[chatbot, status_model, log_vystup]) demo.launch()