Update app.py

app.py

@@ -1,13 +1,16 @@
+import nest_asyncio
+nest_asyncio.apply()
+
+import os
 import gradio as gr
 import logging
 from roboflow import Roboflow
 from PIL import Image, ImageDraw
 import cv2
 import numpy as np
-import os
 from math import atan2, degrees
 import asyncio
-import multiprocessing
+from pyppeteer import launch
 
 # Configure logging
 logging.basicConfig(
@@ -25,72 +28,62 @@ PROJECT_NAME = "model_verification_project"
 VERSION_NUMBER = 2
 
 # ----------------------------
-# Function to generate handwriting image using Pyppeteer in a separate process
+# Asynchronous function to generate handwriting image via Pyppeteer
 # ----------------------------
-def generate_handwriting_image_process(text_prompt, screenshot_path, return_dict):
+async def _generate_handwriting_image(text_prompt, screenshot_path):
     """
-    This function runs in a separate process so that Pyppeteer's signal handling
-    works correctly in its main thread.
+    Launches a headless browser, goes to Calligraphr, types the text, 
+    and takes a screenshot of the rendered handwriting.
     """
-    import asyncio
-    from pyppeteer import launch
-
-    async def _generate():
-        # Launch Chromium with additional flags for containerized environments
-        browser = await launch(
-            headless=True,
-            args=[
-                '--no-sandbox',
-                '--disable-setuid-sandbox',
-                '--disable-dev-shm-usage',
-                '--disable-gpu',
-                '--single-process',
-                '--no-zygote',
-                '--window-size=1920,1080'
-            ]
-        )
+    # Launch Chromium with additional flags for containerized environments
+    browser = await launch(
+        headless=True,
+        handleSIGINT=False,
+        handleSIGTERM=False,
+        handleSIGHUP=False,
+        args=[
+            '--no-sandbox',
+            '--disable-setuid-sandbox',
+            '--disable-dev-shm-usage',
+            '--disable-gpu',
+            '--single-process',
+            '--no-zygote',
+            '--window-size=1920,1080'
+        ]
+    )
+    try:
         page = await browser.newPage()
         await page.goto('https://www.calligraphr.com/en/font/', {'waitUntil': 'networkidle2'})
         await page.waitForSelector('#text-input')
         await page.type('#text-input', text_prompt)
-        await asyncio.sleep(3)  # Increased wait time for the page to render
 
-        # Adjust the clip values as needed to capture the proper area of the page
+        # Give the page time to render the handwriting
+        await asyncio.sleep(3)
+
+        # Screenshot a portion of the page that should contain the handwriting
         await page.screenshot({
             'path': screenshot_path,
             'clip': {'x': 100, 'y': 200, 'width': 600, 'height': 150}
         })
-        await browser.close()
         return screenshot_path
 
-    # Create a new event loop for this process
-    loop = asyncio.new_event_loop()
-    asyncio.set_event_loop(loop)
-    try:
-        result = loop.run_until_complete(_generate())
-        return_dict['result'] = result
-    except Exception as e:
-        logging.error("Error in handwriting generation process: " + str(e))
-        return_dict['result'] = None
     finally:
-        loop.close()
+        await browser.close()
 
-def get_handwriting_image(text_prompt, screenshot_path="/tmp/handwriting.png"):
+def generate_handwriting_image(text_prompt, screenshot_path="/tmp/handwriting.png"):
     """
-    Starts a separate process to generate a handwriting image and returns the image path.
+    Synchronous wrapper around the async Pyppeteer call.
     """
-    manager = multiprocessing.Manager()
-    return_dict = manager.dict()
-    process = multiprocessing.Process(
-        target=generate_handwriting_image_process,
-        args=(text_prompt, screenshot_path, return_dict)
-    )
-    process.start()
-    process.join()
-    return return_dict.get('result', None)
+    try:
+        loop = asyncio.get_event_loop()
+        result = loop.run_until_complete(_generate_handwriting_image(text_prompt, screenshot_path))
+        return result
+    except Exception as e:
+        logging.error(f"Error generating handwriting image: {e}")
+        return None
 
 # ----------------------------
-# Helper: Detect paper angle within bounding box
+# Detect paper angle within bounding box
 # ----------------------------
 def detect_paper_angle(image, bounding_box):
     x1, y1, x2, y2 = bounding_box
@@ -133,20 +126,23 @@ def process_image(image, text):
         prediction = model.predict(input_image_path, confidence=70, overlap=50).json()
         logging.debug(f"Inference result: {prediction}")
 
-        # Convert image for processing
         pil_image = image.convert("RGBA")
         logging.debug("Converted image to RGBA mode.")
 
-        # Process each detected object (assumed to be white paper)
+        # Iterate over detected objects (assumed white paper)
         for obj in prediction['predictions']:
+            # Paper dimensions
             white_paper_width = obj['width']
             white_paper_height = obj['height']
+
+            # Padding
             padding_x = int(white_paper_width * 0.1)
             padding_y = int(white_paper_height * 0.1)
             box_width = white_paper_width - 2 * padding_x
             box_height = white_paper_height - 2 * padding_y
             logging.debug(f"Padded white paper dimensions: width={box_width}, height={box_height}.")
 
+            # Calculate padded coordinates
             x1_padded = int(obj['x'] - white_paper_width / 2 + padding_x)
             y1_padded = int(obj['y'] - white_paper_height / 2 + padding_y)
             x2_padded = int(obj['x'] + white_paper_width / 2 - padding_x)
@@ -156,15 +152,15 @@ def process_image(image, text):
             angle = detect_paper_angle(np.array(image), (x1_padded, y1_padded, x2_padded, y2_padded))
             logging.debug(f"Detected paper angle: {angle} degrees.")
 
-            # (Optional) Save a debug image with the bounding box drawn
+            # (Optional) debug bounding box
             debug_layer = pil_image.copy()
             debug_draw = ImageDraw.Draw(debug_layer)
             debug_draw.rectangle([(x1_padded, y1_padded), (x2_padded, y2_padded)], outline="red", width=3)
             debug_layer.save("/tmp/debug_bounding_box.png")
             logging.debug("Saved bounding box debug image to /tmp/debug_bounding_box.png.")
 
-            # Generate handwriting image using the separate process
-            handwriting_path = get_handwriting_image(text, "/tmp/handwriting.png")
+            # Generate handwriting image
+            handwriting_path = generate_handwriting_image(text, "/tmp/handwriting.png")
             if not handwriting_path:
                 logging.error("Handwriting image generation failed.")
                 continue
@@ -173,7 +169,7 @@ def process_image(image, text):
             handwriting_img = handwriting_img.resize((box_width, box_height), Image.ANTIALIAS)
             rotated_handwriting = handwriting_img.rotate(-angle, resample=Image.BICUBIC, expand=True)
 
-            # Composite the handwriting onto the original image
+            # Composite the handwriting
             text_layer = Image.new("RGBA", pil_image.size, (255, 255, 255, 0))
             paste_x = int(obj['x'] - rotated_handwriting.size[0] / 2)
             paste_y = int(obj['y'] - rotated_handwriting.size[1] / 2)
@@ -181,7 +177,7 @@ def process_image(image, text):
             pil_image = Image.alpha_composite(pil_image, text_layer)
             logging.debug("Handwriting layer composited onto the original image.")
 
-        # Save and return the output image
+        # Save output
         output_image_path = "/tmp/output_image.png"
         pil_image.convert("RGB").save(output_image_path)
         logging.debug(f"Output image saved to {output_image_path}.")
@@ -192,7 +188,7 @@ def process_image(image, text):
         return None
 
 # ----------------------------
-# Gradio interface function
+# Gradio inference function
 # ----------------------------
 def gradio_inference(image, text):
     logging.debug("Starting Gradio inference.")
@@ -204,7 +200,7 @@ def gradio_inference(image, text):
     return None, None, "An error occurred while processing the image. Please check the logs."
 
 # ----------------------------
-# Gradio interface definition
+# Gradio interface
 # ----------------------------
 interface = gr.Interface(
     fn=gradio_inference,
@@ -219,7 +215,9 @@ interface = gr.Interface(
     ],
     title="Roboflow Detection with Handwriting Overlay",
     description="Upload an image and enter text to overlay. The Roboflow model detects the white paper area, and a handwriting image is generated via Calligraphr using Pyppeteer. The output image is composited accordingly.",
-    allow_flagging="never"
+    allow_flagging="never",
+    # Limit concurrency to 1 to reduce potential conflicts with the single event loop
+    concurrency_count=1
 )
 
 if __name__ == "__main__":