Download app.py from simikkk/DAMR58H: direct link, hf CLI and curl.
- Browser
- Download file 13.8 kB
-
https://huggingface.co/spaces/simikkk/DAMR58H/resolve/main/app.py
- Command line
-
hf download hf://spaces/simikkk/DAMR58H/app.py
-
curl -L -o app.py https://huggingface.co/spaces/simikkk/DAMR58H/resolve/main/app.py
13.8 kB
| 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() |