Create app.py

app.py

@@ -0,0 +1,350 @@
+# app.py
+import os
+import io
+import uuid
+from datetime import datetime
+from typing import Tuple, Optional
+
+import gradio as gr
+import pandas as pd
+from PIL import Image
+from reportlab.lib.pagesizes import A4
+from reportlab.pdfgen import canvas
+from transformers import pipeline
+
+# -------------------------
+# Configuration / Filenames
+# -------------------------
+CSV_FILE = "job_cards.csv"
+EXCEL_FILE = "job_cards.xlsx"
+PDF_FOLDER = "pdf_reports"
+os.makedirs(PDF_FOLDER, exist_ok=True)
+
+# -------------------------
+# Load AI models (pipelines)
+# -------------------------
+# NOTE: these models are reasonably capable but can be large.
+# On Hugging Face Spaces choose a GPU runtime for best performance.
+
+print("Loading models... this may take a while on first run.")
+
+# 1) Text classification (zero-shot)
+text_classifier = pipeline(
+    task="zero-shot-classification",
+    model="facebook/bart-large-mnli"
+)
+
+# 2) Image object detection (DETR)
+image_detector = pipeline(
+    task="object-detection",
+    model="facebook/detr-resnet-50"
+)
+
+# 3) Speech-to-text (Whisper)
+asr = pipeline(
+    task="automatic-speech-recognition",
+    model="openai/whisper-large-v2",
+    chunk_length_s=30
+)
+
+print("Models loaded.")
+
+# -------------------------
+# Knowledge bases (starter)
+# -------------------------
+CATEGORIES = [
+    "Engine issue", "Brake issue", "AC problem", "Steering issue",
+    "Electrical issue", "Regular service", "Body damage", "Tire issue"
+]
+
+REPAIR_JOBS = {
+    "Engine issue": ["Engine diagnosis", "Oil change", "Spark plug check"],
+    "Brake issue": ["Brake pad replacement", "Brake fluid top-up", "Disc inspection"],
+    "AC problem": ["AC gas refill", "Compressor check", "Cabin filter replacement"],
+    "Steering issue": ["Power steering fluid check", "Alignment test"],
+    "Electrical issue": ["Battery test", "Alternator check", "Fuse replacement"],
+    "Regular service": ["Oil change", "Filter replacement", "General inspection"],
+    "Body damage": ["Dent repair", "Paint touch-up", "Bumper replacement"],
+    "Tire issue": ["Tire rotation", "Tire replacement", "Wheel balancing"]
+}
+
+COST_RANGES = {
+    "Engine issue": "₹3000–₹12,000",
+    "Brake issue": "₹1500–₹7000",
+    "AC problem": "₹2500–₹10,000",
+    "Steering issue": "₹1500–₹8000",
+    "Electrical issue": "₹1000–₹5000",
+    "Regular service": "₹1500–₹6000",
+    "Body damage": "₹2000–₹30,000",
+    "Tire issue": "₹800–₹8,000"
+}
+
+# Simple parts mapping (starter)
+PARTS_DB = {
+    "Brake pad replacement": ["Brake pads (set)", "Brake cleaner", "Brake fluid"],
+    "AC gas refill": ["R134a refrigerant", "O-rings", "Compressor oil"],
+    "Oil change": ["Engine oil (4L)", "Oil filter"],
+    "Battery test": ["12V battery", "Battery terminal cleaner"]
+}
+
+# -------------------------
+# Utility functions
+# -------------------------
+def ensure_csv_exists():
+    if not os.path.exists(CSV_FILE):
+        df = pd.DataFrame(columns=[
+            "ID", "Date", "Customer", "Vehicle", "Complaint",
+            "Category", "Suggested_Jobs", "Estimated_Cost",
+            "Detected_Damages", "Parts_Recommendations"
+        ])
+        df.to_csv(CSV_FILE, index=False)
+
+def classify_complaint(complaint: str) -> Tuple[str, str, str]:
+    """Return human text, top category, comma-joined suggested jobs, and cost"""
+    if not complaint or not complaint.strip():
+        return ("❌ No complaint text provided.", "", "", "")
+    res = text_classifier(complaint, candidate_labels=CATEGORIES)
+    top_category = res["labels"][0]
+    suggested_jobs = REPAIR_JOBS.get(top_category, ["Workshop inspection required"])
+    cost = COST_RANGES.get(top_category, "Varies")
+    jobs_text = ", ".join(suggested_jobs)
+    human_text = f"🔍 Category: {top_category}\n📋 Jobs: {jobs_text}\n💰 Estimated cost: {cost}"
+    return human_text, top_category, jobs_text, cost
+
+def detect_damage_from_image(image: Image.Image) -> Tuple[str, str]:
+    """Return human text & comma-joined detections"""
+    if image is None:
+        return ("❌ No image uploaded.", "")
+    pil = image.convert("RGB")
+    results = image_detector(pil, threshold=0.3)
+    # results is a list of dicts with label and score
+    if not results:
+        return ("✅ No obvious damage detected.", "")
+    detections = [f"{r['label']} ({r['score']:.2f})" for r in results]
+    return ("🚗 Detected: " + ", ".join(detections), ", ".join(detections))
+
+def speech_to_text(audio) -> str:
+    """Accept a wav/ogg file-like and return cleaned text"""
+    if audio is None:
+        return ""
+    # the pipeline accepts file path or array; Gradio supplies a filepath
+    try:
+        result = asr(audio)
+        text = result.get("text", "")
+        # simple cleanup
+        return text.strip()
+    except Exception as e:
+        return f"⚠️ ASR error: {e}"
+
+def recommend_parts(jobs_list_str: str, damages_str: str) -> str:
+    """Return a simple list of recommended parts. Placeholder for vector/DB search."""
+    parts = set()
+    if jobs_list_str:
+        # heuristic: check if any known job matches
+        for job_key in PARTS_DB:
+            if job_key.lower() in jobs_list_str.lower():
+                for p in PARTS_DB[job_key]:
+                    parts.add(p)
+    # also check damages keywords
+    if damages_str:
+        if "tire" in damages_str.lower():
+            parts.add("Tire(s)")
+        if "dent" in damages_str.lower():
+            parts.add("Body filler / Paint")
+    if not parts:
+        return "No specific parts suggested. Inspect to confirm."
+    return ", ".join(sorted(parts))
+
+# -------------------------
+# Persistence / Job Card CRUD
+# -------------------------
+ensure_csv_exists()
+
+def save_job_card(
+    customer: str, vehicle: str, complaint: str,
+    category: str, suggested_jobs: str, estimated_cost: str,
+    detected_damages: str, parts: str
+) -> str:
+    if not (customer and vehicle and complaint):
+        return "⚠️ Provide Customer, Vehicle number, and Complaint before saving."
+    df = pd.read_csv(CSV_FILE)
+    new_id = str(uuid.uuid4())[:8]
+    entry = {
+        "ID": new_id,
+        "Date": datetime.now().strftime("%Y-%m-%d %H:%M"),
+        "Customer": customer,
+        "Vehicle": vehicle,
+        "Complaint": complaint,
+        "Category": category,
+        "Suggested_Jobs": suggested_jobs,
+        "Estimated_Cost": estimated_cost,
+        "Detected_Damages": detected_damages,
+        "Parts_Recommendations": parts
+    }
+    df = pd.concat([df, pd.DataFrame([entry])], ignore_index=True)
+    df.to_csv(CSV_FILE, index=False)
+    return f"✅ Saved job card (ID: {new_id})"
+
+def view_all_cards() -> pd.DataFrame:
+    df = pd.read_csv(CSV_FILE)
+    return df
+
+def search_cards(customer_query: str, vehicle_query: str) -> pd.DataFrame:
+    df = pd.read_csv(CSV_FILE)
+    if customer_query:
+        df = df[df["Customer"].str.contains(customer_query, case=False, na=False)]
+    if vehicle_query:
+        df = df[df["Vehicle"].str.contains(vehicle_query, case=False, na=False)]
+    if df.empty:
+        return pd.DataFrame([{"Message": "No matching records found"}])
+    return df
+
+def export_excel() -> Optional[str]:
+    if not os.path.exists(CSV_FILE):
+        return None
+    df = pd.read_csv(CSV_FILE)
+    df.to_excel(EXCEL_FILE, index=False, engine="openpyxl")
+    return EXCEL_FILE
+
+def generate_jobcard_pdf(job_id: str) -> Optional[str]:
+    df = pd.read_csv(CSV_FILE)
+    row = df[df["ID"] == job_id]
+    if row.empty:
+        return None
+    row = row.iloc[0]
+    filename = os.path.join(PDF_FOLDER, f"jobcard_{job_id}.pdf")
+    buffer = io.BytesIO()
+    c = canvas.Canvas(buffer, pagesize=A4)
+    w, h = A4
+    margin = 50
+    y = h - margin
+
+    c.setFont("Helvetica-Bold", 16)
+    c.drawString(margin, y, f"Job Card — ID: {job_id}")
+    y -= 30
+    c.setFont("Helvetica", 11)
+    lines = [
+        f"Date: {row['Date']}",
+        f"Customer: {row['Customer']}",
+        f"Vehicle: {row['Vehicle']}",
+        f"Complaint: {row['Complaint']}",
+        f"Category: {row['Category']}",
+        f"Suggested Jobs: {row['Suggested_Jobs']}",
+        f"Estimated Cost: {row['Estimated_Cost']}",
+        f"Detected Damages: {row['Detected_Damages']}",
+        f"Parts Recommendations: {row['Parts_Recommendations']}"
+    ]
+    for line in lines:
+        c.drawString(margin, y, line)
+        y -= 18
+        if y < 80:
+            c.showPage()
+            y = h - margin
+            c.setFont("Helvetica", 11)
+    c.save()
+
+    with open(filename, "wb") as f:
+        f.write(buffer.getvalue())
+    return filename
+
+# -------------------------
+# Gradio UI
+# -------------------------
+with gr.Blocks() as demo:
+    gr.Markdown("# 🚘 Full AI Service Advisor — Multimodal (All features)")
+
+    with gr.Tab("Create Job Card"):
+        with gr.Row():
+            with gr.Column(scale=2):
+                customer = gr.Textbox(label="Customer Name")
+                vehicle = gr.Textbox(label="Vehicle Number")
+                complaint = gr.Textbox(label="Complaint (free text)", lines=3, placeholder="e.g. My steering is stiff when turning left...")
+                analyze_btn = gr.Button("Analyze Complaint (NLP)")
+                nlp_out = gr.Textbox(label="Complaint Analysis (NLP output)", lines=4)
+                category_hidden = gr.Textbox(visible=False)
+                jobs_hidden = gr.Textbox(visible=False)
+                cost_hidden = gr.Textbox(visible=False)
+
+                gr.Markdown("### Voice input")
+                audio_input = gr.Audio(label="Record Complaint (or upload audio)", source="microphone", type="filepath")
+                transcribe_btn = gr.Button("Transcribe Audio → Text")
+                transcription_out = gr.Textbox(label="Transcribed Text", lines=3)
+
+            with gr.Column(scale=2):
+                image_input = gr.Image(type="pil", label="Upload car image (damage/photo)")
+                detect_btn = gr.Button("Detect Damage (Image)")
+                detect_out = gr.Textbox(label="Damage Detection (image)", lines=3)
+                det_hidden = gr.Textbox(visible=False)
+
+                parts_out = gr.Textbox(label="Parts Recommendation", lines=2)
+                save_btn = gr.Button("💾 Save Job Card")
+                save_status = gr.Textbox(label="Save status")
+                gen_pdf_btn = gr.Button("📄 Generate PDF for Last Saved")
+                last_pdf_link = gr.File(label="Download PDF")
+
+    with gr.Tab("View / Search / Export"):
+        view_btn = gr.Button("📋 View All Job Cards")
+        df_view = gr.Dataframe()
+
+        gr.Markdown("### Search job cards")
+        cust_search = gr.Textbox(label="Search by Customer name (partial)")
+        veh_search = gr.Textbox(label="Search by Vehicle number (partial)")
+        search_btn = gr.Button("Search")
+        search_out = gr.Dataframe()
+
+        export_btn = gr.Button("Export all as Excel")
+        excel_file = gr.File(label="Download Excel")
+
+    # -----------------------
+    # Callbacks / wiring
+    # -----------------------
+    def on_analyze(complaint_text):
+        human, top_cat, jobs_text, cost_text = classify_complaint(complaint_text)
+        return human, top_cat, jobs_text, cost_text
+
+    analyze_btn.click(on_analyze, inputs=complaint, outputs=[nlp_out, category_hidden, jobs_hidden, cost_hidden])
+
+    def on_transcribe(audio_path):
+        txt = speech_to_text(audio_path)
+        return txt
+
+    transcribe_btn.click(on_transcribe, inputs=audio_input, outputs=transcription_out)
+
+    def on_detect(image_pil):
+        human, detections = detect_damage_from_image(image_pil)
+        return human, detections
+
+    detect_btn.click(on_detect, inputs=image_input, outputs=[detect_out, det_hidden])
+
+    def on_recommend_parts(jobs_text, det_text):
+        parts = recommend_parts(jobs_text, det_text)
+        return parts
+
+    # when either jobs_hidden or det_hidden change, update parts_out (we'll wire after)
+    jobs_hidden.change(on_recommend_parts, inputs=[jobs_hidden, det_hidden], outputs=parts_out)
+    det_hidden.change(on_recommend_parts, inputs=[jobs_hidden, det_hidden], outputs=parts_out)
+
+    def on_save(customer_name, vehicle_no, complaint_text, category_text, jobs_text, cost_text, det_text, parts_text):
+        return save_job_card(customer_name, vehicle_no, complaint_text, category_text, jobs_text, cost_text, det_text, parts_text)
+
+    save_btn.click(on_save, inputs=[customer, vehicle, complaint, category_hidden, jobs_hidden, cost_hidden, det_hidden, parts_out], outputs=[save_status])
+
+    # Generate PDF for the last saved job (we look up last row)
+    def on_generate_pdf():
+        df = pd.read_csv(CSV_FILE)
+        if df.empty:
+            return None
+        last_id = df.iloc[-1]["ID"]
+        path = generate_jobcard_pdf(last_id)
+        if path:
+            return path
+        return None
+
+    gen_pdf_btn.click(on_generate_pdf, outputs=last_pdf_link)
+
+    # View / search / export
+    view_btn.click(lambda: view_all_cards(), outputs=df_view)
+    search_btn.click(lambda a,b: search_cards(a,b), inputs=[cust_search, veh_search], outputs=search_out)
+    export_btn.click(lambda: export_excel(), outputs=excel_file)
+
+demo.launch()