tdpeuter/2026SEL3-project-Brittle_Star_Controller_Modularity
tdpeuter/2026SEL3-project-Brittle_Star_Controller_Modularity
Archived
1
Fork 0
Code Issues Releases Activity

feat(log): implement LoggerProxy to defer initialization and prevent premature directory creation

Browse source
This commit is contained in:
Tibo De Peuter 2026-04-14 22:53:17 +02:00
parent 71aedc4853
commit eabc64009a
Signed by: tdpeuter
SSH key fingerprint: SHA256:u/h/LVoqKF1Iz02uOyxe6hcjmoZASCGV2HM0TG9ZMoU
3 changed files with 69 additions and 48 deletions
Download patch file Download diff file Expand all files Collapse all files

3
src/experiment_logger/__init__.py
View file

@ -5,7 +5,7 @@ This package provides a unified interface for logging to multiple backends
"""
from experiment_logger.config_utils import load_yaml_config, merge_config_with_cli
from experiment_logger.unified_logger import UnifiedLogger, get_logger
from experiment_logger.unified_logger import UnifiedLogger, get_logger, init_logger
from experiment_logger.simple_logger import SimpleLogger
from experiment_logger.wandb_utils import finish_wandb, init_wandb
@ -13,6 +13,7 @@ __all__ = [
"UnifiedLogger",
"SimpleLogger",
"get_logger",
"init_logger",
"init_wandb",
"finish_wandb",
"load_yaml_config",

24
src/experiment_logger/config_utils.py
View file

@ -8,8 +8,6 @@ from dataclasses import fields, is_dataclass
from experiment_logger.unified_logger import get_logger
log = get_logger()
T = TypeVar("T")
@ -24,7 +22,7 @@ def load_yaml_config(config_path: str) -> Dict[str, Any]:
if config is None:
return {}
log.info(f"Loaded configuration from: {config_path}")
get_logger().info(f"Loaded configuration from: {config_path}")
return config
@ -35,7 +33,7 @@ def save_yaml_config(config: Dict[str, Any], config_path: str):
with open(config_path, "w") as f:
yaml.dump(config, f, default_flow_style=False, indent=2, sort_keys=False)
log.info(f"Saved configuration to: {config_path}")
get_logger().info(f"Saved configuration to: {config_path}")
def dataclass_from_dict(cls: Type[T], config_dict: Dict[str, Any]) -> T:
@ -66,10 +64,10 @@ def dataclass_from_dict(cls: Type[T], config_dict: Dict[str, Any]) -> T:
else:
filtered_config[key] = field.type(value) if value is not None else None # type: ignore
except (ValueError, TypeError) as e:
log.warning(f"Could not convert {key}={value} to {field.type}: {e}")
get_logger().warning(f"Could not convert {key}={value} to {field.type}: {e}")
filtered_config[key] = value
else:
log.warning(f"Unknown configuration parameter: {key}")
get_logger().warning(f"Unknown configuration parameter: {key}")
return cls(**filtered_config)
@ -101,9 +99,9 @@ def merge_config_with_cli(config_class: Type[T], config_file: str | None = None)
yaml_config = {}
if extracted_config_file and os.path.exists(extracted_config_file):
yaml_config = load_yaml_config(extracted_config_file)
log.info(f"Merging YAML config from {extracted_config_file} with CLI args")
get_logger().info(f"Merging YAML config from {extracted_config_file} with CLI args")
elif extracted_config_file:
log.warning(f"Config file not found: {extracted_config_file}, using CLI args only")
get_logger().warning(f"Config file not found: {extracted_config_file}, using CLI args only")
# Create default instance to know what the defaults are
default_instance = config_class()
@ -126,22 +124,22 @@ def merge_config_with_cli(config_class: Type[T], config_file: str | None = None)
if cli_value != default_value:
final_config[field_name] = cli_value
if yaml_value != default_value and yaml_value != cli_value:
log.info(f"CLI override: {field_name}={cli_value} (YAML had {yaml_value})")
get_logger().info(f"CLI override: {field_name}={cli_value} (YAML had {yaml_value})")
else:
final_config[field_name] = yaml_value
if yaml_value != default_value:
log.info(f"YAML config: {field_name}={yaml_value}")
get_logger().info(f"YAML config: {field_name}={yaml_value}")
return config_class(**final_config)
def print_config(config: Any, title: str = "Configuration"):
"""Pretty print configuration."""
log.info(f"{title}:")
get_logger().info(f"{title}:")
if is_dataclass(config):
for field in fields(config):
value = getattr(config, field.name)
log.info(f" {field.name}: {value}")
get_logger().info(f" {field.name}: {value}")
else:
for key, value in vars(config).items():
log.info(f" {key}: {value}")
get_logger().info(f" {key}: {value}")

90
src/experiment_logger/unified_logger.py
View file

@ -21,38 +21,67 @@ import numpy as np
from experiment_logger.wandb_utils import finish_wandb, init_wandb
# Global singleton storage
_global_logger = None
# Global storage for the active logger and the proxy singleton
_active_logger: Optional[Any] = None
_proxy_instance: Optional["LoggerProxy"] = None
def get_logger() -> "UnifiedLogger":
"""Retrieve the global UnifiedLogger. If not initialized, fallback to auto-initialization."""
global _global_logger
if _global_logger is None:
try:
commit_hash = (
subprocess.check_output(
["git", "rev-parse", "--short", "HEAD"], stderr=subprocess.STDOUT
)
.decode("utf-8")
.strip()
)
except Exception:
commit_hash = "unknown"
def get_logger() -> "LoggerProxy":
"""Retrieve the global LoggerProxy.
timestamp_str = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
generic_name = f"{timestamp_str}_{commit_hash}_brittle_star"
This should be used for all logging calls. It returns a proxy that
delegates to the active logger (defaulting to a SimpleLogger until
init_logger is called).
"""
global _proxy_instance, _active_logger
if _proxy_instance is None:
if _active_logger is None:
# Fallback to SimpleLogger to avoid premature directory creation
from experiment_logger.simple_logger import SimpleLogger
# Initialize generic fallback logger without WandB
_global_logger = UnifiedLogger(
run_name=generic_name,
config={"auto_initialized": True},
use_wandb=False,
_set_as_global=False, # Prevent recursive call inside __init__
)
_global_logger.warning(f"UnifiedLogger auto-initialized with name: {generic_name}")
_active_logger = SimpleLogger(run_name="pre_init")
return _global_logger
_proxy_instance = LoggerProxy()
return _proxy_instance
def init_logger(**kwargs) -> "UnifiedLogger":
"""Initialize the full UnifiedLogger and set it as the active logger.
This should be called once the configuration is ready. It will create
the output directories and set up all logging backends.
"""
global _active_logger
logger = UnifiedLogger(_set_as_global=False, **kwargs)
_active_logger = logger
return logger
class LoggerProxy:
"""Proxy that delegates all method calls to the active logger instance.
This allows the logger to be swapped out (e.g., from a SimpleLogger to
a UnifiedLogger) without any clients needing to update their references.
"""
def _get_logger(self) -> Any:
global _active_logger
if _active_logger is None:
# This shouldn't normally happen since get_logger handles it
from experiment_logger.simple_logger import SimpleLogger
_active_logger = SimpleLogger(run_name="pre_init_fallback")
return _active_logger
def __getattr__(self, name: str) -> Any:
return getattr(self._get_logger(), name)
def __enter__(self):
return self._get_logger().__enter__()
def __exit__(self, exc_type, exc_val, exc_tb):
return self._get_logger().__exit__(exc_type, exc_val, exc_tb)
class UnifiedLogger:
@ -68,7 +97,6 @@ class UnifiedLogger:
use_wandb: bool = True,
save_code: bool = True,
log_level: int = logging.INFO,
_set_as_global: bool = True,
):
"""Initialize the unified logger.
@ -80,7 +108,6 @@ class UnifiedLogger:
base_dir: Base directory for local storage
use_wandb: Whether to use WandB logging
save_code: Whether to save code to WandB
_set_as_global: Internal flag to override the global singleton
"""
self.run_name = run_name
self.config = config
@ -119,11 +146,6 @@ class UnifiedLogger:
self._text_logger.addHandler(fh)
self._text_logger.addHandler(ch)
# Set as global singleton
global _global_logger
if _set_as_global:
_global_logger = self
# Save config to disk
self._save_config()