Upload 22 files

LICENSE

@@ -0,0 +1 @@
+MIT License - 2025

README.md

@@ -1,13 +1,3 @@
----
-title: Serviceadvisor
-emoji: 📉
-colorFrom: gray
-colorTo: purple
-sdk: gradio
-sdk_version: 5.44.1
-app_file: app.py
-pinned: false
-license: apache-2.0
----
+# Workshop Car Service Advisor (Hugging Face)
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+See instructions inside.

app.py

@@ -0,0 +1,104 @@
+import os, json, io, base64
+from typing import List, Dict, Any
+from PIL import Image
+import gradio as gr
+
+from car_advisor.vision_model import VisionInference
+from car_advisor.nlp_model import NLPInference
+from car_advisor.fusion import fuse
+from car_advisor.cost_estimator import estimate_costs
+from car_advisor.suggestions import predictive_maintenance, advanced_suggestions
+from car_advisor.reporter import export_pdf, export_json
+from car_advisor.scheduler import create_service_ics
+
+vision = VisionInference()
+nlp = NLPInference()
+
+def _to_image(obj):
+    if isinstance(obj, dict) and "image" in obj:
+        return Image.open(io.BytesIO(base64.b64decode(obj["image"].split(",")[-1])))
+    if isinstance(obj, str):
+        return Image.open(obj)
+    return obj
+
+def analyze(images: list, customer_text: str, make: str, model: str, year: int, mileage_km: int, vin: str, name: str, phone: str):
+    # Vision aggregation
+    agg = None
+    valid = 0
+    for it in images or []:
+        try:
+            img = _to_image(it)
+            vp = vision.predict(img)
+            valid += 1
+            if agg is None:
+                agg = {k: v for k,v in vp.items()}
+            else:
+                for k in agg:
+                    agg[k] += vp.get(k, 0.0)
+        except Exception:
+            pass
+    if agg is None:
+        agg = {k: 0.0 for k in vision.labels}
+    else:
+        for k in agg:
+            agg[k] /= max(1, valid)
+
+    tp = nlp.predict(customer_text or "")
+
+    fused = fuse(agg, tp)
+    top = dict(list(fused.items())[:4])
+    estimate = estimate_costs(top, "configs/parts_costs.yaml", top_k=4)
+    pm = predictive_maintenance(car_year=int(year) if year else None, mileage_km=int(mileage_km) if mileage_km else None)
+    adv = advanced_suggestions(top_issues=top)
+
+    payload = {
+        "customer": {"name": name, "phone": phone},
+        "vehicle": {"make": make, "model": model, "year": year, "mileage_km": mileage_km, "vin": vin},
+        "complaint_text": customer_text,
+        "issues_ranked": fused,
+        "estimate": estimate,
+        "predictive_maintenance": pm,
+        "advanced_suggestions": adv
+    }
+
+    os.makedirs("exports", exist_ok=True)
+    pdf_path = "exports/service_report.pdf"
+    json_path = "exports/service_report.json"
+    ics_path = "exports/service_appointment.ics"
+    export_pdf(payload, pdf_path)
+    export_json(payload, json_path)
+    create_service_ics(ics_path, hours_from_now=48, duration_minutes=60)
+
+    def to_dl(path):
+        with open(path, "rb") as f:
+            return (os.path.basename(path), f.read())
+
+    return payload, to_dl(pdf_path), to_dl(json_path), to_dl(ics_path)
+
+with gr.Blocks(fill_height=True) as demo:
+    gr.Markdown("## 🚗 Workshop Car Service Advisor")
+    with gr.Row():
+        with gr.Column(scale=1):
+            imgs = gr.File(label="Upload car image(s)", file_count="multiple", file_types=["image"])
+            cust = gr.Textbox(label="Customer reported issue", placeholder="Describe the problem...")
+            with gr.Row():
+                make = gr.Textbox(label="Make", value="Toyota")
+                model = gr.Textbox(label="Model", value="Corolla")
+                year = gr.Number(label="Year", value=2017, precision=0)
+            with gr.Row():
+                mileage = gr.Number(label="Mileage (km)", value=60000, precision=0)
+                vin = gr.Textbox(label="VIN", placeholder="Optional")
+            with gr.Row():
+                name = gr.Textbox(label="Customer Name", value="")
+                phone = gr.Textbox(label="Phone", value="")
+            run = gr.Button("Analyze", variant="primary")
+        with gr.Column(scale=1):
+            out_json = gr.JSON(label="Structured output")
+            pdf_file = gr.File(label="Download PDF report")
+            json_file = gr.File(label="Download JSON")
+            ics_file = gr.File(label="Download .ics (appointment)")
+    run.click(analyze, inputs=[imgs, cust, make, model, year, mileage, vin, name, phone],
+              outputs=[out_json, pdf_file, json_file, ics_file])
+
+if __name__ == "__main__":
+    demo.launch()

car_advisor/__init__.py

@@ -0,0 +1 @@
+__version__ = '0.1.0'

car_advisor/config.py

@@ -0,0 +1,34 @@
+from dataclasses import dataclass
+from typing import List, Dict
+
+DEFAULT_LABELS: List[str] = [
+    "scratch_dent","paint_damage","cracked_windshield","flat_tire","engine_leak",
+    "brake_wear","headlight_fault","battery_corrosion","rust","bumper_damage"
+]
+
+SEVERITY_DEFAULTS: Dict[str, int] = {
+    "scratch_dent": 2,
+    "paint_damage": 2,
+    "cracked_windshield": 4,
+    "flat_tire": 3,
+    "engine_leak": 5,
+    "brake_wear": 4,
+    "headlight_fault": 3,
+    "battery_corrosion": 2,
+    "rust": 2,
+    "bumper_damage": 3,
+}
+
+PM_THRESHOLDS = {
+    "engine_oil": 10000,
+    "brake_pads": 30000,
+    "coolant": 40000,
+    "battery_check": 25000,
+    "tire_rotation": 8000,
+}
+
+@dataclass
+class AppSettings:
+    labels: List[str] = DEFAULT_LABELS
+    labor_rate_per_hour: float = 1200.0
+    diagnostic_fee: float = 500.0

car_advisor/cost_estimator.py

@@ -0,0 +1,45 @@
+import yaml
+from typing import Dict, Any
+
+def estimate_costs(fused: Dict[str, Any], parts_yaml_path: str, top_k: int = 4) -> Dict[str, Any]:
+    cfg = yaml.safe_load(open(parts_yaml_path, "r", encoding="utf-8"))
+    labor_rate = float(cfg.get("labor_rate_per_hour", 1200.0))
+    diagnostic_fee = float(cfg.get("diagnostic_fee", 500.0))
+    parts_cfg = cfg.get("parts", {})
+    items = []
+    total = 0.0
+    for i, (label, rec) in enumerate(fused.items()):
+        if i >= top_k:
+            break
+        part_info = parts_cfg.get(label, {})
+        hours = float(part_info.get("hours", 1.0))
+        parts_list = part_info.get("parts_list", [])
+        parts_cost = sum(float(p.get("cost", 0.0)) for p in parts_list)
+        labor_cost = hours * labor_rate
+        line_total = parts_cost + labor_cost
+        items.append({
+            "issue": label,
+            "probability": round(float(rec["prob"]), 3),
+            "severity": int(rec.get("severity", 3)),
+            "labor_hours": hours,
+            "labor_cost": round(labor_cost, 2),
+            "parts": parts_list,
+            "parts_cost": round(parts_cost, 2),
+            "line_total": round(line_total, 2)
+        })
+        total += line_total
+    if not items:
+        items.append({
+            "issue": "diagnostic_only",
+            "probability": 0.3,
+            "severity": 1,
+            "labor_hours": 0.0,
+            "labor_cost": 0.0,
+            "parts": [],
+            "parts_cost": 0.0,
+            "line_total": diagnostic_fee
+        })
+        total += diagnostic_fee
+    tax = round(0.18 * total, 2)
+    grand = round(total + tax, 2)
+    return {"items": items, "subtotal": round(total, 2), "tax": tax, "grand_total": grand}

car_advisor/fusion.py

@@ -0,0 +1,12 @@
+from typing import Dict, Any
+from .config import SEVERITY_DEFAULTS
+import math
+
+def fuse(vision_probs: Dict[str, float], text_probs: Dict[str, float]) -> Dict[str, Any]:
+    fused = {}
+    for label in sorted(set(list(vision_probs.keys()) + list(text_probs.keys()))):
+        pv = max(1e-6, vision_probs.get(label, 0.0))
+        pt = max(1e-6, text_probs.get(label, 0.0))
+        p = math.sqrt(pv * pt) * 1.2 + 0.1 * pt + 0.05 * pv
+        fused[label] = {"prob": float(min(1.0, p)), "severity": SEVERITY_DEFAULTS.get(label, 3)}
+    return dict(sorted(fused.items(), key=lambda kv: kv[1]["prob"], reverse=True))

car_advisor/nlp_model.py

@@ -0,0 +1,55 @@
+from typing import Dict, List
+from .config import DEFAULT_LABELS
+
+KEYWORD_MAP = {
+    "brake_wear": ["brake", "squeal", "screech", "stopping", "pads"],
+    "flat_tire": ["flat", "puncture", "tyre", "tire", "pressure", "air"],
+    "engine_leak": ["leak", "oil", "puddle", "drip", "smell burning"],
+    "cracked_windshield": ["crack", "windshield", "glass"],
+    "paint_damage": ["scratch", "scrape", "paint", "scuff"],
+    "scratch_dent": ["dent", "dented", "bent"],
+    "headlight_fault": ["headlight", "bulb", "beam", "lamp"],
+    "battery_corrosion": ["battery", "corrosion", "terminal", "start"],
+    "rust": ["rust", "oxid"],
+    "bumper_damage": ["bumper", "fender"]
+}
+
+def _contains_any(text: str, keywords: List[str]) -> bool:
+    t = (text or "").lower()
+    return any(kw in t for kw in keywords)
+
+class NLPInference:
+    def __init__(self, labels: List[str] = None, ckpt_dir: str = "checkpoints/nlp"):
+        self.labels = labels or DEFAULT_LABELS
+        self.ckpt_dir = ckpt_dir
+        self.trained = False
+        try:
+            import joblib, os
+            clf_p = os.path.join(ckpt_dir, "best.joblib")
+            mlb_p = os.path.join(ckpt_dir, "mlb.joblib")
+            if os.path.exists(clf_p) and os.path.exists(mlb_p):
+                self.clf = joblib.load(clf_p)
+                self.mlb = joblib.load(mlb_p)
+                self.trained = True
+        except Exception:
+            self.trained = False
+
+    def predict(self, text: str) -> Dict[str, float]:
+        if not text:
+            return {l: 0.0 for l in self.labels}
+        if self.trained:
+            probs = self.clf.predict_proba([text])[0]
+            out = {}
+            for i, lbl in enumerate(self.mlb.classes_):
+                val = probs[i] if isinstance(probs[i], (float,int)) else probs[i][1]
+                out[lbl] = float(val)
+            for l in self.labels:
+                out.setdefault(l, 0.0)
+            return out
+        else:
+            scores = {l: 0.01 for l in self.labels}
+            for label, kws in KEYWORD_MAP.items():
+                if _contains_any(text, kws):
+                    scores[label] += 0.5
+            s = sum(scores.values())
+            return {k: v/s for k,v in scores.items()}

car_advisor/reporter.py

@@ -0,0 +1,69 @@
+import json
+from typing import Dict, Any
+from reportlab.lib.pagesizes import A4
+from reportlab.lib import colors
+from reportlab.lib.styles import getSampleStyleSheet
+from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle
+from datetime import datetime
+
+def export_json(payload: Dict[str, Any], out_path: str) -> str:
+    with open(out_path, "w", encoding="utf-8") as f:
+        json.dump(payload, f, ensure_ascii=False, indent=2)
+    return out_path
+
+def export_pdf(payload: Dict[str, Any], out_path: str) -> str:
+    doc = SimpleDocTemplate(out_path, pagesize=A4)
+    styles = getSampleStyleSheet()
+    story = []
+    story.append(Paragraph("<b>Workshop Car Service Advisor Report</b>", styles["Title"]))
+    story.append(Spacer(1, 10))
+
+    meta = payload.get("vehicle", {})
+    cust = payload.get("customer", {})
+    details = f"""
+    <b>Customer:</b> {cust.get('name','N/A')} | <b>Phone:</b> {cust.get('phone','N/A')}<br/>
+    <b>Vehicle:</b> {meta.get('make','N/A')} {meta.get('model','')} {meta.get('year','')} | <b>VIN:</b> {meta.get('vin','N/A')}<br/>
+    <b>Mileage:</b> {meta.get('mileage_km','N/A')} km | <b>Date:</b> {datetime.now().strftime('%Y-%m-%d %H:%M')}
+    """
+    story.append(Paragraph(details, styles["Normal"]))
+    story.append(Spacer(1, 10))
+
+    story.append(Paragraph("<b>Detected Issues</b>", styles["Heading2"]))
+    data = [["Issue", "Probability", "Severity", "Labor (hrs)", "Labor Cost", "Parts Cost", "Line Total"]]
+    for item in payload["estimate"]["items"]:
+        data.append([
+            item["issue"],
+            f"{item['probability']:.2f}",
+            str(item["severity"]),
+            f"{item['labor_hours']:.2f}",
+            f"₹{item['labor_cost']:.2f}",
+            f"₹{item['parts_cost']:.2f}",
+            f"₹{item['line_total']:.2f}",
+        ])
+    table = Table(data, hAlign="LEFT")
+    table.setStyle(TableStyle([
+        ('BACKGROUND',(0,0),(-1,0),colors.lightblue),
+        ('TEXTCOLOR',(0,0),(-1,0),colors.whitesmoke),
+        ('ALIGN',(0,0),(-1,-1),'CENTER'),
+        ('GRID',(0,0),(-1,-1),0.25,colors.grey),
+        ('FONTNAME',(0,0),(-1,0),'Helvetica-Bold'),
+    ]))
+    story.append(table)
+    story.append(Spacer(1, 8))
+
+    story.append(Paragraph(f"<b>Subtotal:</b> ₹{payload['estimate']['subtotal']:.2f}", styles["Normal"]))
+    story.append(Paragraph(f"<b>Tax:</b> ₹{payload['estimate']['tax']:.2f}", styles["Normal"]))
+    story.append(Paragraph(f"<b>Grand Total:</b> ₹{payload['estimate']['grand_total']:.2f}", styles["Heading3"]))
+
+    story.append(Spacer(1, 10))
+    story.append(Paragraph("<b>Predictive Maintenance</b>", styles["Heading2"]))
+    for tip in payload.get("predictive_maintenance", []):
+        story.append(Paragraph(f"• {tip}", styles["Normal"]))
+
+    story.append(Spacer(1, 10))
+    story.append(Paragraph("<b>Advanced Suggestions</b>", styles["Heading2"]))
+    for tip in payload.get("advanced_suggestions", []):
+        story.append(Paragraph(f"• {tip}", styles["Normal"]))
+
+    doc.build(story)
+    return out_path

car_advisor/scheduler.py

@@ -0,0 +1,15 @@
+from ics import Calendar, Event
+from datetime import datetime, timedelta
+
+def create_service_ics(out_path: str, summary="Car Service Appointment", hours_from_now: int = 48, duration_minutes: int = 60):
+    cal = Calendar()
+    e = Event()
+    start = datetime.now() + timedelta(hours=hours_from_now)
+    e.name = summary
+    e.begin = start
+    e.duration = timedelta(minutes=duration_minutes)
+    e.description = "Auto-suggested appointment from Workshop Car Service Advisor."
+    cal.events.add(e)
+    with open(out_path, "w", encoding="utf-8") as f:
+        f.writelines(cal.serialize_iter())
+    return out_path

car_advisor/suggestions.py

@@ -0,0 +1,33 @@
+from typing import Dict, Any, List
+from .config import PM_THRESHOLDS
+
+def predictive_maintenance(car_year: int = None, mileage_km: int = None) -> List[str]:
+    tips = []
+    if mileage_km is not None:
+        if mileage_km % PM_THRESHOLDS["engine_oil"] > PM_THRESHOLDS["engine_oil"] - 1000:
+            tips.append("Engine oil service due soon based on mileage.")
+        if mileage_km % PM_THRESHOLDS["tire_rotation"] > PM_THRESHOLDS["tire_rotation"] - 500:
+            tips.append("Consider tire rotation and balancing.")
+        if mileage_km > 50000:
+            tips.append("Inspect suspension components (shocks/struts) for wear.")
+        if mileage_km > 80000:
+            tips.append("Check timing belt/chain and water pump as per manufacturer schedule.")
+    if car_year is not None and car_year < 2015:
+        tips.append("Vehicle age suggests comprehensive electrical & rubber parts inspection.")
+    if not tips:
+        tips.append("No immediate predictive maintenance items flagged.")
+    return tips
+
+def advanced_suggestions(top_issues: Dict[str, Any]) -> List[str]:
+    tips = []
+    if "engine_leak" in top_issues:
+        tips.append("After fixing leak, clean engine bay and monitor oil level weekly for 1 month.")
+    if "brake_wear" in top_issues:
+        tips.append("Bed-in new pads and avoid hard braking for first 200 km.")
+    if "flat_tire" in top_issues:
+        tips.append("Check alignment and inspect other tires for embedded nails/screws.")
+    if "rust" in top_issues:
+        tips.append("Apply rust protection and inspect underbody after monsoon season.")
+    if "cracked_windshield" in top_issues:
+        tips.append("Avoid potholes and sudden temperature changes until replacement.")
+    return tips

car_advisor/utils.py

@@ -0,0 +1,11 @@
+import yaml
+
+def load_yaml(path):
+    with open(path, "r", encoding="utf-8") as f:
+        return yaml.safe_load(f)
+
+def softmax(xs):
+    import numpy as np
+    x = np.array(xs, dtype=float)
+    e = np.exp(x - x.max())
+    return (e / e.sum()).tolist()

car_advisor/vision_model.py

@@ -0,0 +1,74 @@
+from typing import List, Dict
+from PIL import Image
+import torch, os
+import torch.nn as nn
+import torchvision.transforms as T
+from .config import DEFAULT_LABELS
+from .utils import softmax
+
+class SimpleVisionModel(nn.Module):
+    """
+    Wrapper around a lightweight classifier. For training, use training/train_vision.py.
+    At inference, if checkpoint absent or downloads fail, we return rule-based scores.
+    """
+    def __init__(self, num_classes: int):
+        super().__init__()
+        try:
+            import timm
+            self.net = timm.create_model("mobilenetv3_small_100", pretrained=True, num_classes=num_classes)
+        except Exception:
+            self.net = nn.Sequential(
+                nn.AdaptiveAvgPool2d((8,8)),
+                nn.Flatten(),
+                nn.Linear(8*8*3, 128),
+                nn.ReLU(),
+                nn.Linear(128, num_classes)
+            )
+
+    def forward(self, x):
+        return self.net(x)
+
+class VisionInference:
+    def __init__(self, labels: List[str] = None, ckpt_path: str = "checkpoints/vision/best.pt"):
+        self.labels = labels or DEFAULT_LABELS
+        self.device = "cuda" if torch.cuda.is_available() else "cpu"
+        self.model = SimpleVisionModel(num_classes=len(self.labels)).to(self.device)
+        self.transform = T.Compose([T.Resize((224,224)), T.ToTensor()])
+        self.ready = False
+        if os.path.exists(ckpt_path):
+            try:
+                state = torch.load(ckpt_path, map_location=self.device)
+                self.model.load_state_dict(state["model"] if "model" in state else state)
+                self.ready = True
+            except Exception:
+                self.ready = False
+
+    @torch.no_grad()
+    def predict(self, image: Image.Image) -> Dict[str, float]:
+        if image is None:
+            return {l: 0.0 for l in self.labels}
+        try:
+            x = self.transform(image.convert("RGB")).unsqueeze(0).to(self.device)
+            logits = self.model(x)[0].detach().cpu().tolist()
+            probs = softmax(logits)
+            return {lbl: float(p) for lbl, p in zip(self.labels, probs)}
+        except Exception:
+            import numpy as np
+            img = image.convert("RGB").resize((64,64))
+            arr = np.array(img).astype("float32")/255.0
+            gray = arr.mean(axis=2)
+            contrast = float(gray.std())
+            red_mean = float(arr[:,:,0].mean())
+            green_mean = float(arr[:,:,1].mean())
+            blue_mean = float(arr[:,:,2].mean())
+            scores = {l: 0.01 for l in self.labels}
+            if contrast > 0.22:
+                scores["scratch_dent"] += 0.2
+                scores["paint_damage"] += 0.15
+                scores["bumper_damage"] += 0.1
+            if blue_mean < 0.35 and green_mean < 0.35:
+                scores["rust"] += 0.2
+            if red_mean > 0.55:
+                scores["engine_leak"] += 0.15
+            s = sum(scores.values())
+            return {k: v/s for k,v in scores.items()}

configs/issues.yaml

@@ -0,0 +1,11 @@
+labels:
+  - scratch_dent
+  - paint_damage
+  - cracked_windshield
+  - flat_tire
+  - engine_leak
+  - brake_wear
+  - headlight_fault
+  - battery_corrosion
+  - rust
+  - bumper_damage

configs/parts_costs.yaml

@@ -0,0 +1,46 @@
+labor_rate_per_hour: 1200
+diagnostic_fee: 500
+parts:
+  scratch_dent:
+    parts_list:
+      - {name: "Body filler/paint kit", cost: 1800}
+    hours: 1.5
+  paint_damage:
+    parts_list:
+      - {name: "Paint & materials", cost: 2500}
+    hours: 2.0
+  cracked_windshield:
+    parts_list:
+      - {name: "Windshield glass", cost: 8000}
+      - {name: "Sealant kit", cost: 900}
+    hours: 2.5
+  flat_tire:
+    parts_list:
+      - {name: "New tire", cost: 4500}
+    hours: 0.6
+  engine_leak:
+    parts_list:
+      - {name: "Gasket/seal kit", cost: 3200}
+      - {name: "Engine oil", cost: 1800}
+    hours: 3.0
+  brake_wear:
+    parts_list:
+      - {name: "Brake pads (pair)", cost: 3500}
+    hours: 1.4
+  headlight_fault:
+    parts_list:
+      - {name: "Headlight bulb/assembly", cost: 2200}
+    hours: 0.8
+  battery_corrosion:
+    parts_list:
+      - {name: "Battery terminals/cleaner", cost: 600}
+    hours: 0.5
+  rust:
+    parts_list:
+      - {name: "Rust converter & primer", cost: 1000}
+    hours: 2.0
+  bumper_damage:
+    parts_list:
+      - {name: "Bumper cover", cost: 7000}
+      - {name: "Clips/fasteners", cost: 500}
+    hours: 2.2

data/sample_data/annotations.csv

@@ -0,0 +1,2 @@
+image_path,issue_label,customer_text,car_make,car_model,car_year,mileage_km
+images/example.jpg,paint_damage,"Scratches on left door, visible scuff marks",Maruti,Swift,2017,65000

requirements.txt

@@ -0,0 +1,15 @@
+torch>=2.1.0
+torchvision>=0.16.0
+torchaudio>=2.1.0
+timm>=1.0.3
+transformers>=4.42.0
+tokenizers>=0.15.2
+gradio>=4.36.1
+pydantic>=2.7.0
+pillow>=10.3.0
+numpy>=1.26.4
+pandas>=2.2.2
+scikit-learn>=1.5.0
+pyyaml>=6.0.1
+reportlab>=4.1.0
+ics>=0.7.2

training/dataset.py

@@ -0,0 +1,37 @@
+import pandas as pd
+from PIL import Image
+from torch.utils.data import Dataset
+import torchvision.transforms as T
+
+class CarIssuesDataset(Dataset):
+    def __init__(self, csv_path, img_root, labels, transform=None, text_col="customer_text"):
+        self.df = pd.read_csv(csv_path)
+        self.img_root = img_root
+        self.labels = labels
+        self.transform = transform or T.Compose([T.Resize((224,224)), T.ToTensor()])
+        self.text_col = text_col
+
+    def __len__(self):
+        return len(self.df)
+
+    def __getitem__(self, idx):
+        row = self.df.iloc[idx]
+        img_path = row['image_path']
+        if not str(img_path).startswith(self.img_root):
+            import os
+            img_path = os.path.join(self.img_root, img_path)
+        try:
+            img = Image.open(img_path).convert("RGB")
+        except Exception:
+            import numpy as np
+            img = Image.fromarray((np.zeros((224,224,3))+255).astype("uint8"))
+        x = self.transform(img)
+        y = [1 if row["issue_label"] == l else 0 for l in self.labels]
+        text = str(row.get(self.text_col, ""))
+        meta = {
+            "car_make": row.get("car_make", ""),
+            "car_model": row.get("car_model", ""),
+            "car_year": row.get("car_year", ""),
+            "mileage_km": row.get("mileage_km", ""),
+        }
+        return x, y, text, meta

training/train_fusion.py

@@ -0,0 +1,42 @@
+import argparse, os, pandas as pd, numpy as np
+from sklearn.linear_model import LogisticRegression
+from sklearn.model_selection import train_test_split
+from sklearn.metrics import f1_score
+import joblib
+
+def main(args):
+    df = pd.read_csv(args.annotations)
+    labels = sorted(df["issue_label"].unique().tolist())
+    label_to_idx = {l:i for i,l in enumerate(labels)}
+
+    X = []
+    y = []
+    for _, row in df.iterrows():
+        text = str(row.get("customer_text","")).lower()
+        features = [
+            len(text),
+            int("brake" in text),
+            int("leak" in text),
+            int("tire" in text or "tyre" in text),
+            int("scratch" in text or "dent" in text),
+        ]
+        X.append(features)
+        y.append(label_to_idx[row["issue_label"]])
+    X = np.array(X); y = np.array(y)
+
+    Xtr, Xv, ytr, yv = train_test_split(X, y, test_size=0.2, random_state=42)
+    clf = LogisticRegression(max_iter=200).fit(Xtr, ytr)
+    yp = clf.predict(Xv)
+    print("fusion macro F1:", f1_score(yv, yp, average="macro"))
+    os.makedirs(args.out_dir, exist_ok=True)
+    joblib.dump({"clf": clf, "labels": labels}, os.path.join(args.out_dir, "best.joblib"))
+    print("Saved", args.out_dir)
+
+if __name__ == "__main__":
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--annotations", required=True)
+    ap.add_argument("--vision_ckpt", required=False)
+    ap.add_argument("--nlp_ckpt", required=False)
+    ap.add_argument("--out_dir", default="checkpoints/fusion")
+    args = ap.parse_args()
+    main(args)

training/train_nlp.py

@@ -0,0 +1,38 @@
+import argparse, os, pandas as pd
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.preprocessing import MultiLabelBinarizer
+from sklearn.multiclass import OneVsRestClassifier
+from sklearn.linear_model import LogisticRegression
+from sklearn.pipeline import Pipeline
+from sklearn.model_selection import train_test_split
+from sklearn.metrics import f1_score
+import joblib
+
+def main(args):
+    df = pd.read_csv(args.annotations)
+    X = df["customer_text"].fillna("")
+    y = df["issue_label"].fillna("").apply(lambda x: [x])
+    mlb = MultiLabelBinarizer()
+    Y = mlb.fit_transform(y)
+
+    X_tr, X_v, Y_tr, Y_v = train_test_split(X, Y, test_size=0.2, random_state=42)
+
+    pipe = Pipeline([
+        ("tfidf", TfidfVectorizer(ngram_range=(1,2), max_features=40000)),
+        ("clf", OneVsRestClassifier(LogisticRegression(max_iter=200)))
+    ])
+    pipe.fit(X_tr, Y_tr)
+    Yp = pipe.predict(X_v)
+    print("macro F1:", f1_score(Y_v, Yp, average="macro"))
+    os.makedirs(args.out_dir, exist_ok=True)
+    joblib.dump(pipe, os.path.join(args.out_dir, "best.joblib"))
+    joblib.dump(mlb, os.path.join(args.out_dir, "mlb.joblib"))
+    print("Saved to", args.out_dir)
+
+if __name__ == "__main__":
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--annotations", required=True)
+    ap.add_argument("--out_dir", default="checkpoints/nlp")
+    ap.add_argument("--epochs", type=int, default=3)
+    args = ap.parse_args()
+    main(args)

training/train_vision.py

@@ -0,0 +1,64 @@
+import argparse, os
+import torch, torch.nn as nn
+from torch.utils.data import DataLoader, random_split
+import torchvision.transforms as T
+import timm
+import pandas as pd
+from training.dataset import CarIssuesDataset
+
+def main(args):
+    labels = pd.read_csv(args.annotations)["issue_label"].unique().tolist()
+    transform = T.Compose([T.Resize((224,224)), T.ToTensor()])
+    ds = CarIssuesDataset(args.annotations, os.path.dirname(args.annotations), labels, transform=transform)
+    n = len(ds)
+    n_val = max(1, int(0.2 * n))
+    tr, val = random_split(ds, [n - n_val, n_val])
+    tl = DataLoader(tr, batch_size=16, shuffle=True)
+    vl = DataLoader(val, batch_size=16)
+
+    model = timm.create_model("mobilenetv3_small_100", pretrained=True, num_classes=len(labels))
+    device = "cuda" if torch.cuda.is_available() else "cpu"
+    model.to(device)
+
+    opt = torch.optim.AdamW(model.parameters(), lr=1e-3)
+    crit = nn.CrossEntropyLoss()
+
+    best = 0.0
+    os.makedirs(args.out_dir, exist_ok=True)
+
+    for epoch in range(args.epochs):
+        model.train()
+        for xb, yb, _, _ in tl:
+            xb = xb.to(device)
+            yb = yb.argmax(dim=1).to(device)
+            opt.zero_grad()
+            out = model(xb)
+            loss = crit(out, yb)
+            loss.backward()
+            opt.step()
+
+        model.eval()
+        correct = 0; total = 0
+        with torch.no_grad():
+            for xb, yb, _, _ in vl:
+                xb = xb.to(device)
+                y_true = yb.argmax(dim=1).to(device)
+                logits = model(xb)
+                preds = logits.argmax(dim=1)
+                correct += (preds == y_true).sum().item()
+                total += y_true.numel()
+        acc = correct/total if total else 0
+        print(f"Epoch {epoch+1}: val_acc={acc:.3f}")
+        if acc > best:
+            best = acc
+            torch.save({"model": model.state_dict(), "labels": labels}, os.path.join(args.out_dir, "best.pt"))
+    print("Done. Best acc:", best)
+
+if __name__ == "__main__":
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--data_root", required=True)
+    ap.add_argument("--annotations", required=True)
+    ap.add_argument("--out_dir", default="checkpoints/vision")
+    ap.add_argument("--epochs", type=int, default=5)
+    args = ap.parse_args()
+    main(args)