Spaces:
Sleeping
Sleeping
| # app.py — Diag Studio (Customer UI) | |
| # AI Destekli Araç Motor Ses Analizi — sade ticari arayüz | |
| # İstekler: Ses oynatıcı, ilerleme çubuğu, kullanıcı dostu akış. | |
| # Teknik sağlayıcı/kitaplık adları ve loglar UI’da gösterilmez. | |
| # Yönetici ayarları yalnızca DEV_MODE=1 iken görünür. | |
| import os | |
| import io | |
| import json | |
| import time | |
| import requests | |
| import streamlit as st | |
| from requests.adapters import HTTPAdapter | |
| from urllib3.util.retry import Retry | |
| # ---------------- Safe defaults ---------------- | |
| os.environ.setdefault("HOME", "/tmp") | |
| os.environ.setdefault("STREAMLIT_BROWSER_GATHER_USAGE_STATS", "false") | |
| os.environ.setdefault("STREAMLIT_SERVER_HEADLESS", "true") | |
| # ---------------- Theme ---------------- | |
| try: | |
| st._config.set_option("theme.base", "dark") | |
| st._config.set_option("theme.primaryColor", "#ff7a1a") | |
| st._config.set_option("theme.backgroundColor", "#0e0f12") | |
| st._config.set_option("theme.secondaryBackgroundColor", "#171a1f") | |
| st._config.set_option("theme.textColor", "#e6e8ea") | |
| except Exception: | |
| pass | |
| # ---------------- Env helpers ---------------- | |
| def _env(k: str) -> str: | |
| return (os.getenv(k) or "").strip().strip("'\"") | |
| DEV_MODE = (_env("DEV_MODE") in {"1", "true", "yes", "on"}) | |
| API_BASE_DEFAULT = (_env("CARDIAG_AI_API") or _env("APP_API") or "http://127.0.0.1:7860").rstrip("/") | |
| TOKEN = _env("CARDIAG_AI_TOKEN") or _env("APP_TOKEN") | |
| HEADERS = {"Authorization": f"Bearer {TOKEN}"} if TOKEN else {} | |
| # ---------------- i18n ---------------- | |
| LANGS = { | |
| "tr": { | |
| "title": "AI Destekli Araç Motor Ses Analizi", | |
| "about": "Motor ses kaydını yükleyin veya bağlantı girin. Sistem sesi analiz eder, olası arızaları ve önerileri özetler.", | |
| "upload": "Ses dosyası yükle", | |
| "or": "veya", | |
| "paste_url": "Ses bağlantısı (opsiyonel)", | |
| "analyze": "Analiz Et", | |
| "reset": "Sıfırla", | |
| "preview": "Önizleme", | |
| "result": "Sonuç", | |
| "commentary": "Özet ve Öneriler", | |
| "json": "Detaylı Bulgular", | |
| "artifacts": "Çıktılar", | |
| "download_summary": "Özet indir", | |
| "download_parsed": "Detay indir (JSON)", | |
| "download_raw": "Ham metin indir", | |
| "upload_hint": "mp3 / wav / ogg önerilir. İsterseniz herkese açık bir bağlantı da girebilirsiniz.", | |
| "warming": "Hazırlanıyor…", | |
| "analyzing": "Analiz ediliyor…", | |
| "health_ok": "Sistem hazır.", | |
| "health_fail": "Sisteme erişilemedi.", | |
| "need_backend": "Sistem adresi yapılandırılmadı.", | |
| "need_input": "Lütfen bir dosya yükleyin veya bağlantı girin.", | |
| "error_generic": "İşlem başarısız. Lütfen tekrar deneyin.", | |
| "try_again": "Yoğunluk olabilir. Biraz sonra tekrar deneyin.", | |
| "timeline_title": "İşlem Durumu", | |
| "step_prepare": "Hazırlık", | |
| "step_upload": "Yükleme", | |
| "step_process": "Analiz", | |
| "step_summarize": "Özet oluşturma", | |
| "admin": "Yönetici", | |
| "admin_desc": "Bu bölüm yalnızca yetkili kullanım içindir.", | |
| "backend_url": "Sistem adresi", | |
| "use_data_uri": "Gizli yükleme", | |
| "warmup": "Ön ısıtma", | |
| "auto_retry": "Otomatik yeniden dene", | |
| "max_wait": "Azami bekleme (sn)", | |
| "show_debug": "Hata ayıklama bilgileri", | |
| }, | |
| "en": { | |
| "title": "AI-Powered Engine Sound Analysis", | |
| "about": "Upload an engine sound or provide a public URL. The system analyzes the audio and summarizes likely issues and next steps.", | |
| "upload": "Upload audio file", | |
| "or": "or", | |
| "paste_url": "Audio URL (optional)", | |
| "analyze": "Analyze", | |
| "reset": "Reset", | |
| "preview": "Preview", | |
| "result": "Result", | |
| "commentary": "Summary & Recommendations", | |
| "json": "Detailed Findings", | |
| "artifacts": "Downloads", | |
| "download_summary": "Download summary", | |
| "download_parsed": "Download details (JSON)", | |
| "download_raw": "Download raw text", | |
| "upload_hint": "mp3 / wav / ogg recommended. You may paste a public URL instead of uploading.", | |
| "warming": "Preparing…", | |
| "analyzing": "Analyzing…", | |
| "health_ok": "Service is ready.", | |
| "health_fail": "Cannot reach the service.", | |
| "need_backend": "Service URL is not configured.", | |
| "need_input": "Please upload a file or enter a URL.", | |
| "error_generic": "Request failed. Please try again.", | |
| "try_again": "Service may be busy. Try again shortly.", | |
| "timeline_title": "Progress", | |
| "step_prepare": "Preparing", | |
| "step_upload": "Uploading", | |
| "step_process": "Processing", | |
| "step_summarize": "Summarizing", | |
| "admin": "Admin", | |
| "admin_desc": "This section is for authorized use only.", | |
| "backend_url": "Service URL", | |
| "use_data_uri": "Private upload", | |
| "warmup": "Warm up", | |
| "auto_retry": "Auto-retry", | |
| "max_wait": "Max wait (sec)", | |
| "show_debug": "Show debug info", | |
| }, | |
| } | |
| # ---------------- Session ---------------- | |
| if "lang" not in st.session_state: st.session_state.lang = "tr" | |
| if "api_base" not in st.session_state: st.session_state.api_base = API_BASE_DEFAULT | |
| if "use_data_uri" not in st.session_state: st.session_state.use_data_uri = True | |
| if "warmup" not in st.session_state: st.session_state.warmup = True | |
| if "auto_retry" not in st.session_state: st.session_state.auto_retry = True | |
| if "max_wait" not in st.session_state: st.session_state.max_wait = 60 | |
| if "show_debug" not in st.session_state: st.session_state.show_debug = False | |
| if "prompt_override" not in st.session_state: st.session_state.prompt_override = "" | |
| if "res" not in st.session_state: st.session_state.res = None | |
| T = LANGS[st.session_state.lang] | |
| # ---------------- Header ---------------- | |
| left, right = st.columns([0.7, 0.3]) | |
| with left: | |
| st.title(T["title"]) | |
| with right: | |
| st.session_state.lang = st.selectbox( | |
| "Language / Dil", ["tr", "en"], | |
| index=(0 if st.session_state.lang == "tr" else 1), | |
| ) | |
| T = LANGS[st.session_state.lang] | |
| st.caption(T["about"]) | |
| # ---------------- Admin (DEV only) ---------------- | |
| if DEV_MODE: | |
| with st.expander(f"🛠 {T['admin']}", expanded=False): | |
| st.write(T["admin_desc"]) | |
| st.session_state.api_base = st.text_input(T["backend_url"], value=st.session_state.api_base) | |
| st.session_state.use_data_uri = st.checkbox(T["use_data_uri"], value=st.session_state.use_data_uri) | |
| st.session_state.warmup = st.checkbox(T["warmup"], value=st.session_state.warmup) | |
| st.session_state.auto_retry = st.checkbox(T["auto_retry"], value=st.session_state.auto_retry) | |
| st.session_state.max_wait = st.slider(T["max_wait"], 10, 240, st.session_state.max_wait, step=10) | |
| st.session_state.show_debug = st.checkbox(T["show_debug"], value=st.session_state.show_debug) | |
| st.session_state.prompt_override = st.text_area("Advanced prompt (optional)", value=st.session_state.prompt_override, height=120) | |
| st.markdown("---") | |
| # ---------------- Upload / URL ---------------- | |
| st.write(T["upload_hint"]) | |
| file = st.file_uploader(T["upload"], type=["mp3", "wav", "ogg"]) | |
| st.write(f"**{T['or']}**") | |
| audio_url = st.text_input(T["paste_url"], value="", placeholder="https://...") | |
| # Audio preview block (player) | |
| if file is not None: | |
| st.markdown(f"**{T['preview']}**") | |
| st.audio(file, format=file.type or "audio/mpeg") | |
| elif audio_url.strip(): | |
| st.markdown(f"**{T['preview']}**") | |
| st.audio(audio_url) | |
| # ---------------- Controls ---------------- | |
| c1, c2, c3 = st.columns([0.22, 0.22, 0.56]) | |
| go = c1.button(f"▶ {T['analyze']}", type="primary") | |
| reset = c2.button(f"↩ {T['reset']}") | |
| health = c3.button("🔄 Health") | |
| # ---------------- HTTP helpers ---------------- | |
| def build_session(auto_retry: bool) -> requests.Session: | |
| s = requests.Session() | |
| s.headers.update(HEADERS) | |
| if auto_retry: | |
| retry = Retry( | |
| total=6, connect=6, read=6, | |
| backoff_factor=0.8, | |
| status_forcelist=[429, 500, 502, 503, 504], | |
| allowed_methods=frozenset(["GET", "POST"]), | |
| raise_on_status=False, | |
| ) | |
| adapter = HTTPAdapter(max_retries=retry) | |
| s.mount("http://", adapter) | |
| s.mount("https://", adapter) | |
| return s | |
| def warm_up(sess: requests.Session, base: str, progress_cb=None, total_budget: int = 25): | |
| start = time.time() | |
| paths = ["/", "/health"] | |
| for i, path in enumerate(paths, 1): | |
| try: | |
| _ = sess.get(f"{base.rstrip('/')}{path}", timeout=8) | |
| except Exception: | |
| pass | |
| if progress_cb: | |
| progress_cb(min(int(i/len(paths)*18), 18)) # 0-18% | |
| if time.time() - start > total_budget: | |
| break | |
| time.sleep(0.4) | |
| def fetch_bytes(sess: requests.Session, base: str, url: str) -> bytes: | |
| full = url if url.startswith("http") else f"{base.rstrip('/')}{url}" | |
| r = sess.get(full, timeout=90) | |
| r.raise_for_status() | |
| return r.content | |
| def post_analyze(sess: requests.Session, base: str, _file, _url: str, progress_cb=None) -> dict: | |
| data = { | |
| "use_data_uri": "true" if st.session_state.use_data_uri else "false", | |
| "lang": st.session_state.lang, | |
| } | |
| if DEV_MODE and st.session_state.prompt_override.strip(): | |
| data["prompt"] = st.session_state.prompt_override.strip() | |
| files = None | |
| if _file is not None: | |
| # read into BytesIO for repeatable upload | |
| content = _file.read() | |
| buf = io.BytesIO(content) | |
| mime = getattr(_file, "type", None) or "application/octet-stream" | |
| files = {"file": (_file.name, buf, mime)} | |
| if progress_cb: progress_cb(30) # after upload prep | |
| elif _url.strip(): | |
| data["audio_url"] = _url.strip() | |
| if not files and "audio_url" not in data: | |
| return {"ok": False, "error": T["need_input"]} | |
| url = f"{base.rstrip('/')}/analyze" | |
| deadline = time.time() + st.session_state.max_wait | |
| last_err = None | |
| while True: | |
| try: | |
| if progress_cb: progress_cb(35) # request start | |
| resp = sess.post(url, data=data, files=files, timeout=240) | |
| if progress_cb: progress_cb(78) # server responded | |
| resp.raise_for_status() | |
| if progress_cb: progress_cb(92) # parse | |
| return resp.json() | |
| except Exception as e: | |
| last_err = e | |
| if time.time() > deadline or not st.session_state.auto_retry: | |
| return {"ok": False, "error": f"{T['error_generic']} ({last_err})"} | |
| time.sleep(1.2) | |
| # ---------------- Actions ---------------- | |
| sess = build_session(st.session_state.auto_retry) | |
| if health: | |
| try: | |
| r = sess.get(f"{st.session_state.api_base.rstrip('/')}/health", timeout=10) | |
| st.success(T["health_ok"] if r.ok else T["health_fail"]) | |
| if DEV_MODE and r.ok: | |
| st.code(json.dumps(r.json(), ensure_ascii=False, indent=2), language="json") | |
| except Exception: | |
| st.error(T["health_fail"]) | |
| if reset: | |
| st.session_state.res = None | |
| st.rerun() | |
| # Progress widgets (player stays above) | |
| progress_box = st.container() | |
| timeline = progress_box.expander(T["timeline_title"], expanded=True) | |
| p_text = progress_box.empty() | |
| p_bar = progress_box.progress(0) | |
| def progress_update(pct: int, label: str = ""): | |
| p_bar.progress(min(max(pct, 0), 100)) | |
| if label: | |
| p_text.write(label) | |
| if go: | |
| if not st.session_state.api_base: | |
| st.error(T["need_backend"]) | |
| elif not file and not audio_url.strip(): | |
| st.error(T["need_input"]) | |
| else: | |
| # timeline | |
| with timeline: | |
| st.write(f"1) {T['step_prepare']}") | |
| st.write(f"2) {T['step_upload']}") | |
| st.write(f"3) {T['step_process']}") | |
| st.write(f"4) {T['step_summarize']}") | |
| # warmup (optional) | |
| if st.session_state.warmup: | |
| progress_update(5, T["warming"]) | |
| warm_up(sess, st.session_state.api_base, progress_cb=lambda v: progress_update(v, T["warming"])) | |
| # analyze | |
| progress_update(22, T["analyzing"]) | |
| res = post_analyze( | |
| sess, st.session_state.api_base, file, audio_url, | |
| progress_cb=lambda v: progress_update(v, T["analyzing"]) | |
| ) | |
| st.session_state.res = res | |
| progress_update(100, "") | |
| # ---------------- Results ---------------- | |
| res = st.session_state.res | |
| if res: | |
| if not res.get("ok", True) and res.get("error"): | |
| st.error(T["error_generic"]) | |
| st.write(res.get("error")) | |
| st.info(T["try_again"]) | |
| if DEV_MODE and st.session_state.show_debug and res.get("fal_debug"): | |
| st.expander("Debug").code(json.dumps(res["fal_debug"], ensure_ascii=False, indent=2), language="json") | |
| else: | |
| st.subheader(T["result"]) | |
| # Summary for drivers | |
| comm = (res.get("commentary") or "").strip() | |
| if comm: | |
| st.markdown(f"**{T['commentary']}**") | |
| st.write(comm) | |
| # Detailed findings (JSON). This is the structured diagnosis — no provider names. | |
| parsed = res.get("fal_parsed") | |
| if parsed: | |
| st.markdown(f"**{T['json']}**") | |
| st.json(parsed) | |
| # Downloads | |
| dls = res.get("downloads") or {} | |
| if any(dls.get(k) for k in ("summary_txt", "fal_parsed_json", "fal_raw_txt")): | |
| st.markdown(f"**{T['artifacts']}**") | |
| col_a, col_b, col_c = st.columns(3) | |
| if dls.get("summary_txt"): | |
| try: | |
| data = fetch_bytes(sess, st.session_state.api_base, dls["summary_txt"]) | |
| col_a.download_button(label=T["download_summary"], data=data, file_name="ozet.txt" if st.session_state.lang=="tr" else "summary.txt") | |
| except Exception: | |
| col_a.error("✖") | |
| if dls.get("fal_parsed_json"): | |
| try: | |
| data = fetch_bytes(sess, st.session_state.api_base, dls["fal_parsed_json"]) | |
| col_b.download_button(label=T["download_parsed"], data=data, file_name="detay.json" if st.session_state.lang=="tr" else "details.json", mime="application/json") | |
| except Exception: | |
| col_b.error("✖") | |
| if dls.get("fal_raw_txt"): | |
| try: | |
| data = fetch_bytes(sess, st.session_state.api_base, dls["fal_raw_txt"]) | |
| col_c.download_button(label=T["download_raw"], data=data, file_name="ham.txt" if st.session_state.lang=="tr" else "raw.txt") | |
| except Exception: | |
| col_c.error("✖") | |
| # ---------------- Footer ---------------- | |
| st.caption("© 2025 Diag Studio — Araç Motor Ses Analizi") | |