File size: 3,153 Bytes
43a90a1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 | #!/usr/bin/env python3
"""
Orqanta-AI Vehicle Dataset Benchmark & Evaluation Tool
Evaluates both Car and Two-Wheeler datasets:
1. Orqanta_AI_Dataset_v1 (Cars: EV & Petrol)
2. Orqanta_TwoWheeler_Dataset_v1 (Two-Wheelers: EV & Petrol Scooters / Motorcycles)
"""
import os
import sys
import json
import time
from collections import Counter
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
import orqanta
DATASETS = {
"🚗 Car (EV & Petrol)": os.path.abspath(os.path.join(os.path.dirname(__file__), "../Orqanta_AI_Dataset_v1")),
"🛵 Two-Wheeler (EV & Petrol)": os.path.abspath(os.path.join(os.path.dirname(__file__), "../Orqanta_TwoWheeler_Dataset_v1"))
}
def load_split(dataset_dir: str, filename: str):
path = os.path.join(dataset_dir, filename)
if not os.path.exists(path):
return []
records = []
with open(path, "r", encoding="utf-8") as f:
for line in f:
if line.strip():
records.append(json.loads(line))
return records
def print_dataset_summary(name: str, dataset_dir: str):
train_data = load_split(dataset_dir, "train.jsonl")
val_data = load_split(dataset_dir, "validation.jsonl")
test_data = load_split(dataset_dir, "test.jsonl")
total = len(train_data) + len(val_data) + len(test_data)
print("=" * 68)
print(f" {name} SFT Dataset Overview")
print("=" * 68)
print(f" • Total Examples: {total}")
print(f" • Training Split: {len(train_data)} ({len(train_data)/total*100:.1f}%)")
print(f" • Validation Split: {len(val_data)} ({len(val_data)/total*100:.1f}%)")
print(f" • Test Split: {len(test_data)} ({len(test_data)/total*100:.1f}%)")
print("-" * 68)
cats = Counter(x.get("category", "unknown") for x in train_data + val_data + test_data)
print(" Category Distribution:")
for cat, count in cats.most_common()[:8]:
print(f" - {cat:<24} : {count:>3} records")
print("=" * 68 + "\n")
def evaluate_samples(dataset_dir: str, label: str, n_samples: int = 2):
test_data = load_split(dataset_dir, "validation.jsonl")
if not test_data:
return
print(f"⚡ Testing Offline Edge Latency on {label}...")
agent = orqanta.Orqanta()
for i, item in enumerate(test_data[:n_samples]):
user_msg = ""
for m in item["messages"]:
if m["role"] == "user":
user_msg = m["content"]
expected_asst = ""
for m in item["messages"]:
if m["role"] == "assistant":
expected_asst = m["content"]
t0 = time.perf_counter()
res = agent.run(user_msg[:120])
elapsed_ms = (time.perf_counter() - t0) * 1000
print(f" [{i+1}/{n_samples}] ID: {item.get('id')} | Category: {item.get('category')}")
print(f" Latency: {elapsed_ms:.1f} ms | RAM: {res.get('peak_ram_mb')} MB | TPS: {res.get('decode_tps')}")
if __name__ == "__main__":
for name, path in DATASETS.items():
print_dataset_summary(name, path)
evaluate_samples(path, name, n_samples=2)
print()
|