Evaluation Framework API¶

class Evaluator(ABC):
    evaluators = {}

    def __init__(self, config: EvalCfg):
        """
        Initialize evaluator.

        Args:
            config: Evaluation configuration
        """

@classmethod
def register(cls, eval_type: str):
    """
    Register evaluator class.

    Args:
        eval_type: Evaluator type identifier

    Returns:
        Decorator function
    """

@classmethod
def init(cls, config: EvalCfg) -> 'Evaluator':
    """
    Create evaluator instance.

    Args:
        config: Evaluation configuration

    Returns:
        Evaluator instance
    """

def eval(self) -> None:
    """
    Run evaluation.

    Results saved to output_path directory.
    """

def eval_episode(self, episode: Dict[str, Any]) -> Dict[str, Any]:
    """
    Evaluate single episode.

    Args:
        episode: Episode data

    Returns:
        Episode result
    """

class Env(ABC):
    envs = {}

    def __init__(self, env_config: EnvCfg, task_config: TaskCfg):
        """
        Initialize environment.

        Args:
            env_config: Environment configuration
            task_config: Task configuration
        """

@classmethod
def register(cls, env_type: str):
    """
    Register environment class.

    Args:
        env_type: Environment type identifier
    """

@classmethod
def init(cls, env_config: EnvCfg, task_config: TaskCfg) -> 'Env':
    """
    Create environment instance.

    Args:
        env_config: Environment configuration
        task_config: Task configuration

    Returns:
        Environment instance
    """

def reset(self, episode: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """
    Reset environment for new episode.

    Args:
        episode: Episode data

    Returns:
        Initial observation
    """

def step(self, action: Dict[str, Any]) -> Dict[str, Any]:
    """
    Execute one action step.

    Args:
        action: Action dictionary {"x": float, "y": float, "yaw": float}

    Returns:
        Dict with observation, reward, done, info
    """

def get_info(self) -> Dict[str, Any]:
    """
    Get current environment info.

    Returns:
        Info dictionary
    """

class Agent(ABC):
    agents = {}

    def __init__(self, config: AgentCfg):
        """
        Initialize agent.

        Args:
            config: Agent configuration
        """

@classmethod
def register(cls, agent_type: str):
    """
    Register agent class.

    Args:
        agent_type: Agent type identifier
    """

@classmethod
def init(cls, config: AgentCfg) -> 'Agent':
    """
    Create agent instance.

    Args:
        config: Agent configuration

    Returns:
        Agent instance
    """

def reset(self, episode: Optional[Dict[str, Any]] = None):
    """
    Reset agent state.

    Args:
        episode: Episode data (optional)
    """

def act(self, observation: Dict[str, Any]) -> Dict[str, Any]:
    """
    Generate action from observation.

    Args:
        observation: Observation dictionary

    Returns:
        Action dictionary {"x": float, "y": float, "yaw": float}
    """

def close(self):
    """Close agent and release resources"""

class Dataset(ABC):
    datasets = {}

    def __init__(self, config: Dict[str, Any]):
        """
        Initialize dataset.

        Args:
            config: Dataset configuration dict (dataset_type, dataset_path, etc.)
        """

def get_episodes(self, num_episodes: Optional[int] = None) -> List[Dict[str, Any]]:
    """
    Get list of episodes.

    Args:
        num_episodes: Limit count; None for all

    Returns:
        List of episode dicts
    """

def get_episode(self, episode_id: str) -> Optional[Dict[str, Any]]:
    """
    Get single episode.

    Args:
        episode_id: Episode identifier

    Returns:
        Episode dict or None
    """

class Metric(ABC):
    metrics = {}

    def __init__(self, **kwargs):
        """Initialize metric"""

@classmethod
def register(cls, metric_name: str):
    """
    Register metric class.

    Args:
        metric_name: Metric name
    """

@classmethod
def init(cls, metric_name: str, **kwargs) -> 'Metric':
    """
    Create metric instance.

    Args:
        metric_name: Metric name
        **kwargs: Metric parameters

    Returns:
        Metric instance
    """

def update(self, episode_result: Dict[str, Any]):
    """
    Update metric value.

    Args:
        episode_result: Episode result
    """

def compute(self) -> float:
    """
    Compute final metric value.

    Returns:
        Metric value
    """

def reset(self):
    """Reset metric state"""

@dataclass
class EvalCfg(BaseConfig):
    eval_type: str = ""  # "pointnav", "objectnav", "imagenav", "vln"
    run_id: str = ""     # Auto-generated from timestamp by default
    env: EnvCfg = field(default_factory=EnvCfg)
    agent: AgentCfg = field(default_factory=AgentCfg)
    task: TaskCfg = field(default_factory=TaskCfg)
    dataset: Optional[Dict[str, Any]] = None  # Dataset config dict
    eval_settings: Dict[str, Any] = field(default_factory=dict)

@dataclass
class EnvCfg(BaseConfig):
    env_type: str = ""
    env_settings: Dict[str, Any] = field(default_factory=dict)

@dataclass
class GSEnvConfig(BaseConfig):
    camera_config: str = ""
    enable_occupancy: bool = True
    success_distance: float = 0.5
    rotation_threshold: float = 0.2
    gpu_id: Optional[int] = None
    enable_depth: bool = True
    enable_rgb: bool = True
    camera_names: list = field(default_factory=lambda: ["face", "left", "right"])
    image_width: int = 640
    image_height: int = 480

@dataclass
class AgentCfg(BaseConfig):
    agent_type: str = ""
    model_settings: Dict[str, Any] = field(default_factory=dict)
    device: Optional[str] = None  # "cuda", "cpu", or null for auto-detect

@dataclass
class TaskCfg(BaseConfig):
    task_type: str = ""  # "pointnav", "objectnav", "imagenav", "vln"

def get_logger(name: str) -> logging.Logger:
    """
    Get logger instance.

    Args:
        name: Logger name

    Returns:
        Logger instance
    """