# streamlit_app.py — LamboVision · AI Car Studio (native-only UI, no padding)
# Flow: Step 1 Gallery (Background Cleaned) → Step 2 AI Configurated (config) → Step 3 AI Generated Video
# Tiles use a uniform aspect. Images/videos fit inside (no overflow). “Generate Video From These Results” button included.
import os, io, base64, zipfile, requests, streamlit as st
from PIL import Image
# ===== Theme (dark + Lamborghini orange) =====
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", "#161a1f")
st._config.set_option("theme.textColor", "#e7e9ed")
except Exception:
pass
# ================= Env & API =================
API_BASE = (os.getenv("AI_LamboVision_API", "http://127.0.0.1:7860") or "http://127.0.0.1:7860").strip().strip("'\"").rstrip("/")
HF_TOKEN = (os.getenv("AI_LamboVision_TOKEN", "") or "").strip().strip("'\"")
HEADERS = {"Authorization": f"Bearer {HF_TOKEN}"} if HF_TOKEN else {}
VIDEO_MAX_PX_DEFAULT = int(os.getenv("VIDEO_MAX_PX", "720"))
VIDEO_RES_OPTIONS = ["512P", "768P"]
VIDEO_DUR_OPTIONS = ["6", "10"]
ASPECT_OPTIONS = ["3:4", "9:16", "1:1", "16:9"] # default 3:4
# ================= Session =================
if "results" not in st.session_state:
st.session_state["results"] = None
if "job_counter" not in st.session_state:
st.session_state["job_counter"] = 0
if "video_ui_max_px" not in st.session_state:
st.session_state["video_ui_max_px"] = VIDEO_MAX_PX_DEFAULT
# ================= Helpers =================
def _needs_auth(url: str) -> bool:
return url.startswith(API_BASE) or url.startswith("outputs/") or url.startswith("/outputs/")
def _abs_backend_url(url_or_path: str) -> str:
if not url_or_path:
return ""
if url_or_path.startswith(("http://","https://")):
return url_or_path
if url_or_path.startswith("/outputs/"):
return f"{API_BASE}{url_or_path}"
if url_or_path.startswith("outputs/"):
return f"{API_BASE}/{url_or_path}"
return f"{API_BASE}/{url_or_path.lstrip('/')}"
def backend_ok() -> bool:
try:
r = requests.get(f"{API_BASE}/health", headers=HEADERS, timeout=10)
r.raise_for_status()
return True
except Exception:
return False
def post_image_edit(data: dict, files_payload):
r = requests.post(f"{API_BASE}/v1/image/edit", data=data, files=files_payload or None,
headers=HEADERS, timeout=600)
r.raise_for_status()
return r.json()
def post_video(image_url: str, duration="6", resolution="768P", preview_aspect="3:4", prompt_text="", prompt_optimizer=False):
payload = {
"image_url": image_url,
"duration": duration,
"resolution": resolution,
"preview_aspect": preview_aspect,
"prompt": prompt_text or "",
"prompt_optimizer": "true" if prompt_optimizer else "false",
}
r = requests.post(f"{API_BASE}/v1/video/from-image", data=payload, headers=HEADERS, timeout=600)
r.raise_for_status()
return r.json()
@st.cache_data(ttl=300, show_spinner=False)
def fetch_bytes(url_or_path: str):
if not url_or_path: return None
try:
url = _abs_backend_url(url_or_path)
headers = HEADERS if _needs_auth(url) else None
r = requests.get(url, headers=headers, timeout=180)
r.raise_for_status()
return r.content
except Exception:
return None
def _sniff_mime_from_bytes(b: bytes) -> str:
try:
fmt = (Image.open(io.BytesIO(b)).format or "JPEG").lower()
return {"jpeg":"image/jpeg","jpg":"image/jpeg","png":"image/png","webp":"image/webp"}.get(fmt,"image/jpeg")
except Exception:
return "image/jpeg"
def image_bytes_to_data_url(b: bytes, mime: str | None = None) -> str:
if not mime:
mime = _sniff_mime_from_bytes(b)
enc = base64.b64encode(b).decode("ascii")
return f"data:{mime};base64,{enc}"
@st.cache_data(ttl=300, show_spinner=False)
def image_size_from_bytes(b: bytes):
try:
im = Image.open(io.BytesIO(b)); return im.size
except Exception:
return None
def infer_ext(data: bytes) -> str:
try:
im = Image.open(io.BytesIO(data))
fmt = (im.format or "JPEG").lower()
return "." + {"jpeg":"jpg","jpg":"jpg","png":"png","webp":"webp"}.get(fmt,"jpg")
except Exception:
if data[:4] == b"\x00\x00\x00\x18" or data[4:8] == b"ftyp":
return ".mp4"
return ".bin"
def make_zip(named_bytes: list[tuple[str, bytes]]) -> bytes:
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", compression=zipfile.ZIP_DEFLATED) as z:
for fname, b in named_bytes:
if b: z.writestr(fname, b)
buf.seek(0); return buf.read()
def _parse_aspect_str(s: str):
try:
a, b = s.split(":"); return max(1, int(a)), max(1, int(b))
except Exception:
return (3, 4)
def inject_base_css(card_fit_mode="contain"):
st.markdown(f"""
""", unsafe_allow_html=True)
def inject_preview_css(aspect: str):
aw, ah = _parse_aspect_str(aspect)
st.markdown(f"""
""", unsafe_allow_html=True)
def show_tile_card_display_download(
col, title,
display_url: str | None = None,
display_bytes: bytes | None = None,
download_url: str | None = None,
download_bytes: bytes | None = None,
filename_stub: str = "image",
key_prefix: str = "",
):
col.markdown(f'
{title}
', unsafe_allow_html=True)
if not display_url and not display_bytes:
col.markdown('', unsafe_allow_html=True)
col.caption("—"); return None, None
if display_bytes is not None:
src = image_bytes_to_data_url(display_bytes)
col.markdown(f'