Spaces:
Build error
Build error
Download app.py from m0hsin123/Accidental_News_Classification: direct link, hf CLI and curl.
- Browser
- Download file 13.7 kB
-
https://huggingface.co/spaces/m0hsin123/Accidental_News_Classification/resolve/main/app.py
- Command line
-
hf download hf://spaces/m0hsin123/Accidental_News_Classification/app.py
-
curl -L -o app.py https://huggingface.co/spaces/m0hsin123/Accidental_News_Classification/resolve/main/app.py
13.7 kB
| import os | |
| os.environ.setdefault("TOKENIZERS_PARALLELISM", "false") | |
| import gc | |
| import html | |
| import json | |
| from functools import lru_cache | |
| from pathlib import Path | |
| import gradio as gr | |
| import h5py | |
| import numpy as np | |
| import torch | |
| from transformers import ( | |
| AutoConfig, | |
| AutoModelForSequenceClassification, | |
| AutoModelForTokenClassification, | |
| AutoTokenizer, | |
| pipeline, | |
| ) | |
| BASE_DIR = Path(__file__).resolve().parent | |
| CFG = json.loads((BASE_DIR / "config.json").read_text(encoding="utf-8")) | |
| DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu") | |
| TASK_A_LABELS = {int(k): v for k, v in CFG["task_a_labels"].items()} | |
| TASK_B_LABELS = {int(k): v for k, v in CFG["task_b_labels"].items()} | |
| # sagorsarker/mbert-bengali-ner semantic labels | |
| NER_ID2LABEL = { | |
| 0: "O", | |
| 1: "B-PER", | |
| 2: "I-PER", | |
| 3: "B-ORG", | |
| 4: "I-ORG", | |
| 5: "B-LOC", | |
| 6: "I-LOC", | |
| } | |
| NER_LABEL2ID = {v: k for k, v in NER_ID2LABEL.items()} | |
| # ============================================================ | |
| # LOCAL H5 WEIGHT LOADING | |
| # ============================================================ | |
| def get_local_weight_path(filename: str) -> Path: | |
| path = BASE_DIR / filename | |
| if not path.exists(): | |
| raise FileNotFoundError( | |
| f"{filename} was not found in the Space repository root. " | |
| f"Upload {filename} beside app.py." | |
| ) | |
| return path | |
| def list_h5_datasets(h5_file): | |
| names = [] | |
| def visitor(name, obj): | |
| if isinstance(obj, h5py.Dataset): | |
| names.append(name) | |
| h5_file.visititems(visitor) | |
| return names | |
| def load_pytorch_h5_weights(model: torch.nn.Module, h5_path: Path): | |
| """ | |
| Load the custom HDF5 format used in training: | |
| with h5py.File(h5_path, "w") as f: | |
| for name, param in model.state_dict().items(): | |
| f.create_dataset(name, data=param.cpu().numpy()) | |
| This is NOT a Keras H5 model. | |
| Tensors are copied directly into the rebuilt PyTorch model state one-by-one. | |
| """ | |
| state = model.state_dict() | |
| model_keys = set(state.keys()) | |
| with h5py.File(h5_path, "r") as f: | |
| file_keys = set(list_h5_datasets(f)) | |
| # Compatibility-only buffers can differ across transformers versions. | |
| def benign(key): | |
| return key.endswith("position_ids") or key.endswith("token_type_ids") | |
| missing = sorted( | |
| key for key in (model_keys - file_keys) | |
| if not benign(key) | |
| ) | |
| unexpected = sorted( | |
| key for key in (file_keys - model_keys) | |
| if not benign(key) | |
| ) | |
| if missing or unexpected: | |
| raise RuntimeError( | |
| "\n".join([ | |
| f"Architecture mismatch for {h5_path.name}.", | |
| f"Missing learned keys ({len(missing)}): {missing[:15]}", | |
| f"Unexpected learned keys ({len(unexpected)}): {unexpected[:15]}", | |
| "The model must be recreated with the same Transformers " | |
| "architecture/version family used during training." | |
| ]) | |
| ) | |
| with torch.no_grad(): | |
| for key in file_keys: | |
| if key not in state: | |
| continue | |
| arr = np.asarray(f[key]) | |
| src = torch.from_numpy(arr) | |
| dst = state[key] | |
| if tuple(src.shape) != tuple(dst.shape): | |
| raise RuntimeError( | |
| f"Shape mismatch at {key}: " | |
| f"H5={tuple(src.shape)} vs model={tuple(dst.shape)}" | |
| ) | |
| dst.copy_(src.to(dtype=dst.dtype, device=dst.device)) | |
| del state | |
| gc.collect() | |
| return model | |
| # ============================================================ | |
| # XLM-R TOKENIZER + CLASSIFIERS | |
| # ============================================================ | |
| def get_xlmr_tokenizer(): | |
| return AutoTokenizer.from_pretrained(CFG["base_model"]) | |
| def build_classifier(num_labels: int, labels: dict): | |
| id2label = {int(i): label for i, label in labels.items()} | |
| label2id = {label: i for i, label in id2label.items()} | |
| # Same architecture as: | |
| # AutoModelForSequenceClassification.from_pretrained( | |
| # "xlm-roberta-base", num_labels=... | |
| # ) | |
| # | |
| # We instantiate from config so we don't first download a second full copy | |
| # of XLM-R base weights. The local H5 contains the COMPLETE fine-tuned state_dict. | |
| config = AutoConfig.from_pretrained( | |
| CFG["base_model"], | |
| num_labels=num_labels, | |
| id2label=id2label, | |
| label2id=label2id, | |
| ) | |
| return AutoModelForSequenceClassification.from_config(config) | |
| def get_task_a_model(): | |
| model = build_classifier(2, TASK_A_LABELS) | |
| model = load_pytorch_h5_weights( | |
| model, | |
| get_local_weight_path(CFG["task_a_file"]) | |
| ) | |
| model.to(DEVICE) | |
| model.eval() | |
| return model | |
| def get_task_b_model(): | |
| # Lazy load: this 10-class model is loaded only when Task A says Accidental. | |
| model = build_classifier(10, TASK_B_LABELS) | |
| model = load_pytorch_h5_weights( | |
| model, | |
| get_local_weight_path(CFG["task_b_file"]) | |
| ) | |
| model.to(DEVICE) | |
| model.eval() | |
| return model | |
| def predict_classifier(text: str, model, labels: dict): | |
| tokenizer = get_xlmr_tokenizer() | |
| encoded = tokenizer( | |
| text, | |
| max_length=int(CFG["max_length"]), | |
| padding="max_length", | |
| truncation=True, | |
| return_tensors="pt", | |
| ) | |
| encoded = {k: v.to(DEVICE) for k, v in encoded.items()} | |
| with torch.inference_mode(): | |
| outputs = model( | |
| input_ids=encoded["input_ids"], | |
| attention_mask=encoded["attention_mask"], | |
| ) | |
| probs = torch.softmax(outputs.logits, dim=-1)[0].cpu().numpy() | |
| pred_id = int(np.argmax(probs)) | |
| scores = { | |
| labels[i]: float(probs[i]) | |
| for i in range(len(labels)) | |
| } | |
| return pred_id, labels[pred_id], scores | |
| # ============================================================ | |
| # BENGALI NER / EXACT LOCATION EXTRACTION | |
| # ============================================================ | |
| def get_ner_pipeline(): | |
| tokenizer = AutoTokenizer.from_pretrained(CFG["ner_model"]) | |
| model = AutoModelForTokenClassification.from_pretrained(CFG["ner_model"]) | |
| # Force semantic NER label mapping so LOC filtering works reliably. | |
| model.config.id2label = NER_ID2LABEL | |
| model.config.label2id = NER_LABEL2ID | |
| return pipeline( | |
| "token-classification", | |
| model=model, | |
| tokenizer=tokenizer, | |
| aggregation_strategy="simple", | |
| device=0 if DEVICE.type == "cuda" else -1, | |
| ) | |
| def extract_locations(text: str): | |
| entities = get_ner_pipeline()(text) | |
| spans = [] | |
| for ent in entities: | |
| group = str(ent.get("entity_group", "")).upper() | |
| if group != "LOC": | |
| continue | |
| start = int(ent["start"]) | |
| end = int(ent["end"]) | |
| score = float(ent.get("score", 0.0)) | |
| if 0 <= start < end <= len(text): | |
| spans.append((start, end, score)) | |
| # Merge overlapping/adjacent location pieces. | |
| spans.sort(key=lambda x: (x[0], x[1])) | |
| merged = [] | |
| for start, end, score in spans: | |
| if merged and start <= merged[-1][1]: | |
| ps, pe, pscore = merged[-1] | |
| merged[-1] = ( | |
| ps, | |
| max(pe, end), | |
| max(pscore, score) | |
| ) | |
| else: | |
| merged.append((start, end, score)) | |
| locations = [] | |
| for start, end, _ in merged: | |
| location = text[start:end].strip() | |
| if location and location not in locations: | |
| locations.append(location) | |
| return locations, merged | |
| def highlight_locations(text: str, spans): | |
| if not spans: | |
| return ( | |
| '<div class="location-box">' | |
| + html.escape(text) | |
| + '<div class="empty-location">No location detected.</div>' | |
| + '</div>' | |
| ) | |
| pieces = [] | |
| cursor = 0 | |
| for start, end, score in spans: | |
| pieces.append(html.escape(text[cursor:start])) | |
| loc = html.escape(text[start:end]) | |
| pieces.append( | |
| f'<mark title="NER confidence: {score:.3f}">{loc}</mark>' | |
| ) | |
| cursor = end | |
| pieces.append(html.escape(text[cursor:])) | |
| return ( | |
| '<div class="location-box">' | |
| + "".join(pieces) | |
| + "</div>" | |
| ) | |
| # ============================================================ | |
| # MAIN PIPELINE | |
| # ============================================================ | |
| def analyze_news(text): | |
| text = (text or "").strip() | |
| if not text: | |
| raise gr.Error("Please enter a Bangla news text.") | |
| # ---------------- Task A ---------------- | |
| a_id, a_label, a_scores = predict_classifier( | |
| text, | |
| get_task_a_model(), | |
| TASK_A_LABELS | |
| ) | |
| # ---------------- Task B ---------------- | |
| if a_id == 1: | |
| _, b_label, b_scores = predict_classifier( | |
| text, | |
| get_task_b_model(), | |
| TASK_B_LABELS | |
| ) | |
| b_result = b_label | |
| else: | |
| b_result = "Not Applicable" | |
| b_scores = { | |
| "Not Applicable — Task A predicted Non-Accidental": 1.0 | |
| } | |
| # ---------------- NER ---------------- | |
| locations, spans = extract_locations(text) | |
| location_text = ( | |
| " | ".join(locations) | |
| if locations | |
| else "No location detected" | |
| ) | |
| summary = f""" | |
| ### Analysis Result | |
| **Subtask A — Accident Identification:** {a_label} | |
| **Subtask B — Accident Category:** {b_result} | |
| **Exact Location(s):** {location_text} | |
| """ | |
| return ( | |
| summary, | |
| a_scores, | |
| b_scores, | |
| location_text, | |
| highlight_locations(text, spans), | |
| ) | |
| # ============================================================ | |
| # GRADIO UI | |
| # ============================================================ | |
| CSS = """ | |
| .gradio-container { | |
| max-width: 1180px !important; | |
| margin: 0 auto !important; | |
| } | |
| .header-card { | |
| border: 1px solid var(--border-color-primary); | |
| border-radius: 18px; | |
| padding: 24px 26px; | |
| margin-bottom: 18px; | |
| background: var(--background-fill-secondary); | |
| } | |
| .header-card h1 { | |
| margin: 0 0 10px 0; | |
| font-size: 1.7rem; | |
| line-height: 1.25; | |
| } | |
| .header-card p { | |
| margin: 0; | |
| opacity: .8; | |
| } | |
| .location-box { | |
| border: 1px solid var(--border-color-primary); | |
| border-radius: 14px; | |
| padding: 18px; | |
| min-height: 110px; | |
| line-height: 2.1; | |
| font-size: 1.05rem; | |
| white-space: pre-wrap; | |
| } | |
| .location-box mark { | |
| background: #facc15; | |
| color: #111827; | |
| padding: 2px 5px; | |
| border-radius: 5px; | |
| font-weight: 700; | |
| } | |
| .empty-location { | |
| opacity: .65; | |
| margin-top: 12px; | |
| font-size: .9rem; | |
| } | |
| """ | |
| with gr.Blocks(css=CSS, title="Accident Causal Identification") as demo: | |
| gr.HTML( | |
| f""" | |
| <div class="header-card"> | |
| <h1>{html.escape(CFG["paper_title"])}</h1> | |
| <p> | |
| Bangla news analysis using fine-tuned XLM-RoBERTa | |
| for accident classification and Bengali mBERT NER | |
| for exact location extraction. | |
| </p> | |
| </div> | |
| """ | |
| ) | |
| with gr.Row(): | |
| with gr.Column(scale=7): | |
| news_text = gr.Textbox( | |
| label="Bangla News Text", | |
| placeholder="এখানে বাংলা সংবাদ লিখুন...", | |
| lines=10, | |
| ) | |
| analyze_button = gr.Button( | |
| "Analyze News", | |
| variant="primary" | |
| ) | |
| with gr.Column(scale=4): | |
| gr.Markdown( | |
| """ | |
| ### Pipeline | |
| **Subtask A** | |
| Accidental / Non-Accidental | |
| **Subtask B** | |
| 10 accident categories | |
| **Exact Location** | |
| Bengali mBERT NER | |
| **XLM-RoBERTa max length:** `256` | |
| """ | |
| ) | |
| result_summary = gr.Markdown() | |
| with gr.Row(): | |
| task_a_probabilities = gr.Label( | |
| label="Subtask A Probabilities", | |
| num_top_classes=2 | |
| ) | |
| task_b_probabilities = gr.Label( | |
| label="Subtask B Probabilities", | |
| num_top_classes=10 | |
| ) | |
| exact_locations = gr.Textbox( | |
| label="Extracted Exact Location(s)", | |
| interactive=False | |
| ) | |
| gr.Markdown("### Location Highlighting") | |
| highlighted_news = gr.HTML() | |
| gr.Examples( | |
| examples=[ | |
| ["গাজীপুরের জয়দেবপুর স্টেশন এলাকায় ট্রেনে কাটা পড়ে এক যুবকের মৃত্যু হয়েছে।"], | |
| ["রাজধানী ঢাকার পল্লবী এলাকায় নির্মাণাধীন ভবন থেকে ইট পড়ে এক পথচারীর মৃত্যু হয়েছে।"], | |
| ["রোয়ালখালীতে হাতির হামলায় তিনজনের মৃত্যু হয়েছে।"], | |
| ["বাংলাদেশের অর্থনীতি নিয়ে নতুন প্রতিবেদন প্রকাশ করেছে একটি আন্তর্জাতিক সংস্থা।"], | |
| ], | |
| inputs=news_text | |
| ) | |
| analyze_button.click( | |
| fn=analyze_news, | |
| inputs=news_text, | |
| outputs=[ | |
| result_summary, | |
| task_a_probabilities, | |
| task_b_probabilities, | |
| exact_locations, | |
| highlighted_news, | |
| ] | |
| ) | |
| if __name__ == "__main__": | |
| demo.launch() | |