Update app.py

app.py

@@ -10,7 +10,6 @@ import pandas as pd
 from PIL import Image, ImageDraw, ImageFont
 
 import cv2
-from urllib.parse import urlparse, parse_qs
 
 # Optional: YOLO for phone detection
 YOLO_MODEL = None
@@ -20,20 +19,17 @@ def load_yolo():
     if YOLO_MODEL is None:
         try:
             from ultralytics import YOLO
-            YOLO_MODEL = YOLO('yolov8n.pt')  # lightweight pretrained model
+            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
 
-# ✅ FIXED iou function
 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])
-    yB = min(boxA[3], boxB[3])   # ✅ fixed
-
+    yB = min(boxA[3], boxB[3])
     interW = max(0, xB - xA)
     interH = max(0, yB - yA)
     interArea = interW * interH
@@ -44,6 +40,7 @@ def iou(boxA, boxB) -> float:
 
 def detect_qr_opencv(image_np: np.ndarray) -> List[Dict[str, Any]]:
     det = cv2.QRCodeDetector()
+    # ✅ Correct unpack
     retval, data_list, points, _ = det.detectAndDecodeMulti(image_np)
     results = []
     if points is None:
@@ -52,9 +49,11 @@ 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
 
     if isinstance(data_list, (list, tuple)):
@@ -67,9 +66,11 @@ def detect_qr_opencv(image_np: np.ndarray) -> List[Dict[str, Any]]:
         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
 
 def detect_phones_yolo(image_np: np.ndarray, conf: float = 0.25) -> List[List[float]]:
@@ -80,11 +81,12 @@ def detect_phones_yolo(image_np: np.ndarray, conf: float = 0.25) -> List[List[fl
     bboxes = []
     for r in results:
         for box, cls in zip(r.boxes.xyxy.cpu().numpy(), r.boxes.cls.cpu().numpy()):
-            if int(cls) == 67:  # COCO: class 67 = cell phone
+            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:
+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:
@@ -108,13 +110,13 @@ def annotate_image(pil_img: Image.Image, qr_boxes: List[Dict[str, Any]], phone_b
         draw.text((qr["bbox"][0], qr["bbox"][1]-12), label, fill=color, font=font)
 
     for i, msgs in flags.items():
-        if not msgs:
-            continue
+        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):
@@ -139,43 +141,23 @@ def read_approved_list(file) -> List[str]:
                 vals = df.iloc[:,0].dropna().astype(str).tolist()
         else:
             content = file.read().decode("utf-8", errors="ignore")
+            file.seek(0)  # ✅ reset pointer so file can be reused
             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 []
 
-# ✅ FIXED payload normalization
-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
-    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:
     if not payload:
         return False
-    norm_payload = normalize_payload(payload)
+    p = payload.strip().lower()
     for a in approved:
-        norm_a = normalize_payload(a)
-        if not norm_a:
-            continue
-        if norm_a in norm_payload or norm_payload in norm_a:
+        a_norm = a.strip().lower()
+        if a_norm in p or p in a_norm:
             return True
     return False
 
-# ---------------- STREAMLIT UI (unchanged) -----------------
 st.set_page_config(page_title="QR Code Anomaly Scanner", layout="wide")
 st.title("🕵️ QR Code Anomaly Scanner (Retail Store 360° CCTV Frames)")
 
@@ -188,6 +170,7 @@ The app will:
   - **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.
+Download the annotated images and a consolidated CSV report at the end.
 """)
 
 with st.sidebar:
@@ -202,4 +185,135 @@ with st.sidebar:
 
 workdir = tempfile.mkdtemp()
 
-# (rest of your original processing loop stays exactly the same…)
+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()
+
+    rows = []
+    annotated_dir = os.path.join(workdir, "annotated")
+    os.makedirs(annotated_dir, exist_ok=True)
+
+    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")
+        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 = {}
+        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")
+            if phone_boxes:
+                qb = qr["bbox"]
+                for pb in phone_boxes:
+                    if iou(qb, pb) >= iou_threshold:
+                        msgs.append("ON_PHONE")
+                        break
+            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))
+
+    status.text("Completed.")
+    df = pd.DataFrame(rows)
+
+    st.subheader("Results")
+    st.dataframe(df, use_container_width=True)
+
+    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
+    })
+
+    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.markdown("""
+**Tips**
+- Approved list can be **TXT** (one payload per line) or **CSV** (use a `payload` column or first column).  
+- For QR-on-phone detection, keep **Detect phones (YOLO)** enabled.  
+- Name frames with timestamps to correlate events later.
+""")