v0.1.2

robogauge/utils/config.py

@@ -0,0 +1,18 @@
+import inspect
+
+class Config:
+    # From https://github.com/leggedrobotics/legged_gym/blob/master/legged_gym/envs/base/base_config.py
+    def __init__(self) -> None:
+        """ Initializes all member classes recursively. Ignores all namse starting with '__' (buit-in methods)."""
+        self.init_member_classes(self)
+    
+    @staticmethod
+    def init_member_classes(obj):
+        for key in dir(obj):
+            if key == "__class__":
+                continue
+            var = getattr(obj, key)
+            if inspect.isclass(var):
+                i_var = var()
+                setattr(obj, key, i_var)
+                Config.init_member_classes(i_var)

robogauge/utils/helpers.py

@@ -0,0 +1,56 @@
+# -*- coding: utf-8 -*-
+'''
+@File    : helpers.py
+@Time    : 2025/11/27 15:26:37
+@Author  : wty-yy
+@Version : 1.0
+@Blog    : https://wty-yy.github.io/
+@Desc    : Helpers for robogauge, include:
+- Argument parsing
+- Class to dict conversion
+- Path parsing
+'''
+from argparse import ArgumentParser
+from pathlib import Path
+from robogauge import ROBOGAUGE_ROOT_DIR
+
+def pares_path(path):
+    if "{ROBOGAUGE_ROOT_DIR}" in str(path):
+        path = str(path).replace("{ROBOGAUGE_ROOT_DIR}", ROBOGAUGE_ROOT_DIR)
+    return path
+
+def class_to_dict(obj) -> dict:
+    if not hasattr(obj, "__dict__"):
+        return obj
+    result = {}
+    for key in dir(obj):
+        if key.startswith("_"):
+            continue
+        element = []
+        val = getattr(obj, key)
+        if isinstance(val, list):
+            for item in val:
+                element.append(class_to_dict(item))
+        else:
+            element = class_to_dict(val)
+        result[key] = element
+    return result
+
+def str2bool(v):
+    if v.lower() in ('yes', 'true', 't', 'y', '1'):
+        return True
+    elif v.lower() in ('no', 'false', 'f', 'n', '0'):
+        return False
+    raise TypeError('Boolean value expected.')
+
+def parse_args():
+    parser = ArgumentParser()
+    parameters = [
+        {"name": "--task-name", "type": str, "default": "base", "help": "Name of the task to run."},
+        {"name": "--experiment-name", "type": str, "default": "base", "help": "Name of the experiment to run."},
+        {"name": "--headless", "action": "store_true", "default": False, "help": "Run in headless mode."},
+        {"name": "--save-video", "action": "store_true", "default": False, "help": "Save video output."},
+    ]
+    for param in parameters:
+        parser.add_argument(param['name'], **{k: v for k, v in param.items() if k != 'name'})
+    return parser.parse_args()

robogauge/utils/logger.py

@@ -0,0 +1,133 @@
+# -*- coding: utf-8 -*-
+'''
+@File    : my_logger.py
+@Time    : 2025/02/26 21:43:47
+@Author  : wty-yy
+@Version : 1.0
+@Blog    : https://wty-yy.github.io/
+@Desc    : A customed logger, support: 
+1. Color level name
+2. Output to console and save log to file
+3. Support vscode file location jump (ctrl+left key)
+'''
+import time
+import logging
+from pathlib import Path
+from robogauge import ROBOGAUGE_ROOT_DIR
+
+class LogColor:
+    """ ANSI color codes """
+    RESET = '\033[0m'
+    RED   = '\033[31m'
+    GREEN = '\033[32m'
+    YELLOW = '\033[33m'
+    BLUE  = '\033[34m'
+    MAGENTA = '\033[35m'
+    CYAN  = '\033[36m'
+    WHITE = '\033[37m'
+    BOLD  = '\033[1m'
+    UNDERLINE = '\033[4m'
+
+LOG_COLORS = {
+    """ Match level name to color """
+    'DEBUG': LogColor.CYAN,
+    'INFO': LogColor.GREEN,
+    'WARNING': LogColor.YELLOW,
+    'ERROR': LogColor.RED,
+    'CRITICAL': LogColor.RED + LogColor.BOLD,
+}
+
+class ColorFormatter(logging.Formatter):
+    """Color Formatter for color_level"""
+    def __init__(self, fmt, datefmt=None, use_color=True):
+        self.formatter = logging.Formatter(fmt, datefmt)
+        self.use_color = use_color
+
+    def format(self, record):
+        record.color_level = f"{LOG_COLORS.get(record.levelname, LogColor.RESET)}{record.levelname}{LogColor.RESET}" if self.use_color else record.levelname
+        return self.formatter.format(record)
+
+
+class Logger:
+    def create(self,
+        experiment_name,
+        console_output=True, color_output=True,
+        log_level=logging.DEBUG, save_file_mode='a'
+    ):
+        """
+        Create customed Logger
+
+        Args:
+            logger_name (str): Logger name
+            console_output (bool, optional): Whether output to console. Defaults to True.
+            color_output (bool, optional): Whether use color output. Defaults to True.
+            log_level (int, optional): Defaults to logging.DEBUG.
+            save_file_mode (str, optional): The mode of saving to path_log_file
+
+        Returns:
+            logging.Logger: logger
+        """
+        self.logger = logging.getLogger(experiment_name + "_logger")
+        self.logger.setLevel(log_level)
+
+        console_formatter = ColorFormatter(    # console output format
+            fmt="%(asctime)s - %(color_level)s - %(filename)s:%(lineno)d - %(message)s",
+            datefmt="%Y-%m-%d %H:%M:%S",
+            use_color=color_output
+        )
+        file_formatter = logging.Formatter(    # file output format
+            fmt="%(asctime)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s",
+            datefmt="%Y-%m-%d %H:%M:%S",
+        )
+
+        if console_output:
+            sh = logging.StreamHandler()
+            sh.setFormatter(console_formatter)
+            self.logger.addHandler(sh)
+
+        self.log_dir = Path(ROBOGAUGE_ROOT_DIR) / "logs" / experiment_name / time.strftime("%Y%m%d-%H-%M-%S")
+        self.log_dir.mkdir(parents=True, exist_ok=True)
+        path_log_file = self.log_dir / "stdout.log"
+        if path_log_file:
+            fh = logging.FileHandler(path_log_file, mode=save_file_mode, encoding='utf-8')
+            fh.setFormatter(file_formatter)
+            self.logger.addHandler(fh)
+    
+    def debug(self, msg, *args, **kwargs):
+        self.logger.debug(msg, *args, **kwargs)
+    
+    def info(self, msg, *args, **kwargs):
+        self.logger.info(msg, *args, **kwargs)
+    
+    def warning(self, msg, *args, **kwargs):
+        self.logger.warning(msg, *args, **kwargs)
+
+    def error(self, msg, *args, **kwargs):
+        self.logger.error(msg, *args, **kwargs)
+
+    def critical(self, msg, *args, **kwargs):
+        self.logger.critical(msg, *args, **kwargs)
+
+logger = Logger()
+
+if __name__ == '__main__':
+    from pathlib import Path
+    path_parent = Path(__file__).parents[0]
+    path_log = path_parent / "app.log"
+
+    logger = Logger()
+    logger.create("my_logger")
+
+    logger.debug("This is a debug message")
+    logger.info("This is an info message")
+    logger.warning("This is a warning message")
+    logger.error("This is an error message")
+    logger.critical("This is a critical message")
+
+    # logger_no_color = Logger()
+    # logger_no_color.create("no_color_logger", console_output=True, color_output=False)
+    # logger_no_color.info("This is a info message without color")
+
+    # logger_file_only = Logger()
+    # logger_file_only.create("file_only_logger", console_output=False)  # save to file only
+    # logger_file_only.error("This is an error message only in file")

robogauge/utils/task_register.py

@@ -0,0 +1,55 @@
+# -*- coding: utf-8 -*-
+'''
+@File    : task_register.py
+@Time    : 2025/11/27 15:59:03
+@Author  : wty-yy
+@Version : 1.0
+@Blog    : https://wty-yy.github.io/
+@Desc    : Task Registration Utility
+'''
+class TaskRegister():
+    def __init__(self):
+        self.pipeline_classes = {}
+        self.sim_cfgs = {}
+        self.gauger_cfgs = {}
+        self.robot_cfgs = {}
+    
+    def register(self, name: str, pipeline_class, sim_cfg, gauger_cfg, robot_cfg):
+        self.pipeline_classes[name] = pipeline_class
+        self.sim_cfgs[name] = sim_cfg
+        self.gauger_cfgs[name] = gauger_cfg
+        self.robot_cfgs[name] = robot_cfg
+    
+    def get_pipeline_class(self, name: str):
+        if name not in self.pipeline_classes:
+            raise ValueError(f"Task '{name}' is not registered.")
+        return self.pipeline_classes[name]
+    
+    def get_cfgs(self, name):
+        if name not in self.sim_cfgs:
+            raise ValueError(f"Task '{name}' is not registered.")
+        sim_cfg = self.sim_cfgs[name]
+        gauger_cfg = self.gauger_cfgs[name]
+        robot_cfg = self.robot_cfgs[name]
+        return sim_cfg, gauger_cfg, robot_cfg
+    
+    def make_pipeline(self, name, args=None, sim_cfg=None, gauger_cfg=None, robot_cfg=None):
+        default_cfgs = self.get_cfgs(name)
+        if sim_cfg is None:
+            sim_cfg = default_cfgs[0]
+        if gauger_cfg is None:
+            gauger_cfg = default_cfgs[1]
+        if robot_cfg is None:
+            robot_cfg = default_cfgs[2]
+        if args is not None:
+            self.update_args_to_cfg(sim_cfg, gauger_cfg, robot_cfg, args)
+        pipeline_class = self.get_pipeline_class(name)
+        return pipeline_class(sim_cfg, robot_cfg, gauger_cfg)
+
+    def update_args_to_cfg(self, sim_cfg, gauger_cfg, robot_cfg, args):
+        if args.headless is not None:
+            sim_cfg.viewer.headless = args.headless
+        if args.save_video is not None:
+            sim_cfg.render.save_video = args.save_video
+
+task_register = TaskRegister()