CPU

app.py

@@ -1,31 +1,56 @@
 import gradio as gr
 from transformers import AutoModel, AutoTokenizer
 import torch
-import spaces
 import os
 import sys
 import tempfile
 import shutil
 from PIL import Image, ImageDraw, ImageFont, ImageOps
-import fitz
+import fitz  # PyMuPDF
 import re
-import warnings
 import numpy as np
 import base64
 from io import StringIO, BytesIO
 
-MODEL_NAME = 'deepseek-ai/DeepSeek-OCR'
+"""
+CPU-friendly version of the DeepSeekOCR Space app.
+
+Changes vs GPU version:
+- Removed `spaces` and @spaces.GPU decorators.
+- Removed FlashAttention2 forcing (`_attn_implementation='flash_attention_2'`).
+- Removed `.cuda()`; model runs on CPU.
+- Uses torch.float32 (CPU-safe). This will be SLOW and may use lots of RAM.
+"""
+
+# Force CPU usage (helps avoid accidental CUDA paths)
+os.environ["CUDA_VISIBLE_DEVICES"] = ""
+
+MODEL_NAME = "deepseek-ai/DeepSeek-OCR"
+
+# Optional: limit CPU threads if your machine spikes (tweak as you like)
+try:
+    torch.set_num_threads(max(1, min(8, os.cpu_count() or 1)))
+except Exception:
+    pass
 
 tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME, trust_remote_code=True)
-model = AutoModel.from_pretrained(MODEL_NAME, _attn_implementation='flash_attention_2', torch_dtype=torch.bfloat16, trust_remote_code=True, use_safetensors=True)
-model = model.eval().cuda()
+
+# CPU-safe load: no flash-attn, float32
+model = AutoModel.from_pretrained(
+    MODEL_NAME,
+    torch_dtype=torch.float32,
+    trust_remote_code=True,
+    use_safetensors=True,
+)
+model = model.eval()  # keep on CPU
 
 MODEL_CONFIGS = {
-    "Gundam": {"base_size": 1024, "image_size": 640, "crop_mode": True},
+    # On CPU, prefer smaller modes for speed/memory.
     "Tiny": {"base_size": 512, "image_size": 512, "crop_mode": False},
     "Small": {"base_size": 640, "image_size": 640, "crop_mode": False},
     "Base": {"base_size": 1024, "image_size": 1024, "crop_mode": False},
-    "Large": {"base_size": 1280, "image_size": 1280, "crop_mode": False}
+    "Gundam": {"base_size": 1024, "image_size": 640, "crop_mode": True},
+    "Large": {"base_size": 1280, "image_size": 1280, "crop_mode": False},
 }
 
 TASK_PROMPTS = {
@@ -33,7 +58,7 @@ TASK_PROMPTS = {
     "📝 Free OCR": {"prompt": "<image>\nFree OCR.", "has_grounding": False},
     "📍 Locate": {"prompt": "<image>\nLocate <|ref|>text<|/ref|> in the image.", "has_grounding": True},
     "🔍 Describe": {"prompt": "<image>\nDescribe this image in detail.", "has_grounding": False},
-    "✏️ Custom": {"prompt": "", "has_grounding": False}
+    "✏️ Custom": {"prompt": "", "has_grounding": False},
 }
 
 def extract_grounding_references(text):
@@ -44,39 +69,60 @@ def draw_bounding_boxes(image, refs, extract_images=False):
     img_w, img_h = image.size
     img_draw = image.copy()
     draw = ImageDraw.Draw(img_draw)
-    overlay = Image.new('RGBA', img_draw.size, (0, 0, 0, 0))
+    overlay = Image.new("RGBA", img_draw.size, (0, 0, 0, 0))
     draw2 = ImageDraw.Draw(overlay)
-    font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 30)
+    # Fallback font if path doesn't exist
+    font_path = "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf"
+    try:
+        font = ImageFont.truetype(font_path, 30)
+    except Exception:
+        font = ImageFont.load_default()
+
     crops = []
-    
     color_map = {}
     np.random.seed(42)
 
     for ref in refs:
         label = ref[1]
         if label not in color_map:
-            color_map[label] = (np.random.randint(50, 255), np.random.randint(50, 255), np.random.randint(50, 255))
+            color_map[label] = (
+                int(np.random.randint(50, 255)),
+                int(np.random.randint(50, 255)),
+                int(np.random.randint(50, 255)),
+            )
 
         color = color_map[label]
-        coords = eval(ref[2])
+        try:
+            coords = eval(ref[2])
+        except Exception:
+            continue
         color_a = color + (60,)
-        
+
         for box in coords:
-            x1, y1, x2, y2 = int(box[0]/999*img_w), int(box[1]/999*img_h), int(box[2]/999*img_w), int(box[3]/999*img_h)
-            
-            if extract_images and label == 'image':
+            x1, y1, x2, y2 = (
+                int(box[0] / 999 * img_w),
+                int(box[1] / 999 * img_h),
+                int(box[2] / 999 * img_w),
+                int(box[3] / 999 * img_h),
+            )
+
+            if extract_images and label == "image":
                 crops.append(image.crop((x1, y1, x2, y2)))
-            
-            width = 5 if label == 'title' else 3
+
+            width = 5 if label == "title" else 3
             draw.rectangle([x1, y1, x2, y2], outline=color, width=width)
             draw2.rectangle([x1, y1, x2, y2], fill=color_a)
-            
-            text_bbox = draw.textbbox((0, 0), label, font=font)
-            tw, th = text_bbox[2] - text_bbox[0], text_bbox[3] - text_bbox[1]
+
+            try:
+                text_bbox = draw.textbbox((0, 0), label, font=font)
+                tw, th = text_bbox[2] - text_bbox[0], text_bbox[3] - text_bbox[1]
+            except Exception:
+                tw, th = (len(label) * 10, 20)
+
             ty = max(0, y1 - 20)
             draw.rectangle([x1, ty, x1 + tw + 4, ty + th + 4], fill=color)
             draw.text((x1 + 2, ty + 2), label, font=font, fill=(255, 255, 255))
-    
+
     img_draw.paste(overlay, (0, 0), overlay)
     return img_draw, crops
 
@@ -86,17 +132,17 @@ def clean_output(text, include_images=False):
     pattern = r'(<\|ref\|>(.*?)<\|/ref\|><\|det\|>(.*?)<\|/det\|>)'
     matches = re.findall(pattern, text, re.DOTALL)
     img_num = 0
-    
+
     for match in matches:
-        if '<|ref|>image<|/ref|>' in match[0]:
+        if "<|ref|>image<|/ref|>" in match[0]:
             if include_images:
-                text = text.replace(match[0], f'\n\n**[Figure {img_num + 1}]**\n\n', 1)
+                text = text.replace(match[0], f"\n\n**[Figure {img_num + 1}]**\n\n", 1)
                 img_num += 1
             else:
-                text = text.replace(match[0], '', 1)
+                text = text.replace(match[0], "", 1)
         else:
-            text = re.sub(rf'(?m)^[^\n]*{re.escape(match[0])}[^\n]*\n?', '', text)
-    
+            text = re.sub(rf"(?m)^[^\n]*{re.escape(match[0])}[^\n]*\n?", "", text)
+
     return text.strip()
 
 def embed_images(markdown, crops):
@@ -106,123 +152,160 @@ def embed_images(markdown, crops):
         buf = BytesIO()
         img.save(buf, format="PNG")
         b64 = base64.b64encode(buf.getvalue()).decode()
-        markdown = markdown.replace(f'**[Figure {i + 1}]**', f'\n\n![Figure {i + 1}](data:image/png;base64,{b64})\n\n', 1)
+        markdown = markdown.replace(
+            f"**[Figure {i + 1}]**",
+            f"\n\n![Figure {i + 1}](data:image/png;base64,{b64})\n\n",
+            1,
+        )
     return markdown
 
-@spaces.GPU(duration=60)
+def _infer_with_model(prompt, jpg_path, out_dir, base_size, image_size, crop_mode):
+    # DeepSeek model prints progress to stdout; capture it like original.
+    stdout = sys.stdout
+    sys.stdout = StringIO()
+    try:
+        model.infer(
+            tokenizer=tokenizer,
+            prompt=prompt,
+            image_file=jpg_path,
+            output_path=out_dir,
+            base_size=base_size,
+            image_size=image_size,
+            crop_mode=crop_mode,
+        )
+        raw = sys.stdout.getvalue()
+    finally:
+        sys.stdout = stdout
+    return raw
+
 def process_image(image, mode, task, custom_prompt):
     if image is None:
-        return " Error Upload image", "", "", None, []
+        return "Error: Upload image", "", "", None, []
+
     if task in ["✏️ Custom", "📍 Locate"] and not custom_prompt.strip():
-        return "Enter prompt", "", "", None, []
-    
-    if image.mode in ('RGBA', 'LA', 'P'):
-        image = image.convert('RGB')
+        return "Error: Enter prompt", "", "", None, []
+
+    if image.mode in ("RGBA", "LA", "P"):
+        image = image.convert("RGB")
     image = ImageOps.exif_transpose(image)
-    
+
     config = MODEL_CONFIGS[mode]
-    
+
     if task == "✏️ Custom":
         prompt = f"<image>\n{custom_prompt.strip()}"
-        has_grounding = '<|grounding|>' in custom_prompt
+        has_grounding = "<|grounding|>" in custom_prompt
     elif task == "📍 Locate":
         prompt = f"<image>\nLocate <|ref|>{custom_prompt.strip()}<|/ref|> in the image."
         has_grounding = True
     else:
         prompt = TASK_PROMPTS[task]["prompt"]
         has_grounding = TASK_PROMPTS[task]["has_grounding"]
-    
-    tmp = tempfile.NamedTemporaryFile(delete=False, suffix='.jpg')
-    image.save(tmp.name, 'JPEG', quality=95)
+
+    tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".jpg")
+    image.save(tmp.name, "JPEG", quality=95)
     tmp.close()
     out_dir = tempfile.mkdtemp()
-    
-    stdout = sys.stdout
-    sys.stdout = StringIO()
-    
-    model.infer(tokenizer=tokenizer, prompt=prompt, image_file=tmp.name, output_path=out_dir,
-                base_size=config["base_size"], image_size=config["image_size"], crop_mode=config["crop_mode"])
-    
-    result = '\n'.join([l for l in sys.stdout.getvalue().split('\n') 
-                        if not any(s in l for s in ['image:', 'other:', 'PATCHES', '====', 'BASE:', '%|', 'torch.Size'])]).strip()
-    sys.stdout = stdout
-    
-    os.unlink(tmp.name)
-    shutil.rmtree(out_dir, ignore_errors=True)
-    
-    if not result:
-        return "No text", "", "", None, []
-    
-    cleaned = clean_output(result, False)
-    markdown = clean_output(result, True)
-    
-    img_out = None
-    crops = []
-    
-    if has_grounding and '<|ref|>' in result:
-        refs = extract_grounding_references(result)
-        if refs:
-            img_out, crops = draw_bounding_boxes(image, refs, True)
-    
-    markdown = embed_images(markdown, crops)
-    
-    return cleaned, markdown, result, img_out, crops
-
-@spaces.GPU(duration=60)
+
+    try:
+        raw_stdout = _infer_with_model(
+            prompt=prompt,
+            jpg_path=tmp.name,
+            out_dir=out_dir,
+            base_size=config["base_size"],
+            image_size=config["image_size"],
+            crop_mode=config["crop_mode"],
+        )
+
+        # Filter noisy lines
+        result = "\n".join(
+            [
+                l
+                for l in raw_stdout.split("\n")
+                if not any(
+                    s in l
+                    for s in [
+                        "image:",
+                        "other:",
+                        "PATCHES",
+                        "====",
+                        "BASE:",
+                        "%|",
+                        "torch.Size",
+                    ]
+                )
+            ]
+        ).strip()
+
+        if not result:
+            return "No text", "", "", None, []
+
+        cleaned = clean_output(result, False)
+        markdown = clean_output(result, True)
+
+        img_out = None
+        crops = []
+
+        if has_grounding and "<|ref|>" in result:
+            refs = extract_grounding_references(result)
+            if refs:
+                img_out, crops = draw_bounding_boxes(image, refs, True)
+
+        markdown = embed_images(markdown, crops)
+        return cleaned, markdown, result, img_out, crops
+
+    except Exception as e:
+        return f"Runtime error: {type(e).__name__}: {e}", "", "", None, []
+    finally:
+        try:
+            os.unlink(tmp.name)
+        except Exception:
+            pass
+        shutil.rmtree(out_dir, ignore_errors=True)
+
 def process_pdf(path, mode, task, custom_prompt, page_num):
     doc = fitz.open(path)
     total_pages = len(doc)
-    
-    # Process all pages
-    all_cleaned = []
-    all_markdown = []
-    all_raw = []
-    all_crops = []
+
+    all_cleaned, all_markdown, all_raw, all_crops = [], [], [], []
     img_out = None
-    
-    for page_idx in range(total_pages):
-        page = doc.load_page(page_idx)
-        pix = page.get_pixmap(matrix=fitz.Matrix(300/72, 300/72), alpha=False)
-        img = Image.open(BytesIO(pix.tobytes("png")))
-        
-        cleaned, markdown, result, page_img_out, page_crops = process_image(img, mode, task, custom_prompt)
-        
-        if page_idx == 0:
-            # Use first page's error message if there's an error
-            if cleaned.startswith(" Error") or cleaned.startswith("Enter prompt") or cleaned == "No text":
-                doc.close()
+
+    try:
+        for page_idx in range(total_pages):
+            page = doc.load_page(page_idx)
+            pix = page.get_pixmap(matrix=fitz.Matrix(300 / 72, 300 / 72), alpha=False)
+            img = Image.open(BytesIO(pix.tobytes("png")))
+
+            cleaned, markdown, result, page_img_out, page_crops = process_image(img, mode, task, custom_prompt)
+
+            if page_idx == 0 and (cleaned.startswith("Error") or cleaned == "No text"):
                 return cleaned, "", "", None, []
-        
-        all_cleaned.append(cleaned)
-        all_markdown.append(markdown)
-        all_raw.append(result)
-        all_crops.extend(page_crops)
-        
-        # Use the last page's bounding boxes image, or first if available
-        if page_img_out is not None:
-            img_out = page_img_out
-    
-    doc.close()
-    
-    # Combine results from all pages
-    combined_cleaned = "\n\n--- Page Break ---\n\n".join(all_cleaned)
-    combined_markdown = "\n\n--- Page Break ---\n\n".join(all_markdown)
-    combined_raw = "\n\n--- Page Break ---\n\n".join(all_raw)
-    
-    return combined_cleaned, combined_markdown, combined_raw, img_out, all_crops
+
+            all_cleaned.append(cleaned)
+            all_markdown.append(markdown)
+            all_raw.append(result)
+            all_crops.extend(page_crops)
+
+            if page_img_out is not None:
+                img_out = page_img_out
+
+        combined_cleaned = "\n\n--- Page Break ---\n\n".join(all_cleaned)
+        combined_markdown = "\n\n--- Page Break ---\n\n".join(all_markdown)
+        combined_raw = "\n\n--- Page Break ---\n\n".join(all_raw)
+        return combined_cleaned, combined_markdown, combined_raw, img_out, all_crops
+    finally:
+        doc.close()
 
 def process_file(path, mode, task, custom_prompt, page_num):
     if not path:
-        return "Error Upload file", "", "", None, []
-    if path.lower().endswith('.pdf'):
+        return "Error: Upload file", "", "", None, []
+    if path.lower().endswith(".pdf"):
         return process_pdf(path, mode, task, custom_prompt, page_num)
-    else:
-        return process_image(Image.open(path), mode, task, custom_prompt)
+    return process_image(Image.open(path), mode, task, custom_prompt)
 
 def toggle_prompt(task):
     if task == "✏️ Custom":
         return gr.update(visible=True, label="Custom Prompt", placeholder="Add <|grounding|> for boxes")
-    elif task == "📍 Locate":
+    if task == "📍 Locate":
         return gr.update(visible=True, label="Text to Locate", placeholder="Enter text")
     return gr.update(visible=False)
 
@@ -232,53 +315,65 @@ def select_boxes(task):
     return gr.update()
 
 def get_pdf_page_count(file_path):
-    if not file_path or not file_path.lower().endswith('.pdf'):
+    if not file_path or not file_path.lower().endswith(".pdf"):
         return 1
     doc = fitz.open(file_path)
-    count = len(doc)
-    doc.close()
-    return count
+    try:
+        return len(doc)
+    finally:
+        doc.close()
 
 def load_image(file_path, page_num=1):
     if not file_path:
         return None
-    if file_path.lower().endswith('.pdf'):
+    if file_path.lower().endswith(".pdf"):
         doc = fitz.open(file_path)
-        page_idx = max(0, min(int(page_num) - 1, len(doc) - 1))
-        page = doc.load_page(page_idx)
-        pix = page.get_pixmap(matrix=fitz.Matrix(300/72, 300/72), alpha=False)
-        img = Image.open(BytesIO(pix.tobytes("png")))
-        doc.close()
-        return img
-    else:
-        return Image.open(file_path)
+        try:
+            page_idx = max(0, min(int(page_num) - 1, len(doc) - 1))
+            page = doc.load_page(page_idx)
+            pix = page.get_pixmap(matrix=fitz.Matrix(300 / 72, 300 / 72), alpha=False)
+            return Image.open(BytesIO(pix.tobytes("png")))
+        finally:
+            doc.close()
+    return Image.open(file_path)
 
 def update_page_selector(file_path):
     if not file_path:
         return gr.update(visible=False)
-    if file_path.lower().endswith('.pdf'):
+    if file_path.lower().endswith(".pdf"):
         page_count = get_pdf_page_count(file_path)
-        return gr.update(visible=True, maximum=page_count, value=1, minimum=1,
-                        label=f"Select Page (1-{page_count})")
+        return gr.update(
+            visible=True,
+            maximum=page_count,
+            value=1,
+            minimum=1,
+            label=f"Select Page (1-{page_count})",
+        )
     return gr.update(visible=False)
 
-with gr.Blocks(theme=gr.themes.Soft(), title="DeepSeek-OCR") as demo:
-    gr.Markdown("""
-    # 🚀 DeepSeek-OCR
-    
-    **Document parser with OCR capabilities. Process multi-page PDFs and images to extract text, convert to markdown, or locate specific content with bounding boxes.**
-    """)
-    
+with gr.Blocks(theme=gr.themes.Soft(), title="DeepSeek-OCR (CPU)") as demo:
+    gr.Markdown(
+        """
+# 🐢 DeepSeek-OCR (CPU)
+
+⚠️ **CPU mode is very slow** and may fail on large documents due to RAM/time limits.
+- Prefer **Tiny/Small** modes on CPU.
+- For best results/latency, use GPU.
+
+This Space processes images and multi-page PDFs: extract text, convert to markdown, or locate content with bounding boxes.
+"""
+    )
+
     with gr.Row():
         with gr.Column(scale=1):
             file_in = gr.File(label="Upload Image or PDF", file_types=["image", ".pdf"], type="filepath")
             input_img = gr.Image(label="Input Image", type="pil", height=300)
             page_selector = gr.Number(label="Select Page", value=1, minimum=1, step=1, visible=False)
-            mode = gr.Dropdown(list(MODEL_CONFIGS.keys()), value="Gundam", label="Mode")
-            task = gr.Dropdown(list(TASK_PROMPTS.keys()), value="📋 Markdown", label="Task")
+            mode = gr.Dropdown(list(MODEL_CONFIGS.keys()), value="Tiny", label="Mode (CPU recommend: Tiny/Small)")
+            task = gr.Dropdown(list(TASK_PROMPTS.keys()), value="📝 Free OCR", label="Task")
             prompt = gr.Textbox(label="Prompt", lines=2, visible=False)
             btn = gr.Button("Extract", variant="primary", size="lg")
-        
+
         with gr.Column(scale=2):
             with gr.Tabs() as tabs:
                 with gr.Tab("Text", id="tab_text"):
@@ -291,23 +386,27 @@ with gr.Blocks(theme=gr.themes.Soft(), title="DeepSeek-OCR") as demo:
                     gallery = gr.Gallery(show_label=False, columns=3, height=400)
                 with gr.Tab("Raw Text", id="tab_raw"):
                     raw_out = gr.Textbox(lines=20, show_copy_button=True, show_label=False)
-    
+
     file_in.change(load_image, [file_in, page_selector], [input_img])
     file_in.change(update_page_selector, [file_in], [page_selector])
     page_selector.change(load_image, [file_in, page_selector], [input_img])
     task.change(toggle_prompt, [task], [prompt])
     task.change(select_boxes, [task], [tabs])
-    
+
     def run(image, file_path, mode, task, custom_prompt, page_num):
         if file_path:
             return process_file(file_path, mode, task, custom_prompt, int(page_num))
         if image is not None:
             return process_image(image, mode, task, custom_prompt)
-        return "Error uploading file or image", "", "", None, []
+        return "Error: uploading file or image", "", "", None, []
 
-    submit_event = btn.click(run, [input_img, file_in, mode, task, prompt, page_selector],
-                             [text_out, md_out, raw_out, img_out, gallery])
+    submit_event = btn.click(
+        run,
+        [input_img, file_in, mode, task, prompt, page_selector],
+        [text_out, md_out, raw_out, img_out, gallery],
+    )
     submit_event.then(select_boxes, [task], [tabs])
 
 if __name__ == "__main__":
-    demo.queue(max_size=20).launch()
+    # Keep queue modest on CPU
+    demo.queue(max_size=10).launch(server_name="0.0.0.0", server_port=7860, ssr_mode=False)