Add PDF/MD/JSON reporting helpers and GPT‑OSS wrapper for strong analysis

app.py

@@ -145,7 +145,216 @@ def _extract_path(d):
     return (d.get("path") if isinstance(d, dict) else d)
 
 # ────────────────────────────
-# Interfaz Gradio
+# Helpers for multimodal reporting (PDF/MD/JSON)
+# ────────────────────────────
+
+def _write_pdf(path: str, title: str, narrative: str, frames):
+    if REPORTLAB_AVAILABLE:
+        c = canvas.Canvas(path, pagesize=A4)
+        width, height = A4
+        margin = 40
+        y = height - margin
+        c.setFont("Helvetica-Bold", 16)
+        c.drawString(margin, y, title)
+        y -= 30
+        c.setFont("Helvetica", 11)
+        for line in (narrative or "").splitlines():
+            if y < margin + 50:
+                c.showPage()
+                y = height - margin
+                c.setFont("Helvetica", 11)
+            c.drawString(margin, y, line)
+            y -= 16
+        y -= 10
+        c.setFont("Helvetica-Bold", 12)
+        c.drawString(margin, y, "Per-frame detections:")
+        y -= 18
+        c.setFont("Helvetica", 10)
+        for f in frames:
+            if y < margin + 80:
+                c.showPage()
+                y = height - margin
+                c.setFont("Helvetica", 10)
+            c.drawString(margin, y, f"Frame {f.get('frame_index')}:")
+            y -= 14
+            dets = f.get("detections", [])
+            if not dets:
+                c.drawString(margin + 12, y, "No detections")
+                y -= 12
+            else:
+                for d in dets:
+                    line = f"- {d.get('label')} | conf={d.get('confidence')} | bbox={d.get('bbox')}"
+                    c.drawString(margin + 12, y, line)
+                    y -= 12
+        c.save()
+    else:
+        # Fallback: plain text file (saved with .pdf extension)
+        with open(path, "w", encoding="utf-8") as f:
+            f.write(title + "\n\n")
+            f.write((narrative or "") + "\n\n")
+            f.write("Per-frame detections:\n")
+            for fr in frames:
+                f.write(f"Frame {fr.get('frame_index')}:\n")
+                dets = fr.get("detections", [])
+                if not dets:
+                    f.write("  No detections\n")
+                else:
+                    for d in dets:
+                        f.write(f"  - {d}\n")
+
+def _load_gptoss_wrapper():
+    """
+    Load the blade-inspection-demo/gptoss_wrapper.py module by filepath so we don't rely on package imports.
+    """
+    try:
+        base = os.path.dirname(__file__)
+        wrapper_path = os.path.join(base, "blade-inspection-demo", "gptoss_wrapper.py")
+        if not os.path.exists(wrapper_path):
+            # fallback: maybe file already at project root
+            wrapper_path = os.path.join(base, "gptoss_wrapper.py")
+        spec = importlib.util.spec_from_file_location("gptoss_wrapper", wrapper_path)
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        return getattr(module, "GPTOSSWrapper", None)
+    except Exception:
+        return None
+
+def _build_prompt(frames):
+    lines = []
+    lines.append("You are an expert inspection assistant for wind turbine blade images/videos.")
+    lines.append("Given per-frame detections (label, confidence, bbox), write a concise inspection report with:")
+    lines.append("- Summary of main findings")
+    lines.append("- Suggested severity (low/medium/high) when appropriate")
+    lines.append("- Recommended next steps for inspection/repair")
+    lines.append("")
+    lines.append("Frame detections follow:")
+    for f in frames:
+        fid = f.get("frame_index")
+        dets = f.get("detections", [])
+        if not dets:
+            lines.append(f"Frame {fid}: No detections")
+        else:
+            det_texts = []
+            for d in dets:
+                conf = d.get("confidence")
+                conf_s = f"{conf:.2f}" if isinstance(conf, float) else str(conf)
+                det_texts.append(f"{d.get('label')}({conf_s})")
+            lines.append(f"Frame {fid}: " + ", ".join(det_texts))
+    lines.append("")
+    lines.append("Produce the report in plain text, 6-10 short paragraphs.")
+    return "\n".join(lines)
+
+def generar_analisis_fuerte(media_path):
+    """Generate strong analysis (PDF/MD/JSON) from a given media file path."""
+    if not media_path:
+        return {"status": "no_input", "report_pdf": None, "report_md": None, "report_json": None}
+
+    tmpdir = tempfile.mkdtemp()
+    frames = []
+
+    try:
+        ext = os.path.splitext(media_path)[1].lower()
+        # attempt to extract up to 3 frames/detections using the loaded YOLO model
+        if ext in [".mp4", ".mov", ".avi", ".mkv"]:
+            cap = cv2.VideoCapture(media_path)
+            idx = 0
+            grabbed = 0
+            while grabbed < 3:
+                ret, frame = cap.read()
+                if not ret:
+                    break
+                tmpf = os.path.join(tmpdir, f"frame_{idx}.jpg")
+                cv2.imwrite(tmpf, frame)
+                results = model.predict(source=tmpf, conf=0.25, iou=0.45)
+                dets = []
+                if results and len(results) > 0:
+                    for box in results[0].boxes:
+                        try:
+                            cls_id = int(box.cls[0])
+                            label = model.names[cls_id]
+                        except Exception:
+                            label = "object"
+                        try:
+                            x1, y1, x2, y2 = map(int, box.xyxy[0])
+                        except Exception:
+                            x1 = y1 = x2 = y2 = 0
+                        try:
+                            confv = float(box.conf[0])
+                        except Exception:
+                            confv = None
+                        dets.append({"label": label, "confidence": confv, "bbox": [x1, y1, x2, y2]})
+                frames.append({"frame_index": idx, "detections": dets})
+                idx += 1
+                grabbed += 1
+            cap.release()
+        else:
+            # single image
+            results = model.predict(source=media_path, conf=0.25, iou=0.45)
+            dets = []
+            if results and len(results) > 0:
+                for box in results[0].boxes:
+                    try:
+                        cls_id = int(box.cls[0])
+                        label = model.names[cls_id]
+                    except Exception:
+                        label = "object"
+                    try:
+                        x1, y1, x2, y2 = map(int, box.xyxy[0])
+                    except Exception:
+                        x1 = y1 = x2 = y2 = 0
+                    try:
+                        confv = float(box.conf[0])
+                    except Exception:
+                        confv = None
+                    dets.append({"label": label, "confidence": confv, "bbox": [x1, y1, x2, y2]})
+            frames.append({"frame_index": 0, "detections": dets})
+
+        prompt = _build_prompt(frames)
+        GPTClass = _load_gptoss_wrapper()
+        narrative = None
+        if GPTClass:
+            try:
+                wrapper = GPTClass(model="gpt-oss-120")
+                narrative = wrapper.generate(prompt)
+            except Exception as e:
+                narrative = f"(GPT call failed) {e}"
+        else:
+            narrative = "(GPT wrapper unavailable) Fallback summary:\n"
+            counts = {}
+            for f in frames:
+                for d in f.get("detections", []):
+                    counts[d["label"]] = counts.get(d["label"], 0) + 1
+            narrative += "Detected classes: " + ", ".join([f"{k}({v})" for k, v in counts.items()]) if counts else "No detections"
+
+        # Write Markdown
+        report_md = os.path.join(tmpdir, "report.md")
+        with open(report_md, "w", encoding="utf-8") as md:
+            md.write("# Informe de inspección (Generar analisis fuerte)\n\n")
+            md.write(narrative or "Sin narrativa disponible.\n\n")
+            md.write("\n## Per-frame detections\n\n")
+            for f in frames:
+                md.write(f"- Frame {f.get('frame_index')}: ")
+                dets = f.get("detections", [])
+                if not dets:
+                    md.write("No detections\n")
+                else:
+                    md.write("; ".join([f\"{d['label']}({d['confidence']}) bbox={d['bbox']}\" for d in dets]) + "\n")
+
+        # Write JSON
+        report_json = os.path.join(tmpdir, "report.json")
+        with open(report_json, "w", encoding="utf-8") as jf:
+            json.dump({"narrative": narrative, "frames": frames}, jf, indent=2)
+
+        # Write PDF
+        report_pdf = os.path.join(tmpdir, "report.pdf")
+        _write_pdf(report_pdf, "Informe de inspección - Generar analisis fuerte", narrative, frames)
+
+        return {"status": "done", "report_pdf": report_pdf, "report_md": report_md, "report_json": report_json}
+    except Exception as e:
+        return {"status": f"error: {e}", "report_pdf": None, "report_md": None, "report_json": None}
+
+# ────────────────────────────
+# Interfaz Gradio (extendida)
 # ────────────────────────────
 with gr.Blocks(title="Kesherat · Inspección de palas eólicas") as demo:
     gr.Markdown("## Inspección de palas eólicas con YOLOv8\n"
@@ -180,8 +389,29 @@ with gr.Blocks(title="Kesherat · Inspección de palas eólicas") as demo:
     txt_classes = gr.Textbox(label="Clases cargadas", interactive=False)
     btn_classes.click(fn=show_classes, outputs=txt_classes)
 
+    # New reporting outputs and button
+    btn_report = gr.Button("Generar analisis fuerte")
+    status = gr.Textbox(label="Estado", interactive=False)
+    pdf_out = gr.File(label="Reporte PDF")
+    md_out = gr.File(label="Reporte Markdown")
+    json_out = gr.File(label="Reporte JSON")
+
+    def _on_report(vid, img):
+        path = None
+        # gr.Video returns path-like; gr.Image type="filepath" returns path
+        if vid:
+            path = vid
+        elif img:
+            # if gr.Image returns an object with .name (local runs)
+            path = img if isinstance(img, str) else getattr(img, "name", None)
+        if not path:
+            return "No media provided", None, None, None
+        res = generar_analisis_fuerte(path)
+        return res.get("status", "error"), (res.get("report_pdf") if res.get("report_pdf") else None), (res.get("report_md") if res.get("report_md") else None), (res.get("report_json") if res.get("report_json") else None)
+
+    btn_report.click(fn=_on_report, inputs=[video_input, image_input], outputs=[status, pdf_out, md_out, json_out])
+
 # Habilitar cola para ZeroGPU
-# Nota: en Gradio 4.x el parámetro es "max_size" / "default_concurrency_limit"; sin kwargs específicos, queue() basta.
 demo.queue()
 
 if __name__ == "__main__":