Spaces:
Running
Running
File size: 10,633 Bytes
93bc112 5ed3eee 93bc112 | 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 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 | import glob
import json
import os
from typing import List, Optional
import numpy as np
import pandas as pd
BASIC_METRICS = [
"Image Quality",
"Aesthetic Quality",
"JEPA Similarity",
"Dynamic Degree",
"Flow Score",
"Motion Smoothness",
"Subject Consistency",
"Background Consistency",
"Photometric Consistency",
"Interaction Quality",
"Trajectory Accuracy",
"Depth Accuracy",
"Perspectivity",
"Instruction Following",
"Semantic Alignment",
]
DIMENSION_MAP = {
"Visual Quality": ["Image Quality", "Aesthetic Quality", "JEPA Similarity"],
"Motion Quality": ["Dynamic Degree", "Flow Score", "Motion Smoothness"],
"Content Consistency": [
"Subject Consistency",
"Background Consistency",
"Photometric Consistency",
],
"Physics Adherence": ["Interaction Quality", "Trajectory Accuracy"],
"3D Accuracy": ["Depth Accuracy", "Perspectivity"],
"Controllability": [
"Instruction Following",
"Semantic Alignment",
],
}
ALL_METRICS = BASIC_METRICS + list(DIMENSION_MAP) + ["EWMScore"]
DIMENSION_METRICS = list(DIMENSION_MAP)
METRIC_CHOICES = sorted(ALL_METRICS)
TASK_AGGREGATES = {
"Data Engine": "Data Engine(",
"Action Planner": "Action Planner(",
"RL Environment": "RL Environment(",
"Visuo-Tactile Success Rate": "Visuo-Tactile(",
"Real Action Planner": "Real Action Planner(",
}
class DataLoader:
def __init__(self, results_dir: str = "./worldarena-results"):
self.results_dir = results_dir
self.df_all: Optional[pd.DataFrame] = None
self.BASIC_METRICS = BASIC_METRICS
self.DIMENSION_MAP = DIMENSION_MAP
self.DIMENSION_METRICS = DIMENSION_METRICS
self.ALL_METRICS = ALL_METRICS
self.METRIC_CHOICES = METRIC_CHOICES
def load_results(self) -> pd.DataFrame:
rows = []
file_patterns = ("*.json", "*.xlsx", "*.csv")
all_files = [
path
for pattern in file_patterns
for path in sorted(glob.glob(os.path.join(self.results_dir, pattern)))
]
for file_path in all_files:
model_name = os.path.splitext(os.path.basename(file_path))[0]
if file_path.endswith(".json"):
row = self._load_json_file(file_path, model_name)
else:
row = self._load_table_file(file_path, model_name)
if row:
rows.append(row)
df = pd.DataFrame(rows)
return self._process_data(df) if not df.empty else df
def _load_json_file(self, file_path: str, model_name: str) -> dict:
try:
with open(file_path, "r", encoding="utf-8") as file:
data = json.load(file)
row = {
"Model": data.get("Model", model_name),
"open_source": data.get(
"open source", data.get("open_source", "unknown")
),
"year": data.get("year", data.get("date", "unknown")),
}
def extract_values(data_dict, prefix=""):
for key, value in data_dict.items():
metric_name = f"{prefix}{key}" if prefix else key
if isinstance(value, dict):
extract_values(value, f"{metric_name}_")
elif isinstance(value, (int, float)):
row[metric_name] = value
extract_values(data.get("Metrics", {}))
return row
except Exception as exc:
print(f"Error loading JSON file {file_path}: {exc}")
return {}
def _load_table_file(self, file_path: str, model_name: str) -> dict:
try:
if file_path.endswith(".xlsx"):
table = pd.read_excel(file_path)
else:
table = pd.read_csv(file_path)
row = {"Model": model_name, "open_source": "unknown", "year": "unknown"}
for metadata_key in ("open source", "open_source"):
if metadata_key in table.columns:
row["open_source"] = table[metadata_key].iloc[0]
break
for metadata_key in ("year", "date"):
if metadata_key in table.columns:
row["year"] = table[metadata_key].iloc[0]
break
aliases = {
metric: [
metric,
metric.lower().replace(" ", "_"),
metric.replace(" ", "_"),
]
for metric in BASIC_METRICS
}
aliases["Image Quality"].append("imaging_quality")
aliases["JEPA Similarity"].append("JEPA_normalized")
for metric, candidates in aliases.items():
matching = next(
(candidate for candidate in candidates if candidate in table.columns),
None,
)
row[metric] = (
round(table[matching].mean(), 4) if matching else np.nan
)
return row
except Exception as exc:
print(f"Error loading table file {file_path}: {exc}")
return {}
@staticmethod
def _normalize_metadata(df: pd.DataFrame) -> pd.DataFrame:
if "open_source" in df.columns:
source_map = {
"yes": "Open-source",
"open-source": "Open-source",
"opensource": "Open-source",
"true": "Open-source",
"1": "Open-source",
"no": "Closed-source",
"closed-source": "Closed-source",
"closedsource": "Closed-source",
"false": "Closed-source",
"0": "Closed-source",
}
df["open_source"] = (
df["open_source"].astype(str).str.lower().map(source_map).fillna(
df["open_source"].astype(str).str.lower()
)
)
if "year" in df.columns:
df["year"] = df["year"].astype(str).str.extract(r"(\d{4})")[0]
return df
@staticmethod
def _to_percentage(series: pd.Series) -> pd.Series:
values = pd.to_numeric(series, errors="coerce")
valid = values.dropna()
if not valid.empty and valid.min() >= 0 and valid.max() <= 1:
values = values * 100
values = values.mask(values < 0, 0)
return values.round(2)
def _process_task_success_data(self, df: pd.DataFrame) -> pd.DataFrame:
aggregate_columns = {}
task_cols = []
for aggregate_name, prefix in TASK_AGGREGATES.items():
matching_cols = [
column for column in df.columns if column.startswith(prefix)
]
if matching_cols:
aggregate_columns[aggregate_name] = matching_cols
task_cols.extend(matching_cols)
for column in task_cols:
df[column] = self._to_percentage(df[column])
for aggregate_name, columns in aggregate_columns.items():
df[aggregate_name] = df[columns].mean(axis=1, skipna=True).round(2)
aggregate_cols = list(aggregate_columns)
df["EWMScore"] = df[aggregate_cols].mean(axis=1, skipna=True).round(2)
df = self._normalize_metadata(df).dropna(subset=["EWMScore"])
meta_cols = ["Model", "open_source", "year"]
return df[meta_cols + aggregate_cols + ["EWMScore"] + task_cols]
def _process_policy_evaluator_data(self, df: pd.DataFrame) -> pd.DataFrame:
df["Policy Evaluator"] = self._to_percentage(df["Policy Evaluator"])
df = self._normalize_metadata(df).dropna(subset=["Policy Evaluator"])
return df[["Model", "open_source", "year", "Policy Evaluator"]]
def _process_video_quality_data(self, df: pd.DataFrame) -> pd.DataFrame:
for metric in BASIC_METRICS:
if metric not in df.columns:
df[metric] = np.nan
else:
df[metric] = self._to_percentage(df[metric])
for dimension, sub_metrics in DIMENSION_MAP.items():
df[dimension] = df[sub_metrics].mean(axis=1, skipna=True).round(2)
df["EWMScore"] = df[BASIC_METRICS].mean(axis=1, skipna=True).round(2)
df = self._normalize_metadata(df)
df = df.dropna(subset=["EWMScore"])
meta_cols = ["Model", "open_source", "year"]
metric_cols = BASIC_METRICS + list(DIMENSION_MAP) + ["EWMScore"]
return df[meta_cols + metric_cols]
def _process_data(self, df: pd.DataFrame) -> pd.DataFrame:
metric_cols = [
column
for column in df.columns
if column not in ["Model", "open_source", "year"]
]
if any(
column.startswith(prefix)
for column in metric_cols
for prefix in TASK_AGGREGATES.values()
):
return self._process_task_success_data(df)
if "Policy Evaluator" in metric_cols and not any(
metric in metric_cols for metric in BASIC_METRICS
):
return self._process_policy_evaluator_data(df)
return self._process_video_quality_data(df)
def reload_data(self) -> str:
self.df_all = self.load_results()
if self.df_all is None or self.df_all.empty:
return (
f"No JSON or table files found in {self.results_dir}. "
"Please upload some results."
)
return f"Loaded {len(self.df_all)} models from {self.results_dir}"
def get_open_source_choices(self) -> List[str]:
if self.df_all is None or "open_source" not in self.df_all.columns:
return ["All"]
choices = sorted(
str(value)
for value in self.df_all["open_source"].dropna().unique()
if value != ""
)
return ["All"] + choices
def get_year_choices(self) -> List[str]:
if self.df_all is None or "year" not in self.df_all.columns:
return ["All"]
years = sorted(
(
str(value)
for value in self.df_all["year"].dropna().unique()
if value != ""
),
reverse=True,
)
return ["All"] + years
|