oshyryn/HVAC_Service_Analysis

1# app.py2 3import os, json, traceback4import pandas as pd5import matplotlib.pyplot as plt6import torch7import gradio as gr8import assemblyai as aai9from transformers import pipeline10 11# 0) Device & MNLI loader12CUDA_OK = torch.cuda.is_available()13DEVICE_PREF = (0, -1)14 15def load_mnli():16    for dev in DEVICE_PREF:17        try:18            print(f"Loading MNLI on device {dev}…")19            return pipeline(20                "zero-shot-classification",21                model="facebook/bart-large-mnli",22                tokenizer="facebook/bart-large-mnli",23                device=dev24            )25        except Exception as e:26            print(f"  ↳ failed on {dev}: {e}")27    raise RuntimeError("Could not load MNLI")28 29clf = load_mnli()30_sales_clf = clf  # reuse31 32# 1) AssemblyAI transcription + diarization33def transcribe_and_diarize(audio_fp: str, api_key: str):34    aai.settings.api_key = api_key.strip()35    tx = aai.Transcriber().transcribe(36        audio_fp,37        config=aai.TranscriptionConfig(speaker_labels=True)38    )39    if tx.status == "error":40        raise RuntimeError(tx.error)41 42    segments = []43    for utt in tx.utterances:44        speaker = "Agent" if utt.speaker == 0 else "Customer"45        segments.append({46            "speaker": speaker,47            "start":   utt.start / 1000.0,48            "end":     utt.end   / 1000.0,49            "text":    utt.text.strip()50        })51    return tx.text, segments52 53def md_transcript(segments):54    return "\n\n".join(f"**{s['speaker']}**: {s['text']}" for s in segments)55 56# 2) Compliance check57CRITERIA = {58    "Introduction":        "Did the technician greet the customer and introduce themselves and the company?",59    "Problem Diagnosis":   "Did the technician ask diagnostic questions to understand the customer's issue?",60    "Solution Explanation":"Did the technician clearly explain the solution or service performed?",61    "Upsell Attempts":     "Did the technician attempt to upsell additional services or products?",62    "Maintenance Plan":    "Did the technician offer any maintenance plans or service agreements?",63    "Closing & Thank You": "Did the technician thank the customer and close courteously?"64}65 66def compliance_report(segments, threshold=0.6):67    report = {}68    for stage, question in CRITERIA.items():69        scores = [70            clf(seg["text"], [question])["scores"][0]71            for seg in segments if seg["text"]72        ]73        ok = any(score >= threshold for score in scores)74        examples = [75            seg["text"][:200]76            for seg, score in zip(segments, scores)77            if score >= threshold78        ][:2]79        report[stage] = {80            "compliant": bool(ok),81            "examples": examples82        }83    return report84 85# 3) Sales‑insights block86INTENT = ["Inquiry","Information","Purchase","Complaint","Support","Greeting","Closing"]87TONE   = ["Interested","Neutral","Disinterested","Frustrated","Satisfied"]88SAT    = ["Very satisfied","Satisfied","Neutral","Dissatisfied","Very dissatisfied"]89QUAL   = ["Excellent workmanship","Adequate workmanship","Poor workmanship"]90 91sat_weights = {92    "Very satisfied": .2, "Satisfied": .1, "Neutral": 0,93    "Dissatisfied": - .15, "Very dissatisfied": - .2594}95qual_weights = {96    "Excellent workmanship": .2,97    "Adequate workmanship": .05,98    "Poor workmanship": - .299}100 101def score_metric(turns, label, weights, neutral="Neutral"):102    if not turns:103        return 0.5104    s = 0.5105    for t in turns:106        tag = t.get(label) or neutral107        s += weights.get(tag, 0)108        if t.get("tone") == "Frustrated":109            s -= 0.05110    return max(0.0, min(1.0, 0.5 + (s - 0.5) / len(turns)))111 112def sales_block(segments):113    analysis = []114    for seg in segments:115        txt = seg["text"]116        if not txt:117            continue118        intent = _sales_clf(txt, INTENT)["labels"][0]119        tone   = _sales_clf(txt, TONE)["labels"][0]120        sat = qual = None121        if seg["speaker"] == "Customer":122            sat  = _sales_clf(txt, SAT)["labels"][0]123            qual = _sales_clf(txt, QUAL)["labels"][0]124        analysis.append({125            **seg, "intent": intent, "tone": tone,126            "satisfaction": sat, "work_quality": qual127        })128 129    cust = [t for t in analysis if t["speaker"] == "Customer"]130    sat_score  = score_metric(cust, "satisfaction", sat_weights)131    qual_score = score_metric(cust, "work_quality", qual_weights,132                              neutral="Adequate workmanship")133 134    intents = pd.Series([t["intent"] for t in analysis]).value_counts().to_dict()135    return analysis, sat_score, qual_score, intents136 137# 4) Plot helpers138def plot_gauges(sat, qual):139    fig, axes = plt.subplots(1, 2, figsize=(8, 2.5))140    for ax, val, label in zip(axes, (sat, qual), ("Satisfaction", "Work Quality")):141        c = "green" if val >= 0.7 else "orange" if val >= 0.4 else "red"142        ax.bar([label], [val], color=c)143        ax.set_ylim(0, 1)144        ax.set_title(f"{label}: {val:.1%}")145    plt.tight_layout()146    return fig147 148def plot_intents(intents):149    fig, ax = plt.subplots(figsize=(5, 3))150    pd.Series(intents).plot(kind="bar", color="#60A5FA", ax=ax)151    ax.set_ylabel("Count")152    ax.set_title("Intent Distribution")153    plt.tight_layout()154    return fig155 156# 5) Full pipeline157def full_pipeline(audio_path, api_key):158    try:159        transcript_txt, segments = transcribe_and_diarize(audio_path, api_key)160        comp = compliance_report(segments)161        analysis, sat, qual, intents = sales_block(segments)162 163        sales_summary = {164            "satisfaction_score": sat,165            "work_quality_score": qual,166            "intent_counts": intents167        }168 169        # Build plots170        fig_g = plot_gauges(sat, qual)171        fig_i = plot_intents(intents)172 173        return (174            md_transcript(segments),175            comp,176            sales_summary,177            fig_g,178            fig_i179        )180    except Exception:181        err = traceback.format_exc()182        return f"❌ Error:\n{err}", None, None, None, None183 184# 6) Gradio UI185with gr.Blocks(title="HVAC Call Analyzer") as demo:186    gr.Markdown("## 🛠️ HVAC Service‑Call Analyzer")187 188    with gr.Row():189        audio_in = gr.Audio(label="Upload audio", type="filepath")190        key_in   = gr.Textbox(label="AssemblyAI API key", type="password")191 192    run = gr.Button("Analyze Call")193 194    transcript_md = gr.Markdown()195    comp_json     = gr.JSON(label="Compliance Report")196    sales_json    = gr.JSON(label="Sales Insights")197    p_gauge       = gr.Plot(label="Satisfaction & Quality")198    p_intent      = gr.Plot(label="Intent Distribution")199 200    run.click(201        fn=full_pipeline,202        inputs=[audio_in, key_in],203        outputs=[transcript_md, comp_json, sales_json, p_gauge, p_intent]204    )205 206demo.launch(server_name="0.0.0.0", share=True)207