Spaces:
Sleeping
Sleeping
| """ | |
| AETHER-TaskFlow Grader — Explicit per-difficulty scoring | |
| Ensures scores are always in [0.0, 1.0] and deterministic. | |
| """ | |
| from __future__ import annotations | |
| from typing import Any, Dict | |
| def grade_easy(result: Dict[str, Any]) -> float: | |
| """Easy: Strong reward for completion and resource conservation.""" | |
| return _compute_score( | |
| result, | |
| {"efficiency": 0.50, "resource": 0.25, "health": 0.15, "speed": 0.10}, | |
| ) | |
| def grade_medium(result: Dict[str, Any]) -> float: | |
| """Medium: Heavier penalty on missed deadlines.""" | |
| base = _compute_score( | |
| result, | |
| {"efficiency": 0.45, "resource": 0.20, "health": 0.25, "speed": 0.10}, | |
| ) | |
| missed_ratio = result.get("tasks_failed", 0) / max(result.get("total_tasks", 1), 1) | |
| return max(0.0, base - missed_ratio * 0.20) | |
| def grade_hard(result: Dict[str, Any]) -> float: | |
| """Hard: Strong penalties for system collapse and failures.""" | |
| base = _compute_score( | |
| result, | |
| {"efficiency": 0.40, "resource": 0.15, "health": 0.35, "speed": 0.10}, | |
| ) | |
| collapse_penalty = 0.15 if result.get("system_health", 1.0) < 0.3 else 0.0 | |
| missed_ratio = result.get("tasks_failed", 0) / max(result.get("total_tasks", 1), 1) | |
| return max(0.0, base - collapse_penalty - missed_ratio * 0.25) | |
| def _compute_score(result: Dict[str, Any], weights: Dict[str, float]) -> float: | |
| total = max(result.get("total_tasks", 1), 1) | |
| efficiency = result.get("tasks_completed", 0) / total | |
| rem_time = result.get("remaining_time", 0) | |
| rem_energy = result.get("remaining_energy", 0) | |
| rem_budget = result.get("remaining_budget", 0) | |
| init_time = max(result.get("initial_time", 10), 1) | |
| init_energy = max(result.get("initial_energy", 12), 1) | |
| init_budget = max(result.get("initial_budget", 60), 1) | |
| resource_score = ( | |
| (rem_time / init_time) * 0.3 | |
| + (rem_energy / init_energy) * 0.35 | |
| + (rem_budget / init_budget) * 0.35 | |
| ) | |
| health = max(0.0, min(1.0, result.get("system_health", 1.0))) | |
| steps_used = max(result.get("steps_used", 10), 1) | |
| max_steps = max(result.get("max_steps", 10), 1) | |
| speed = 1.0 - (steps_used / max_steps) | |
| raw = ( | |
| weights.get("efficiency", 0.4) * efficiency | |
| + weights.get("resource", 0.2) * resource_score | |
| + weights.get("health", 0.2) * health | |
| + weights.get("speed", 0.1) * speed | |
| ) | |
| return round(max(0.0, min(1.0, raw)), 4) | |
| def grade(difficulty: str, result: Dict[str, Any]) -> float: | |
| graders = { | |
| "easy": grade_easy, | |
| "medium": grade_medium, | |
| "hard": grade_hard, | |
| } | |
| fn = graders.get(difficulty, grade_easy) | |
| return fn(result) | |