import os, sys, time, threading, subprocess import cv2 import gradio as gr import pandas as pd import PIL.Image as Image from ultralytics import YOLO try: import requests as _requests except ImportError: _requests = None sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "RPi")) try: import debugging as hw HW = True except Exception as e: hw, HW = None, False print(f"Hardware not available: {e}") # ── Load YOLO model ──────────────────────────────────────────────────────────── MODEL_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "Laptop", "best.pt") try: model = YOLO(MODEL_PATH) MODEL_OK = True print("✓ Model loaded.") except Exception as e: model, MODEL_OK = None, False print(f"Model not loaded: {e}") # ── DE WASSTRAAT: Forceert YOLO's ruwe output naar schone, harde strings ── def wasstraat(ruwe_naam): s = str(ruwe_naam).strip().lower() if "mud" in s or "dirt" in s: return "Mud" if "stick" in s: return "Sticker" if "struct" in s: return "Structural" if "organ" in s: return "Organic" if "good" in s or "clean" in s: return "Clean" if "bad" in s: return "bad_crate" return "Unknown" # Hardcoded en foutloos voor je grafiek: CLASS_NAMES = ["Clean", "Structural", "Organic", "Mud", "Sticker", "bad_crate"] history_log = [] scan_count = 0 _last_result_image = None _factory_mode = "AUTO" _API_BASE = os.getenv("CRATESCAN_API_URL", "http://localhost:8000") _LOG_SCAN_URL = f"{_API_BASE}/api/log_scan" _STATS_URL = f"{_API_BASE}/api/dashboard/stats" def _log_scan_async(overall_status: str, yolo_result) -> None: if _requests is None: return def _post(): try: boxes = yolo_result.boxes clean_names = [wasstraat(model.names.get(int(c), "")) for c in boxes.cls.cpu().tolist()] if len(boxes) > 0 else [] has_filth = any(n not in ["Clean", "bad_crate", "Unknown"] for n in clean_names) detections = [] if len(boxes) > 0: for cls_id, conf, box in zip(boxes.cls.cpu().tolist(), boxes.conf.cpu().tolist(), boxes.xyxy.cpu().tolist()): name = wasstraat(model.names.get(int(cls_id), "")) if name == "bad_crate" or name == "Unknown": continue if name == "Clean" and has_filth: continue detections.append({ "defect_name": name, "confidence_score": round(float(conf), 4), "bounding_box": [round(v, 1) for v in box], }) payload = {"overall_status": overall_status, "detections": detections} resp = _requests.post(_LOG_SCAN_URL, json=payload, timeout=3) if resp.status_code == 422: print(f"[API ERROR] Database weigert de data: {resp.text}") except Exception as exc: print(f"[DB log] Failed to reach API: {exc}") threading.Thread(target=_post, daemon=True).start() def _fetch_db_stats(): if _requests is None: return _counter_html(0), gr.update(value=_bar_df({})), {} try: resp = _requests.get(_STATS_URL, timeout=3) resp.raise_for_status() data = resp.json() counts = data.get("defect_distribution", {}) counts["bad_crate"] = data.get("bad_crates", 0) return _counter_html(data.get("total_crates", 0)), gr.update(value=_bar_df(counts)), counts except Exception as exc: print(f"[DB stats] Could not reach API: {exc}") return _counter_html(0), gr.update(value=_bar_df({})), {} def _lcd_refresh_fn(): if _requests is None: return try: resp = _requests.get(_STATS_URL, timeout=3).json() counts = resp.get("defect_distribution", {}) counts["bad_crate"] = resp.get("bad_crates", 0) hw.show_results_on_lcd(generate_dashboard_image(counts, resp.get("total_crates", 0))) except Exception as exc: print(f"[LCD refresh] {exc}") def _toggle_factory_mode() -> str: global _factory_mode if _factory_mode == "AUTO": subprocess.run(["sudo", "systemctl", "stop", "conveyor.service"], check=False) _factory_mode = "MANUAL" return "Manual Mode: Hardware released. Run 'python RPi/debugging.py' in terminal." else: if HW: try: _, _, c = _fetch_db_stats() hw.show_results_on_lcd(generate_dashboard_image(c, sum(v for k,v in c.items() if k != "bad_crate"), mode="STARTING")) except Exception: pass subprocess.run(["sudo", "systemctl", "restart", "conveyor.service"], check=False) _factory_mode = "AUTO" if HW: try: _, _, c = _fetch_db_stats() hw.show_results_on_lcd(generate_dashboard_image(c, sum(v for k,v in c.items() if k != "bad_crate"), mode=_factory_mode)) except Exception: pass return "System: RUNNING" if HW: hw.set_lcd_refresh_callback(_lcd_refresh_fn) hw.set_factory_mode_callback(_toggle_factory_mode) def capture_usb_camera_frame(device_index=0, width=640, height=480): if HW: time.sleep(0.8) backend = cv2.CAP_DSHOW if os.name == "nt" else cv2.CAP_V4L2 cap = cv2.VideoCapture(device_index, backend) if not cap.isOpened(): raise gr.Error(f"Cannot open USB camera.") cap.set(cv2.CAP_PROP_FRAME_WIDTH, width) cap.set(cv2.CAP_PROP_FRAME_HEIGHT, height) for _ in range(10): cap.read() ok, frame = cap.read() cap.release() if not ok or frame is None: raise gr.Error("Failed to capture frame.") return Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)) def run_yolo(img, conf, iou, counts): if img is None: raise gr.Error("Capture a camera frame first.") if not MODEL_OK: raise gr.Error("YOLO model not loaded.") results = model.predict(source=img, conf=conf, iou=iou, show_labels=True, show_conf=True, imgsz=512, agnostic_nms=True) r = results[0] result_img = Image.fromarray(r.plot()[..., ::-1]) global _last_result_image _last_result_image = result_img pred_classes = r.boxes.cls.cpu().tolist() if len(r.boxes) > 0 else [] clean_names = [wasstraat(model.names.get(int(c), "")) for c in pred_classes] has_filth = any(n not in ["Clean", "bad_crate", "Unknown"] for n in clean_names) new_counts = counts.copy() for name in clean_names: if name == "Unknown": continue if name == "Clean" and has_filth: continue new_counts[name] = new_counts.get(name, 0) + 1 text = "Detected objects:\n─────────────────\n" for cls_name in set(clean_names): if cls_name == "Unknown": continue n = clean_names.count(cls_name) if n > 0: text += f"• {cls_name}: {n}\n" if not clean_names or all(c == "Unknown" for c in clean_names): text += "No bad crates found." ts = time.strftime("%H:%M:%S") history_log.append(f"[{ts}] {text.splitlines()[0]}") history = "\n".join(reversed(history_log[-20:])) global scan_count scan_count += 1 if HW: if "Structural" in clean_names: threading.Thread(target=hw.beep_structural, daemon=True).start() threading.Thread(target=hw.show_results_on_lcd, args=(result_img,), daemon=True).start() _counts_snapshot = new_counts.copy() _total_snapshot = scan_count def _after_10s(): try: if _requests is not None: data = _requests.get(_STATS_URL, timeout=3).json() db_counts = data.get("defect_distribution", {}) db_counts["bad_crate"] = data.get("bad_crates", 0) db_total = data.get("total_crates", _total_snapshot) else: db_counts = _counts_snapshot db_total = _total_snapshot hw.show_results_on_lcd(generate_dashboard_image(db_counts, db_total)) except Exception: pass threading.Timer(10.0, _after_10s).start() if "bad_crate" in clean_names: _status = "Rejected" elif "Clean" in clean_names and not has_filth: _status = "Passed" else: _defects = set(clean_names) - {"bad_crate", "Unknown"} _status = "Rejected" if "Structural" in _defects else ("Warning" if _defects - {"Clean"} else "Passed") _log_scan_async(_status, r) return result_img, text, new_counts, _bar_df(new_counts), _counter_html(scan_count), history def _bar_df(counts): lower_counts = {str(k).lower(): v for k, v in counts.items()} return pd.DataFrame({ "Class": CLASS_NAMES, "Count": [lower_counts.get(n.lower(), 0) for n in CLASS_NAMES] }) def _counter_html(n): return f"
Live Counter
{n}
Total crates scanned
" def generate_dashboard_image(counts, total, mode=None): import io, matplotlib, matplotlib.pyplot as plt matplotlib.use("Agg") mode = mode or _factory_mode fig, (ax_left, ax_right) = plt.subplots(1, 2, figsize=(12.8, 7.2)) fig.patch.set_facecolor("#0f1624") ax_left.set_facecolor("#0f1624"); ax_left.axis("off") ax_left.text(0.5, 0.55, str(total), ha="center", va="center", fontsize=110, fontweight="bold", color="white", transform=ax_left.transAxes) ax_left.text(0.5, 0.28, "Total Crates Scanned", ha="center", va="center", fontsize=16, color="#94a3b8", transform=ax_left.transAxes) ax_right.set_facecolor("#0f1624") values = [counts.get(n, 0) for n in CLASS_NAMES] bars = ax_right.barh(CLASS_NAMES, values, color=["#22c55e", "#ef4444", "#22d3ee", "#b45309", "#a855f7", "#f97316"][:len(CLASS_NAMES)], height=0.55) ax_right.set_title("Defect Distribution", color="white", fontsize=20, pad=14) ax_right.tick_params(colors="white", labelsize=18) ax_right.spines[["top", "right"]].set_visible(False) ax_right.spines[["bottom", "left"]].set_color("#2a3a50") ax_right.invert_yaxis() for bar, val in zip(bars, values): ax_right.text(bar.get_width() + 0.2, bar.get_y() + bar.get_height() / 2, str(val), va="center", color="white", fontsize=15) mode_color, mode_label = {"AUTO": ("#22c55e", "⚙ RUNNING"), "STARTING": ("#3b82f6", "⚙ Starting..."), "MANUAL": ("#f59e0b", "⚙ MANUAL")}.get(mode, ("#f59e0b", "⚙ MANUAL")) fig.text(0.02, 0.95, mode_label, color=mode_color, fontsize=19, va="top", ha="left", bbox=dict(boxstyle="round,pad=0.5", facecolor="#1a2840", edgecolor=mode_color, linewidth=2)) fig.text(0.02, 0.03, "↻ Refresh", color="#94a3b8", fontsize=19, va="bottom", ha="left", bbox=dict(boxstyle="round,pad=0.5", facecolor="#1a2840", edgecolor="#475569", linewidth=2)) plt.tight_layout(pad=1.5) buf = io.BytesIO(); fig.savefig(buf, format="png", facecolor=fig.get_facecolor(), dpi=100); plt.close(fig); buf.seek(0) return Image.open(buf).copy() def _show(n): return [gr.update(visible=(i == n)) for i in range(4)] def show_last_image(): if _last_result_image is None: raise gr.Error("No scan has run yet.") return gr.update(visible=False), gr.update(visible=True, value=_last_result_image), gr.update(visible=False), gr.update(visible=True) def back_to_graph(): return gr.update(visible=True), gr.update(visible=False), gr.update(visible=True), gr.update(visible=False) def _safe(fn_name): if not HW: return "⚠ Not on Raspberry Pi." try: getattr(hw, fn_name)(); return f"✓ {fn_name} test complete." except Exception as e: return f"✗ {e}" def hw_snapshot(): try: return "✓ Frame captured.", capture_usb_camera_frame() except Exception as e: return f"✗ {e}", None def hw_single_dist(): if not HW: return "⚠ Not on Raspberry Pi." try: val = hw.get_single_distance(); return f"{val:.1f} cm" if val is not None else "No object in range." except Exception as e: return f"✗ {e}" CSS = """#sidebar{background:linear-gradient(to bottom,#0b1120 0%,#0f1624 12%,#132030 30%,#1a2a3a 50%,#263547 68%,#3d4f63 82%,#6b7f95 93%,#e8edf3 100%)!important;min-height:100vh;padding:20px 10px!important;border-right:1px solid #1e2d40}#sidebar button{display:block;width:100%;text-align:left!important;background:transparent!important;color:#94a3b8!important;border:none!important;border-radius:7px!important;padding:10px 14px!important;font-size:14px!important;margin-bottom:4px;box-shadow:none!important;transition:background 0.15s,color 0.15s}#sidebar button:hover{background:#1a2840!important;color:#e2e8f0!important}#content{background:#f1f5f9!important;min-height:100vh;padding:32px 36px!important}#content .block{padding-top:0!important;margin-top:0!important}.step-card{background:white;border:1px solid #e2e8f0;border-radius:10px;padding:14px 20px;margin-bottom:12px;font-size:14px;color:#334155}.led-row{display:flex;gap:14px;flex-wrap:wrap;margin-top:10px}.led-box{display:flex;align-items:center;gap:10px;padding:11px 22px;border:1.5px solid #e2e8f0;border-radius:10px;background:white;font-size:14px;font-weight:500;color:#1e293b}.dot{width:14px;height:14px;border-radius:50%;display:inline-block;flex-shrink:0}.dot-green{background:#22c55e}.dot-yellow{background:#eab308}.dot-red{background:#ef4444}""" STATUS_HTML = f"
Status : ● {'Live' if MODEL_OK else 'Offline'}
" with gr.Blocks(title="Crate-Scan Pro") as demo: counts_state = gr.State({n: 0 for n in CLASS_NAMES}) with gr.Row(equal_height=False): with gr.Column(scale=1, min_width=200, elem_id="sidebar"): gr.HTML("
Crate-Scan Pro.
") nav0, nav1, nav2, nav3 = gr.Button("⊞ Onboarding"), gr.Button("☰ Data"), gr.Button("◎ Operating"), gr.Button("⚙ Debugging") with gr.Column(scale=4, elem_id="content"): with gr.Group(visible=True) as page0: gr.HTML("

OPERATING GUIDE

1. Place the crate centrally on the conveyor rollers.
2. Wait for the conveyor to start or click the button.
3. The distance sensor triggers the camera when the crate is directly below it.

Led signals

Green : Clean Crate
Yellow : Filthy Crate
Red : Defect Crate

") go_debug_btn = gr.Button("→ Go to Debugging", variant="secondary") with gr.Group(visible=False) as page1: with gr.Row(): gr.HTML("

DATA

"); gr.HTML(STATUS_HTML) gr.HTML("
") counter_box = gr.HTML(_counter_html(0)) bar_plot = gr.BarPlot(value=_bar_df({}), x="Class", y="Count", title="Defect Distribution Per Class Type", color="Class", height=360, min_width=560) with gr.Row(): refresh_btn, last_image_btn, back_to_graph_btn = gr.Button("↻ Refresh from Database", variant="secondary", size="sm"), gr.Button("🖼 Show Last Image", variant="secondary", size="sm"), gr.Button("← Back to Graph", variant="secondary", size="sm", visible=False) last_image_view = gr.Image(label="Most Recent Scan", type="pil", interactive=False, visible=False) with gr.Group(visible=False) as page2: gr.HTML("

Operating

") with gr.Row(equal_height=False): with gr.Column(scale=1): gr.HTML("

Camera Input

") captured_frame = gr.Image(label="Captured Frame", type="pil", height=280, interactive=False) with gr.Tabs(): with gr.Tab("USB Camera (Pi)"): capture_btn = gr.Button("📷 Capture Frame", variant="secondary") with gr.Tab("Upload"): upload_img = gr.Image(label="Upload an image", type="pil", sources=["upload"]) conf_slider, iou_slider = gr.Slider(0, 1, value=0.25, step=0.01, label="Confidence"), gr.Slider(0, 1, value=0.45, step=0.01, label="IOU") run_btn = gr.Button("▶ Run YOLO", variant="primary") with gr.Column(scale=1): gr.HTML("

Last YOLO Result

") result_img = gr.Image(label="Detection view", type="pil", height=280, interactive=False) count_text, history_box = gr.Textbox(label="Count summary", lines=4, interactive=False), gr.Textbox(label="AI Response history", lines=5, interactive=False) with gr.Group(visible=False) as page3: gr.HTML("

Debugging

") with gr.Row(): dbg_led, dbg_motor, dbg_buzzer = gr.Button("🔴 Test RGB LED"), gr.Button("⚙️ Test Motor"), gr.Button("🔊 Test Buzzer") with gr.Row(): dbg_cam, dbg_dist_once, dbg_lcd = gr.Button("📷 Camera Snapshot"), gr.Button("📏 Read Distance"), gr.Button("🖥️ Test LCD Screen") dbg_output, dbg_img, dbg_dist_txt = gr.Textbox(label="Test Output", lines=3, interactive=False), gr.Image(label="Camera Snapshot", type="pil", height=300, interactive=False), gr.Textbox(label="Distance Reading", interactive=False) pages = [page0, page1, page2, page3] nav0.click(fn=lambda: _show(0), outputs=pages); nav1.click(fn=lambda: _show(1), outputs=pages); nav2.click(fn=lambda: _show(2), outputs=pages); nav3.click(fn=lambda: _show(3), outputs=pages); go_debug_btn.click(fn=lambda: _show(3), outputs=pages) refresh_btn.click(fn=_fetch_db_stats, outputs=[counter_box, bar_plot, counts_state]); nav1.click(fn=_fetch_db_stats, outputs=[counter_box, bar_plot, counts_state]) last_image_btn.click(fn=show_last_image, outputs=[bar_plot, last_image_view, last_image_btn, back_to_graph_btn]); back_to_graph_btn.click(fn=back_to_graph, outputs=[bar_plot, last_image_view, last_image_btn, back_to_graph_btn]) capture_btn.click(fn=capture_usb_camera_frame, outputs=captured_frame); upload_img.change(fn=lambda x: x, inputs=upload_img, outputs=captured_frame) run_btn.click(fn=run_yolo, inputs=[captured_frame, conf_slider, iou_slider, counts_state], outputs=[result_img, count_text, counts_state, bar_plot, counter_box, history_box]) dbg_led.click(fn=lambda: _safe("test_rgb_led"), outputs=dbg_output); dbg_motor.click(fn=lambda: _safe("test_motor"), outputs=dbg_output); dbg_buzzer.click(fn=lambda: _safe("test_buzzer"), outputs=dbg_output) def _test_lcd(): res = _safe("test_lcd_screen") if HW: threading.Timer(3.0, lambda: hw.show_results_on_lcd(generate_dashboard_image({k:0 for k in CLASS_NAMES}, scan_count))).start() return res dbg_lcd.click(fn=_test_lcd, outputs=dbg_output); dbg_cam.click(fn=hw_snapshot, outputs=[dbg_output, dbg_img]); dbg_dist_once.click(fn=hw_single_dist, outputs=dbg_dist_txt) if __name__ == "__main__": if HW: try: _, _, boot_counts = _fetch_db_stats() hw.show_results_on_lcd(generate_dashboard_image(boot_counts, sum(v for k,v in boot_counts.items() if k != "bad_crate"))) except Exception: pass demo.launch(server_name="0.0.0.0", share=False, css=CSS, theme=gr.themes.Base())