import os import re from typing import List, Tuple, Dict from dataclasses import dataclass import gradio as gr import torch from transformers import AutoTokenizer, AutoModelForCausalLM # ---------------- Language detection ---------------- ARABIC_RE = re.compile(r"[\u0600-\u06FF]") def detect_lang(text: str) -> str: return "ur" if ARABIC_RE.search(text) else "en" # ---------------- Guardrails ---------------- @dataclass class Guardrails: refusal_msg_ur: str refusal_msg_en: str blocked_patterns: List[str] soft_patterns: List[str] helplines: List[dict] @classmethod def from_yaml(cls, path: str): import yaml with open(path, "r", encoding="utf-8") as f: data = yaml.safe_load(f) return cls( refusal_msg_ur=data.get("refusal_msg_ur", ""), refusal_msg_en=data.get("refusal_msg_en", ""), blocked_patterns=data.get("blocked_patterns", []), soft_patterns=data.get("soft_patterns", []), helplines=data.get("helplines", []), ) def check(self, text: str) -> Tuple[str, str]: low = text.lower() for p in self.blocked_patterns: if re.search(p, low): return ("BLOCK", p) for p in self.soft_patterns: if re.search(p, low): return ("SOFT", p) return ("OK", "") GUARD = Guardrails.from_yaml("guardrails.yaml") # ---------------- Model ---------------- MODEL_ID = os.environ.get("SAFEPak_MODEL_ID", "Qwen/Qwen2-0.5B-Instruct") try: tok = AutoTokenizer.from_pretrained(MODEL_ID) model = AutoModelForCausalLM.from_pretrained(MODEL_ID, torch_dtype=torch.float32) except Exception as e: print("LLM load failed, fallback active:", e) tok = model = None def llm_reply(prompt: str, max_new_tokens: int = 140, temperature: float = 0.45) -> str: if not (tok and model): return "" ids = tok.encode(prompt, return_tensors="pt") out = model.generate( ids, max_new_tokens=max_new_tokens, do_sample=True, temperature=temperature, top_p=0.9, repetition_penalty=1.12, pad_token_id=tok.eos_token_id, ) gen_ids = out[0][ids.shape[1]:] return tok.decode(gen_ids, skip_special_tokens=True).strip() # ---------------- Helpline rendering ---------------- def _clean_phone_for_links(p: str) -> str: return "+" + "".join(ch for ch in p if ch.isdigit()) def phone_tel_link(p: str) -> str: return f"tel:{_clean_phone_for_links(p)}" def phone_wa_link(p: str) -> str: num = _clean_phone_for_links(p).lstrip("+") return f"https://wa.me/{num}" def render_quick_helplines_md(lang: str = "ur") -> str: if not GUARD.helplines: return "" title = "### فوری ہیلپ لائنز" if lang == "ur" else "### Quick Helplines" rows = [title] for h in GUARD.helplines: name = h.get("name","").strip() phone = h.get("phone","").strip() if not (name and phone): continue tel = phone_tel_link(phone) wa = phone_wa_link(phone) rows.append(f"**{name}** \n{phone}\n[📞 Call]({tel})   [💬 WhatsApp]({wa})") return "\n\n".join(rows) 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: tel = phone_tel_link(ph) wa = phone_wa_link(ph) lines.append(f"**{nm}**\n{ph}\n[📞 Call]({tel}) | [💬 WhatsApp]({wa})") return "\n\n".join(lines) # ---------------- Core prompt ---------------- SAFEPAK_SYSTEM = """ You are SafePak – 1122 Guide, a multilingual emergency disaster assistant. Always reply in user's language (Urdu/English/regional). Return medium-length, action-oriented checklists (5–7 bullets) and, if evacuation is implied or requested, append a Go-Bag sublist (4–8 items). Never panic. Provide offline-safe steps. End with a calming line. No personal data storage. No medical diagnoses (first aid only). """ # ---------------- Variety helpers ---------------- def _fingerprint(text: str) -> set: toks = re.findall(r"[A-Za-z\u0600-\u06FF]{3,}", text.lower()) return set(toks) def diversify_if_similar(prev: str, current: str, user_msg: str, lang: str) -> str: if not prev: return current fp_prev = _fingerprint(prev) fp_cur = _fingerprint(current) overlap = len(fp_prev & fp_cur) if overlap > max(12, int(len(fp_cur) * 0.55)): prompt = ( f"\nAvoid repeating earlier phrasing. Keep the same structure but change wording and add 2 scenario-specific, non-generic tips.\n" f"Use user's language, stay concise and practical.\n\n" f"\nMessage:\n{user_msg}\nCurrent answer:\n{current}\n" ) alt = llm_reply(prompt, max_new_tokens=160, temperature=0.6) return alt if alt else current return current # ---------------- Prebaked Urdu FAQs (exact templates you provided) ---------------- # ہم regex-based ہلکا میچ استعمال کر رہے ہیں تاکہ اردو عبارت کے عام انداز پکڑ سکیں PREBAKED_UR: List[Dict[str,str]] = [ { "q_re": r"(سیلاب).*(گھر).*(ہوں|ہو)", # سیلاب آ گیا! میں گھر میں ہوں "answer": "🚨 فوری اقدامات:\n- فوراً اونچی جگہ (2+ فلور) کی طرف ہٹیں۔\n- گہرے گھاٹ، ندیوں کے کنارے، یا چھوٹے دریاؤں کے قریب نہ رہیں۔\n- اگر گاڑی میں ہیں، تو فوراً گاڑی چھوڑ کر ہٹیں — گاڑی بہہ سکتی ہے۔\n\n📌 اگلے اقدامات:\n- 1122 یا 1500 پر فون کریں۔\n- اپنے فون کو ہائی ہولڈ رکھیں — پانی میں بھیگنے سے بچائیں ۔\n\n💡 کلیدی نکات:\n- پانی میں ہاتھ پاؤں نہ ڈالیں — گہرائی یا تیز دھارا کا خطرہ ہے۔\n- سیلاب کے بعد گھر میں داخل نہ ہوں جب تک کہ پانی نہ روکا گیا ہو۔" }, { "q_re": r"(زلزلہ).*(آ).*(رہا|رہی|گیا|گئی)|(جھٹک|ہل رہا)", "answer": "🚨 فوری اقدامات:\n- 'ڈاون ڈاؤن' (Drop, Cover, Hold On):\n - گریں (Drop)\n - ٹیبل یا مضبوط دیوار کے نیچے چھپیں (Cover)\n - سر کو ہاتھوں سے چھپائیں (Hold On)\n- کھڑے ہو کر دروازے یا کھڑکیوں کی طرف نہ بھاگیں — شیشے ٹوٹ سکتے ہیں۔\n\n📌 اگلے اقدامات:\n- زلزلہ ختم ہونے کے بعد فوراً گھر سے باہر نکلیں — گری سے گری ہوئی عمارت گر سکتی ہے۔\n- بجلی، گیس، پانی کے پائپ بند کریں (اگر گیس کی بو آ رہی ہو)۔\n\n💡 کلیدی نکات:\n- زلزلہ کے دوران کسی چیز کو ہلائیں یا چھوڑیں — ہلاکت خیز ہو سکتا ہے۔\n- گاڑی میں ہوں تو گاڑی کے اندر بیٹھیں — باہر نہ نکلیں۔" }, { "q_re": r"(آگ|جل).*(لگ|گئی|گیا)", "answer": "🚨 فوری اقدامات:\n- فوراً گھر سے باہر نکلیں — دھواں سانس کے لیے خطرہ ہے۔\n- ہاتھ سے چہرہ ڈھانپیں اور زمین پر چلتے رہیں (دھواں اوپر ہوتا ہے)۔\n- کوئی بجلی، فون، یا شمع نہ چلائیں — چمک آتشباری کر سکتی ہے۔\n\n📌 اگلے اقدامات:\n- 1122 پر فون کریں — آگ کے بارے میں مطلع کریں۔\n- رشتہ داروں کو مطلع کریں کہ آپ کہاں ہیں۔\n\n💡 کلیدی نکات:\n- زندگی کو بچانا آگ کو بچانے سے زیادہ اہم ہے۔\n- آگ کے بعد گھر میں داخل نہ ہوں جب تک کہ ٹیم کا اجازت نہ ہو۔" }, { "q_re": r"(گیس).*(لیک|بو|رس|smell)", "answer": "🚨 فوری اقدامات:\n- گیس کا سوئچ بند کریں (اسٹو کے بالکل نیچے)۔\n- کوئی بجلی، فون، یا شمع نہ چلائیں — چمک آتشباری کر سکتی ہے۔\n- فوراً گھر سے باہر نکلیں — گیس ہلکی سی بھی ہو سکتی ہے۔\n\n📌 اگلے اقدامات:\n- 1122 یا 1500 پر فون کریں — گیس کمپنی کو مطلع کریں۔\n- گھر میں داخل نہ ہوں جب تک کہ گیس چیک نہ ہو جائے۔\n\n💡 کلیدی نکات:\n- گیس لیک کا خطرہ ہر وقت موجود ہے — ہر ماہ چیک کروائیں۔\n- گیس ڈیٹیکٹر لگائیں — اگر ہے، تو اسے فوراً چالو کریں۔" }, { "q_re": r"(کلاؤڈ\s*برسٹ|cloud\s*burst)", "answer": "🚨 فوری اقدامات:\n- فوراً اونچی جگہ (پہاڑی، چوٹی) کی طرف ہٹیں۔\n- گہرے گھاٹ، ندیوں کے کنارے، یا چھوٹے دریاؤں کے قریب نہ رہیں۔\n- اگر گاڑی میں ہیں، تو فوراً گاڑی چھوڑ کر ہٹیں — گاڑی بہہ سکتی ہے۔\n\n📌 اگلے اقدامات:\n- 1122 پر فون کریں — ہیلپ مانگیں۔\n- اپنے فون کو ہائی ہولڈ رکھیں — پانی میں نہ ڈالیں۔\n\n💡 کلیدی نکات:\n- کلاؤڈ برسٹ کے بعد سیلاب فوری ہوتا ہے — اس لیے وقت ضائع نہ کریں۔\n- اگر زمین ہل رہی ہو، تو یہ سیلاب کا اشارہ ہے — فوراً ہٹیں۔" }, { "q_re": r"(بجلی).*(گر|گری|گر گئی|گرگئی)|lightning|electr", "answer": "🚨 فوری اقدامات:\n- فوراً بجلی کے سوئچ بند کریں (اگر ممکن ہو)۔\n- کوئی بجلی کا آلہ نہ چلائیں — شرارت یا چوٹ لگ سکتی ہے۔\n- ہاتھ پاؤں نہ ڈالیں بلکہ ہلکی روشنی (ٹارچ) استعمال کریں۔\n\n📌 اگلے اقدامات:\n- 1122 پر فون کریں — بجلی کمپنی کو مطلع کریں۔\n- ایمرجنسی ٹارچ یا فون کی روشنی استعمال کریں۔\n\n💡 کلیدی نکات:\n- بجلی گرنے کے بعد کسی بھی الیکٹریکل آلے کو چلانے سے گریز کریں۔\n- ہر 6 ماہ بجلی کے سوئچ اور کیبلز چیک کروائیں۔" }, { "q_re": r"(^|\s)SOS(\s|!|۔)|مدد\s*کرو", "answer": "🚨 SOS فائر ہو گیا!\n- فوراً 1122 پر فون کریں — ہیلپ ڈیسک آپ کو مدد کرے گی۔\n- اگر فون کنکشن ہو، تو میں فوراً 1122 پر فون کرتا ہوں...\n- اپنے آپ کو ٹھیک رکھیں — آپ کے ساتھ ہے۔\n\n💡 کلیدی نکات:\n- SOS کا وقت ہے — فوراً 1122 پر فون کریں۔\n- ہر ہفتے ایمرجنسی بیگ چیک کریں: پانی، فوڈ، ٹارچ، فون چارجر، فارمیسی۔" }, { "q_re": r"(سیلاب).*(بعد|ختم|اتر)", "answer": "🚨 ایمرجنسی کے بعد:\n- گھر میں داخل نہ ہوں جب تک کہ پانی نہ روکا گیا ہو۔\n- گیس، بجلی، پانی کے پائپ چیک کروائیں — خطرہ ہو سکتا ہے۔\n- اپنے فون کو چارج کریں — ہیلپ ڈیسک سے رابطہ کریں۔\n\n📌 اگلے اقدامات:\n- 1122 پر فون کریں — اپنی حالت بتائیں۔\n- رشتہ داروں کو مطلع کریں کہ آپ کہاں ہیں۔\n\n💡 کلیدی نکات:\n- پانی کے بعد گرد و غبار، کیمیکل، یا بیکٹیریا ہو سکتے ہیں — کپڑے بدلیں۔\n- ایمرجنسی بیگ میں پانی، فوڈ، فارمیسی، ٹارچ وغیرہ رکھیں" }, ] def match_prebaked_ur(user_msg: str) -> str: """اگر یوزر میسج اردو میں ہو اور کسی پری باکڈ ٹیمپلیٹ سے میچ کرے تو وہی جواب لوٹائیں۔""" if detect_lang(user_msg) != "ur": return "" text = user_msg.strip() for item in PREBAKED_UR: pattern = item.get("q_re") try: if re.search(pattern, text, flags=re.IGNORECASE): return item.get("answer","").strip() except re.error: continue return "" # ---------------- Base checklist + Go-Bag ---------------- def base_checklist_and_bag(user_msg: str, lang: str) -> Tuple[str, List[str]]: low = user_msg.lower() evac = False if any(k in low for k in ["fire","آگ","smoke"]): evac = True checklist = ("✅ ابھی یہ کریں:\n" "- جھک جائیں/Stay low — دھواں اوپر جاتا ہے\n" "- ناک/منہ پر گیلا کپڑا رکھیں\n" "- اگر محفوظ ہو تو بجلی/گیس مین بند کریں\n" "- گرم دروازہ مت کھولیں؛ دوسرا راستہ لیں\n" "- سیڑھیاں استعمال کریں — لفٹ نہیں\n" "- بچوں/بزرگوں کو ساتھ رکھیں\n" "🕊️ گھبرائیں نہیں — میں آپ کے ساتھ ہوں۔" ) if lang=="ur" else ( "✅ Do this now:\n" "- Stay low — smoke rises\n" "- Wet cloth over nose/mouth\n" "- Cut power/gas if safe\n" "- Do not open hot doors; use alternate exit\n" "- Use stairs, never elevators\n" "- Keep kids/elderly close\n" "🕊️ Stay calm — I’m with you." ) bag = (["شناختی کارڈ/کاغذات","نقدی","موبائل+پاور بینک","پانی","خشک کھانا","دوائیں","ہلکی چادر","ٹارچ"] if lang=="ur" else ["IDs/documents","Cash","Phone+power bank","Water","Dry food","Medicines","Light blanket","Flashlight"]) elif any(k in low for k in ["زلزلہ","earthquake","tremor","shaking"]): evac = True checklist = ("✅ ابھی یہ کریں:\n" "- Drop, Cover, Hold — جھکیں، ڈھکیں، پکڑیں\n" "- کھڑکیوں/بھاری اشیاء سے دور رہیں\n" "- جھٹکے رکنے پر سیڑھیوں سے باہر نکلیں (لفٹ نہیں)\n" "- کھلی جگہ میں رکیں\n" "🕊️ حوصلہ رکھیں — میں آپ کے ساتھ ہوں۔" ) if lang=="ur" else ( "✅ Do this now:\n" "- Drop, Cover, Hold On\n" "- Stay away from windows/heavy objects\n" "- After shaking, exit via stairs (no elevators)\n" "- Wait in an open safe area\n" "🕊️ Stay calm — I’m with you." ) bag = (["شناختی کارڈ/کاغذات","موبائل+پاور بینک","پانی","خشک کھانا","دوائیں","وسل","چھوٹی فرسٹ ایڈ","ٹارچ"] if lang=="ur" else ["IDs/documents","Phone+power bank","Water","Dry food","Medicines","Whistle","Small first-aid","Flashlight"]) elif any(k in low for k in ["سیلاب","flood","overflow","water entering","طوفان","storm","آندھی"]): evac = True checklist = ("✅ ابھی یہ کریں:\n" "- اگر محفوظ ہو تو بجلی مین سوئچ بند کریں\n" "- اونچی جگہ/چھت پر منتقل ہوں\n" "- پانی میں گاڑی نہ چلائیں\n" "- بچوں/بزرگوں کو پہلے نکالیں\n" "🕊️ گھبرائیں نہیں — آپ محفوظ نکل سکتے ہیں۔" ) if lang=="ur" else ( "✅ Do this now:\n" "- Turn off main power if safe\n" "- Move to higher ground/roof\n" "- Do not drive through water\n" "- Evacuate kids/elderly first\n" "🕊️ Stay calm — you can get to safety." ) bag = (["شناختی کارڈ/کاغذات","نقدی","موبائل+پاور بینک","پانی","خشک کھانا","دوائیں","واٹر پروف بیگ","چپل/بوٹس"] if lang=="ur" else ["IDs/documents","Cash","Phone+power bank","Water","Dry food","Medicines","Waterproof pouches","Boots/Sandals"]) elif any(k in low for k in ["gas","گیس","leak","smell gas","بو"]): checklist = ("✅ ابھی یہ کریں:\n" "- کوئی چنگاری/سوئچ/لائٹر استعمال نہ کریں\n" "- کھڑکیاں/دروازے کھول کر ہوا دار کریں\n" "- گیس مین والو بند کریں\n" "- محفوظ فاصلے پر چلے جائیں\n" "🕊️ پرسکون رہیں — میں رہنمائی کر رہا/رہی ہوں۔" ) if lang=="ur" else ( "✅ Do this now:\n" "- No sparks (no switches/lighters/flash)\n" "- Ventilate: open windows/doors\n" "- Shut the main gas valve\n" "- Move to a safe distance\n" "🕊️ Stay calm — I’m guiding you." ) bag = (["ماسک/گیلا کپڑا","شناختی کارڈ","موبائل+پاور بینک","پانی","دوائیں"] if lang=="ur" else ["Mask/wet cloth","IDs","Phone+power bank","Water","Medicines"]) elif any(k in low for k in ["injury","زخمی","bleeding","cut","fracture","burn"]): checklist = ("✅ ابھی یہ کریں:\n" "- خون ہو تو مضبوط دباؤ دیں (صاف کپڑا)\n" "- زخم صاف پانی سے دھوئیں\n" "- گہرا زخم/زیادہ خون ہو تو فوری مدد لیں\n" "- گردن/کمر چوٹ میں مریض کو نہ ہلائیں\n" "🕊️ پرسکون رہیں — آپ سنبھال سکتے ہیں۔" ) if lang=="ur" else ( "✅ Do this now:\n" "- Apply firm direct pressure (clean cloth)\n" "- Rinse with clean water\n" "- Deep/heavy bleeding → seek urgent help\n" "- Suspected spine injury: do not move\n" "🕊️ Stay calm — you’ve got this." ) bag = (["فرسٹ ایڈ کٹ","صاف کپڑا/بینڈیج","پانی","درد کش دوا","شناختی کارڈ"] if lang=="ur" else ["First-aid kit","Clean cloth/bandage","Water","Pain reliever","ID"]) else: checklist = ("✅ فوری رہنمائی:\n" "- اردگرد خطرات پہچانیں (بجلی/گیس/دھواں/پانی)\n" "- محفوظ کور یا اخراج کا راستہ منتخب کریں\n" "- بچوں/بزرگوں کو ساتھ رکھیں\n" "- لفٹ سے پرہیز کریں\n" "🕊️ گھبرائیں نہیں — میں آپ کے ساتھ ہوں۔" ) if lang=="ur" else ( "✅ General guidance:\n" "- Identify hazards (electricity/gas/smoke/water)\n" "- Choose safe cover or exit route\n" "- Keep kids/elderly close\n" "- Avoid elevators\n" "🕊️ Stay calm — I’m with you." ) evac = True bag = (["شناختی کارڈ/کاغذات","نقدی","فون+پاور بینک","پانی","خشک کھانا","دوائیں"] if lang=="ur" else ["IDs/documents","Cash","Phone+power bank","Water","Dry food","Medicines"]) if evac: if lang=="ur": checklist += "\n\n🎒 ساتھ کیا لیں (Go-Bag):\n" + "\n".join(f"- {i}" for i in bag) else: checklist += "\n\n🎒 Take with you (Go-Bag):\n" + "\n".join(f"- {i}" for i in bag) return checklist, bag # ---------------- Refinement (more question-specific) ---------------- def format_response(base: str, user_msg: str, lang: str) -> str: words = re.findall(r"[A-Za-z\u0600-\u06FF0-9_]+", user_msg)[:20] key_hint = ", ".join(words) system_rules = ( "Return a medium-length emergency guide in the user's language: 5–7 direct action bullets, " "then a Go-Bag sublist (4–8 items) IF evacuation is implied or requested. " "Be specific to the user's message, location constraints, and actors (e.g., children, elderly). " "Use if-then phrasing where helpful. Do NOT repeat boilerplate from earlier answers." ) refine_prompt = ( f"\n{SAFEPAK_SYSTEM}\n{system_rules}\n\n" f"\nMessage (keywords): {key_hint}\nFull message:\n{user_msg}\n\n" f"Draft to improve:\n{base}\n" ) refined = llm_reply(refine_prompt, max_new_tokens=140, temperature=0.45) return refined if refined else base # ---------------- Chat logic (messages format) ---------------- def chat_fn(user_msg: str, chat_messages: List[Dict[str,str]], sim_state: dict): if not user_msg: return chat_messages, sim_state lang = detect_lang(user_msg) status, _ = GUARD.check(user_msg) # BLOCK if status == "BLOCK": msg = GUARD.refusal_msg_ur if lang == "ur" else GUARD.refusal_msg_en msg += "\n\n" + helplines_block(lang) return chat_messages + [ {"role":"user","content":user_msg}, {"role":"assistant","content":msg} ], sim_state # ✅ NEW: Prebaked Urdu templates first pre_ur = match_prebaked_ur(user_msg) if pre_ur: final = pre_ur + "\n\n" + helplines_block("ur") return chat_messages + [ {"role":"user","content":user_msg}, {"role":"assistant","content":final} ], sim_state # Base checklist + Go-Bag base, _ = base_checklist_and_bag(user_msg, lang) # SOFT prepend note if status == "SOFT": base = ("⚠️ محتاط رہیں: یہ موضوع حساس ہے۔\n\n" + base) if lang=="ur" else ("⚠️ Note: sensitive topic.\n\n" + base) # Refine with LLM (and append helplines) final = format_response(base, user_msg, lang) final += "\n\n" + helplines_block(lang) # Diversify vs previous assistant message prev_assistant = "" for msg in reversed(chat_messages): if msg.get("role") == "assistant": prev_assistant = msg.get("content","") break final = diversify_if_similar(prev_assistant, final, user_msg, lang) return chat_messages + [ {"role":"user","content":user_msg}, {"role":"assistant","content":final} ], sim_state # ---------------- UI ---------------- with gr.Blocks(title="SafePak – 1122 Guide") as demo: gr.Markdown("## SafePak – 1122 Guide\nMultilingual emergency assistant (CPU-only).") gr.Markdown(render_quick_helplines_md("ur")) state = gr.State({"active": False, "step": 0, "scenario": None}) chatbot = gr.Chatbot(height=420, type="messages") msg = gr.Textbox(label="پیغام / Message", placeholder="یہاں لکھیں…", lines=2) send = gr.Button("Send / بھیجیں") clear = gr.Button("Clear") 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, chatbot, state], [chatbot, msg, state]) send.click(on_submit, [msg, chatbot, state], [chatbot, msg, state]) clear.click(lambda: ([], {"active": False, "step": 0, "scenario": None}), [], [chatbot, state]) if __name__ == "__main__": demo.queue().launch( server_name="0.0.0.0", server_port=7860, share=False )