tcod/tests/utils/registry_test.py

# -*- coding: utf-8 -*-
"""Test cases for workflows registry mapping."""
import unittest

import ray
import torch

from trinity.algorithm import (
    ADVANTAGE_FN,
    ALGORITHM_TYPE,
    ENTROPY_LOSS_FN,
    KL_FN,
    POLICY_LOSS_FN,
    SAMPLE_STRATEGY,
    AdvantageFn,
    AlgorithmType,
    EntropyLossFn,
    KLFn,
    PolicyLossFn,
    SampleStrategy,
)
from trinity.buffer.buffer_reader import BufferReader
from trinity.buffer.operators import EXPERIENCE_OPERATORS, ExperienceOperator
from trinity.buffer.reader import READER
from trinity.buffer.schema import FORMATTER, SQL_SCHEMA
from trinity.buffer.selector import SELECTORS, BaseSelector
from trinity.buffer.storage import PRIORITY_FUNC
from trinity.buffer.storage.queue import PriorityFunction
from trinity.common.rewards import REWARD_FUNCTIONS, RewardFn
from trinity.common.workflows import WORKFLOWS, Workflow
from trinity.utils.monitor import MONITOR, Monitor


@ENTROPY_LOSS_FN.register_module("dummy_entropy_loss_fn")
class DummyEntropyLossFn(EntropyLossFn):
    def __init__(self, entropy_coef: float):
        self.entropy_coef = entropy_coef

    def __call__(
        self,
        entropy,
        action_mask,
        **kwargs,
    ):
        return torch.tensor(0.0), {}


class ImportUtils:
    def run(self):
        from trinity.common.workflows import WORKFLOWS, Workflow

        workflow_cls = WORKFLOWS.get("tests.utils.plugins.main.MainDummyWorkflow")
        assert issubclass(workflow_cls, Workflow)
        workflow = workflow_cls(task=None, model=None)
        res = workflow.run()
        assert res[0] == 0
        assert res[1] == "0"


class TestRegistryWithRay(unittest.TestCase):
    def setUp(self):
        ray.init(ignore_reinit_error=True)

    def tearDown(self):
        ray.shutdown()

    def test_dynamic_import(self):
        # test local import
        ImportUtils().run()
        # test remote import
        ray.get(ray.remote(ImportUtils).remote().run.remote())


class TestRegistry(unittest.TestCase):
    """Test registry functionality."""

    def test_common_module_registry_mapping(self):
        """Test registry mapping in common module"""
        # test workflow
        workflow_names = list(WORKFLOWS._default_mapping.keys())
        for workflow_name in workflow_names:
            with self.subTest(workflow_name=workflow_name):
                workflow_cls = WORKFLOWS.get(workflow_name)
                self.assertIsNotNone(
                    workflow_cls, f"{workflow_name} should be retrievable from registry"
                )
                self.assertTrue(
                    issubclass(workflow_cls, Workflow),
                    f"{workflow_name} should be a subclass of Workflow",
                )
        with self.assertRaises(ValueError):
            WORKFLOWS.get("non_existent_workflow")

        # test reward function
        reward_fn_names = list(REWARD_FUNCTIONS._default_mapping.keys())
        for reward_fn_name in reward_fn_names:
            with self.subTest(reward_fn_name=reward_fn_name):
                reward_fn_cls = REWARD_FUNCTIONS.get(reward_fn_name)
                self.assertIsNotNone(
                    reward_fn_cls, f"{reward_fn_name} should be retrievable from registry"
                )
                self.assertTrue(
                    issubclass(reward_fn_cls, RewardFn),
                    f"{reward_fn_name} should be a subclass of RewardFn",
                )
        with self.assertRaises(ValueError):
            REWARD_FUNCTIONS.get("non_existent_reward_fn")

    def test_algorithm_registry_mapping(self):
        """Test registry mapping in algorithm module"""
        # test algorithm
        algorithm_names = list(ALGORITHM_TYPE._default_mapping.keys())
        for algorithm_name in algorithm_names:
            with self.subTest(algorithm_name=algorithm_name):
                algorithm_cls = ALGORITHM_TYPE.get(algorithm_name)
                self.assertIsNotNone(
                    algorithm_cls, f"{algorithm_name} should be retrievable from registry"
                )
                self.assertTrue(
                    issubclass(algorithm_cls, AlgorithmType),
                    f"{algorithm_name} should be a subclass of AlgorithmType",
                )
        with self.assertRaises(ValueError):
            ALGORITHM_TYPE.get("non_existent_algorithm")

        # test advantage function
        advantage_fn_names = list(ADVANTAGE_FN._default_mapping.keys())
        for advantage_fn_name in advantage_fn_names:
            with self.subTest(advantage_fn_name=advantage_fn_name):
                advantage_fn_cls = ADVANTAGE_FN.get(advantage_fn_name)
                self.assertIsNotNone(
                    advantage_fn_cls, f"{advantage_fn_name} should be retrievable from registry"
                )
                self.assertTrue(
                    issubclass(advantage_fn_cls, AdvantageFn),
                    f"{advantage_fn_name} should be a subclass of AdvantageFn",
                )
        with self.assertRaises(ValueError):
            ADVANTAGE_FN.get("non_existent_advantage_fn")

        # test entropy loss function
        entropy_loss_fn_names = list(ENTROPY_LOSS_FN._default_mapping.keys())
        for entropy_loss_fn_name in entropy_loss_fn_names:
            with self.subTest(entropy_loss_fn_name=entropy_loss_fn_name):
                entropy_loss_fn_cls = ENTROPY_LOSS_FN.get(entropy_loss_fn_name)
                self.assertIsNotNone(
                    entropy_loss_fn_cls,
                    f"{entropy_loss_fn_name} should be retrievable from registry",
                )
                self.assertTrue(
                    issubclass(entropy_loss_fn_cls, EntropyLossFn),
                    f"{entropy_loss_fn_name} should be a subclass of EntropyLossFn",
                )
        with self.assertRaises(ValueError):
            ENTROPY_LOSS_FN.get("non_existent_entropy_loss_fn")

        # test kl function
        kl_fn_names = list(KL_FN._default_mapping.keys())
        for kl_fn_name in kl_fn_names:
            with self.subTest(kl_fn_name=kl_fn_name):
                kl_fn_cls = KL_FN.get(kl_fn_name)
                self.assertIsNotNone(kl_fn_cls, f"{kl_fn_name} should be retrievable from registry")
                self.assertTrue(
                    issubclass(kl_fn_cls, KLFn), f"{kl_fn_name} should be a subclass of KLFn"
                )
        with self.assertRaises(ValueError):
            KL_FN.get("non_existent_kl_fn")

        # test policy loss function
        policy_loss_fn_names = list(POLICY_LOSS_FN._default_mapping.keys())
        for policy_loss_fn_name in policy_loss_fn_names:
            with self.subTest(policy_loss_fn_name=policy_loss_fn_name):
                policy_loss_fn_cls = POLICY_LOSS_FN.get(policy_loss_fn_name)
                self.assertIsNotNone(
                    policy_loss_fn_cls, f"{policy_loss_fn_name} should be retrievable from registry"
                )
                self.assertTrue(
                    issubclass(policy_loss_fn_cls, PolicyLossFn),
                    f"{policy_loss_fn_name} should be a subclass of PolicyLossFn",
                )
        with self.assertRaises(ValueError):
            POLICY_LOSS_FN.get("non_existent_policy_loss_fn")

        # test sample strategy
        sample_strategy_names = list(SAMPLE_STRATEGY._default_mapping.keys())
        for sample_strategy_name in sample_strategy_names:
            with self.subTest(sample_strategy_name=sample_strategy_name):
                sample_strategy_cls = SAMPLE_STRATEGY.get(sample_strategy_name)
                self.assertIsNotNone(
                    sample_strategy_cls,
                    f"{sample_strategy_name} should be retrievable from registry",
                )
                self.assertTrue(
                    issubclass(sample_strategy_cls, SampleStrategy),
                    f"{sample_strategy_name} should be a subclass of SampleStrategy",
                )
        with self.assertRaises(ValueError):
            SAMPLE_STRATEGY.get("non_existent_sample_strategy")

    def test_buffer_module_registry_mapping(self):
        """Test registry mapping in buffer module"""
        # test experience operator
        operator_names = list(EXPERIENCE_OPERATORS._default_mapping.keys())
        for operator_name in operator_names:
            with self.subTest(operator_name=operator_name):
                operator_cls = EXPERIENCE_OPERATORS.get(operator_name)
                self.assertIsNotNone(
                    operator_cls, f"{operator_name} should be retrievable from registry"
                )
                self.assertTrue(
                    issubclass(operator_cls, ExperienceOperator),
                    f"{operator_name} should be a subclass of ExperienceOperator",
                )
        with self.assertRaises(ValueError):
            EXPERIENCE_OPERATORS.get("non_existent_operator")

        # test reader
        reader_names = list(READER._default_mapping.keys())
        for reader_name in reader_names:
            with self.subTest(reader_name=reader_name):
                reader_cls = READER.get(reader_name)
                self.assertIsNotNone(
                    reader_cls, f"{reader_name} should be retrievable from registry"
                )
                self.assertTrue(
                    issubclass(reader_cls, BufferReader),
                    f"{reader_name} should be a subclass of BufferReader",
                )
        with self.assertRaises(ValueError):
            READER.get("non_existent_reader")

        # test formatter
        formatter_names = list(FORMATTER._default_mapping.keys())
        for formatter_name in formatter_names:
            with self.subTest(formatter_name=formatter_name):
                formatter_cls = FORMATTER.get(formatter_name)
                self.assertIsNotNone(
                    formatter_cls, f"{formatter_name} should be retrievable from registry"
                )
        with self.assertRaises(ValueError):
            FORMATTER.get("non_existent_formatter")

        # test sql schema
        schema_names = list(SQL_SCHEMA._default_mapping.keys())
        for schema_name in schema_names:
            with self.subTest(schema_name=schema_name):
                schema_cls = SQL_SCHEMA.get(schema_name)
                self.assertIsNotNone(
                    schema_cls, f"{schema_name} should be retrievable from registry"
                )
        with self.assertRaises(ValueError):
            SQL_SCHEMA.get("non_existent_schema")

        # test selector
        selector_names = list(SELECTORS._default_mapping.keys())
        for selector_name in selector_names:
            with self.subTest(selector_name=selector_name):
                selector_cls = SELECTORS.get(selector_name)
                self.assertIsNotNone(
                    selector_cls, f"{selector_name} should be retrievable from registry"
                )
                self.assertTrue(
                    issubclass(selector_cls, BaseSelector),
                    f"{selector_name} should be a subclass of BaseSelector",
                )
        with self.assertRaises(ValueError):
            SELECTORS.get("non_existent_selector")

        # test priority function
        priority_fn_names = list(PRIORITY_FUNC._default_mapping.keys())

        for priority_fn_name in priority_fn_names:
            with self.subTest(priority_fn_name=priority_fn_name):
                priority_fn_cls = PRIORITY_FUNC.get(priority_fn_name)
                self.assertIsNotNone(
                    priority_fn_cls, f"{priority_fn_name} should be retrievable from registry"
                )
                self.assertTrue(
                    issubclass(priority_fn_cls, PriorityFunction),
                    f"{priority_fn_name} should be a subclass of PriorityFunction",
                )
        with self.assertRaises(ValueError):
            PRIORITY_FUNC.get("non_existent_priority_fn")

    def test_utils_module_registry_mapping(self):
        """Test registry mapping in utils module"""
        # test monitor
        monitor_names = list(MONITOR._default_mapping.keys())
        for monitor_name in monitor_names:
            with self.subTest(monitor_name=monitor_name):
                monitor_cls = MONITOR.get(monitor_name)
                self.assertIsNotNone(
                    monitor_cls, f"{monitor_name} should be retrievable from registry"
                )
                self.assertTrue(
                    issubclass(monitor_cls, Monitor),
                    f"{monitor_name} should be a subclass of Monitor",
                )
        with self.assertRaises(ValueError):
            MONITOR.get("non_existent_monitor")

    def test_register_module(self):
        """Test register module functionality"""
        # Test that the registered class can be retrieved from registry
        retrieved_cls = ENTROPY_LOSS_FN.get("dummy_entropy_loss_fn")
        self.assertIsNotNone(
            retrieved_cls, "dummy_entropy_loss_fn should be retrievable from registry"
        )
        self.assertTrue(
            issubclass(retrieved_cls, EntropyLossFn),
            "dummy_entropy_loss_fn should be a subclass of EntropyLossFn",
        )
        self.assertEqual(
            retrieved_cls, DummyEntropyLossFn, "Retrieved class should be DummyEntropyLossFn"
        )

        # Test that the registered class can be instantiated and used
        instance = retrieved_cls(entropy_coef=0.1)
        self.assertIsInstance(instance, EntropyLossFn)
        self.assertEqual(instance.entropy_coef, 0.1)

        # Test that the instance can be called (basic functionality)
        loss, metrics = instance(entropy=torch.tensor(1.0), action_mask=torch.tensor(1.0))
        self.assertEqual(loss.item(), 0.0)
        self.assertIsInstance(metrics, dict)