`nemo_gym.cli`#

Module Contents#

Classes#

`RunConfig`
`TestConfig`
`ServerInstanceDisplayConfig`
`RunHelper`
`TestAllConfig`

Functions#

`_setup_env_command`
`_run_command`
`run`
`_validate_data_single`
`_test_single`
`test`
`_display_list_of_paths`
`_format_pct`
`test_all`
`dev_test`
`init_resources_server`
`dump_config`
`display_help`

API#

nemo_gym.cli._setup_env_command( dir_path: pathlib.Path, global_config_dict: omegaconf.DictConfig, ) → str#

nemo_gym.cli._run_command( command: str, working_directory: pathlib.Path, ) → subprocess.Popen#

class nemo_gym.cli.RunConfig#

Bases: nemo_gym.config_types.BaseNeMoGymCLIConfig

entrypoint: str#: ‘Field(…)’

class nemo_gym.cli.TestConfig#

Bases: nemo_gym.cli.RunConfig

should_validate_data: bool#: ‘Field(…)’

_dir_path: pathlib.Path#: None

model_post_init(context)#

property dir_path: pathlib.Path#

class nemo_gym.cli.ServerInstanceDisplayConfig#

Bases: pydantic.BaseModel

process_name: str#: None

server_type: str#: None

name: str#: None

dir_path: pathlib.Path#: None

entrypoint: str#: None

host: Optional[str]#: None

port: Optional[int]#: None

pid: Optional[int]#: None

config_path: str#: None

url: Optional[str]#: None

class nemo_gym.cli.RunHelper#

_head_server: uvicorn.Server#: None

_head_server_thread: threading.Thread#: None

_processes: Dict[str, subprocess.Popen]#: None

_server_instance_display_configs: List[nemo_gym.cli.ServerInstanceDisplayConfig]#: None

_server_client: nemo_gym.server_utils.ServerClient#: None

start( global_config_dict_parser_config: nemo_gym.global_config.GlobalConfigDictParserConfig, ) → None#

display_server_instance_info() → None#

poll() → None#

wait_for_spinup() → None#

shutdown() → None#

run_forever() → None#

check_http_server_statuses() → List[nemo_gym.server_utils.ServerStatus]#

nemo_gym.cli.run( global_config_dict_parser_config: Optional[nemo_gym.global_config.GlobalConfigDictParserConfig] = None, )#

nemo_gym.cli._validate_data_single(test_config: nemo_gym.cli.TestConfig) → None#

nemo_gym.cli._test_single( test_config: nemo_gym.cli.TestConfig, global_config_dict: omegaconf.DictConfig, ) → subprocess.Popen#

nemo_gym.cli.test()#

nemo_gym.cli._display_list_of_paths(paths: List[pathlib.Path]) → str#

nemo_gym.cli._format_pct(count: int, total: int) → str#

class nemo_gym.cli.TestAllConfig#