import glob
import shutil
import numpy as np
import json
from typing import Any , Callable , Optional , List , NamedTuple
from typing import Callable , Optional , List , NamedTuple , Dict
import mlagents.trainers
import mlagents_envs
from mlagents_envs.side_channel.engine_configuration_channel import EngineConfig
class CommandLineOptions ( NamedTuple ) :
debug : bool
seed : int
env_path : str
run_id : str
load_model : bool
train_model : bool
save_freq : int
keep_checkpoints : int
base_port : int
num_envs : int
curriculum_folder : Optional [ str ]
lesson : int
no_graphics : bool
multi_gpu : bool # ?
trainer_config_path : str
sampler_file_path : Optional [ str ]
docker_target_name : Optional [ str ]
env_args : Optional [ List [ str ] ]
cpu : bool
width : int
height : int
quality_level : int
time_scale : float
target_frame_rate : int
@staticmethod
def from_argparse ( args : Any ) - > " CommandLineOptions " :
return CommandLineOptions ( * * vars ( args ) )
def get_version_string ( ) - > str :
# pylint: disable=no-member
return f """ Version information:
ml - agents : { mlagents . trainers . __version__ } ,
ml - agents - envs : { mlagents_envs . __version__ } ,
Communicator API : { UnityEnvironment . API_VERSION } ,
TensorFlow : { tf_utils . tf . __version__ } """
def parse_command_line ( argv : Optional [ List [ str ] ] = None ) - > CommandLineOptions :
parser = argparse . ArgumentParser (
def _create_parser ( ) :
argparser = argparse . ArgumentParser (
parser . add_argument ( " trainer_config_path " )
parser . add_argument (
argparser . add_argument ( " trainer_config_path " )
argparser . add_argument (
parser . add_argument (
argparser . add_argument (
dest = " curriculum_folder " ,
help = " Curriculum json directory for environment " ,
dest = " curriculum_config_path " ,
help = " Curriculum config yaml file for environment " ,
parser . add_argument (
argparser . add_argument (
parser . add_argument (
argparser . add_argument (
parser . add_argument (
argparser . add_argument (
parser . add_argument (
argparser . add_argument (
" --load " ,
default = False ,
dest = " load_model " ,
parser . add_argument (
argparser . add_argument (
parser . add_argument (
argparser . add_argument (
parser . add_argument (
argparser . add_argument (
parser . add_argument (
argparser . add_argument (
" --train " ,
default = False ,
dest = " train_model " ,
parser . add_argument (
argparser . add_argument (
parser . add_argument (
argparser . add_argument (
parser . add_argument (
argparser . add_argument (
parser . add_argument (
argparser . add_argument (
parser . add_argument (
argparser . add_argument (
parser . add_argument (
argparser . add_argument (
parser . add_argument (
argparser . add_argument (
parser . add_argument (
argparser . add_argument (
parser . add_argument ( " --version " , action = " version " , version = " " )
argparser . add_argument ( " --version " , action = " version " , version = " " )
eng_conf = parser . add_argument_group ( title = " Engine Configuration " )
eng_conf = argparser . add_argument_group ( title = " Engine Configuration " )
eng_conf . add_argument (
" --width " ,
default = 84 ,
type = int ,
help = " The target frame rate of the Unity environment(s) " ,
)
return argparser
parser = _create_parser ( )
class RunOptions ( NamedTuple ) :
trainer_config : Dict
debug : bool = parser . get_default ( " debug " )
seed : int = parser . get_default ( " seed " )
env_path : Optional [ str ] = parser . get_default ( " env_path " )
run_id : str = parser . get_default ( " run_id " )
load_model : bool = parser . get_default ( " load_model " )
train_model : bool = parser . get_default ( " train_model " )
save_freq : int = parser . get_default ( " save_freq " )
keep_checkpoints : int = parser . get_default ( " keep_checkpoints " )
base_port : int = parser . get_default ( " base_port " )
num_envs : int = parser . get_default ( " num_envs " )
curriculum_config : Optional [ Dict ] = None
lesson : int = parser . get_default ( " lesson " )
no_graphics : bool = parser . get_default ( " no_graphics " )
multi_gpu : bool = parser . get_default ( " multi_gpu " )
sampler_config : Optional [ Dict ] = None
docker_target_name : Optional [ str ] = parser . get_default ( " docker_target_name " )
env_args : Optional [ List [ str ] ] = parser . get_default ( " env_args " )
cpu : bool = parser . get_default ( " cpu " )
width : int = parser . get_default ( " width " )
height : int = parser . get_default ( " height " )
quality_level : int = parser . get_default ( " quality_level " )
time_scale : float = parser . get_default ( " time_scale " )
target_frame_rate : int = parser . get_default ( " target_frame_rate " )
@staticmethod
def from_argparse ( args : argparse . Namespace ) - > " RunOptions " :
"""
Takes an argparse . Namespace as specified in `parse_command_line` , loads input configuration files
from file paths , and converts to a CommandLineOptions instance .
: param args : collection of command - line parameters passed to mlagents - learn
: return : CommandLineOptions representing the passed in arguments , with trainer config , curriculum and sampler
configs loaded from files.
"""
argparse_args = vars ( args )
docker_target_name = argparse_args [ " docker_target_name " ]
trainer_config_path = argparse_args [ " trainer_config_path " ]
curriculum_config_path = argparse_args [ " curriculum_config_path " ]
if docker_target_name is not None :
trainer_config_path = f " /{docker_target_name}/{trainer_config_path} "
if curriculum_config_path is not None :
curriculum_config_path = (
f " /{docker_target_name}/{curriculum_config_path} "
)
argparse_args [ " trainer_config " ] = load_config ( trainer_config_path )
if curriculum_config_path is not None :
argparse_args [ " curriculum_config " ] = load_config ( curriculum_config_path )
if argparse_args [ " sampler_file_path " ] is not None :
argparse_args [ " sampler_config " ] = load_config (
argparse_args [ " sampler_file_path " ]
)
# Since argparse accepts file paths in the config options which don't exist in CommandLineOptions,
# these keys will need to be deleted to use the **/splat operator below.
argparse_args . pop ( " sampler_file_path " )
argparse_args . pop ( " curriculum_config_path " )
argparse_args . pop ( " trainer_config_path " )
return RunOptions ( * * vars ( args ) )
def get_version_string ( ) - > str :
# pylint: disable=no-member
return f """ Version information:
ml - agents : { mlagents . trainers . __version__ } ,
ml - agents - envs : { mlagents_envs . __version__ } ,
Communicator API : { UnityEnvironment . API_VERSION } ,
TensorFlow : { tf_utils . tf . __version__ } """
def parse_command_line ( argv : Optional [ List [ str ] ] = None ) - > RunOptions :
return CommandLineOptions . from_argparse ( args )
return RunOptions . from_argparse ( args )
def run_training ( run_seed : int , options : CommandLineOptions ) - > None :
def run_training ( run_seed : int , options : RunOptions ) - > None :
"""
Launches training session .
: param options : parsed command line arguments
# Docker Parameters
trainer_config_path = options . trainer_config_path
curriculum_folder = options . curriculum_folder
trainer_config_path = f " /{options.docker_target_name}/{trainer_config_path} "
if curriculum_folder is not None :
curriculum_folder = f " /{options.docker_target_name}/{curriculum_folder} "
trainer_config = load_config ( trainer_config_path )
port = options . base_port
# Configure CSV, Tensorboard Writers and StatsReporter
)
env_manager = SubprocessEnvManager ( env_factory , engine_config , options . num_envs )
maybe_meta_curriculum = try_create_meta_curriculum (
curriculum_folder , env_manager , options . lesson
options . curriculum_config , env_manager , options . lesson
options . sampler_file_path , run_seed
options . sampler_config , run_seed
trainer_config ,
options . trainer_config ,
summaries_dir ,
options . run_id ,
model_path ,
env_manager . close ( )
def create_sampler_manager ( sampler_file_path , run_seed = None ) :
sampler_config = None
def create_sampler_manager ( sampler_config , run_seed = None ) :
if sampler_file_path is not None :
sampler_config = load_config ( sampler_file_path )
if sampler_config is not None :
if " resampling-interval " in sampler_config :
# Filter arguments that do not exist in the environment
resample_interval = sampler_config . pop ( " resampling-interval " )
def try_create_meta_curriculum (
curriculum_folder : Optional [ str ] , env : SubprocessEnvManager , lesson : int
curriculum_config : Optional [ Dict ] , env : SubprocessEnvManager , lesson : int
if curriculum_folder is None :
if curriculum_config is None :
meta_curriculum = MetaCurriculum . from_directory ( curriculum_folder )
meta_curriculum = MetaCurriculum ( curriculum_config )
return meta_curriculum
def create_environment_factory (
env_path : str ,
env_path : Optional [ str ] ,
docker_target_name : Optional [ str ] ,
no_graphics : bool ,
seed : Optional [ int ] ,
return create_unity_environment
def main ( ) :
def run_cli ( options : RunOptions ) - > None :
try :
print (
"""
except Exception :
print ( " \n \n \t Unity Technologies \n " )
print ( get_version_string ( ) )
options = parse_command_line ( )
trainer_logger . info ( options )
if options . debug :
trainer_logger . setLevel ( " DEBUG " )
env_logger . setLevel ( " DEBUG " )
trainer_logger . debug ( " Configuration for this run: " )
trainer_logger . debug ( json . dumps ( options . _asdict ( ) , indent = 4 ) )
run_seed = options . seed
if options . cpu :
os . environ [ " CUDA_VISIBLE_DEVICES " ] = " -1 "
run_training ( run_seed , options )
def main ( ) :
run_cli ( parse_command_line ( ) )
# For python debugger to directly run this script