Competition agent integration with zoo

smarts/zoo/agent_spec.py

@@ -74,6 +74,8 @@ class AgentSpec:
     """An adaptor that allows shaping of the reward (default lambda obs, reward: reward)"""
     info_adapter: Callable = lambda obs, reward, info: info
     """An adaptor that allows shaping of info (default lambda obs, reward, info: info)"""
+    adapt_env: Optional[Callable]= lambda env: env
+    """And adaptor that allows configurating the env (default lambda env:env)"""
 
     def __post_init__(self):
         # make sure we can pickle ourselves

zoo/policies/__init__.py

@@ -1,3 +1,7 @@
+import sys
+import os
+import importlib.util
+from pathlib import Path
 from typing import Any, Dict
 
 from smarts.core.agent_interface import AgentInterface, AgentType
@@ -100,3 +104,72 @@ def human_keyboard_entrypoint(*arg, **kwargs):
 
 
 register(locator="human-in-the-loop-v0", entry_point=human_keyboard_entrypoint)
+
+
+from smarts.env.multi_scenario_env import resolve_agent_interface
+
+
+def load_config(path):
+    import yaml
+
+    config = None
+    if path.exists():
+        with open(path, "r") as file:
+            config = yaml.safe_load(file)
+    return config
+
+
+def competition_entry(**kwargs):
+    policy_path = kwargs.get("policy_path", None)
+
+    from .competition_agent import CompetitionAgent
+
+    def env_wrapper(env):
+        import gym
+
+        # import policy.py module
+        wrapper_path = str(os.path.join(policy_path, "policy.py"))
+        wrapper_spec = importlib.util.spec_from_file_location(
+            "competition_wrapper", wrapper_path
+        )
+        wrapper_module = importlib.util.module_from_spec(wrapper_spec)
+        sys.modules["competition_wrapper"] = wrapper_module
+        if wrapper_spec:
+            wrapper_spec.loader.exec_module(wrapper_module)
+
+        wrappers = wrapper_module.submitted_wrappers()
+        env = gym.Wrapper(env)
+        for wrapper in wrappers:
+            env = wrapper(env)
+
+        # delete competition wrapper module
+        sys.modules.pop("competition_wrapper")
+        del wrapper_module
+
+        return env
+
+    config = load_config(Path(os.path.join(policy_path, "config.yaml")))
+
+    spec = AgentSpec(
+        interface=resolve_agent_interface(
+            img_meters=int(config["img_meters"]),
+            img_pixels=int(config["img_pixels"]),
+            action_space="TargetPose",
+        ),
+        agent_params={
+            "policy_path": policy_path,
+        },
+        adapt_env=env_wrapper,
+        agent_builder=CompetitionAgent,
+    )
+
+    return spec
+
+
+root_path = str(Path(__file__).absolute().parents[2])
+
+register(
+    "competition_agent-v0",
+    entry_point=competition_entry,
+    policy_path=os.path.join(root_path, "competition/track1/submission"),
+)

zoo/policies/competition_agent.py

@@ -0,0 +1,37 @@
+import subprocess
+import sys
+import os
+import importlib.util
+
+from pathlib import Path, PurePath
+from smarts.core.agent import Agent
+
+
+class CompetitionAgent(Agent):
+    def __init__(self, policy_path):
+        req_file = os.path.join(policy_path, "requirements.txt")
+        try:
+            subprocess.check_call(
+                [sys.executable, "-m", "pip", "install", "-r", req_file]
+            )
+        except:
+            print("Failed to install requirement for Competition Agent")
+
+        # import policy.py module
+        self._policy_path = str(os.path.join(policy_path, "policy.py"))
+        policy_spec = importlib.util.spec_from_file_location(
+            "competition_policy", self._policy_path
+        )
+        policy_module = importlib.util.module_from_spec(policy_spec)
+        sys.modules["competition_policy"] = policy_module
+        if policy_spec:
+            policy_spec.loader.exec_module(policy_module)
+
+        self._policy = policy_module.Policy()
+
+        # delete competition policy module
+        sys.modules.pop("competition_policy")
+        del policy_module
+
+    def act(self, obs):
+        return self._policy.act(obs)