Update app.py

app.py

@@ -1,9 +1,9 @@
-
 import io
 import os
 import zipfile
 import tempfile
-from typing import List, Tuple, Dict, Any
+import time
+from typing import List, Dict, Any
 
 import streamlit as st
 import numpy as np
@@ -14,20 +14,39 @@ import cv2
 # Optional: YOLO for phone detection
 YOLO_MODEL = None
 
+# ROI Zones (x1, y1, x2, y2)
+ROI_ZONES = [(100, 100, 400, 400)]  # example, configurable later
+ABSENCE_THRESHOLD_SEC = 15
+ALERT_LOG_FILE = "alerts.json"
+
+# --- Alert persistence ---
+def load_alerts():
+    if os.path.exists(ALERT_LOG_FILE):
+        return pd.read_json(ALERT_LOG_FILE).to_dict(orient="records")
+    return []
+
+def save_alert(alert: Dict[str, Any]):
+    alerts = load_alerts()
+    alerts.append(alert)
+    pd.DataFrame(alerts).to_json(ALERT_LOG_FILE, orient="records", indent=2)
+
+def log_alert(message, frame_name):
+    alert = {"time": time.ctime(), "alert": message, "frame": frame_name}
+    save_alert(alert)
+
+# --- YOLO Loader ---
 def load_yolo():
     global YOLO_MODEL
     if YOLO_MODEL is None:
         try:
             from ultralytics import YOLO
-            # Use lightweight pretrained model; supports "cell phone" class via COCO.
-            YOLO_MODEL = YOLO("yolov8n.pt")  # will use local file if present, or download on first run
+            YOLO_MODEL = YOLO('yolov8n.pt')
         except Exception as e:
             st.warning(f"YOLO model could not be loaded: {e}")
             YOLO_MODEL = None
     return YOLO_MODEL
 
 def iou(boxA, boxB) -> float:
-    # boxes in [x1,y1,x2,y2]
     xA = max(boxA[0], boxB[0])
     yA = max(boxA[1], boxB[1])
     xB = min(boxA[2], boxB[2])
@@ -40,396 +59,181 @@ def iou(boxA, boxB) -> float:
     denom = areaA + areaB - interArea + 1e-6
     return interArea / denom
 
+# --- QR Detection ---
 def detect_qr_opencv(image_np: np.ndarray) -> List[Dict[str, Any]]:
-    """
-    Use OpenCV's QRCodeDetector to find and decode QR codes.
-    Returns list of dicts: {bbox: [x1,y1,x2,y2], data: str, points: np.ndarray}
-    """
     det = cv2.QRCodeDetector()
-    data, points, _ = det.detectAndDecodeMulti(image_np)
+    retval, data_list, points, _ = det.detectAndDecodeMulti(image_np)
     results = []
-
     if points is None:
-        # Try single QR fallback
         data_single, points_single, _ = det.detectAndDecode(image_np)
         if points_single is not None and data_single:
             pts = np.array(points_single, dtype=np.float32).reshape(-1, 2)
-            x1, y1 = np.min(pts[:, 0]), np.min(pts[:, 1])
-            x2, y2 = np.max(pts[:, 0]), np.max(pts[:, 1])
-            results.append({
-                "bbox": [float(x1), float(y1), float(x2), float(y2)],
-                "data": data_single,
-                "points": pts.tolist()
-            })
+            x1, y1 = np.min(pts[:,0]), np.min(pts[:,1])
+            x2, y2 = np.max(pts[:,0]), np.max(pts[:,1])
+            results.append({"bbox": [float(x1), float(y1), float(x2), float(y2)],"data": data_single,"points": pts.tolist()})
         return results
-
-    # points shape: (N,4,2), data is list/tuple of strings (may be '' for undecodable)
-    if isinstance(data, (list, tuple)):
-        decoded_list = data
-    else:
-        decoded_list = [data] * len(points)
-
+    decoded_list = data_list if isinstance(data_list, (list, tuple)) else [data_list] * len(points)
     for i, quad in enumerate(points):
-        pts = np.array(quad, dtype=np.float32).reshape(-1, 2)
-        x1, y1 = np.min(pts[:, 0]), np.min(pts[:, 1])
-        x2, y2 = np.max(pts[:, 0]), np.max(pts[:, 1])
+        pts = np.array(quad, dtype=np.float32).reshape(-1,2)
+        x1, y1 = np.min(pts[:,0]), np.min(pts[:,1])
+        x2, y2 = np.max(pts[:,0]), np.max(pts[:,1])
         payload = decoded_list[i] if i < len(decoded_list) else ""
-        results.append({
-            "bbox": [float(x1), float(y1), float(x2), float(y2)],
-            "data": payload,
-            "points": pts.tolist()
-        })
+        results.append({"bbox": [float(x1), float(y1), float(x2), float(y2)],"data": payload,"points": pts.tolist()})
     return results
 
+# --- Phone Detection ---
 def detect_phones_yolo(image_np: np.ndarray, conf: float = 0.25) -> List[List[float]]:
-    """
-    Detect cell phones with YOLO. Returns list of [x1,y1,x2,y2].
-    """
     model = load_yolo()
     if model is None:
         return []
-    # YOLO expects RGB image; ultralytics handles numpy arrays
     results = model.predict(source=image_np, conf=conf, verbose=False)
     bboxes = []
     for r in results:
-        # handle cases where no boxes are returned
-        if getattr(r, "boxes", None) is None:
-            continue
-        # collect cell phone boxes (COCO class 67)
-        try:
-            for box, cls in zip(r.boxes.xyxy.cpu().numpy(), r.boxes.cls.cpu().numpy()):
-                if int(cls) == 67:
-                    bboxes.append([float(box[0]), float(box[1]), float(box[2]), float(box[3])])
-        except Exception:
-            # Fail-open if structure is different
-            continue
+        for box, cls in zip(r.boxes.xyxy.cpu().numpy(), r.boxes.cls.cpu().numpy()):
+            if int(cls) == 67:  # COCO: "cell phone"
+                bboxes.append([float(box[0]), float(box[1]), float(box[2]), float(box[3])])
     return bboxes
 
-def annotate_image(pil_img: Image.Image, qr_boxes: List[Dict[str, Any]], phone_boxes: List[List[float]], flags: Dict[int, List[str]]) -> Image.Image:
-    img = pil_img.copy().convert("RGB")
-    draw = ImageDraw.Draw(img)
-    # Try to load a default font
-    try:
-        font = ImageFont.load_default()
-    except Exception:
-        font = None
-
-    # Draw phone boxes
-    for pb in phone_boxes:
-        draw.rectangle(pb, outline=(255, 165, 0), width=3)  # orange
-        draw.text((pb[0], pb[1] - 12), "PHONE", fill=(255, 165, 0), font=font)
-
-    # Draw QR boxes
-    for i, qr in enumerate(qr_boxes):
-        color = (0, 255, 0)  # green
-        if i in flags and any("UNAPPROVED" in f or "ON_PHONE" in f for f in flags[i]):
-            color = (255, 0, 0)  # red for anomaly
-        draw.rectangle(qr["bbox"], outline=color, width=3)
-        label = "QR"
-        if qr.get("data"):
-            snippet = qr["data"][:32].replace("\n", " ")
-            label += f": {snippet}"
-        draw.text((qr["bbox"][0], qr["bbox"][1] - 12), label, fill=color, font=font)
-
-    # Add flags text
-    for i, msgs in flags.items():
-        if not msgs:
-            continue
-        x1, y1, x2, y2 = qr_boxes[i]["bbox"]
-        y_text = y2 + 4
-        for msg in msgs:
-            draw.text((x1, y_text), f"[{msg}]", fill=(255, 0, 0), font=font)
-            y_text += 12
-    return img
-
-def unpack_zip(uploaded_file, workdir):
-    zf = zipfile.ZipFile(uploaded_file)
-    out_paths = []
-    for name in zf.namelist():
-        if name.lower().endswith((".jpg", ".jpeg", ".png", ".bmp", ".webp")):
-            p = os.path.join(workdir, os.path.basename(name))
-            with open(p, "wb") as f:
-                f.write(zf.read(name))
-            out_paths.append(p)
-    return out_paths
-
-def read_approved_list(file) -> List[str]:
-    """
-    Accepts CSV or TXT. One payload per line or in a 'payload' column.
-    Payloads can be full strings or partial substrings to match.
-    """
-    name = file.name.lower()
-    try:
-        if name.endswith(".csv"):
-            df = pd.read_csv(file)
-            if "payload" in df.columns:
-                vals = df["payload"].dropna().astype(str).tolist()
-            else:
-                # take first column
-                vals = df.iloc[:, 0].dropna().astype(str).tolist()
-        else:
-            # plain text
-            content = file.read().decode("utf-8", errors="ignore")
-            vals = [line.strip() for line in content.splitlines() if line.strip()]
-        # Normalize
-        return [v.strip() for v in vals if v.strip()]
-
-    except Exception as e:
-        st.error(f"Failed to parse approved list: {e}")
-        return []
+# --- Tampering detection ---
+def detect_tampering(image_np: np.ndarray, bbox: List[float]) -> bool:
+    x1, y1, x2, y2 = map(int, bbox)
+    roi = image_np[y1:y2, x1:x2]
+    if roi.size == 0:
+        return False
+    gray = cv2.cvtColor(roi, cv2.COLOR_BGR2GRAY)
+    edges = cv2.Canny(gray, 100, 200)
+    edge_density = np.sum(edges > 0) / (roi.shape[0] * roi.shape[1])
+    return edge_density < 0.01
+
+# --- Payload normalization & UPI parsing ---
+from urllib.parse import urlparse, parse_qs
+
+def normalize_payload(payload: str) -> str:
+    if not payload:
+        return ""
+    p = payload.strip().lower()
+    if p.startswith("upi://"):
+        try:
+            parsed = urlparse(p)
+            qs = parse_qs(parsed.query)
+            if "pa" in qs:
+                return qs["pa"][0].strip().lower()
+        except Exception:
+            pass
+    if "pa=" in p:
+        try:
+            part = p.split("pa=")[1].split("&")[0]
+            return part.strip().lower()
+        except Exception:
+            pass
+    return p
 
 def match_payload(payload: str, approved: List[str]) -> bool:
-    """
-    Return True if payload matches an approved entry.
-    We allow substring match either way to account for embedded metadata/UTMs.
-    """
     if not payload:
         return False
-    p = payload.strip()
+    norm_payload = normalize_payload(payload)
     for a in approved:
-        if a in p or p in a:
+        norm_a = normalize_payload(a)
+        if norm_payload == norm_a:
             return True
     return False
 
-# -------------------- UI --------------------
-st.set_page_config(page_title="QR Code Anomaly Scanner", page_icon="🕵", layout="wide")
-st.title("🕵 QR Code Anomaly Scanner (Retail Store 360° CCTV Frames)")
-
-st.markdown("""
-Upload a set of frame images (multiple files **or** a ZIP), plus the approved QR list (CSV/TXT).  
-The app will:
-- Detect and decode QR codes in each frame.
-- Detect **cell phones** via YOLO to infer if a QR is shown on a phone.
-- Flag anomalies:
-  - **UNAPPROVED_QR**: decoded payload not in the approved list.
-  - **ON_PHONE**: QR bounding box overlaps a detected phone.
-  - **UNDECODED_QR**: QR detected but not decodable (could be suspicious/obstructed).
-- Download the annotated images and a consolidated CSV report at the end.
-""")
+# --- ROI Helper ---
+def inside_roi(bbox, roi):
+    x1, y1, x2, y2 = bbox
+    rx1, ry1, rx2, ry2 = roi
+    return (x1 >= rx1 and y1 >= ry1 and x2 <= rx2 and y2 <= ry2)
 
-with st.sidebar:
-    st.header("Inputs")
-    approved_file = st.file_uploader("Approved QR List (CSV/TXT)", type=["csv", "txt"])
-    frames = st.file_uploader("Frames (images) — select multiple", type=["jpg", "jpeg", "png", "bmp", "webp"], accept_multiple_files=True)
-    frames_zip = st.file_uploader("Or upload a ZIP of frames", type=["zip"])
-    run_phone_detection = st.checkbox("Detect phones (YOLO)", value=True)
-    phone_conf = st.slider("Phone detection confidence", 0.1, 0.8, 0.25, 0.05)
-    iou_threshold = st.slider("QR–Phone overlap IoU threshold", 0.05, 0.8, 0.2, 0.05)
-    process_btn = st.button("Run Scan", use_container_width=True)
+# --- UI ---
+st.set_page_config(page_title="QR Code Anomaly Scanner", page_icon="🕵️", layout="wide")
 
-workdir = tempfile.mkdtemp()
+st.title("🕵️ QR Code Anomaly Scanner with Extended Compliance")
+
+with st.sidebar:
+    st.header("⚙️ Inputs")
+    approved_file = st.file_uploader("📑 Approved QR List (CSV/TXT)", type=["csv","txt"])
+    frames = st.file_uploader("🖼️ Frames (images)", type=["jpg","jpeg","png","bmp","webp"], accept_multiple_files=True)
+    run_phone_detection = st.checkbox("📱 Detect phones (YOLO)", value=True)
+    phone_conf = st.slider("📏 Phone detection confidence", 0.1, 0.8, 0.25, 0.05)
+    iou_threshold = st.slider("🎯 QR–Phone overlap IoU threshold", 0.05, 0.8, 0.2, 0.05)
+    process_btn = st.button("🚀 Run Scan", use_container_width=True)
 
 if process_btn:
     if not approved_file:
         st.error("Please upload the Approved QR List first.")
         st.stop()
 
-    # Read approved list
-    approved_list = read_approved_list(approved_file)
-    if not approved_list:
-        st.warning("Approved list is empty or failed to parse. All decoded QR payloads will be treated as UNAPPROVED.")
-    else:
-        st.success(f"Loaded {len(approved_list)} approved entries.")
-
-    # Gather images
-    img_paths = []
-    # Save multi-file uploads
-    for f in frames or []:
-        out = os.path.join(workdir, f.name)
-        with open(out, "wb") as g:
-            g.write(f.read())
-        img_paths.append(out)
-    # Or unpack ZIP
-    if frames_zip is not None:
-        img_paths.extend(unpack_zip(frames_zip, workdir))
-
-    img_paths = sorted(set(img_paths))
-    if not img_paths:
-        st.error("Please upload at least one frame image (or a ZIP).")
-        st.stop()
-
-    if run_phone_detection:
-        load_yolo()  # try to initialize early to show warnings
+    approved_list = pd.read_csv(approved_file).iloc[:,0].astype(str).tolist() if approved_file.name.endswith(".csv") else approved_file.read().decode().splitlines()
+    st.success(f"✅ Loaded {len(approved_list)} approved entries.")
 
     rows = []
-    annotated_dir = os.path.join(workdir, "annotated")
-    os.makedirs(annotated_dir, exist_ok=True)
+    absence_counter = 0
 
-    progress = st.progress(0.0)
-    status = st.empty()
-
-    for idx, path in enumerate(img_paths):
-        status.text(f"Processing {os.path.basename(path)} ({idx + 1}/{len(img_paths)})")
-        pil = Image.open(path).convert("RGB")
+    for f in frames or []:
+        pil = Image.open(f).convert("RGB")
         np_img = np.array(pil)
 
         qr_results = detect_qr_opencv(np_img)
         phone_boxes = detect_phones_yolo(np_img, conf=phone_conf) if run_phone_detection else []
 
         flags = {}
+        if len(qr_results) > 1:
+            log_alert("Multiple QRs detected", f.name)
+
+        if not qr_results:
+            absence_counter += 1
+            if absence_counter * 1 > ABSENCE_THRESHOLD_SEC:  # assuming 1s per frame approx
+                log_alert("QR absent for threshold duration", f.name)
+        else:
+            absence_counter = 0
+
         for i, qr in enumerate(qr_results):
             msgs = []
             payload = qr.get("data", "")
-
             if not payload:
                 msgs.append("UNDECODED_QR")
             elif not match_payload(payload, approved_list):
                 msgs.append("UNAPPROVED_QR")
-
-            # Check overlap with phones
             if phone_boxes:
                 qb = qr["bbox"]
                 for pb in phone_boxes:
                     if iou(qb, pb) >= iou_threshold:
                         msgs.append("ON_PHONE")
                         break
-
+            if not any(inside_roi(qr["bbox"], roi) for roi in ROI_ZONES):
+                msgs.append("OUTSIDE_ROI")
+            if detect_tampering(np_img, qr["bbox"]):
+                msgs.append("TAMPERING")
+            if msgs:
+                log_alert("|".join(msgs), f.name)
             flags[i] = msgs
 
-            # Append a row
-            rows.append({
-                "frame": os.path.basename(path),
-                "qr_index": i,
-                "payload": payload,
-                "approved_match": (payload and match_payload(payload, approved_list)),
-                "on_phone": ("ON_PHONE" in msgs),
-                "undecoded": ("UNDECODED_QR" in msgs),
-                "anomalies": "|".join(msgs) if msgs else "",
-                "qr_bbox": qr["bbox"],
-                "phone_boxes": phone_boxes
-            })
-
-        # If no QR detected, still log the frame
-        if not qr_results:
-            rows.append({
-                "frame": os.path.basename(path),
-                "qr_index": -1,
-                "payload": "",
-                "approved_match": False,
-                "on_phone": False,
-                "undecoded": False,
-                "anomalies": "NO_QR_FOUND",
-                "qr_bbox": None,
-                "phone_boxes": phone_boxes
-            })
-
-        annotated = annotate_image(pil, qr_results, phone_boxes, flags)
-        out_path = os.path.join(annotated_dir, os.path.basename(path))
-        annotated.save(out_path)
-
-        progress.progress((idx + 1) / len(img_paths))
-
-    status.text("Completed.")
+            rows.append({"frame": f.name,"qr_index": i,"payload": payload,"approved_match": (payload and match_payload(payload, approved_list)),"anomalies": "|".join(msgs) if msgs else ""})
 
     df = pd.DataFrame(rows)
-    st.subheader("Results")
-    st.dataframe(df, use_container_width=True)
-
-    # Summary
-    st.markdown("### Summary")
-    total_frames = len(img_paths)
-    total_qr = int((df["qr_index"] >= 0).sum())
-    unapproved = int((df["anomalies"].str.contains("UNAPPROVED_QR", na=False)).sum())
-    on_phone = int((df["anomalies"].str.contains("ON_PHONE", na=False)).sum())
-    undecoded = int((df["anomalies"].str.contains("UNDECODED_QR", na=False)).sum())
-    no_qr = int((df["anomalies"] == "NO_QR_FOUND").sum())
-    st.write({
-        "frames_processed": total_frames,
-        "qr_detections": total_qr,
-        "unapproved_qr": unapproved,
-        "qr_on_phone": on_phone,
-        "undecoded_qr": undecoded,
-        "frames_with_no_qr": no_qr
-    })
-
-    # --- Dashboard Cards (Microsoft-style colored cards) ---
+    st.dataframe(df)
+
+    # --- Dashboard Cards ---
     st.markdown("### 📊 Compliance Dashboard")
+    col1, col2, col3 = st.columns(3)
+    col4, col5, col6 = st.columns(3)
 
-    # minimal CSS injected once for card styling
-    card_css = """
-    <style>
-    .dashboard-grid { display: grid; grid-template-columns: repeat(3, 1fr); gap: 12px; }
-    .card {
-      border-radius: 12px;
-      border: 2px solid rgba(0,0,0,0.12);
-      padding: 16px;
-      background: #fff;
-      box-shadow: 0 2px 6px rgba(0,0,0,0.06);
-      text-align: center;
-    }
-    .card .label { font-size: 14px; font-weight: 600; opacity: 0.8; margin-bottom: 6px; }
-    .card .value { font-size: 28px; font-weight: 800; }
-    .card.red    { border-color: #B71C1C; background: #FFEBEE; }
-    .card.orange { border-color: #EF6C00; background: #FFF3E0; }
-    .card.purple { border-color: #6A1B9A; background: #F3E5F5; }
-    .card.blue   { border-color: #01579B; background: #E1F5FE; }
-    .card.green  { border-color: #1B5E20; background: #E8F5E9; }
-    .card.gray   { border-color: #263238; background: #ECEFF1; }
-    </style>
-    """
-    st.markdown(card_css, unsafe_allow_html=True)
-
-    # compute counts from the current df (keeping existing anomaly labels)
     unapproved_count = int((df["anomalies"].str.contains("UNAPPROVED_QR", na=False)).sum())
-    on_phone_count   = int((df["anomalies"].str.contains("ON_PHONE", na=False)).sum())
-    tampering_count  = int((df["anomalies"].str.contains("TAMPERING", na=False)).sum())  # likely 0 if not used
-    roi_count        = int((df["anomalies"].str.contains("OUTSIDE_ROI", na=False)).sum()) # likely 0 if not used
-    absence_count    = int((df["anomalies"].str.contains("NO_QR_FOUND", na=False)).sum())
-    undecoded_count  = int((df["anomalies"].str.contains("UNDECODED_QR", na=False)).sum())
-
-    cards_html = f"""
-    <div class="dashboard-grid">
-      <div class="card red">
-        <div class="label">❌ Unauthorized QRs</div>
-        <div class="value">{unapproved_count}</div>
-      </div>
-      <div class="card orange">
-        <div class="label">📱 On Phone</div>
-        <div class="value">{on_phone_count}</div>
-      </div>
-      <div class="card purple">
-        <div class="label">⚠️ Tampered</div>
-        <div class="value">{tampering_count}</div>
-      </div>
-      <div class="card blue">
-        <div class="label">🚫 Outside ROI</div>
-        <div class="value">{roi_count}</div>
-      </div>
-      <div class="card green">
-        <div class="label">⏳ QR Missing</div>
-        <div class="value">{absence_count}</div>
-      </div>
-      <div class="card gray">
-        <div class="label">🔍 Undecoded</div>
-        <div class="value">{undecoded_count}</div>
-      </div>
-    </div>
-    """
-    st.markdown(cards_html, unsafe_allow_html=True)
-    # --- end Dashboard Cards ---
-
-    # Downloads: CSV + ZIP of annotated images
-    csv_bytes = df.to_csv(index=False).encode("utf-8")
-    st.download_button("⬇️ Download CSV Report", data=csv_bytes, file_name="qr_anomaly_report.csv", mime="text/csv")
-
-    # Create ZIP of annotations
-    mem = io.BytesIO()
-    with zipfile.ZipFile(mem, mode="w", compression=zipfile.ZIP_DEFLATED) as z:
-        for fname in sorted(os.listdir(annotated_dir)):
-            z.write(os.path.join(annotated_dir, fname), arcname=fname)
-    mem.seek(0)
-    st.download_button(
-        "⬇️ Download Annotated Images (ZIP)",
-        data=mem.getvalue(),
-        file_name="annotated_frames.zip",
-        mime="application/zip"
-    )
-
-else:
-    st.info("Upload inputs on the left and click **Run Scan** to begin.")
-    st.markdown("""
-**Tips**
-- Your approved list can be **TXT** (one payload per line) or **CSV** (use a `payload` column or the first column).
-- For mobile QR misuse detection, keep **Detect phones (YOLO)** enabled.
-- Name frames with timestamps if you want to correlate events later.
-""")
+    on_phone_count = int((df["anomalies"].str.contains("ON_PHONE", na=False)).sum())
+    tampering_count = int((df["anomalies"].str.contains("TAMPERING", na=False)).sum())
+    roi_count = int((df["anomalies"].str.contains("OUTSIDE_ROI", na=False)).sum())
+    absence_count = int((df["anomalies"].str.contains("ABSENCE", na=False)).sum())
+    undecoded_count = int((df["anomalies"].str.contains("UNDECODED_QR", na=False)).sum())
+
+    col1.metric("❌ Unauthorized QRs", unapproved_count)
+    col2.metric("📱 On Phone", on_phone_count)
+    col3.metric("⚠️ Tampered", tampering_count)
+
+    col4.metric("🚫 Outside ROI", roi_count)
+    col5.metric("⏳ QR Missing", absence_count)
+    col6.metric("🔍 Undecoded", undecoded_count)
+
+    # --- Alert History ---
+    st.subheader("📜 Alert History")
+    for a in load_alerts()[-10:][::-1]:
+        st.write(f"{a['time']} [{a['frame']}] → {a['alert']}")