import logging import platform import etils.epath as epath import jax import openpi.training.config as _config import openpi.training.data_loader as _data_loader import openpi.training.sharding as sharding import openpi.training.utils as training_utils def init_logging(): """Custom logging format for better readability.""" level_mapping = {"DEBUG": "D", "INFO": "I", "WARNING": "W", "ERROR": "E", "CRITICAL": "C"} class CustomFormatter(logging.Formatter): def format(self, record): record.levelname = level_mapping.get(record.levelname, record.levelname) return super().format(record) formatter = CustomFormatter( fmt="%(asctime)s.%(msecs)03d [%(levelname)s] %(message)-80s (%(process)d:%(filename)s:%(lineno)s)", datefmt="%H:%M:%S", ) logger = logging.getLogger() logger.setLevel(logging.INFO) logger.handlers[0].setFormatter(formatter) def main(config: _config.TrainConfig): init_logging() logging.info(f"Running on: {platform.node()}") if config.batch_size % jax.device_count() != 0: raise ValueError( f"Batch size {config.batch_size} must be divisible by the number of devices {jax.device_count()}." ) jax.config.update("jax_compilation_cache_dir", str(epath.Path("~/.cache/jax").expanduser())) rng = jax.random.key(config.seed) train_rng, init_rng = jax.random.split(rng) mesh = sharding.make_mesh(config.fsdp_devices) data_sharding = jax.sharding.NamedSharding(mesh, jax.sharding.PartitionSpec(sharding.DATA_AXIS)) replicated_sharding = jax.sharding.NamedSharding(mesh, jax.sharding.PartitionSpec()) data_loader = _data_loader.create_behavior_data_loader( config, sharding=data_sharding, shuffle=True, skip_norm_stats=False ) data_iter = iter(data_loader) batch = next(data_iter) logging.info(f"Initialized data loader:\n{training_utils.array_tree_to_info(batch)}") breakpoint() if __name__ == "__main__": main(_config.cli())