Blame for utils/config.py

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import argparse
import os
from omegaconf import OmegaConf

from torch.distributed.fsdp import MixedPrecisionPolicy, CPUOffloadPolicy
import torch


class DeviceType(str, Enum):
	CUDA: str = 'cuda'
	CPU: str = 'cpu'

@dataclass
class FSDPMixedPrecisionConfig:
    param_dtype: str = "bfloat16"
    reduce_dtype: str = "float32"
    output_dtype: str = "float32"
    cast_forward_inputs: bool = True

    def _resolve_dtype(self, value: str | torch.dtype) -> torch.dtype:
        if isinstance(value, torch.dtype):
            return value
        try:
            return getattr(torch, value)
        except AttributeError as exc:
            raise ValueError(f"Unknown torch dtype '{value}'") from exc

    def to_policy(self) -> MixedPrecisionPolicy:
        return MixedPrecisionPolicy(
            param_dtype=self._resolve_dtype(self.param_dtype),
            reduce_dtype=self._resolve_dtype(self.reduce_dtype),
            output_dtype=self._resolve_dtype(self.output_dtype),
            cast_forward_inputs=self.cast_forward_inputs,
        )


@dataclass
class DistributedConfig:
	use_ddp: bool = False
	use_fsdp: bool = False
	reshard_after_forward: bool = False
	fsdp_mixed_precision: FSDPMixedPrecisionConfig | None = field(default_factory=FSDPMixedPrecisionConfig)
	offload_policy: CPUOffloadPolicy | None = None

	def __post_init__(self) -> None:
		if self.use_ddp and self.use_fsdp:
			raise ValueError("DistributedConfig cannot enable both DDP and FSDP; choose only one.")

	def get_mixed_precision_policy(self) -> MixedPrecisionPolicy | None:
		if self.fsdp_mixed_precision is None:
			return None
		if isinstance(self.fsdp_mixed_precision, MixedPrecisionPolicy):
			return self.fsdp_mixed_precision
		return self.fsdp_mixed_precision.to_policy()


def _validate_amp_fsdp(amp: bool, distributed: DistributedConfig) -> None:
	if amp and distributed.use_fsdp:
		raise ValueError(
			"Disable AMP when using FSDP; configure mixed precision via distributed.fsdp_mixed_precision instead."
		)


def _validate_distibuted_training(nproc_per_node: int, distributed: DistributedConfig) -> None:
	if nproc_per_node > 1 and not (distributed.use_ddp or distributed.use_fsdp):
		raise ValueError(
			"nproc_per_node > 1 requires enabling distributed.use_ddp or distributed.use_fsdp."
		)


def _validate_distributed_device(device: DeviceType, distributed: DistributedConfig) -> None:
	current_device = device if isinstance(device, DeviceType) else DeviceType(device)
	if (distributed.use_ddp or distributed.use_fsdp) and current_device is not DeviceType.CUDA:
		raise ValueError("Distributed training (DDP/FSDP) requires device=cuda.")


@dataclass
class VideoTokenizerConfig:
	# Training
	batch_size_per_gpu: int
	gradient_accumulation_steps: int
	n_updates: int # number of optimizer.step(), excluding grad_accum_step
	learning_rate: float
	log_interval: int
	dataset: str
	context_length: int
	frame_size: int
	# Model
	patch_size: int
	embed_dim: int
	num_heads: int
	hidden_dim: int
	num_blocks: int
	latent_dim: int
	num_bins: int
	amp: bool
	tf32: bool
	compile: bool
	# distributed
	distributed: DistributedConfig
	nproc_per_node: int
	standalone: bool
	# W&B
	use_wandb: bool
	wandb_project: str
	# resume from checkpoint
	checkpoint: Optional[str]
	# Optimizer
	optimizer: str = "adamw"
	muon_momentum: float = 0.95
	muon_backend_steps: int = 5
	# device
	device: DeviceType = DeviceType.CUDA
	# other params
	fps: Optional[int] = None
	preload_ratio: Optional[float] = None
	
	def __post_init__(self) -> None:
		_validate_amp_fsdp(self.amp, self.distributed)
		_validate_distibuted_training(self.nproc_per_node, self.distributed)
		_validate_distributed_device(self.device, self.distributed)


@dataclass
class LatentActionsConfig:
	# Training
	batch_size_per_gpu: int
	gradient_accumulation_steps: int
	n_updates: int # number of optimizer.step(), excluding grad_accum_step
	learning_rate: float
	log_interval: int
	dataset: str
	context_length: int
	frame_size: int
	# Model
	n_actions: int
	patch_size: int
	embed_dim: int
	num_heads: int
	hidden_dim: int
	num_blocks: int
	amp: bool
	tf32: bool
	compile: bool
	# distributed
	distributed: DistributedConfig
	nproc_per_node: int
	standalone: bool
	# W&B
	use_wandb: bool
	wandb_project: str
	# resume from checkpoint
	checkpoint: Optional[str]
	# Optimizer
	optimizer: str = "adamw"
	muon_momentum: float = 0.95
	muon_backend_steps: int = 5
	# device
	device: DeviceType = DeviceType.CUDA
	# other params
	fps: Optional[int] = None
	preload_ratio: Optional[float] = None
	
	def __post_init__(self) -> None:
		_validate_amp_fsdp(self.amp, self.distributed)
		_validate_distibuted_training(self.nproc_per_node, self.distributed)
		_validate_distributed_device(self.device, self.distributed)


@dataclass
class DynamicsConfig:
	# Training
	batch_size_per_gpu: int
	gradient_accumulation_steps: int
	n_updates: int # number of optimizer.step(), excluding grad_accum_step
	learning_rate: float
	log_interval: int
	dataset: str
	context_length: int
	frame_size: int
	# Model (must match tokenizer)
	patch_size: int
	embed_dim: int
	num_heads: int
	hidden_dim: int
	num_blocks: int
	latent_dim: int
	num_bins: int
	n_actions: int
	use_actions: bool
	# Paths
	video_tokenizer_path: Optional[str]
	latent_actions_path: Optional[str]
	# Perf
	amp: bool
	tf32: bool
	compile: bool
	# distributed
	distributed: DistributedConfig
	nproc_per_node: int
	standalone: bool
	# W&B
	use_wandb: bool
	wandb_project: str
	# resume from checkpoint
	checkpoint: Optional[str]
	# MoE
	use_moe: bool = False
	num_experts: int = 4
	top_k_experts: int = 2
	moe_aux_loss_coeff: float = 0.01
	# Optimizer
	optimizer: str = "adamw"
	muon_momentum: float = 0.95
	muon_backend_steps: int = 5
	# device
	device: DeviceType = DeviceType.CUDA
	# other params
	fps: Optional[int] = None
	preload_ratio: Optional[float] = None
	
	def __post_init__(self) -> None:
		_validate_amp_fsdp(self.amp, self.distributed)
		_validate_distibuted_training(self.nproc_per_node, self.distributed)
		_validate_distributed_device(self.device, self.distributed)


@dataclass
class TrainingConfig:
	# WandB
	use_wandb: bool
	wandb_project: str
	# Dataset
	dataset: str
	# Config paths for stages
	video_tokenizer_config: str
	latent_actions_config: str
	dynamics_config: str
	# Which stages to run
	run_video_tokenizer: bool
	run_latent_actions: bool
	run_dynamics: bool
	# Shared model params
	patch_size: int
	context_length: int
	frame_size: int
	latent_dim: int
	num_bins: int
	n_actions: int
	# Performance
	amp: bool
	tf32: bool
	compile: bool
	# distributed
	distributed: DistributedConfig
	nproc_per_node: int
	standalone: bool
	# device
	device: DeviceType = DeviceType.CUDA
	# These can vary per model
	embed_dim: Optional[int] = None
	num_heads: Optional[int] = None
	hidden_dim: Optional[int] = None
	num_blocks: Optional[int] = None
	learning_rate: Optional[float] = None
	batch_size_per_gpu: Optional[int] = None
	gradient_accumulation_steps: Optional[int] = None
	log_interval: Optional[int] = None
	n_updates: Optional[int] = None # number of optimizer.step(), excluding grad_accum_step
	fps: Optional[int] = None
	preload_ratio: Optional[float] = None
	# MoE (dynamics only)
	use_moe: bool = False
	num_experts: int = 4
	top_k_experts: int = 2
	moe_aux_loss_coeff: float = 0.01
	# Optimizer
	optimizer: str = "adamw"
	muon_momentum: float = 0.95
	muon_backend_steps: int = 5
	
	def __post_init__(self) -> None:
		_validate_amp_fsdp(self.amp, self.distributed)
		_validate_distibuted_training(self.nproc_per_node, self.distributed)
		_validate_distributed_device(self.device, self.distributed)


@dataclass
class InferenceConfig:
	video_tokenizer_path: Optional[str]
	latent_actions_path: Optional[str]
	dynamics_path: Optional[str]
	device: str
	generation_steps: int
	context_window: int
	fps: int
	temperature: float
	use_actions: bool
	teacher_forced: bool
	use_latest_checkpoints: bool
	prediction_horizon: int
	dataset: str
	use_gt_actions: bool
	# Inference performance options
	amp: bool
	tf32: bool
	compile: bool
	# Interactive mode (user enters action ids)
	use_interactive_mode: bool
	preload_ratio: Optional[float] = None


def load_config(config_cls, default_config_path: Optional[str] = None):
	parser = argparse.ArgumentParser(add_help=True)
	parser.add_argument("--config", type=str, default=default_config_path)
	# Accept dotlist overrides like key=value
	parser.add_argument("overrides", nargs=argparse.REMAINDER)
	args = parser.parse_args()

	# Build a structured schema from the dataclass TYPE, not an instance.
	# This allows Python-side defaults to be omitted and provided solely via YAML.
	base = OmegaConf.structured(config_cls)
	cfg = base

	if args.config is not None:
		if not os.path.isfile(args.config):
			raise FileNotFoundError(f"Config file not found: {args.config} (cwd: {os.getcwd()})")
		file_cfg = OmegaConf.load(args.config)
		cfg = OmegaConf.merge(cfg, file_cfg)

	# Merge dotlist overrides if any (ignore leading '--')
	dot_overrides = [s.lstrip('-') for s in (args.overrides or []) if '=' in s]
	if dot_overrides:
		cli_cfg = OmegaConf.from_dotlist(dot_overrides)
		cfg = OmegaConf.merge(cfg, cli_cfg)

	# Return a typed dataclass instance
	return OmegaConf.to_object(cfg)


def load_stage_config_merged(config_cls, default_config_path: Optional[str] = None, training_config_path: Optional[str] = None):
	"""Load a stage config YAML, then overlay values from training_config.yaml (priority),
	restricted to keys that exist in the stage schema. CLI dotlist overrides still have highest priority.
	"""
	parser = argparse.ArgumentParser(add_help=True)
	parser.add_argument("--config", type=str, default=default_config_path)
	parser.add_argument("--training_config", type=str, default=training_config_path or os.path.join(os.getcwd(), 'configs', 'training.yaml'))
	parser.add_argument("overrides", nargs=argparse.REMAINDER)
	args = parser.parse_args()

	# Build a structured schema from the dataclass TYPE, not an instance.
	base = OmegaConf.structured(config_cls)
	cfg = base

	# Load stage file
	if args.config is not None:
		if not os.path.isfile(args.config):
			raise FileNotFoundError(f"Config file not found: {args.config} (cwd: {os.getcwd()})")
		stage_file_cfg = OmegaConf.load(args.config)
		cfg = OmegaConf.merge(cfg, stage_file_cfg)

	# Load training config and filter to known keys
	if args.training_config and os.path.isfile(args.training_config):
		training_file_cfg = OmegaConf.load(args.training_config)
		# Convert to plain dict to filter
		training_container = OmegaConf.to_container(training_file_cfg, resolve=True) or {}
		allowed_keys = set(cfg.keys())
		filtered_training = {k: v for k, v in training_container.items() if k in allowed_keys and v is not None}
		if filtered_training:
			training_cfg_filtered = OmegaConf.create(filtered_training)
			# Training config takes priority over stage
			cfg = OmegaConf.merge(cfg, training_cfg_filtered)

	# Merge CLI overrides if any (ignore leading '--')
	dot_overrides = [s.lstrip('-') for s in (args.overrides or []) if '=' in s]
	if dot_overrides:
		cli_cfg = OmegaConf.from_dotlist(dot_overrides)
		cfg = OmegaConf.merge(cfg, cli_cfg)

	# Return a typed dataclass instance
	return OmegaConf.to_object(cfg)