feat: implement gemma vision pipeline for form population and scene descriptions

Author: utkarshqz

api/db/models.py

@@ -1,42 +1,53 @@
-from sqlmodel import SQLModel, Field
-from sqlalchemy import Column, JSON
-from datetime import datetime
-
-class Template(SQLModel, table=True):
-    id: int | None = Field(default=None, primary_key=True)
-    name: str
-    fields: dict = Field(sa_column=Column(JSON))
-    pdf_path: str
-    created_at: datetime = Field(default_factory=datetime.utcnow)
-
-
-class FormSubmission(SQLModel, table=True):
-    id: int | None = Field(default=None, primary_key=True)
-    template_id: int
-    input_text: str
-    output_pdf_path: str
-    created_at: datetime = Field(default_factory=datetime.utcnow)
-
-# ADD THIS TO api/db/models.py
-# (append to existing file — don't replace)
-
-from sqlmodel import SQLModel, Field
-from typing import Optional
-from datetime import datetime
-
-
-class IncidentMasterData(SQLModel, table=True):
-    """
-    The Incident Data Lake.
-    Stores all extracted data from one incident as a master JSON blob.
-    Any agency can generate their PDF from this single record — zero new LLM calls.
-    """
-    id: Optional[int] = Field(default=None, primary_key=True)
-    incident_id: str = Field(index=True)        # INC-2026-0321-4821
-    master_json: str                             # JSON string — all extracted fields
-    transcript_text: str                         # original transcript
-    location_lat: Optional[float] = None        # from PWA GPS
-    location_lng: Optional[float] = None        # from PWA GPS
-    officer_notes: Optional[str] = None         # additional context
-    created_at: datetime = Field(default_factory=datetime.utcnow)
-    updated_at: datetime = Field(default_factory=datetime.utcnow)
+from sqlmodel import SQLModel, Field
+from sqlalchemy import Column, JSON
+from typing import Optional
+from datetime import datetime
+
+class Template(SQLModel, table=True):
+    id: int | None = Field(default=None, primary_key=True)
+    name: str
+    fields: dict = Field(sa_column=Column(JSON))
+    pdf_path: str
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+
+
+class FormSubmission(SQLModel, table=True):
+    id: int | None = Field(default=None, primary_key=True)
+    template_id: int
+    input_text: str
+    output_pdf_path: str
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+
+
+class IncidentMasterData(SQLModel, table=True):
+    """
+    The Incident Data Lake.
+    Stores all extracted data from one incident as a master JSON blob.
+    Any agency can generate their PDF from this single record — zero new LLM calls.
+    """
+    id: Optional[int] = Field(default=None, primary_key=True)
+    incident_id: str = Field(index=True)        # INC-2026-0321-4821
+    master_json: str                             # JSON string — all extracted fields
+    transcript_text: str                         # original transcript
+    location_lat: Optional[float] = None        # from PWA GPS
+    location_lng: Optional[float] = None        # from PWA GPS
+    officer_notes: Optional[str] = None         # additional context
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    updated_at: datetime = Field(default_factory=datetime.utcnow)
+
+
+class FormFieldCoordinates(SQLModel, table=True):
+    """
+    Stores visual field positions detected by the vision model for a given template.
+    Coordinates are stored as percentages (0-100) of the page dimensions.
+    """
+    id: int | None = Field(default=None, primary_key=True)
+    template_id: int = Field(foreign_key="template.id", index=True)
+    field_label: str                        # e.g. "patient_name", "incident_date"
+    page_number: int = Field(default=0)     # 0-indexed PDF page
+    x: float                                # % from left edge (0–100)
+    y: float                                # % from top edge (0–100)
+    width: float                            # % of page width
+    height: float                           # % of page height
+    field_type: str = Field(default="text") # "text", "checkbox", "image"
+    scanned_at: datetime = Field(default_factory=datetime.utcnow)

api/db/repositories.py

@@ -114,4 +114,26 @@ def update_incident_json(db, incident_id: str, new_data: dict, new_transcript: s
     db.add(incident)
     db.commit()
     db.refresh(incident)
-    return incident
+    return incident
+
+
+# ── Vision Coordinates ────────────────────────────────────────
+
+from api.db.models import FormFieldCoordinates
+
+def create_field_coordinates(db: Session, coords: list[FormFieldCoordinates]) -> list[FormFieldCoordinates]:
+    for coord in coords:
+        db.add(coord)
+    db.commit()
+    for coord in coords:
+        db.refresh(coord)
+    return coords
+
+
+def get_template_coordinates(db: Session, template_id: int) -> list[FormFieldCoordinates]:
+    from sqlmodel import select
+    return db.exec(
+        select(FormFieldCoordinates).where(
+            FormFieldCoordinates.template_id == template_id
+        )
+    ).all()

api/routes/incidents.py

@@ -275,4 +275,64 @@ def generate_narrative(incident_id: str, db: Session = Depends(get_db)):
         "narrative": narrative,
         "format": "markdown",
         "generated_at": datetime.utcnow().isoformat()
+    }
+
+
+# ── Vision Model Endpoints ──────────────────────────────
+
+from fastapi import UploadFile, File
+
+@router.post("/{incident_id}/scene-photo")
+async def add_scene_photo(
+    incident_id: str,
+    file: UploadFile = File(...),
+    db: Session = Depends(get_db)
+):
+    """
+    Multimodal Vision Integration.
+    Upload a scene photo to the incident. Uses Gemma 3 vision model to generate
+    a professional narrative of the scene and merges it into the Data Lake Master JSON.
+    """
+    if not incident_id:
+        raise AppError("Missing incident_id", status_code=400)
+        
+    incident = get_incident(db, incident_id)
+    if not incident:
+        # Auto-create if it doesn't exist
+        incident = IncidentMasterData(
+            incident_id=incident_id,
+            master_json="{}",
+            transcript_text="[VISION] Image submitted."
+        )
+        create_incident(db, incident)
+
+    import base64
+    image_bytes = await file.read()
+    
+    # Run the vision model (async)
+    from src.llm import LLM
+    llm = LLM()
+    try:
+        scene_narrative = await llm.async_vision_describe_scene(image_bytes)
+        print(f"[VISION] Scene Narrative generated: {scene_narrative[:50]}...")
+    except Exception as e:
+        raise AppError(f"Vision model failed: {e}", status_code=500)
+    
+    # Update the data lake using the smart consensus merge
+    new_data = {
+        "scene_visual_analysis": scene_narrative
+    }
+    
+    update_incident_json(
+        db, 
+        incident_id, 
+        new_data, 
+        new_transcript=f"[VISUAL OBSERVATION]: {scene_narrative}"
+    )
+
+    return {
+        "incident_id": incident_id,
+        "status": "photo_analyzed_and_merged",
+        "narrative_preview": scene_narrative[:100] + "...",
+        "message": "Scene analysis merged into master data lake."
     }

src/filler.py

@@ -274,4 +274,64 @@ def fill_form_with_data(self, pdf_form: str, data: dict) -> str:
         PdfWriter().write(output_pdf, pdf)
         print("\nlog extracted successfully")
         print(f"along with what it extracted accordingly, pdf file : {output_pdf}")
+        return output_pdf
+
+    def fill_static_pdf(self, pdf_form: str, coordinates: list, data: dict) -> str:
+        """
+        Uses PyMuPDF to draw text directly onto absolute pixel coordinates
+        for static (non-fillable) forms.
+        coordinates: list of FormFieldCoordinates (from DB)
+        data: dictionary of extracted values (from Data Lake)
+        """
+        import fitz  # PyMuPDF
+        import os
+
+        output_pdf = (
+            pdf_form[:-4]
+            + "_"
+            + datetime.now().strftime("%Y%m%d_%H%M%S")
+            + "_static_filled.pdf"
+        )
+        
+        doc = fitz.open(pdf_form)
+
+        # Draw each field
+        for coord in coordinates:
+            # Match data (case-insensitive fallback)
+            raw_val = data.get(coord.field_label)
+            if raw_val is None:
+                for k, v in data.items():
+                    if k.lower() == coord.field_label.lower():
+                        raw_val = v
+                        break
+                        
+            if raw_val is None or str(raw_val).strip() == "":
+                continue
+                
+            val_str = str(raw_val).strip()
+
+            if coord.page_number < len(doc):
+                page = doc[coord.page_number]
+                page_rect = page.rect
+                
+                # Convert percentages (0-100) to actual PDF points
+                x_pts = (coord.x / 100.0) * page_rect.width
+                y_pts = (coord.y / 100.0) * page_rect.height
+                w_pts = (coord.width / 100.0) * page_rect.width
+                h_pts = (coord.height / 100.0) * page_rect.height
+                
+                rect = fitz.Rect(x_pts, y_pts, x_pts + w_pts, y_pts + h_pts)
+
+                if coord.field_type == "text":
+                    # fitz.TEXT_ALIGN_LEFT = 0
+                    page.insert_textbox(rect, val_str, fontsize=10, fontname="helv", color=(0, 0, 0), align=0)
+                elif coord.field_type == "checkbox":
+                    page.draw_rect(rect, color=(0,0,0), width=1)
+                    if val_str.lower() in TRUTHY_VALUES:
+                        # Draw X
+                        page.draw_line(rect.top_left, rect.bottom_right, color=(0,0,0), width=1.5)
+                        page.draw_line(rect.bottom_left, rect.top_right, color=(0,0,0), width=1.5)
+                        
+        doc.save(output_pdf)
+        doc.close()
         return output_pdf

src/llm.py

@@ -292,6 +292,93 @@ def handle_plural_values(self, plural_value):
     def get_data(self):
         return self._json
 
+    # ── Vision Model Methods ────────────────────────────────────────
+
+    import base64
+    
+    async def async_vision_scan_fields(self, image_bytes: bytes) -> list[dict]:
+        """
+        Passes a page image to the Vision model to map out form fields.
+        Returns a list of dicts: {"label": "...", "x": ..., "y": ..., "w": ..., "h": ...}
+        where x, y, w, h are percentages (0-100) of the page width/height.
+        """
+        import base64
+        ollama_host = os.getenv("OLLAMA_HOST", "http://localhost:11434").rstrip("/")
+        ollama_url = f"{ollama_host}/api/generate"
+        
+        # We need the vision model here, e.g. gemma3:4b
+        vision_model = os.getenv("FIREFORM_VISION_MODEL", "gemma3:4b")
+        b64_image = base64.b64encode(image_bytes).decode("utf-8")
+        
+        prompt = '''
+SYSTEM PROMPT:
+You are an expert form layout analysis AI. Your task is to extract all visual input fields (text blanks, checkboxes) from the provided image of a document.
+Return ONLY valid JSON. The JSON should be a list of objects with the following keys:
+- "label": A descriptive semantic name for the field (e.g. "patient_name", "date_of_incident"). Use snake_case.
+- "x": The X coordinate of the top-left corner of the blank space (as a percentage of the image width, 0.0 to 100.0).
+- "y": The Y coordinate of the top-left corner of the blank space (as a percentage of the image height, 0.0 to 100.0).
+- "w": The width of the blank space (as a percentage of the image width, 0.0 to 100.0).
+- "h": The height of the blank space (as a percentage of the image height).
+- "type": "text" or "checkbox".
+Do not include markdown blocks, just the JSON list.
+        '''
+        
+        payload = {
+            "model": vision_model,
+            "prompt": prompt,
+            "images": [b64_image],
+            "stream": False,
+            "format": "json" # Force JSON mode if supported
+        }
+
+        try:
+            response = requests.post(ollama_url, json=payload, timeout=300)
+            response.raise_for_status()
+            res_text = response.json().get("response", "[]").strip()
+            
+            # Simple cleanup in case it returns markdown
+            if res_text.startswith("```json"):
+                res_text = res_text[7:]
+            if res_text.endswith("```"):
+                res_text = res_text[:-3]
+                
+            data = json.loads(res_text.strip())
+            return data if isinstance(data, list) else []
+        except Exception as e:
+            print(f"[ERROR] Vision Scan Failed: {e}")
+            return []
+
+    async def async_vision_describe_scene(self, image_bytes: bytes) -> str:
+        """
+        Takes a scene photo and generates a professional incident narrative to feed into the Data Lake.
+        """
+        import base64
+        ollama_host = os.getenv("OLLAMA_HOST", "http://localhost:11434").rstrip("/")
+        ollama_url = f"{ollama_host}/api/generate"
+        vision_model = os.getenv("FIREFORM_VISION_MODEL", "gemma3:4b")
+        b64_image = base64.b64encode(image_bytes).decode("utf-8")
+        
+        prompt = '''
+You are a professional first responder describing an incident scene. 
+Look at the provided image and generate a concise but highly detailed incident narrative.
+Describe any structures, vehicles, hazards, visible injuries, number of units involved, and overall context.
+Use professional, objective reporting language.
+        '''
+        
+        payload = {
+            "model": vision_model,
+            "prompt": prompt,
+            "images": [b64_image],
+            "stream": False,
+        }
+
+        try:
+            response = requests.post(ollama_url, json=payload, timeout=300)
+            response.raise_for_status()
+            return response.json().get("response", "").strip()
+        except Exception as e:
+            print(f"[ERROR] Scene Description Failed: {e}")
+            return "Failed to process image scene."
     @staticmethod
     async def async_semantic_map(master_json: dict, target_pdf_fields: list) -> dict:
         """