ComputerAssignments/CA20_distributed_memory_systems/src/utils.py

import os
import sys
import logging
import yaml
from pathlib import Path
from typing import Dict, Any, Optional
import warnings

def setup_logging(config: Dict[str, Any]) -> logging.Logger:
    log_config = config.get("logging", {})


    log_file = log_config.get("file", "logs/ca20_analysis.log")
    log_dir = Path(log_file).parent
    log_dir.mkdir(parents=True, exist_ok=True)


    logging.basicConfig(
        level=getattr(logging, log_config.get("level", "INFO")),
        format=log_config.get(
            "format", "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
        ),
        handlers=[
            logging.FileHandler(log_file),
            (
                logging.StreamHandler(sys.stdout)
                if log_config.get("console", True)
                else logging.NullHandler()
            ),
        ],
    )

    logger = logging.getLogger("CA20")
    logger.info(f"Logging initialized. Log file: {log_file}")

    return logger

def load_config(config_path: str = "config.yaml") -> Dict[str, Any]:
    try:
        with open(config_path, "r") as file:
            config = yaml.safe_load(file)


        required_sections = ["project", "system", "benchmark", "output"]
        for section in required_sections:
            if section not in config:
                raise ValueError(f"Missing required configuration section: {section}")

        return config

    except FileNotFoundError:
        raise FileNotFoundError(f"Configuration file not found: {config_path}")
    except yaml.YAMLError as e:
        raise ValueError(f"Error parsing configuration file: {e}")

def create_output_directories(config: Dict[str, Any]) -> Dict[str, Path]:
    output_config = config.get("output", {})
    base_dir = Path(output_config.get("base_dir", "./results"))


    base_dir.mkdir(parents=True, exist_ok=True)


    subdirs = output_config.get("subdirs", {})
    directories = {"base": base_dir}

    for name, subdir in subdirs.items():
        dir_path = base_dir / subdir
        dir_path.mkdir(parents=True, exist_ok=True)
        directories[name] = dir_path

    return directories

def validate_system_requirements() -> Dict[str, Any]:
    import psutil
    import torch

    system_info = {
        "cpu_count": os.cpu_count(),
        "total_memory_gb": psutil.virtual_memory().total / (1024**3),
        "available_memory_gb": psutil.virtual_memory().available / (1024**3),
        "python_version": sys.version,
        "torch_version": torch.__version__,
        "cuda_available": torch.cuda.is_available(),
        "cuda_version": torch.version.cuda if torch.cuda.is_available() else None,
        "gpu_count": torch.cuda.device_count() if torch.cuda.is_available() else 0,
    }

    if torch.cuda.is_available():
        system_info["gpu_names"] = [
            torch.cuda.get_device_name(i) for i in range(torch.cuda.device_count())
        ]
        system_info["gpu_memory_gb"] = [
            torch.cuda.get_device_properties(i).total_memory / (1024**3)
            for i in range(torch.cuda.device_count())
        ]

    return system_info

def print_system_info(system_info: Dict[str, Any]) -> None:
    print("🔧 System Information:")
    print(f"  Python version: {system_info['python_version']}")
    print(f"  PyTorch version: {system_info['torch_version']}")
    print(f"  CPU cores: {system_info['cpu_count']}")
    print(f"  Total RAM: {system_info['total_memory_gb']:.1f} GB")
    print(f"  Available RAM: {system_info['available_memory_gb']:.1f} GB")

    if system_info["cuda_available"]:
        print(f"  GPU: {system_info['gpu_count']} device(s) available")
        for i, (name, memory) in enumerate(
            zip(system_info["gpu_names"], system_info["gpu_memory_gb"])
        ):
            print(f"    GPU {i}: {name} ({memory:.1f} GB)")
    else:
        print("  GPU: Not available (CPU-only mode)")

    print("-" * 60)

def check_dependencies() -> bool:
    required_packages = [
        "numpy",
        "scipy",
        "pandas",
        "matplotlib",
        "seaborn",
        "torch",
        "torchvision",
        "psutil",
        "plotly",
        "yaml",
    ]

    missing_packages = []

    for package in required_packages:
        try:
            __import__(package)
        except ImportError:
            missing_packages.append(package)

    if missing_packages:
        print(f"❌ Missing required packages: {', '.join(missing_packages)}")
        print("Please install them using: pip install -r requirements.txt")
        return False

    print("✅ All required dependencies are available")
    return True

def suppress_warnings() -> None:
    warnings.filterwarnings("ignore", category=UserWarning)
    warnings.filterwarnings("ignore", category=FutureWarning)
    warnings.filterwarnings("ignore", category=DeprecationWarning)


    import torch

    torch.autograd.set_detect_anomaly(False)

def get_project_root() -> Path:
    return Path(__file__).parent.parent

def format_bytes(bytes_value: int) -> str:
    for unit in ["B", "KB", "MB", "GB", "TB"]:
        if bytes_value < 1024.0:
            return f"{bytes_value:.1f} {unit}"
        bytes_value /= 1024.0
    return f"{bytes_value:.1f} PB"

def format_time(seconds: float) -> str:
    if seconds < 1:
        return f"{seconds*1000:.1f} ms"
    elif seconds < 60:
        return f"{seconds:.2f} s"
    elif seconds < 3600:
        return f"{seconds/60:.1f} min"
    else:
        return f"{seconds/3600:.1f} hours"

def create_progress_bar(total: int, desc: str = "Processing") -> "tqdm.tqdm":
    try:
        from tqdm import tqdm

        return tqdm(
            total=total,
            desc=desc,
            unit="item",
            bar_format="{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}]",
        )
    except ImportError:

        return range(total)

def safe_divide(numerator: float, denominator: float, default: float = 0.0) -> float:
    return numerator / denominator if denominator != 0 else default