import json import os from dataclasses import dataclass try: from src.db import mongo_get_all_certificates except ImportError: from db import mongo_get_all_certificates try: from src.display.formatting import make_clickable_model except ImportError: try: from src.formatting import make_clickable_model except ImportError: try: from display.formatting import make_clickable_model except ImportError: from formatting import make_clickable_model try: from src.display.utils import AutoEvalColumn, ModelType, Precision, Tasks except ImportError: try: from src.utils import AutoEvalColumn, ModelType, Precision, Tasks except ImportError: try: from display.utils import AutoEvalColumn, ModelType, Precision, Tasks except ImportError: from utils import AutoEvalColumn, ModelType, Precision, Tasks @dataclass class EvalResult: eval_name: str full_model: str org: str model: str revision: str results: dict composite_score: float = 50.0 precision: Precision = Precision.bfloat16 model_type: ModelType = ModelType.IFT architecture: str = "CausalLM" license: str = "apache-2.0" num_params: float = 1.0 date: str = "" sample_adequate: bool = True @classmethod def init_from_dict(cls, data: dict): if not isinstance(data, dict) or data.get("status") not in ("ok", None): return None config = data.get("config", {}) spectral = data.get("spectral", {}) observer = data.get("observer", data.get("hso", {})) behavioral = data.get("behavioral", {}) composite = data.get("composite", data.get("risk", {})) full_model = config.get("model_name", "model/unknown").strip() parts = full_model.split("/", 1) org = parts[0] if len(parts) > 1 else "Community" model = parts[1] if len(parts) > 1 else parts[0] def parse_num(val, default=0.0): if val is None: return default if isinstance(val, (int, float)): return float(val) if isinstance(val, str): try: return float(val.replace("%", "").strip()) except ValueError: return default return default raw_blind = observer.get("blind_fraction", 0.0) blind_val = parse_num(raw_blind) if isinstance(raw_blind, (float, int)) and raw_blind <= 1.0: blind_val = round(raw_blind * 100, 2) raw_fact = behavioral.get("factual_accuracy", 0.0) fact_val = parse_num(raw_fact) if isinstance(raw_fact, (float, int)) and raw_fact <= 1.0: fact_val = round(raw_fact * 100, 1) raw_stab = behavioral.get("paraphrase_fidelity", behavioral.get("paraphrase_stability", 0.0)) stab_val = parse_num(raw_stab) if isinstance(raw_stab, (float, int)) and raw_stab <= 1.0: stab_val = round(raw_stab * 100, 2) comp_score = parse_num( composite.get("unvalidated_composite_score", composite.get("hallucination_potential", 50.0)) ) r_struct = parse_num(composite.get("structural_risk", 0.0)) r_behav = parse_num(composite.get("behavioral_risk", 0.0)) results = { "xray_score": round(comp_score, 1), "composite_score": round(comp_score, 1), "structural_risk": round(r_struct, 1), "struct_risk": round(r_struct, 1), "behavioral_risk": round(r_behav, 1), "behav_risk": round(r_behav, 1), "stable_rank": round(parse_num(spectral.get("stable_rank_mean", 0.0)), 2), "effective_rank": round(parse_num(spectral.get("effective_rank_mean", 0.0)), 2), "condition_number": round(parse_num(spectral.get("condition_number_mean", 0.0)), 1), "cond_num": round(parse_num(spectral.get("condition_number_mean", 0.0)), 1), "blind_fraction": blind_val, "token_ratio": round(parse_num(observer.get("token_to_dim_ratio", 0.0)), 2), "factuality": fact_val, "paraphrase_fidelity": stab_val, "paraphrase_stability": stab_val, "paraphrase_stab": stab_val, } prec_str = config.get("model_dtype", config.get("precision", "bfloat16")) precision_enum = Precision.from_str(str(prec_str)) raw_params = parse_num(config.get("params"), 0.5) num_params = round(raw_params, 3) if raw_params > 0 else 0.5 return cls( eval_name=f"{full_model}_{config.get('model_sha', 'main')}", full_model=full_model, org=org, model=model, revision=str(config.get("model_sha", "main"))[:8], results=results, composite_score=results["composite_score"], architecture=config.get("architecture", "CausalLM"), num_params=num_params, license=config.get("license", "apache-2.0"), date=data.get("audited_at", ""), sample_adequate=observer.get("sample_adequate", results["token_ratio"] >= 5.0), precision=precision_enum, ) @classmethod def init_from_json_file(cls, json_filepath): try: with open(json_filepath, "r", encoding="utf-8") as fp: data = json.load(fp) return cls.init_from_dict(data) except Exception: return None def to_dict(self): symbol_link = f'{self.model_type.value.symbol}' data_dict = { "eval_name": self.eval_name, AutoEvalColumn.model_type_symbol.name: symbol_link, AutoEvalColumn.model.name: make_clickable_model(self.full_model), AutoEvalColumn.sample_adequate.name: "✅ Yes" if self.sample_adequate else "⚠️ Low", AutoEvalColumn.architecture.name: self.architecture, AutoEvalColumn.precision.name: self.precision.value.name if hasattr(self.precision, "value") else str(self.precision), AutoEvalColumn.revision.name: self.revision, AutoEvalColumn.params.name: self.num_params, AutoEvalColumn.license.name: self.license, } for task in Tasks: val = self.results.get(task.value.benchmark, self.results.get(task.name, 0.0)) data_dict[task.value.col_name] = val return data_dict def get_raw_eval_results(results_path: str, requests_path: str = "") -> list[EvalResult]: # Deduplicate models by normalized name so rows never duplicate model_map = {} # 1. Recall from MongoDB Atlas mongo_docs = mongo_get_all_certificates() for doc in mongo_docs: res = EvalResult.init_from_dict(doc) if res and res.full_model: key = res.full_model.lower().strip() model_map[key] = res # 2. Merge local files if os.path.exists(results_path): for root, _, files in os.walk(results_path): for file in files: if file.endswith(".json"): try: res = EvalResult.init_from_json_file(os.path.join(root, file)) if res and res.full_model: key = res.full_model.lower().strip() if key not in model_map: model_map[key] = res except Exception as e: print(f"Error reading {file}: {e}") return list(model_map.values())