Merge pull request #59 from duyminh1998/develop

Release PyCMO v1.4.0

README.md

@@ -12,7 +12,7 @@ Read about the project in detail [here](https://minhhua.com/pycmo/).
 
 # Quick Start Guide
 ## Get PyCMO
-1. Make sure the following settings are enabled in your Command Modern Operations' configurations (in `CPE.ini`):  
+1. (Non-Steam, Premium edition only) Make sure the following settings are enabled in your Command Modern Operations' configurations (in `CPE.ini`):  
 ```
 [Lua]
 EnableSocket = 1
@@ -22,13 +22,15 @@ EncodingMode = 8
 ```
 2. Click on "Clone or download", and then "Download Zip". 
 3. Unzip the repo anywhere.
-4. Edit the project's `pycmo/configs/config_template.py` file to fit your system's paths, then rename it as `pycmo/configs/config.py` (IMPORTANT). You only need to edit the lines 8 - 10:
+4. Edit the project's `pycmo/configs/config_template.py` file to fit your system's paths, then rename it as `pycmo/configs/config.py` (IMPORTANT). You only need to edit the lines 8 - 11:
 ```python
 pycmo_path = os.path.join("path/to", "pycmo")
 cmo_path = os.path.join("path/to/steam/installation/of", "Command - Modern Operations")
-command_mo_version = "Command v1.06 - Build 1328.11"
+command_mo_version = "Command v1.06 - Build 1328.12"
+use_gymnasium = False
 ```
-5. Navigate to the folder than contains `setup.py` and install the repository using `pip install .` Anytime you make changes to the files in the project folder, you need to reinstall the package using `pip install .`. Alternatively, use `pip install -e .` to install the package in editable mode. After doing this you can change the code without needing to continue to install it.
+5. Navigate to the folder than contains `setup.py` and install the repository using `pip install .` Anytime you make changes to the files in the project folder, you need to reinstall the package using `pip install .`. Alternatively, use `pip install -e .` to install the package in editable mode. After doing this you can change the code without needing to continue to install it. 
+6. From PyCMO v1.4.0, [gymnasium](https://gymnasium.farama.org/) became an optional dependency for users who want to use PyCMO as a Gym environment. In this case, use `pip install .[gym]` or `pip install -e .[gym]` for setup. Remember to set `use_gymnasium = True` in the `pycmo/configs/config.py` file.
 
 ## Run an agent (Steam edition only)
 1. Load the provided demo scenario `scen/steam_demo.scen` in the game.

pycmo/__init__.py

@@ -0,0 +1,12 @@
+from pycmo.configs.config import get_config
+
+# open config and set important files and folder paths
+config = get_config()
+
+if config["gymnasium"]:
+     from gymnasium.envs.registration import register
+
+     register(
+          id="FloridistanPycmoGymEnv-v0",
+          entry_point="pycmo.env.cmo_gym_env:FloridistanPycmoGymEnv",
+     )

pycmo/agents/base_agent.py

@@ -1,11 +1,11 @@
-from pycmo.lib import actions
+from pycmo.lib.actions import AvailableFunctions
 from pycmo.lib.features import Features, FeaturesFromSteam
 
 class BaseAgent:
     def __init__(self, player_side):
         self.player_side = player_side
 
-    def action(self, observation: Features | FeaturesFromSteam, VALID_FUNCTIONS:actions.AvailableFunctions) -> str:
+    def action(self, observation: Features | FeaturesFromSteam, VALID_FUNCTIONS:AvailableFunctions) -> str:
         ...
 
     def reset(self) -> None:

pycmo/configs/config_template.py

@@ -7,7 +7,8 @@
 def get_config():
     pycmo_path = os.path.join("path/to", "pycmo")
     cmo_path = os.path.join("path/to/steam/installation/of", "Command - Modern Operations")
-    command_mo_version = "Command v1.06 - Build 1328.11"
+    command_mo_version = "Command v1.06 - Build 1328.12"
+    use_gymnasium = False
 
     return {
     "command_path": cmo_path,
@@ -17,6 +18,7 @@ def get_config():
     "scen_ended": os.path.join(pycmo_path, "pycmo", "configs", "scen_has_ended.txt"),
     "pickle_path": os.path.join(pycmo_path, "pickle"),
     "scripts_path": os.path.join(pycmo_path, "scripts"),
-    "command_mo_version": command_mo_version
+    "command_mo_version": command_mo_version,
+    "gymnasium": use_gymnasium,
     # "command_cli_output_path": "C:\\ProgramData\\Command Professional Edition 2\\Analysis_Int", # only applicable to Premium version so we update this later
     }

pycmo/env/cmo_env.py

@@ -263,8 +263,11 @@ def __init__(self,
         with open(self.scen_ended, 'w') as file:
             file.writelines(data)
 
-    def reset(self) -> TimeStep:
+    def reset(self, close_scenario_end_and_player_eval_messages:bool=False) -> TimeStep:
         try:
+            if close_scenario_end_and_player_eval_messages:
+                self.client.close_scenario_end_and_player_eval_messages()
+
             restart_result = self.client.restart_scenario()
 
             # check that the scenario loaded event has fired correctly in CMO, and if not, restart the scenario
@@ -346,7 +349,7 @@ def get_obs(self) -> FeaturesFromSteam:
                 if get_obs_retries > max_get_obs_retries:
                     raise TimeoutError("CMOEnv unable to get observation.")
 
-    def action_spec(self, observation:Features) -> AvailableFunctions:    
+    def action_spec(self, observation:Features | FeaturesFromSteam) -> AvailableFunctions:    
         return AvailableFunctions(observation)
 
     def check_game_ended(self) -> bool:
@@ -365,4 +368,3 @@ def end_game(self) -> TimeStep:
         export_observation_event_name = 'Export observation'
         action = f"ScenEdit_RunScript('{pycmo_lua_lib_path}', true)\nteardown_and_end_scenario('{export_observation_event_name}', true)"
         return self.step(action)
-

pycmo/env/cmo_gym_env.py

@@ -0,0 +1,123 @@
+from typing import Tuple
+import gymnasium as gym
+from gymnasium import spaces
+import numpy as np
+
+from pycmo.lib.features import FeaturesFromSteam
+from pycmo.env.cmo_env import CMOEnv, StepType
+from pycmo.lib.protocol import SteamClientProps
+
+class BasePycmoGymEnv(gym.Env):
+    metadata = {"render_modes": [None]}
+
+    def __init__(
+            self,
+            player_side: str,
+            steam_client_props:SteamClientProps,
+            observation_path: str, 
+            action_path: str,
+            scen_ended_path: str,
+            pycmo_lua_lib_path: str | None = None,
+            max_resets: int = 20,
+            render_mode=None,
+    ):
+        self.cmo_env = CMOEnv(
+            player_side=player_side,
+            steam_client_props=steam_client_props,
+            observation_path=observation_path,
+            action_path=action_path,
+            scen_ended_path=scen_ended_path,
+            pycmo_lua_lib_path=pycmo_lua_lib_path,
+            max_resets=max_resets
+        )
+
+        assert render_mode is None or render_mode in self.metadata["render_modes"]
+        self.render_mode = render_mode
+
+    def _get_obs(self, observation:FeaturesFromSteam) -> dict:
+        ...
+
+    def _get_info(self) -> dict:
+        ...
+
+    def reset(self, seed:int=None, options:dict=None) -> Tuple[dict, dict]:
+        state = self.cmo_env.reset(close_scenario_end_and_player_eval_messages=options['close_scenario_end_and_player_eval_messages'])
+        observation = self._get_obs(observation=state.observation)
+        info = self._get_info()
+
+        return observation, info
+
+    def step(self, action) -> Tuple[dict, int, bool, bool, dict]:
+        state = self.cmo_env.step(action)
+        terminated = self.cmo_env.check_game_ended() or state.step_type == StepType(2)
+        truncated = False
+        reward = state.reward
+        observation = self._get_obs(observation=state.observation)
+        info = self._get_info()
+
+        return observation, reward, terminated, truncated, info
+
+    def close(self) -> None:
+        self.cmo_env.end_game()
+
+class FloridistanPycmoGymEnv(BasePycmoGymEnv):
+    def __init__(
+            self,
+            observation_space:spaces.Space,
+            action_space:spaces.Space,
+            player_side: str,
+            steam_client_props:SteamClientProps,
+            observation_path: str, 
+            action_path: str,
+            scen_ended_path: str,
+            pycmo_lua_lib_path: str | None = None,
+            max_resets: int = 20,
+            render_mode=None,
+    ):
+        super().__init__(
+            player_side=player_side,
+            steam_client_props=steam_client_props,
+            observation_path=observation_path,
+            action_path=action_path,
+            scen_ended_path=scen_ended_path,
+            pycmo_lua_lib_path=pycmo_lua_lib_path,
+            max_resets=max_resets,
+            render_mode=render_mode
+        )
+
+        self.observation_space = observation_space
+        self.action_space = action_space
+
+    def _get_obs(self, observation:FeaturesFromSteam) -> dict:
+        _observation = {}
+
+        unit_name = "Thunder #1"
+        for unit in observation.units:
+            if unit.Name == unit_name:
+                break
+        _observation[unit_name] = {}
+
+        for key in self.observation_space[unit_name].keys():
+            obs_value = getattr(unit, key)
+            if isinstance(obs_value, float):
+                _observation[unit_name][key] = np.array((obs_value,), dtype=np.float64)
+            else:
+                _observation[unit_name][key] = obs_value        
+
+        contact_name = "BTR-82V"
+        for contact in observation.contacts:
+            if contact.Name == contact_name:
+                break
+        _observation[contact_name] = {}
+
+        for key in self.observation_space[contact_name].keys():
+            obs_value = getattr(contact, key)
+            if isinstance(obs_value, float):
+                _observation[contact_name][key] = np.array((obs_value,), dtype=np.float64)
+            else:
+                _observation[contact_name][key] = obs_value
+
+        return _observation        
+
+    def _get_info(self) -> dict:
+        return {}

pycmo/lib/actions.py

@@ -16,13 +16,13 @@
 def no_op():
   return ""
 
-def launch_aircraft(side:str, unit_name:str, launch_yn:str) -> str:
-  return f"ScenEdit_SetUnit({{side = '{side}', name = '{unit_name}', Launch = {launch_yn}}})"
+def launch_aircraft(side:str, unit_name:str, launch:bool=True) -> str:
+  return f"ScenEdit_SetUnit({{side = '{side}', name = '{unit_name}', Launch = {'true' if launch else 'false'}}})"
 
 def set_unit_course(side:str, unit_name:str, latitude:float, longitude:float) -> str:
   return f"ScenEdit_SetUnit({{side = '{side}', name = '{unit_name}', course = {{{{longitude = {longitude}, latitude = {latitude}, TypeOf = 'ManualPlottedCourseWaypoint'}}}}}})"
 
-def manual_attack_contact(attacker_id:str, contact_id:str, weapon_id:str, qty:int, mount_id:str=None) -> str:
+def manual_attack_contact(attacker_id:str, contact_id:str, weapon_id:int, qty:int, mount_id:int=None) -> str:
   return f"ScenEdit_AttackContact('{attacker_id}', '{contact_id}' , {{mode='1', " + (f"mount='{mount_id}', " if mount_id else "") + f"weapon='{weapon_id}', qty='{qty}'}})"
 
 def auto_attack_contact(attacker_id:str, contact_id:str) -> str:
@@ -34,8 +34,8 @@ def refuel_unit(side:str, unit_name:str, tanker_name:str) -> str:
 def auto_refuel_unit(side:str, unit_name:str) -> str:
   return f"ScenEdit_RefuelUnit({{side='{side}', unitname='{unit_name}'}})"
 
-def rtb(side:str, unit_name:str) -> str:
-  return f"ScenEdit_SetUnit({{side = '{side}', name = '{unit_name}', RTB = true}})"
+def rtb(side:str, unit_name:str, return_to_base:bool=True) -> str:
+  return f"ScenEdit_SetUnit({{side = '{side}', name = '{unit_name}', RTB = {'true' if return_to_base else 'false'}}})"
 
 ARG_TYPES = {
   'no_op': ['NoneChoice'],
@@ -45,7 +45,7 @@ def rtb(side:str, unit_name:str) -> str:
   'auto_attack_contact': ['EnumChoice', 'EnumChoice'],
   'refuel_unit': ['EnumChoice', 'EnumChoice', 'EnumChoice'],
   'auto_refuel_unit': ['EnumChoice', 'EnumChoice'],
-  'rtb': ['EnumChoice', 'EnumChoice'],
+  'rtb': ['EnumChoice', 'EnumChoice', 'EnumChoice'],
 }
 
 class AvailableFunctions():
@@ -73,7 +73,7 @@ def get_unit_ids_and_names(self, features:Features|FeaturesFromSteam) -> Tuple[l
   def get_contact_ids(self, features:Features|FeaturesFromSteam) -> list[str]:
     return [contact.ID for contact in features.contacts]
 
-  def get_weapons(self, features:Features|FeaturesFromSteam) -> Tuple[list[str], list[str], list[str], list[int]]:
+  def get_weapons(self, features:Features|FeaturesFromSteam) -> Tuple[list[int], list[int], list[int], list[int]]:
     mount_ids = []
     loadout_ids = []
     weapon_ids = []
@@ -87,7 +87,7 @@ def get_weapons(self, features:Features|FeaturesFromSteam) -> Tuple[list[str], l
       weapon_qtys += unit_mount_weapon_qtys + unit_loadout_weapon_qtys
     return mount_ids, loadout_ids, weapon_ids, weapon_qtys
 
-  def get_mount_ids_weapon_ids_and_qtys(self, unit:Unit) -> Tuple[list[str], list[str], list[int]]:
+  def get_mount_ids_weapon_ids_and_qtys(self, unit:Unit) -> Tuple[list[int], list[int], list[int]]:
     mount_ids = []
     weapon_ids = []
     weapon_qtys = []
@@ -99,7 +99,7 @@ def get_mount_ids_weapon_ids_and_qtys(self, unit:Unit) -> Tuple[list[str], list[
           weapon_qtys.append(weapon.QuantRemaining)
     return mount_ids, weapon_ids, weapon_qtys
 
-  def get_loadout_id_weapon_ids_and_qtys(self, unit:Unit) -> Tuple[str | None, list[str], list[int]]:
+  def get_loadout_id_weapon_ids_and_qtys(self, unit:Unit) -> Tuple[int | None, list[int], list[int]]:
     loadout_id = None
     weapon_ids = []
     weapon_qtys = []
@@ -112,15 +112,18 @@ def get_loadout_id_weapon_ids_and_qtys(self, unit:Unit) -> Tuple[str | None, lis
     return loadout_id, weapon_ids, weapon_qtys
 
   def get_valid_function_args(self) -> dict[str, list]:
+    boolean_list = [True, False]
+    latitude_ranges = [-90.0, 90.0]
+    longitude_ranges = [-180.0, 180.0]
     return {
       'no_op': [],
-      'launch_aircraft': [self.sides, self.unit_names, ["true", "false"]],
-      'set_unit_course': [self.sides, self.unit_names, [-90, 90], [-180, 180]],
+      'launch_aircraft': [self.sides, self.unit_names, boolean_list],
+      'set_unit_course': [self.sides, self.unit_names, latitude_ranges, longitude_ranges],
       'manual_attack_contact': [self.unit_ids, self.contact_ids, self.weapon_ids, self.weapon_qtys, self.mount_ids],
       'auto_attack_contact': [self.unit_ids, self.contact_ids],
       'refuel_unit': [self.sides, self.unit_names, self.unit_names],
       'auto_refuel_unit': [self.sides, self.unit_names],
-      'rtb': [self.sides, self.unit_names]
+      'rtb': [self.sides, self.unit_names, boolean_list]
     }
 
   def get_valid_functions(self) -> list[Function]: