Spaces:
Sleeping
Sleeping
File size: 5,269 Bytes
8f155fd 3f62288 8f155fd e30538d 8f155fd e30538d 8f155fd e30538d 8f155fd e30538d 8f155fd e30538d 8f155fd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | """Streamlit runtime helpers for long-running jobs on Hugging Face Spaces."""
from __future__ import annotations
import os
import time
from typing import Any, Callable, Optional
def is_huggingface_space() -> bool:
"""Detect Hugging Face Spaces runtime."""
return bool(
os.getenv("SPACE_ID")
or os.getenv("HF_HOME")
or os.getenv("SYSTEM")
)
def embed_html(content: str, height: int = 0) -> Any:
"""Render raw HTML/JS in an iframe.
Uses ``st.iframe`` (Streamlit >= 1.5x) and falls back to the deprecated
``st.components.v1.html`` on older versions. ``st.iframe`` rejects a height
of 0, so script-only snippets get a 1px frame instead.
"""
import streamlit as st
if hasattr(st, "iframe"):
size = height if isinstance(height, int) and height > 0 else 1
return st.iframe(content, height=size)
from streamlit.components.v1 import html as _components_html
return _components_html(content, height=height)
class AnalysisProgressReporter:
"""Keeps the Streamlit websocket alive during long Gemini calls."""
def __init__(
self,
progress_bar: Any = None,
status_text: Any = None,
time_estimate: Any = None,
step_indicator: Any = None,
working_indicator: Any = None,
stream_indicator: Any = None,
base_progress: float = 70.0,
progress_span: float = 25.0,
):
self.progress_bar = progress_bar
self.status_text = status_text
self.time_estimate = time_estimate
self.step_indicator = step_indicator
self.working_indicator = working_indicator
self.stream_indicator = stream_indicator
self.base_progress = base_progress
self.progress_span = progress_span
self._last_flush = 0.0
self._total_steps = 1
def on_phase(self, message: str, fraction: float = 0.0) -> None:
self._update(
progress=self.base_progress + (self.progress_span * fraction),
status=message,
working=message,
)
def on_step_start(self, step_number: int, step_title: str, total_steps: int) -> None:
self._total_steps = max(total_steps, 1)
fraction = max(0.0, (step_number - 1) / self._total_steps)
self.on_phase(
f"🤖 **AI Step {step_number}/{total_steps}:** {step_title}...",
fraction=fraction,
)
def on_stream_chunk(self, step_number: int, chars_received: int) -> None:
now = time.time()
if now - self._last_flush < 0.75:
return
self._last_flush = now
fraction = min(0.99, ((step_number - 1) + min(chars_received / 4000.0, 0.95)) / self._total_steps)
progress = int(self.base_progress + (self.progress_span * fraction))
stream_msg = f"Receiving AI output... {chars_received:,} characters"
self._update(
progress=progress,
status=f"🤖 **AI Step {step_number}/{self._total_steps}:** {stream_msg}",
working=stream_msg,
stream=stream_msg,
)
def on_step_complete(self, step_number: int) -> None:
fraction = min(1.0, step_number / self._total_steps)
self.on_phase(
f"✅ **AI Step {step_number}/{self._total_steps}** complete",
fraction=fraction,
)
def _update(
self,
progress: float,
status: str,
working: Optional[str] = None,
stream: Optional[str] = None,
) -> None:
try:
if self.progress_bar is not None:
self.progress_bar.progress(min(max(int(progress), 0), 100))
if self.status_text is not None:
self.status_text.markdown(status)
if self.working_indicator is not None and working:
self.working_indicator.markdown(f"🔄 **Processing:** {working}")
if self.stream_indicator is not None and stream:
self.stream_indicator.caption(stream)
if self.time_estimate is not None:
self.time_estimate.markdown(
"⏱️ Analysis is still running — keep this tab open."
)
if self.step_indicator is not None:
self.step_indicator.markdown("📋 AI analysis in progress")
except Exception:
pass
def make_progress_callback(
progress_bar: Any = None,
status_text: Any = None,
time_estimate: Any = None,
step_indicator: Any = None,
working_indicator: Any = None,
stream_indicator: Any = None,
) -> AnalysisProgressReporter:
return AnalysisProgressReporter(
progress_bar=progress_bar,
status_text=status_text,
time_estimate=time_estimate,
step_indicator=step_indicator,
working_indicator=working_indicator,
stream_indicator=stream_indicator,
)
def is_connection_abort_error(error: Exception | str) -> bool:
message = str(error).lower()
return any(
token in message
for token in (
"bodystreambuffer",
"aborted",
"connection reset",
"connection interrupted",
"client disconnected",
"stream closed",
)
) |