Spaces:
Sleeping
Sleeping
File size: 9,995 Bytes
ff119bb a4ebbec ff119bb a4ebbec 7d56bcc ff119bb 7d56bcc a4ebbec ff119bb a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc ff119bb a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc ff119bb 7d56bcc ff119bb a4ebbec 7d56bcc ff119bb 7d56bcc ff119bb 7d56bcc ff119bb 7d56bcc a4ebbec ff119bb 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec ff119bb 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec 7d56bcc a4ebbec ff119bb a4ebbec 7d56bcc ff119bb a4ebbec 7d56bcc a4ebbec ff119bb a4ebbec 7d56bcc a4ebbec ff119bb 7d56bcc ff119bb a4ebbec |
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 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 |
import sys
import os
import subprocess # For calling generate.py
import tempfile # For handling temporary image files
from typing import Optional
from PIL import Image as PILImage
import gradio as gr
import time # For timing
# Add the cloned nanoVLM directory to Python's system path
NANOVLM_REPO_PATH = "/app/nanoVLM"
if NANOVLM_REPO_PATH not in sys.path:
print(f"DEBUG: Adding {NANOVLM_REPO_PATH} to sys.path")
sys.path.insert(0, NANOVLM_REPO_PATH)
print(f"DEBUG: Python sys.path: {sys.path}")
print(f"DEBUG: Gradio version: {gr.__version__}") # Log Gradio version
GENERATE_SCRIPT_PATH = "/app/nanoVLM/generate.py"
MODEL_REPO_ID = "lusxvr/nanoVLM-222M"
print(f"DEBUG: Using generate.py script at: {GENERATE_SCRIPT_PATH}")
print(f"DEBUG: Using model repo ID: {MODEL_REPO_ID}")
def call_generate_script(image_path: str, prompt_text: str) -> str:
print(f"\n--- DEBUG (call_generate_script) ---")
print(f"Timestamp: {time.strftime('%Y-%m-%d %H:%M:%S')}")
print(f"Calling with image_path='{image_path}', prompt='{prompt_text}'")
# Arguments for nanoVLM's generate.py
# Using low max_new_tokens for CPU testing.
cmd_args = [
"python", "-u", GENERATE_SCRIPT_PATH, # -u for unbuffered output
"--hf_model", MODEL_REPO_ID,
"--image_path", image_path, # Corrected: nanoVLM generate.py uses --image_path
"--prompt", prompt_text,
"--num_samples", "1", # Corrected: Corresponds to --generations
"--max_new_tokens", "30", # Keep it low for testing
"--device", "cpu" # Explicitly set device for generate.py
# Optional args for generate.py:
# "--temperature", "0.7",
# "--top_k", "50"
]
print(f"Executing command: {' '.join(cmd_args)}")
# Realistic timeout for the subprocess. HF Spaces free tier usually times out requests around 60s.
# Set this shorter to catch issues within app.py.
SCRIPT_TIMEOUT_SECONDS = 55
start_time = time.time()
process_details = "Process details not available." # Placeholder
try:
process = subprocess.run(
cmd_args,
capture_output=True,
text=True,
check=False, # Set to False to manually check returncode and log output
timeout=SCRIPT_TIMEOUT_SECONDS
)
process_details = f"PID {process.pid if hasattr(process, 'pid') else 'N/A'}"
duration = time.time() - start_time
print(f"Subprocess ({process_details}) finished in {duration:.2f} seconds.")
print(f"generate.py RETURN CODE: {process.returncode}")
stdout = process.stdout.strip() if process.stdout else "[No STDOUT from generate.py]"
stderr = process.stderr.strip() if process.stderr else "[No STDERR from generate.py]"
print(f"---------- generate.py STDOUT ({process_details}) START ----------\n{stdout}\n---------- generate.py STDOUT ({process_details}) END ----------")
if stderr or process.returncode != 0:
print(f"---------- generate.py STDERR ({process_details}) START ----------\n{stderr}\n---------- generate.py STDERR ({process_details}) END ----------")
if process.returncode != 0:
error_message = f"Error: Generation script failed (code {process.returncode})."
if "out of memory" in stderr.lower(): error_message += " Potential OOM in script."
print(error_message) # Log it before returning
return error_message + f" See Space logs for full STDOUT/STDERR from script ({process_details})."
# --- Parse the output from nanoVLM's generate.py ---
# Expected format:
# Outputs:
# > Sample 1: <generated text>
output_lines = stdout.splitlines()
generated_text = "[No parsable output from generate.py]" # Default
found_output_line = False
for line_idx, line in enumerate(output_lines):
stripped_line = line.strip()
# print(f"Parsing STDOUT line {line_idx}: '{stripped_line}'") # Can be very verbose
if stripped_line.startswith("> Sample 1:") or stripped_line.startswith(">> Generation 1:"):
prefix_to_remove = ""
if stripped_line.startswith("> Sample 1:"): prefix_to_remove = "> Sample 1:"
elif stripped_line.startswith(">> Generation 1: "): prefix_to_remove = ">> Generation 1: " # Note double space
elif stripped_line.startswith(">> Generation 1: "): prefix_to_remove = ">> Generation 1: " # Note single space
if prefix_to_remove:
generated_text = stripped_line.replace(prefix_to_remove, "", 1).strip()
found_output_line = True
print(f"Parsed generated text: '{generated_text}'")
break
if not found_output_line:
print(f"Could not find 'Sample 1' or 'Generation 1' line in generate.py output.")
# Return a snippet of STDOUT if parsing fails, to help debug output format
generated_text = f"[Parsing failed] STDOUT (first 200 chars): {stdout[:200]}"
print(f"Returning parsed text: '{generated_text}'")
return generated_text
except subprocess.TimeoutExpired as e:
duration = time.time() - start_time
print(f"ERROR: generate.py ({process_details}) timed out after {duration:.2f} seconds (limit: {SCRIPT_TIMEOUT_SECONDS}s).")
stdout_on_timeout = e.stdout.strip() if e.stdout else "[No STDOUT on timeout]"
stderr_on_timeout = e.stderr.strip() if e.stderr else "[No STDERR on timeout]"
print(f"STDOUT on timeout:\n{stdout_on_timeout}")
print(f"STDERR on timeout:\n{stderr_on_timeout}")
return f"Error: Generation script timed out after {SCRIPT_TIMEOUT_SECONDS}s. Model loading and generation may be too slow for CPU."
except Exception as e:
duration = time.time() - start_time
print(f"ERROR: An unexpected error occurred ({process_details}) after {duration:.2f}s: {type(e).__name__} - {e}")
import traceback; traceback.print_exc()
return f"Unexpected error calling script: {str(e)}"
finally:
print(f"--- END (call_generate_script) ---")
def gradio_interface_fn(image_input_pil: Optional[PILImage.Image], prompt_input_str: Optional[str]) -> str:
print(f"\nDEBUG (gradio_interface_fn): Timestamp: {time.strftime('%Y-%m-%d %H:%M:%S')}")
print(f"Received prompt: '{prompt_input_str}', Image type: {type(image_input_pil)}")
if image_input_pil is None:
return "Please upload an image."
cleaned_prompt = prompt_input_str.strip() if prompt_input_str else ""
if not cleaned_prompt:
return "Please provide a non-empty prompt."
tmp_image_path = None
try:
if image_input_pil.mode != "RGB":
print(f"Converting image from {image_input_pil.mode} to RGB.")
image_input_pil = image_input_pil.convert("RGB")
with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp_image_file:
image_input_pil.save(tmp_image_file, format="JPEG")
tmp_image_path = tmp_image_file.name
print(f"Temporary image saved to: {tmp_image_path}")
result_text = call_generate_script(tmp_image_path, cleaned_prompt)
print(f"Result from call_generate_script: '{result_text}'")
return result_text
except Exception as e:
print(f"ERROR (gradio_interface_fn): Error processing image or calling script: {type(e).__name__} - {e}")
import traceback; traceback.print_exc()
return f"An error occurred in Gradio interface function: {str(e)}"
finally:
if tmp_image_path and os.path.exists(tmp_image_path):
try:
os.remove(tmp_image_path)
print(f"Temporary image {tmp_image_path} removed.")
except Exception as e_remove:
print(f"WARN: Could not remove temporary image {tmp_image_path}: {e_remove}")
print(f"DEBUG (gradio_interface_fn): Exiting.")
# --- Gradio Interface Definition ---
description_md = """
## nanoVLM-222M Interactive Demo (via generate.py)
Upload an image and type a prompt. This interface calls the `generate.py` script from
`huggingface/nanoVLM` under the hood to perform inference.
**Note:** Each request re-loads the model via the script, so it might be slow on CPU.
"""
print("DEBUG: Defining Gradio interface...")
iface = None
try:
iface = gr.Interface(
fn=gradio_interface_fn,
inputs=[
gr.Image(type="pil", label="Upload Image"),
gr.Textbox(label="Your Prompt / Question", info="e.g., 'describe this image in detail'")
],
outputs=gr.Textbox(label="Generated Text", show_copy_button=True, lines=5),
title="nanoVLM-222M Demo (via Script)",
description=description_md,
allow_flagging="never"
)
print("DEBUG: Gradio interface defined successfully.")
except Exception as e:
print(f"CRITICAL ERROR defining Gradio interface: {e}")
import traceback; traceback.print_exc()
# --- Launch Gradio App ---
if __name__ == "__main__":
print("DEBUG: Entered __main__ block for Gradio launch.")
if not os.path.exists(GENERATE_SCRIPT_PATH):
print(f"CRITICAL ERROR: The script {GENERATE_SCRIPT_PATH} was not found. Cannot launch app.")
iface = None
if iface is not None:
print("DEBUG: Attempting to launch Gradio interface...")
try:
iface.launch(server_name="0.0.0.0", server_port=7860)
print("DEBUG: Gradio launch command issued. UI should be accessible.")
except Exception as e:
print(f"CRITICAL ERROR launching Gradio interface: {e}")
import traceback; traceback.print_exc()
else:
print("CRITICAL ERROR: Gradio interface (iface) is None or not defined. Cannot launch.") |