Init ptb

PytorchBoot/utils/__init__.py

@@ -0,0 +1,3 @@
+from PytorchBoot.utils.log_util import Log
+from PytorchBoot.utils.tensorboard_util import TensorboardWriter
+from PytorchBoot.utils.timer_util import Timer

PytorchBoot/utils/log_util.py

@@ -0,0 +1,81 @@
+import time
+import PytorchBoot.namespace as namespace
+from PytorchBoot.status import status_manager
+
+class Log:
+    MAX_TITLE_LENGTH:int = 7
+    TYPE_COLOR_MAP = {
+        namespace.LogType.INFO: "\033[94m",
+        namespace.LogType.ERROR: "\033[91m",
+        namespace.LogType.WARNING: "\033[93m",
+        namespace.LogType.SUCCESS: "\033[92m",
+        namespace.LogType.DEBUG: "\033[95m",
+        namespace.LogType.TERMINATE: "\033[96m"
+    }
+    def get_time():
+        return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
+    
+    def blue(message):
+        # blue
+        print(f"\033[94m{message}\033[0m")
+    def red(message):
+        # red
+        print(f"\033[91m{message}\033[0m")
+    def yellow(message):
+        # yellow
+        print(f"\033[93m{message}\033[0m")
+    def green(message):
+        # green
+        print(f"\033[92m{message}\033[0m")
+        
+    def log(message, log_type: str):
+        time_str = Log.get_time()
+        space = ""
+        if len(log_type) < Log.MAX_TITLE_LENGTH:
+            space = " " * (Log.MAX_TITLE_LENGTH - len(log_type))
+        
+        print (f"\033[1m\033[4m({time_str})\033[0m \033[1m{Log.TYPE_COLOR_MAP[log_type]}[{log_type.capitalize()}]\033[0m{space} {Log.TYPE_COLOR_MAP[log_type]}{message}\033[0m")  
+        status_manager.add_log(time_str, log_type, message)
+    
+    def bold(message):
+        print(f"\033[1m{message}\033[0m")
+    def underline(message):
+        print(f"\033[4m{message}\033[0m")
+        
+    def info(message):
+        Log.log(message, namespace.LogType.INFO)
+        
+    def error(message, terminate=False):
+        Log.log(message, namespace.LogType.ERROR)
+        if terminate:
+            Log.terminate("Application Terminated.")
+            
+            
+    def warning(message):
+        Log.log(message, namespace.LogType.WARNING)
+    def success(message):
+        Log.log(message, namespace.LogType.SUCCESS)
+        
+    def debug(message):
+        Log.log(message, namespace.LogType.DEBUG)
+
+    def terminate(message):
+        Log.log(message, namespace.LogType.TERMINATE)
+        exit(1)  
+
+if __name__ == "__main__":
+    Log.info("This is a info message")
+    Log.error("This is a error message")
+    Log.warning("This is a warning message")
+    Log.success("This is a success message")
+    Log.debug("This is a debug message")
+    Log.blue("This is a blue message")
+    Log.red("This is a red message")
+    Log.yellow("This is a yellow message")
+    Log.green("This is a green message")
+    
+    Log.bold("This is a bold message")
+    Log.underline("This is a underline message")
+    Log.error("This is a terminate message", True)
+        
+        

PytorchBoot/utils/project_util.py

@@ -0,0 +1,50 @@
+import os
+import sys
+import yaml
+import importlib
+
+class ProjectUtil:
+    @staticmethod
+    def scan_project(root_path):
+        sys.path.append(root_path)
+        if not os.path.exists(root_path) or not os.path.isdir(root_path):
+            raise ValueError(f"The provided root_path '{root_path}' is not a valid directory.")
+    
+        parent_dir = os.path.dirname(root_path)
+        sys.path.insert(0, parent_dir)
+        
+        def import_all_modules(path, package_name):
+            for root, dirs, files in os.walk(path):
+                relative_path = os.path.relpath(root, root_path)
+                if relative_path == '.':
+                    module_package = package_name
+                else:
+                    module_package = f"{package_name}.{relative_path.replace(os.sep, '.')}"
+                for file in files:
+                    if file.endswith(".py") and file != "__init__.py":
+                        module_name = file[:-3]
+                        full_module_name = f"{module_package}.{module_name}"
+                        if full_module_name not in sys.modules:
+                            importlib.import_module(full_module_name)
+                dirs[:] = [d for d in dirs if not d.startswith('.')]
+        
+        package_name = os.path.basename(root_path)
+        import_all_modules(root_path, package_name)
+        
+    @staticmethod
+    def scan_configs(root_path):
+        configs = {}
+        for root, dirs, files in os.walk(root_path):
+            for file in files:
+                if file.endswith(('.yaml', '.yml')):
+                    if file.startswith('__'):
+                        continue
+                    file_path = os.path.join(root, file)
+                    with open(file_path, 'r', encoding='utf-8') as f:
+                        try:
+                            content = yaml.safe_load(f)
+                            configs[os.path.splitext(file)[0]] = content
+                        except yaml.YAMLError as e:
+                            print(f"Error reading {file_path}: {e}")
+        
+        return configs

PytorchBoot/utils/tensorboard_util.py

@@ -0,0 +1,44 @@
+import torch
+import PytorchBoot.namespace as namespace
+
+class TensorboardWriter:
+    @staticmethod
+    def write_tensorboard(writer, panel, data_dict, step, simple_scalar = False):
+        
+        if simple_scalar:
+            TensorboardWriter.write_scalar_tensorboard(writer, panel, data_dict, step)
+
+        if namespace.TensorBoard.SCALAR in data_dict:
+            scalar_data_dict = data_dict[namespace.TensorBoard.SCALAR]
+            TensorboardWriter.write_scalar_tensorboard(writer, panel, scalar_data_dict, step)
+        if namespace.TensorBoard.IMAGE in data_dict:
+            image_data_dict = data_dict[namespace.TensorBoard.IMAGE]
+            TensorboardWriter.write_image_tensorboard(writer, panel, image_data_dict, step)
+        if namespace.TensorBoard.POINT in data_dict:
+            point_data_dict = data_dict[namespace.TensorBoard.POINT]
+            TensorboardWriter.write_points_tensorboard(writer, panel, point_data_dict, step)
+
+    @staticmethod
+    def write_scalar_tensorboard(writer, panel, data_dict, step):
+        for key, value in data_dict.items():
+            if isinstance(value, dict):
+                writer.add_scalars(f'{panel}/{key}', value, step)
+            else:
+                writer.add_scalar(f'{panel}/{key}', value, step)
+
+    @staticmethod
+    def write_image_tensorboard(writer, panel, data_dict, step):
+        pass
+
+    @staticmethod
+    def write_points_tensorboard(writer, panel, data_dict, step):
+        for key, value in data_dict.items():
+            if value.shape[-1] == 3:
+                colors = torch.zeros_like(value)
+                vertices = torch.cat([value, colors], dim=-1)
+            elif value.shape[-1] == 6:
+                vertices = value
+            else:
+                raise ValueError(f'Unexpected value shape: {value.shape}')
+            faces = None
+            writer.add_mesh(f'{panel}/{key}', vertices=vertices, faces=faces, global_step=step)

PytorchBoot/utils/timer_util.py

@@ -0,0 +1,32 @@
+import time
+
+class Timer:
+    MILLI_SECONDS = "milliseconds"
+    SECONDS = "seconds"
+    MINUTES = "minutes"
+    HOURS = "hours"
+    def __init__(self, name=None):
+        self.start_time = None
+        self.end_time = None
+        self.name = name
+
+    def start(self):
+        self.start_time = time.time()
+
+    def stop(self):
+        self.end_time = time.time()
+
+    def elapsed_time(self):
+        return int(self.end_time - self.start_time)
+    
+    def get_elasped_time_str(self, format):
+        if format == Timer.SECONDS:
+            return f"Elapsed time in <{self.name}>: {self.elapsed_time()} seconds"
+        elif format == Timer.MINUTES:
+            return f"Elapsed time in <{self.name}>: {self.elapsed_time() // 60} minutes, {self.elapsed_time() % 60} seconds"
+        elif format == Timer.HOURS:
+            return f"Elapsed time in <{self.name}>: {self.elapsed_time() // 3600} hours, {(self.elapsed_time() % 3600)//60} minutes, {self.elapsed_time() % 60} seconds"
+        elif format == Timer.MILLI_SECONDS:
+            return f"Elapsed time in <{self.name}>: {(self.end_time - self.start_time) * 1000} milliseconds"
+        else:
+            return f"Invalid format: {format}"