Update app.py

app.py

@@ -1,34 +1,38 @@
 import os
-import json
-import time
-import gradio as gr
 import re
+import gradio as gr
 from dataclasses import dataclass
 from typing import List, Dict, Tuple
 
-# ---- Language detection ----
+# ----------------------------
+# Language detection (Urdu/Arabic script vs Latin)
+# ----------------------------
 ARABIC_RE = re.compile(r"[\u0600-\u06FF]")
 def detect_lang(text: str) -> str:
     return "ur" if ARABIC_RE.search(text) else "en"
 
-# ---- Guardrails ----
+# ----------------------------
+# Guardrails
+# ----------------------------
 @dataclass
 class Guardrails:
     refusal_msg_ur: str
     refusal_msg_en: str
     blocked_patterns: List[str]
     soft_patterns: List[str]
+    helplines: List[Dict[str, str]]
 
     @classmethod
     def from_yaml(cls, path: str):
         import yaml
-        with open(path, 'r', encoding='utf-8') as f:
+        with open(path, "r", encoding="utf-8") as f:
             data = yaml.safe_load(f)
         return cls(
-            refusal_msg_ur=data['refusal_msg_ur'],
-            refusal_msg_en=data['refusal_msg_en'],
-            blocked_patterns=data['blocked_patterns'],
-            soft_patterns=data['soft_patterns'],
+            refusal_msg_ur=data["refusal_msg_ur"],
+            refusal_msg_en=data["refusal_msg_en"],
+            blocked_patterns=data["blocked_patterns"],
+            soft_patterns=data["soft_patterns"],
+            helplines=data.get("helplines", []),
         )
 
     def check(self, text: str) -> Tuple[str, str]:
@@ -41,92 +45,387 @@ class Guardrails:
                 return ("SOFT", p)
         return ("OK", "")
 
-# ---- Tiny placeholder model ----
-def tiny_response(user_msg: str, history: List[Tuple[str,str]], lang: str) -> str:
+GUARD = Guardrails.from_yaml("guardrails.yaml")
+
+# ----------------------------
+# Small CPU LLM (Qwen2-0.5B-Instruct) with fallback
+# ----------------------------
+from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
+import torch
+
+MODEL_ID = os.environ.get("SAFEPak_MODEL_ID", "Qwen/Qwen2-0.5B-Instruct")
+try:
+    TOKENIZER = AutoTokenizer.from_pretrained(MODEL_ID)
+    MODEL = AutoModelForCausalLM.from_pretrained(MODEL_ID, torch_dtype=torch.float32)
+    GEN = pipeline(
+        "text-generation",
+        model=MODEL,
+        tokenizer=TOKENIZER,
+        device_map="auto" if torch.cuda.is_available() else None,
+    )
+    TINY_FALLBACK = False
+except Exception as e:
+    print("LLM load failed, tiny fallback active:", e)
+    TOKENIZER = None
+    GEN = None
+    TINY_FALLBACK = True
+
+def llm_reply(prompt: str, max_new_tokens: int = 160, temperature: float = 0.6) -> str:
+    if TINY_FALLBACK:
+        return "Stay calm. Take a deep breath. Move to safety now. (fallback)"
+    out = GEN(
+        prompt,
+        max_new_tokens=max_new_tokens,
+        do_sample=True,
+        temperature=temperature,
+        pad_token_id=TOKENIZER.eos_token_id,
+    )[0]["generated_text"]
+    return out[len(prompt):].strip()
+
+# ----------------------------
+# SafePak – 1122 Guide (System Instructions)
+# ----------------------------
+SAFEPAK_SYSTEM = """You are a multilingual emergency disaster response assistant named “SafePak – 1122 Guide”.
+Your role is to provide immediate, complete, to-the-point survival and rescue instructions for users facing any natural disaster, accident, war, or panic scenario in Pakistan or similar regions.
+Behave like a trained emergency coach — calm, practical, never panicked.
+
+CORE RULES:
+- Auto-detect language and answer in the user's language (Urdu/English/Punjabi/Sindhi/Pashto/Balochi/Saraiki).
+- Do not ask clarifying questions in normal mode. Assume urgency and provide full, specific, actionable guidance.
+- Use clear bullet points/checklists. Keep responses concise and life-saving.
+- If the user is panicked or crying, begin with a calming line.
+- Trigger relevant response module immediately based on keywords (e.g., fire, flood, earthquake, injury, trapped, explosion).
+- Provide offline-safe advice (assume no electricity/GPS/mobile signal).
+- If the user says “practice with me” or “drill mode,” start Simulation Mode (step-by-step Q&A): ask what they do next, then correct and guide them.
+- Do not store personal data. Do not give medical diagnoses; only emergency first aid.
+
+MODULES:
+1) Risk Identification & Immediate Response
+2) First Aid & Basic Medical Help
+3) Fire, Explosion, and Gas Leak Safety
+4) Earthquake, Flood, Storm, and War Survival
+5) Terrorism, Bombings, and Conflict Safety
+6) Evacuation Planning & Go-Bag Checklist
+7) Mental & Emotional First Aid
+8) SOPs for Police, Rescue, NGOs
+9) Home Equipment Safety (Solar, Gas, Water)
+"""
+
+# ----------------------------
+# Emergency keyword routing (simple templates)
+# ----------------------------
+KW = {
+    "fire": ["fire", "آگ", "جل", "burn", "smoke"],
+    "gas": ["gas", "لیک", "leak", "بو", "smell gas"],
+    "earthquake": ["earthquake", "زلزلہ", "jhatka"],
+    "flood": ["flood", "سیلاب", "barish", "overflow"],
+    "injury": ["injury", "bleeding", "زخمی", "خون"],
+    "explosion": ["explosion", "دھماکا", "blast"],
+    "conflict": ["gunfire", "firing", "گولیاں", "terror", "bomb"],
+    "trapped": ["trapped", "پھنس", "نکل نہیں"],
+    "child": ["child", "بچہ", "رو رہا"],
+    "storm": ["storm", "طوفان", "آندھی"],
+}
+
+def has_kw(text: str, key: str) -> bool:
+    low = text.lower()
+    for token in KW.get(key, []):
+        if token in low:
+            return True
+    return False
+
+def module_response(user_msg: str, lang: str) -> str:
+    bullets = []
+    calm_ur = "گھبرائیں نہیں — ایک گہری سانس لیں، میں آپ کے ساتھ ہوں۔"
+    calm_en = "Don't worry — take a deep breath; I'm with you."
+
     if lang == "ur":
-        return "safePak: میں آپ کی رہنمائی کے لیے حاضر ہوں۔ براہِ کرم سوال واضح کریں یا کوئی مثال دیں۔"
+        # Fire
+        if has_kw(user_msg, "fire"):
+            bullets += [
+                "فوراً جھک جائیں، دھواں اوپر جاتا ہے۔",
+                "ناک اور منہ پر گیلا کپڑا رکھیں۔",
+                "بجلی/گیس کے مین سوئچ بند کریں (اگر محفوظ ہو تو).",
+                "دروازے کھولنے سے پہلے ہاتھ سے گرمی چیک کریں؛ گرم ہو تو دوسرا راستہ لیں۔",
+                "سیڑھیاں استعمال کریں؛ لفٹ ہرگز نہیں۔",
+                "بچوں/بزرگوں کو ساتھ رکھیں، حساب رکھیں۔",
+            ]
+        # Gas Leak
+        elif has_kw(user_msg, "gas"):
+            bullets += [
+                "کسی بھی قسم کی چنگاری نہ بنائیں (سوئچ/لائٹر/فون فلیش نہیں).",
+                "کھڑکیاں/دروازے کھول کر ہوا دار کریں۔",
+                "گیس مین والو بند کریں۔",
+                "عمارت خالی کریں، محفوظ فاصلے پر جائیں۔",
+            ]
+        # Earthquake
+        elif has_kw(user_msg, "earthquake"):
+            bullets += [
+                "جھکیں، ڈھکیں، اور پکڑیں (Drop, Cover, Hold).",
+                "کھڑکیوں، شیشوں اور بھاری الماریوں سے دور رہیں۔",
+                "زلزلہ رکتے ہی عمارت کا محفوظ راستہ دیکھ کر باہر جائیں۔",
+                "سیڑھیاں استعمال کریں؛ لفٹ نہیں۔",
+            ]
+        # Flood / Storm
+        elif has_kw(user_msg, "flood") or has_kw(user_msg, "storm"):
+            bullets += [
+                "بجلی مین سوئچ بند کریں (اگر محفوظ ہو تو).",
+                "اونچی جگہ پر منتقل ہوں — کرنٹ لگنے کا خطرہ۔",
+                "گاڑی کو پانی سے نہ گزاریں؛ 12 انچ پانی بھی خطرناک ہے۔",
+                "گو-بیگ، شناختی دستاویزات، دوائیں ساتھ رکھیں۔",
+            ]
+        # Injury / bleeding
+        elif has_kw(user_msg, "injury"):
+            bullets += [
+                "خون بہہ رہا ہو تو دباؤ ڈال کر روکیں (clean cloth).",
+                "زخم کو صاف پانی سے دھوئیں؛ گہرا ہو تو فوری مدد لیں۔",
+                "سر/گردن/کمر چوٹ میں مریض کو نہ ہلائیں (spinal risk).",
+            ]
+        # Explosion / conflict
+        elif has_kw(user_msg, "explosion") or has_kw(user_msg, "conflict"):
+            bullets += [
+                "زمین پر لیٹ جائیں، کور لیں، کھڑکیوں سے دور رہیں۔",
+                "فائرنگ کی سمت کے مخالف کور ڈھونڈیں (دیوار/پکی رکاوٹ).",
+                "فوراً محفوظ راستہ دیکھ کر نکلیں — رش اور شور سے بچیں۔",
+            ]
+        # Trapped / child crying
+        elif has_kw(user_msg, "trapped") or has_kw(user_msg, "child"):
+            bullets += [
+                "خاموش اور آہستہ سانس — توانائی بچائیں۔",
+                "زور سے کھانسیں/آواز دیں صرف جب مدد قریب لگے۔",
+                "ہوا/روشنی کی سمت کا اندازہ رکھیں؛ دھواں ہو تو نیچے رہیں۔",
+            ]
+        else:
+            bullets += [
+                "فوراً اپنے آس پاس کے خطرات پہچانیں (بجلی، گیس، دھواں، پانی).",
+                "محفوظ جگہ/کور لیں، بچوں/بزرگوں کو ساتھ رکھیں۔",
+                "ایگزٹ راستہ ذہن میں رکھیں، لفٹ سے پرہیز کریں۔",
+            ]
+        lines = ["✅ یہاں آپ کیا کریں:"] + [f"- {b}" for b in bullets]
+        lines.append("🕊️ گھبرائیں نہیں — میں آپ کے ساتھ ہوں۔")
+        return "\n".join(lines)
     else:
-        return "safePak: I'm here to help. Please clarify your question or share an example."
-
-# ---- Toy RAG (in-memory) ----
-RAG_DOCS: List[Dict] = []
-def add_doc(text: str):
-    if not text.strip():
-        return 0
-    RAG_DOCS.append({"text": text, "lang": detect_lang(text)})
-    return len(RAG_DOCS)
-
-def retrieve(query: str, k: int = 3) -> List[str]:
-    q_tokens = set(re.findall(r"\w+", query.lower()))
-    scored = []
-    for d in RAG_DOCS:
-        d_tokens = set(re.findall(r"\w+", d["text"].lower()))
-        score = len(q_tokens & d_tokens)
-        if score:
-            scored.append((score, d["text"]))
-    scored.sort(reverse=True, key=lambda x: x[0])
-    return [t for _, t in scored[:k]]
-
-# ---- Load guardrails ----
-GUARD = Guardrails.from_yaml("guardrails.yaml")
+        if has_kw(user_msg, "fire"):
+            bullets += [
+                "Stay low — smoke rises.",
+                "Cover nose/mouth with a wet cloth.",
+                "Turn off electricity/gas if safe.",
+                "Check door heat with back of hand; use alternate exit if hot.",
+                "Use stairs, never elevators.",
+                "Keep children/elderly with you; headcount.",
+            ]
+        elif has_kw(user_msg, "gas"):
+            bullets += [
+                "No sparks (no switches/lighters/flash).",
+                "Ventilate — open windows/doors.",
+                "Shut the main gas valve.",
+                "Evacuate to a safe distance.",
+            ]
+        elif has_kw(user_msg, "earthquake"):
+            bullets += [
+                "Drop, Cover, and Hold On.",
+                "Stay away from windows/heavy furniture.",
+                "After shaking stops, exit via safe route.",
+                "Use stairs; avoid elevators.",
+            ]
+        elif has_kw(user_msg, "flood") or has_kw(user_msg, "storm"):
+            bullets += [
+                "Turn off main power (if safe).",
+                "Move to higher ground.",
+                "Do not drive through water; 12 inches can stall/float cars.",
+                "Grab go-bag, IDs, medicines.",
+            ]
+        elif has_kw(user_msg, "injury"):
+            bullets += [
+                "Apply direct pressure to stop bleeding.",
+                "Rinse wounds with clean water; seek help if deep.",
+                "Suspected spine injury — do not move the person.",
+            ]
+        elif has_kw(user_msg, "explosion") or has_kw(user_msg, "conflict"):
+            bullets += [
+                "Get low, take cover, away from windows.",
+                "Move opposite to gunfire; use solid cover.",
+                "Exit safely; avoid crowds and noise.",
+            ]
+        elif has_kw(user_msg, "trapped") or has_kw(user_msg, "child"):
+            bullets += [
+                "Stay calm, breathe slowly — conserve energy.",
+                "Signal only when rescuers are near.",
+                "Stay low if smoke; move toward fresh air/light.",
+            ]
+        else:
+            bullets += [
+                "Identify immediate hazards (electricity, gas, smoke, water).",
+                "Shelter/cover; keep kids/elderly close.",
+                "Know exits; avoid elevators.",
+            ]
+        lines = ["✅ Do this now:"] + [f"- {b}" for b in bullets]
+        lines.append("🕊️ Stay calm — I’m with you.")
+        return "\n".join(lines)
+
+# ----------------------------
+# Helplines formatter (always added for SOFT/BLOCK)
+# ----------------------------
+def helplines_block(lang: str) -> str:
+    if not GUARD.helplines:
+        return ""
+    hdr = "ہیلپ لائنز:" if lang == "ur" else "Helplines:"
+    lines = [hdr]
+    for h in GUARD.helplines:
+        nm = h.get("name", "").strip()
+        ph = h.get("phone", "").strip()
+        if nm and ph:
+            lines.append(f"{nm}:\n{ph}")
+    return "\n".join(lines)
+
+# ----------------------------
+# Simulation Mode (Step-by-step Q&A)
+# ----------------------------
+SIM_TRIGGER = re.compile(r"(practice with me|drill mode|drill|پریکٹس|ڈرل)", re.I)
+
+def start_sim_scenario(user_msg: str) -> Dict:
+    # pick scenario by keyword; default to earthquake
+    if has_kw(user_msg, "fire"):
+        scen = "fire"
+    elif has_kw(user_msg, "flood"):
+        scen = "flood"
+    elif has_kw(user_msg, "earthquake"):
+        scen = "earthquake"
+    else:
+        scen = "earthquake"
+    return {"active": True, "step": 1, "scenario": scen}
+
+def sim_prompt(state: Dict, lang: str) -> str:
+    scen = state.get("scenario", "earthquake")
+    if lang == "ur":
+        if state["step"] == 1:
+            if scen == "fire":
+                return "ڈرل شروع: کمرے میں آگ لگی ہے اور دھواں بڑھ رہا ہے۔ آپ سب سے پہلے کیا کریں گے؟"
+            if scen == "flood":
+                return "ڈرل شروع: پانی تیزی سے گھر میں داخل ہو رہا ہے۔ آپ سب سے پہلے کیا کریں گے؟"
+            return "ڈرل شروع: زلزلے کے جھٹکے محسوس ہو رہے ہیں۔ آپ سب سے پہلے کیا کریں گے؟"
+        elif state["step"] == 2:
+            return "اچھا—اب دوسرا قدم کیا ہوگا؟ (مختصر ایک لائن میں بتائیں)"
+        else:
+            return "آخری قدم کیا ہوگا جس سے آپ/گھر والے محفوظ رہیں گے؟"
+    else:
+        if state["step"] == 1:
+            if scen == "fire":
+                return "Drill: There is a room fire and smoke is rising. What is your FIRST action?"
+            if scen == "flood":
+                return "Drill: Water is rapidly entering the house. What is your FIRST action?"
+            return "Drill: You feel earthquake shaking. What is your FIRST action?"
+        elif state["step"] == 2:
+            return "Good — what is your SECOND step? (one short line)"
+        else:
+            return "What is the FINAL step to keep you/your family safe?"
+
+def sim_feedback(state: Dict, user_msg: str, lang: str) -> str:
+    scen = state.get("scenario", "earthquake")
+    tips_ur = {
+        "fire": ["جھک جائیں، دھوئیں سے نیچے رہیں", "گیلا کپڑا ناک/منہ پر", "لفٹ ہرگز نہیں — سیڑھیاں", "گرم دروازہ مت کھولیں"],
+        "flood": ["بجلی مین سوئچ بند", "اونچی جگہ/چھت پر جانا", "گاڑی/پانی سے گزرنے سے پرہیز", "گو-بیگ/شناختی دستاویزات ساتھ"],
+        "earthquake": ["Drop, Cover, Hold", "کھڑکیوں/بھاری اشیاء سے دور", "جھٹکے رکنے پر سیڑھیوں سے باہر", "لفٹ نہیں"],
+    }
+    tips_en = {
+        "fire": ["Stay low under smoke", "Wet cloth over nose/mouth", "Use stairs, not elevators", "Do not open hot doors"],
+        "flood": ["Turn off main power", "Move to higher ground/roof", "Avoid driving through water", "Grab go-bag/IDs"],
+        "earthquake": ["Drop, Cover, Hold", "Stay away from windows/heavy objects", "Exit via stairs after shaking", "No elevators"],
+    }
+    good = tips_ur if lang == "ur" else tips_en
+    # very light scoring by keyword overlap
+    score = 0
+    low = user_msg.lower()
+    for hint in good[scen]:
+        for tok in re.findall(r"\w+", hint.lower()):
+            if tok in low:
+                score += 1
+                break
+    if lang == "ur":
+        fb = "درست سمت میں ہیں۔ " if score >= 1 else "بہتر کریں: "
+        checklist = "\n- " + "\n- ".join(good[scen])
+        return fb + "محفوظ قدم یہ ہیں:" + checklist
+    else:
+        fb = "Good direction. " if score >= 1 else "Improve: "
+        checklist = "\n- " + "\n- ".join(good[scen])
+        return fb + "Follow these safe steps:" + checklist
+
+# ----------------------------
+# Chat logic
+# ----------------------------
+def build_prompt(user_msg: str, lang: str) -> str:
+    return f"<system>\n{SAFEPAK_SYSTEM}\n</system>\n<user>\n{user_msg}\n</user>"
 
-# ---- Chat logic ----
-def chat_fn(user_msg: str, chat_history: List[Tuple[str,str]], use_rag: bool):
+def chat_fn(user_msg: str, chat_history: List[Tuple[str, str]], sim_state: Dict):
     if not user_msg:
-        return chat_history
+        return chat_history, sim_state
 
     lang = detect_lang(user_msg)
     status, _ = GUARD.check(user_msg)
 
+    # Simulation trigger
+    if SIM_TRIGGER.search(user_msg):
+        sim_state = start_sim_scenario(user_msg)
+        reply = sim_prompt(sim_state, lang)
+        return chat_history + [(user_msg, reply)], sim_state
+
+    # Simulation step flow
+    if sim_state.get("active"):
+        # give feedback and advance
+        feedback = sim_feedback(sim_state, user_msg, lang)
+        sim_state["step"] += 1
+        if sim_state["step"] > 3:
+            sim_state = {"active": False, "step": 0, "scenario": None}
+            end_line = "ڈرل مکمل — شاباش! آپ نے بہتر فیصلہ سازی دیکھی۔ محفوظ رہیں۔" if lang == "ur" else "Drill complete — well done. Stay safe."
+            return chat_history + [(user_msg, feedback + "\n\n" + end_line)], sim_state
+        else:
+            prompt_q = sim_prompt(sim_state, lang)
+            return chat_history + [(user_msg, feedback + "\n\n" + prompt_q)], sim_state
+
+    # Normal emergency routing (no questions)
     if status == "BLOCK":
-        msg = GUARD.refusal_msg_ur if lang=="ur" else GUARD.refusal_msg_en
-        return chat_history + [(user_msg, msg)]
+        msg = GUARD.refusal_msg_ur if lang == "ur" else GUARD.refusal_msg_en
+        helpb = helplines_block(lang)
+        if helpb:
+            msg += "\n\n" + helpb
+        return chat_history + [(user_msg, msg)], sim_state
 
-    context = retrieve(user_msg, k=3) if use_rag else []
-    reply = tiny_response(user_msg, chat_history, lang)
+    # Immediate checklist from modules
+    checklist = module_response(user_msg, lang)
 
-    if context:
-        if lang=="ur":
-            reply += "\n\n(متعلقہ مواد)\n- " + "\n- ".join(context)
-        else:
-            reply += "\n\n(Related)\n- " + "\n- ".join(context)
+    # Compose with system prompt for any extra refinement (short)
+    prompt = build_prompt(user_msg, lang)
+    model_tail = llm_reply(prompt, max_new_tokens=80, temperature=0.5)
 
+    # Soft case: prepend a gentle note + add helplines
     if status == "SOFT":
-        note = ("احتیاط: آپ کے سوال میں حساس موضوعات شامل ہو سکتے ہیں۔ معلومات عمومی رہنمائی کی حد تک دی جا رہی ہے۔\n\n"
-                if lang=="ur" else
-                "Note: Your question may include sensitive topics. Responding with general guidance only.\n\n")
-        reply = note + reply
-
-    return chat_history + [(user_msg, reply)]
-
-# ---- Ingest files ----
-def ingest_files(files: List[gr.File]):
-    for f in files or []:
-        try:
-            text = open(f.name, 'r', encoding='utf-8', errors='ignore').read()
-            add_doc(text)
-        except Exception as e:
-            print("Failed to read", f.name, e)
-    return f"Docs in store: {len(RAG_DOCS)}"
-
-# ---- UI ----
-with gr.Blocks(title="safePak – Starter", css=".wrap {max-width: 900px; margin: 0 auto}") as demo:
-    gr.Markdown("# safePak (Starter)\n**لوکل-فرسٹ** ڈیزائن کے ساتھ ایک سادہ ڈیمو۔")
-
-    with gr.Row():
-        use_rag = gr.Checkbox(label="Use local knowledge (RAG)", value=True)
-        with gr.Column():
-            file_uploader = gr.Files(label="Upload .txt files", file_types=[".txt"], file_count="multiple")
-            ingest_btn = gr.Button("Ingest files")
-            rag_status = gr.Markdown("Docs in store: 0")
-
-    chatbot = gr.Chatbot(height=420)
-    msg = gr.Textbox(label="پیغام / Message", placeholder="یہاں لکھیں…", lines=2)
+        note = "احتیاط: یہ حساس موضوع ہے۔\n\n" if lang == "ur" else "Note: This is a sensitive topic.\n\n"
+        checklist = note + checklist
+        helpb = helplines_block(lang)
+        if helpb:
+            checklist += "\n\n" + helpb
+
+    final = checklist  # keep concise; tail is optional if needed
+    return chat_history + [(user_msg, final)], sim_state
+
+# ----------------------------
+# UI
+# ----------------------------
+with gr.Blocks(title="SafePak – 1122 Guide") as demo:
+    gr.Markdown("## SafePak – 1122 Guide\nMultilingual emergency assistant (CPU-only).")
+
+    state = gr.State({"active": False, "step": 0, "scenario": None})
+
+    chat = gr.Chatbot(height=480)
+    msg = gr.Textbox(label="Message / پیغام", placeholder="Type here… / یہاں لکھیں…", lines=2)
     clear = gr.Button("Clear")
 
-    ingest_btn.click(ingest_files, [file_uploader], [rag_status])
-    msg.submit(lambda u,h,r: (chat_fn(u,h,r), ""), [msg,chatbot,use_rag], [chatbot,msg])
-    clear.click(lambda: ([], "Docs in store: 0"), [], [chatbot, rag_status])
+    def on_submit(u, h, s):
+        new_h, new_s = chat_fn(u, h, s)
+        return new_h, "", new_s
+
+    msg.submit(on_submit, [msg, chat, state], [chat, msg, state])
+    clear.click(lambda: ([], {"active": False, "step": 0, "scenario": None}), [], [chat, state])
 
 if __name__ == "__main__":
     demo.launch()