Added wrapper capabilities

Author: mattia-gramuglia

acsl_pychrono/config/__init__.py

@@ -0,0 +1,4 @@
+# acsl_pychrono/config/__init__.py
+
+from .config import MissionConfig, VehicleConfig, EnvironmentConfig, WrapperParams
+from .config import SimulationConfig

acsl_pychrono/config/config.py

@@ -1,9 +1,10 @@
+import warnings
 from dataclasses import dataclass
 
 @dataclass
-class SimulationConfig:
+class MissionConfig:
   # Total simulation duration in seconds
-  simulation_duration_seconds: float = 21.5
+  simulation_duration_seconds: float = 21.5 # 21.5
   # Run the simulator in Wrapper mode (more simulations automatically run sequentially)
   wrapper_flag: bool = False
   # If True, perform real-time rendering of the simulation with Irrlicht
@@ -22,7 +23,7 @@ class SimulationConfig:
   # Controller types:
   # "PID",
   # "MRAC",
-  controller_type: str = "PID"
+  controller_type: str = "MRAC"
 
   # User-defined trajectory types:
   # "circular_trajectory",
@@ -44,6 +45,17 @@ class SimulationConfig:
   # "many_steel_balls_in_random_position"
   payload_type: str = "two_steel_balls"
 
+  # Unique wrapper batch folder passed to the function used for running many parallel wrapper simulations 
+  wrapper_batch_dir: str = "" # LEAVE BLANK!!!
+
+  # Number of parallel simulations (one per CPU) to be run in "wrapper" mode
+  wrapper_max_parallel: int = 20
+
+  def __post_init__(self):
+    if self.wrapper_flag and self.visualization_flag:
+      warnings.warn("Visualization is disabled because wrapper mode is enabled.")
+      self.visualization_flag = False
+
 @dataclass
 class VehicleConfig:
   # Path relative to 'current_working_directory/assets/vehicles'
@@ -55,4 +67,15 @@ class EnvironmentConfig:
   include: bool = False
   # Path relative to 'current_working_directory/assets/environments'
   model_relative_path: str = "environmentA/environmentA.py" 
+
+@dataclass
+class WrapperParams: # Add here the params to be sweeped by the wrapper with their default values
+  my_ball_density: float = 7850
+
+@dataclass
+class SimulationConfig:
+  mission_config: MissionConfig = MissionConfig()
+  vehicle_config: VehicleConfig = VehicleConfig()
+  environment_config: EnvironmentConfig = EnvironmentConfig()
+  wrapper_params: WrapperParams = WrapperParams()
 

acsl_pychrono/control/MRAC/mrac.py

@@ -2,8 +2,8 @@
 import numpy as np  
 from acsl_pychrono.control.outerloop_safetymech import OuterLoopSafetyMechanism
 from acsl_pychrono.control.MRAC.mrac_gains import MRACGains
-from acsl_pychrono.ode_input import OdeInput
-from acsl_pychrono.flight_params import FlightParams
+from acsl_pychrono.simulation.ode_input import OdeInput
+from acsl_pychrono.simulation.flight_params import FlightParams
 from acsl_pychrono.control.control import Control
 from acsl_pychrono.control.base_mrac import BaseMRAC
 from acsl_pychrono.control.MRAC.m_mrac import M_MRAC

acsl_pychrono/control/MRAC/mrac_gains.py

@@ -2,7 +2,7 @@
 import numpy as np
 from numpy import linalg as LA
 from scipy import linalg
-from acsl_pychrono.flight_params import FlightParams
+from acsl_pychrono.simulation.flight_params import FlightParams
 
 class MRACGains:
   def __init__(self, flight_params: FlightParams):

acsl_pychrono/control/PID/pid.py

@@ -2,8 +2,8 @@
 import numpy as np  
 from acsl_pychrono.control.outerloop_safetymech import OuterLoopSafetyMechanism
 from acsl_pychrono.control.PID.pid_gains import PIDGains
-from acsl_pychrono.ode_input import OdeInput
-from acsl_pychrono.flight_params import FlightParams
+from acsl_pychrono.simulation.ode_input import OdeInput
+from acsl_pychrono.simulation.flight_params import FlightParams
 from acsl_pychrono.control.control import Control
 
 class PID(Control):

acsl_pychrono/control/PID/pid_gains.py

@@ -3,7 +3,7 @@
 from numpy import linalg as LA
 import scipy
 from scipy import linalg
-from acsl_pychrono.flight_params import FlightParams
+from acsl_pychrono.simulation.flight_params import FlightParams
 
 class PIDGains:
   def __init__(self, flight_params: FlightParams):

acsl_pychrono/control/PID/pid_logger.py

@@ -2,8 +2,8 @@
 import numpy as np  
 from acsl_pychrono.control.PID.pid_gains import PIDGains
 from acsl_pychrono.control.PID.pid import PID
-from acsl_pychrono.ode_input import OdeInput
-from acsl_pychrono.flight_params import FlightParams
+from acsl_pychrono.simulation.ode_input import OdeInput
+from acsl_pychrono.simulation.flight_params import FlightParams
 
 class PIDLogger:
   def __init__(self, gains: PIDGains) -> None:

acsl_pychrono/control/base_mrac.py

@@ -2,8 +2,8 @@
 import numpy as np
 from numpy import linalg as LA
 
-from acsl_pychrono.ode_input import OdeInput
-from acsl_pychrono.flight_params import FlightParams
+from acsl_pychrono.simulation.ode_input import OdeInput
+from acsl_pychrono.simulation.flight_params import FlightParams
 from acsl_pychrono.control.control import Control
 
 class BaseMRAC():  

acsl_pychrono/control/control.py

@@ -3,9 +3,9 @@
 from numpy.typing import NDArray
 from abc import ABC, abstractmethod
 
-from acsl_pychrono.functions import rk4singlestep
-from acsl_pychrono.ode_input import OdeInput
-from acsl_pychrono.flight_params import FlightParams
+from acsl_pychrono.simulation.functions import rk4singlestep
+from acsl_pychrono.simulation.ode_input import OdeInput
+from acsl_pychrono.simulation.flight_params import FlightParams
 
 class Control:
   def __init__(self, odein: OdeInput) -> None:

acsl_pychrono/control/logging.py

@@ -1,31 +1,50 @@
 import os
 import datetime
+import subprocess
+import warnings
 import numpy as np
 from scipy.io import savemat
+from dataclasses import is_dataclass
+import acsl_pychrono.config.config as Cfg
 
 class Logging:
   @staticmethod
-  def saveMatlabWorkspaceLog(log_dict, gains, controller_type):
-    """
-    Save the log and gains data into a .mat file in a structured directory.
-
-    Args:
-      log_dict (dict): Dictionary containing the log data.
-      gains (object): Object containing gain parameters as attributes.
-      controller_type (str): The name/type of the controller.
-    """
+  def getOutputDir(sim_cfg: Cfg.SimulationConfig) -> str:
+    controller_type = Cfg.MissionConfig.controller_type
+    wrapper_flag = Cfg.MissionConfig.wrapper_flag
 
     # Get current time
     now = datetime.datetime.now()
-    timestamp = now.strftime("%Y%m%d_%H%M%S")
     year = now.strftime("%Y")
     month = now.strftime("%m")
     full_date = now.strftime("%Y%m%d")
 
     # Construct the directory path
-    dir_path = os.path.join("logs", year, month, full_date, controller_type, "workspaces")
+    if wrapper_flag:
+      dir_path = os.path.join(sim_cfg.mission_config.wrapper_batch_dir)
+    else:
+      dir_path = os.path.join("logs", year, month, full_date, controller_type, "workspaces")
     os.makedirs(dir_path, exist_ok=True)  # Create all directories if not present
 
+    return dir_path
+  
+  @staticmethod
+  def generateUniqueFilename(base_name: str, extension: str, dir_path: str, use_suffix: bool) -> str:
+    if not use_suffix:
+      # No suffix, allow overwrite
+      return os.path.join(dir_path, f"{base_name}.{extension}")
+    
+    # Suffix to avoid overwrite
+    run_id = 1
+    while True:
+      filename = f"{base_name}-{run_id}.{extension}"
+      full_path = os.path.join(dir_path, filename)
+      if not os.path.exists(full_path):
+        return full_path
+      run_id += 1
+
+  @staticmethod
+  def extractGainsDict(gains) -> dict:
     # Create a dictionary from instance variables
     gains_dict = {
       key: value for key, value in gains.__dict__.items()
@@ -40,16 +59,102 @@ def saveMatlabWorkspaceLog(log_dict, gains, controller_type):
     for key in gains_dict_shortened:
       if isinstance(gains_dict_shortened[key], np.matrix):
         gains_dict_shortened[key] = np.array(gains_dict_shortened[key])
+    
+    return gains_dict_shortened
+
+  @staticmethod
+  def dataclassToDict(obj, truncate_keys: bool = True):
+    """
+    Recursively convert a dataclass to a nested dictionary,
+    handling nested dataclasses, NumPy arrays/matrices,
+    truncating field names to 31 characters if specified,
+    and filtering out unsupported types.
+    """
+    if is_dataclass(obj):
+      result = {}
+      for key, value in obj.__dict__.items():
+        # Filter to only include serializable fields
+        if isinstance(value, (int, float, str, bool, np.ndarray, np.matrix)) or is_dataclass(value):
+          converted_value = Logging.dataclassToDict(value, truncate_keys=truncate_keys)
+          key = key[:31] if truncate_keys else key
+          if isinstance(converted_value, np.matrix):
+            converted_value = np.array(converted_value)
+          result[key] = converted_value
+      return result
+    elif isinstance(obj, dict):
+      return {k[:31] if truncate_keys else k: Logging.dataclassToDict(v, truncate_keys=truncate_keys) for k, v in obj.items()}
+    elif isinstance(obj, (list, tuple)):
+      return [Logging.dataclassToDict(v, truncate_keys=truncate_keys) for v in obj]
+    elif isinstance(obj, np.matrix):
+      return np.array(obj)
+    else:
+      return obj
+    
+  @staticmethod
+  def getGitRepoInfo() -> dict:
+    def run_git_cmd(args):
+      try:
+        return subprocess.check_output(['git'] + args, cwd=repo_dir, stderr=subprocess.DEVNULL).decode().strip()
+      except subprocess.CalledProcessError:
+        return ""
+      except FileNotFoundError:
+        return ""
+      
+    def get_github_url(repo_url: str, commit_hash: str) -> str:
+      if repo_url.endswith(".git"):
+        repo_url = repo_url[:-4]
+      return f"{repo_url}/tree/{commit_hash}"
+    
+    repo_dir = os.getcwd()
+
+    if not os.path.exists(os.path.join(repo_dir, '.git')):
+      warnings.warn("⚠️ Git repository info cannot be tracked. No .git directory found.")
+      return {
+        "repo_path": "",
+        "remote_url": "",
+        "commit_hash": "",
+        "commit_tag": "",
+        "branch": "",
+        "dirty": 0,
+        "commit_url": ""
+      }
+    
+    remote_url = run_git_cmd(['remote', 'get-url', 'origin']) or ""
+    commit_hash = run_git_cmd(['rev-parse', 'HEAD']) or ""
+
+    git_info = {
+      "repo_path": os.path.abspath(repo_dir),
+      "remote_url": remote_url,
+      "commit_hash": commit_hash,
+      "commit_tag": run_git_cmd(['describe', '--tags', '--exact-match']) or "",
+      "branch": run_git_cmd(['rev-parse', '--abbrev-ref', 'HEAD']) or "",
+      "dirty": 1 if run_git_cmd(['status', '--porcelain']) else 0,
+      "commit_url": get_github_url(remote_url, commit_hash) or ""
+    }
+
+    return git_info
+
+  @staticmethod
+  def saveMatlabWorkspaceLog(log_dict, gains, sim_cfg: Cfg.SimulationConfig, git_info: dict | None = None):
+    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
+    dir_path = Logging.getOutputDir(sim_cfg)
+    os.makedirs(dir_path, exist_ok=True)
+
+    base_filename = f"workspace_log_{timestamp}"
+    full_path_log = Logging.generateUniqueFilename(
+      base_filename,
+      "mat",
+      dir_path,
+      Cfg.MissionConfig.wrapper_flag
+    )
 
-    # Nest gains inside the log structure
     mat_dict = {
-      "log": log_dict,  # Log data
-      "gains": gains_dict_shortened  # Nested gains struct
+      "log": log_dict,
+      "gains": Logging.extractGainsDict(gains),
+      "sim_cfg": Logging.dataclassToDict(sim_cfg)
     }
 
-    # Construct the full file path for the log file
-    filename_log = f"workspace_log_{timestamp}.mat"
-    full_path_log = os.path.join(dir_path, filename_log)
+    if git_info is not None:
+      mat_dict["git_info"] = git_info
 
-    # Save the .mat file with the nested gains structure
     savemat(full_path_log, mat_dict)