Update app.py

app.py

@@ -2,6 +2,7 @@ import io
 import os
 import zipfile
 import tempfile
+import time
 from typing import List, Dict, Any
 
 import streamlit as st
@@ -13,6 +14,27 @@ 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:
@@ -37,6 +59,7 @@ 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]]:
     det = cv2.QRCodeDetector()
     retval, data_list, points, _ = det.detectAndDecodeMulti(image_np)
@@ -47,26 +70,18 @@ def detect_qr_opencv(image_np: np.ndarray) -> List[Dict[str, Any]]:
             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()
-            })
+            results.append({"bbox": [float(x1), float(y1), float(x2), float(y2)],"data": data_single,"points": pts.tolist()})
         return results
-
     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])
         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]]:
     model = load_yolo()
     if model is None:
@@ -79,71 +94,20 @@ def detect_phones_yolo(image_np: np.ndarray, conf: float = 0.25) -> List[List[fl
                 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:
-        font = ImageFont.load_default()
-    except:
-        font = None
-
-    for pb in phone_boxes:
-        draw.rectangle(pb, outline=(255, 165, 0), width=3)
-        draw.text((pb[0], pb[1]-12), "PHONE", fill=(255,165,0), font=font)
-
-    for i, qr in enumerate(qr_boxes):
-        color = (0,255,0)
-        if i in flags and any("UNAPPROVED" in f or "ON_PHONE" in f for f in flags[i]):
-            color = (255,0,0)
-        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)
-
-    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]:
-    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:
-                vals = df.iloc[:,0].dropna().astype(str).tolist()
-        else:
-            content = file.read().decode("utf-8", errors="ignore")
-            file.seek(0)  # reset pointer
-            vals = [line.strip() for line in content.splitlines() if line.strip()]
-        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
 
-# --- FIXED match with normalization ---
+# --- Payload normalization & UPI parsing ---
 from urllib.parse import urlparse, parse_qs
+
 def normalize_payload(payload: str) -> str:
     if not payload:
         return ""
@@ -162,9 +126,6 @@ def normalize_payload(payload: str) -> str:
             return part.strip().lower()
         except Exception:
             pass
-    for prefix in ["upi://", "http://", "https://"]:
-        if p.startswith(prefix):
-            p = p[len(prefix):]
     return p
 
 def match_payload(payload: str, approved: List[str]) -> bool:
@@ -177,75 +138,55 @@ def match_payload(payload: str, approved: List[str]) -> bool:
             return True
     return False
 
+# --- 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)
+
 # --- UI ---
 st.set_page_config(page_title="QR Code Anomaly Scanner", page_icon="🕵️", layout="wide")
 
-st.markdown(
-    """
-    <div style="background-color:#4B8BBE;padding:15px;border-radius:10px;margin-bottom:20px;">
-        <h1 style="color:white;text-align:center;">🕵️ QR Code Anomaly Scanner</h1>
-        <p style="color:white;text-align:center;">AI-Powered Surveillance for Retail Store 360° CCTV Frames</p>
-    </div>
-    """,
-    unsafe_allow_html=True
-)
+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)
-    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)
 
-workdir = tempfile.mkdtemp()
-
 if process_btn:
     if not approved_file:
         st.error("Please upload the Approved QR List first.")
         st.stop()
 
-    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.")
-
-    img_paths = []
-    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)
-    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()
+    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)
-
-    progress = st.progress(0.0)
-    status = st.empty()
+    absence_counter = 0
 
-    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", "")
@@ -259,93 +200,19 @@ if process_btn:
                     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
 
-            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 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))
+            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 ""})
 
-    status.text("✅ Completed.")
     df = pd.DataFrame(rows)
+    st.dataframe(df)
 
-    # --- Summary metrics ---
-    st.markdown("### 📊 Summary Dashboard")
-    col1, col2, col3, col4, col5, col6 = st.columns(6)
-    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())
-
-    col1.metric("Frames Processed", total_frames)
-    col2.metric("QR Detections", total_qr)
-    col3.metric("❌ Unapproved", unapproved)
-    col4.metric("📱 On Phone", on_phone)
-    col5.metric("⚠️ Undecoded", undecoded)
-    col6.metric("🚫 No QR Found", no_qr)
-
-    # --- Results Table ---
-    st.markdown("### 📝 Detailed Results")
-    def highlight_anomalies(val):
-        if "UNAPPROVED" in str(val):
-            return "background-color: #FFB6C1; color: black;"
-        elif "ON_PHONE" in str(val):
-            return "background-color: #FFD580; color: black;"
-        elif "UNDECODED" in str(val):
-            return "background-color: #B0C4DE; color: black;"
-        elif "NO_QR_FOUND" in str(val):
-            return "background-color: #D3D3D3; color: black;"
-        return ""
-
-    st.dataframe(df.style.applymap(highlight_anomalies, subset=["anomalies"]), use_container_width=True)
-
-    # --- Image Gallery ---
-    st.markdown("### 🖼️ Annotated Frames Preview")
-    cols = st.columns(3)
-    for idx, fname in enumerate(sorted(os.listdir(annotated_dir))):
-        with cols[idx % 3]:
-            st.image(os.path.join(annotated_dir, fname), caption=fname, use_container_width=True)
-
-    # --- Downloads ---
-    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")
-
-    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.subheader("📜 Alert History")
+    for a in load_alerts()[-10:][::-1]:
+        st.write(f"{a['time']} [{a['frame']}] → {a['alert']}")