API Reference

class omni.repo.man.BaseRepoToolConfig(*args: Any, **kwargs: Any)

Bases: _BaseConfig

Base class for repo_* tool config.

command: str = Ellipsis

enabled: bool = Ellipsis

entry_point: str = Ellipsis

class omni.repo.man.CmdPreprocessor(repo_root: str = '')

Bases: object

add_extra_args(options: Namespace)

apply_overrides(config: Dict)

consume_argv(

argv: List[str],

) → List[str]

Returns a new argv list with the processes cmd args:

1. consumes everything after – argument as “extra arguments”

2. consumes config override arguments (–/foo=2) removed

exception omni.repo.man.ConfigurationError(message: str, emit_stack: bool = False)

Bases: ExpectedError

A configuration error has been encountered. Typically a key from a toml config file was referenced but not set, or a command-line argument was incorrectly used.

exception omni.repo.man.ExpectedError(message: str, emit_stack: bool = False)

Bases: RepoToolError

exit_code = 2

exception omni.repo.man.FatalNetworkError(message: str, emit_stack: bool = False)

Bases: NetworkError

A fatal network error has occurred. We’ve likely encountered a retry loop timeout.

exception omni.repo.man.FormatError(message: str, emit_stack: bool = False)

Bases: ExpectedError

A formatting error has occurred. Typically a format verification run has failed.

exception omni.repo.man.NetworkError(message: str, emit_stack: bool = False)

Bases: RepoToolError

A Network error has occurred.

class omni.repo.man.OmniConfig

Bases: object

Implements the same “File locations” OV RFC, used in the Omniverse Kit, Launcher, nucleus, omni-config-py, etc.

data_path

The root path for storing OV data.

Type:

str

cache_path

The root path for storing OV caches.

Type:

str

library_path

The root path for storing OV library packages.

Type:

str

logs_path

The root path for storing OV log files.

Type:

str

config_path

The path to the configuration directory, where omniverse.toml is located.

Type:

str

config

The configuration dictionary loaded from the omniverse.toml file.

Type:

dict

reload()

exception omni.repo.man.PipError(message: str, emit_stack: bool = False)

Bases: WrappedToolError

class omni.repo.man.PushAsEnclosingScopeMixin

Bases: object

A mixin class to push the data given to this class as a new scope.

class omni.repo.man.PushEnv

Bases: object

A commonly reused OS environment variables context manager.

exception omni.repo.man.QuietExpectedError(message: str, emit_stack: bool = False)

Bases: ExpectedError

An expected error that will not print anything to the console or file log when raised.

class omni.repo.man.RepoManConfig(*args: Any, **kwargs: Any)

Bases: _BaseConfig

Section for repo_man config

folders: RepoManFolders = Ellipsis

logging: str = Ellipsis

repo_tool_section: ClassVar[str] = 'repo'

tokens: Dict[str, str] = Ellipsis

class omni.repo.man.RepoTool(

name: str,

merged_tool_config: Dict,

tool_config_path: str,

)

Bases: object

Class representing a single tool in the repo_tools.toml file, a single command to repo, e.g. repo test.

load(subparsers)

Load the tool by importing the module and calling setup function to setup the tool parser.

run(options: Namespace)

Generic tool entrypoint. Execute the tool and cascade through some try/catch blocks.

An attempt is made to intelligently dump a stack trace to the logfile wheenver we exit uncleanly.

Parameters:

options – the argparse.Namespace dumped from our parsers

exception omni.repo.man.RepoToolError(message: str, emit_stack: bool = False)

Bases: Exception

exit_code = 1

property name

exception omni.repo.man.RetryableNetworkError(message: str, emit_stack: bool = False)

Bases: NetworkError

A retryable network error has occurred. It’s possible a longer than default timeout is needed to work around a temporary network outage, or a networked resource is present but not responding in an expected manner.

exception omni.repo.man.StorageError(message: str, emit_stack: bool = False)

Bases: RepoToolError

Encountered a storage related problem. Either a file or directory was not in the state we expected, or was missing. Commonly read/write/execute permissions are problematic.

exception omni.repo.man.TestError(message: str, emit_stack: bool = False)

Bases: ExpectedError

A test has failed in an expected manner to the test framework.

class omni.repo.man.ThreadsafeOpen(path, *args, **kwargs)

Bases: object

Class for ensuring that all file operations are atomic, treat initialization like a standard call to ‘open’ that happens to be atomic. This file opener must be used in a “with” block.

omni.repo.man.UnvalidatedData

alias of MutableMapping

exception omni.repo.man.WrappedToolError(message: str, emit_stack: bool = False)

Bases: RepoToolError

A wrapped tool, like 7za or git, has failed in an unpredictable manner.

omni.repo.man.add_config_arg(

tool_parser: ArgumentParser,

flags: List[str] = ['-c', '--config'],

allow_multiple: bool = False,

set_default_config: bool = True,

)

Add config argument to argparse parser

All default values for arguments are recommended standards across tools on how to handle build configs.

Parameters:

• tool_parser (argparse.ArgumentParser) – Parser to add argument to.

• flags (List[str], optional) – Flags to use for argument. Defaults to [“-c”, “–config”].

• allow_multiple (bool, optional) – Allow to specify multiple configs. Defaults to False.

• set_default_config (bool, optional) – Set default config from tool config. Defaults to True.

Returns:

Parser with added argument.

Return type:

argparse.ArgumentParser

omni.repo.man.add_subparsers(

parser: _ArgumentParserWithHelpOnError,

tools: List[RepoTool],

) → _ArgumentParserWithHelpOnError

Add subparsers for repoman subtools like test and format.

Parameters:

• parser – the base argparse parser configured with repoman’s arguments

• tools – A dict mapping path’s to a subtool’s toml file to dicts containing the any configured defaults in repoman’s repo.toml.

• repo_root – A string path representing the root of the repository that called repo

Returns:

An argparse parser that is configured with all of the subtools and their associated arguments.

omni.repo.man.apply_argparse_arguments(parser: ArgumentParser)

Helper function to add all our argparse arguments to a provided parser. This enables the main + bootstrap parser to share the same setup.

omni.repo.man.apply_dict_filter(

d: Dict | List | Tuple,

filterkey: str,

)

Recursively look for key ‘filterkey’, when found merge it’s value into dict where this key was found.

Updates dict in place.

omni.repo.man.apply_token_filter(

d: Dict | List | Tuple,

filters: dict,

)

Iterate through each key in dict d, and check against the keys within filters for instances where a filter matches a key within d.

filters keys like env: should match to a nested dict of keys that are possibly evaluated against e.g.: filters = {env: os.environ, token: {“token_a”: “a”, “token_b”: “b” }}

This is exceedingly similar to apply_dict_filter but with a equality check against tokens.

Updates dict in place.

Parameters:

• d (Union[Dict, List, Tuple]) – object to be evalauted and selectively filtered.

• filters (dict) – tokens or envvar dict to be checked against for selective config filtration.

omni.repo.man.are_paths_equal(path1, path2)

omni.repo.man.assemble_config(

repo_root: str,

cmd_preprocessor: CmdPreprocessor | None = None,

cli_token_overrides: dict = {},

) → Dict

Assemble the combined configuration for repoman.

This overlays the customized config file, repo.toml, over the base defaults in repo_tools.toml. This is primarily used for configuring repo_man, load_tool is used to layer these config files with specific tool config files.

Parameters:

• repo_root – A string path representing the root of the repository that called repo

• cli_token_overrides – A dict containing optional token key/value pairings. These will be set in the global token dict.

Returns:

A dict containing the customized and default configuration values that affect repoman’s behavior. This is primarily logging configuration for subtools.

omni.repo.man.build_bootstrap_parser() → ArgumentParser

This throwaway parser is used for extracting any bootstrapping flags like –set-token.

Returns:

An argparse.ArgumentParser with our –set-token parser added.

omni.repo.man.build_label_filename(name: str, version: str)

Build packman label filename from package name and version.

Builds label name by using only major part of version (before first ‘.’).

Parameters:

• name (str) – Package name.

• version (str) – Package version.

Returns:

Label filename.

omni.repo.man.build_labels(name: str, version: str) → List[str]

Build few packman label filename from package name and version.

Builds labels lile: name@latest.txt, name@MAJOR.latest.txt, name@MAJOR.MINOR.latest.txt

Parameters:

• name (str) – Package name.

• version (str) – Package version.

Returns:

List of labels.

omni.repo.man.build_parser() → _ArgumentParserWithHelpOnError

Build the top-level arg parser.

Returns:

An argparse.ArgumentParser populated with some default args for dumping the customized configuration file + base configuratiion file + adjust verbosity.

omni.repo.man.call_git(

args: list,

cwd: str | None = None,

timeout: int = 60,

max_tries: int = 5,

retry_delay: int = 5,

silence_errors: bool = False,

)

omni.repo.man.call_test()

omni.repo.man.call_with_retry(

op_name: str,

func: ~typing.Callable,

retry_count: int = 3,

retry_delay: float = 5,

exception_types: Exception = (<class 'OSError'>,

<class 'OSError'>),

) → Any

Execute func a number of times with a delay between attempts.

Shamelessly stolen from Packman since Packman does not present this in the API.

Parameters:

• op_name (str) – Human understandable name for the operation that func is trying to accomplish.

• func (Callable) – function object to be called/retried.

• retry_count (int, optional) – Number of times func should be retried. Defaults to 3.

• retry_delay (float, optional) – Duration in seconds to wait between tries. Defaults to 5.

Returns:

return whatever the heck func returns.

Return type:

Any

omni.repo.man.cast(typ, val)

Cast a value to a type.

This returns the value unchanged. To the type checker this signals that the return value has the designated type, but at runtime we intentionally don’t check anything (we want this to be as fast as possible).

omni.repo.man.change_cwd(path: str)

Change current working directory (cwd) for the execution block and then revert it back.

This function is a context manager.

Example:

with omni.repo.man.change_cwd("C:/"):
    print(os.getcwd())

Parameters:

path (str) – Path change cwd to.

omni.repo.man.change_envvar(name: str, value: str)

Change environment variable for the execution block and then revert it back.

This function is a context manager.

Example:

with omni.repo.man.change_envvar("PYTHONPATH", "C:/hello"):
    print(os.environ.get("PYTHONPATH"))

Parameters:

• name (str) – Env var to change.

• value – new value

omni.repo.man.change_syspath(paths: List[str])

Change sys.path for the execution block and then revert it back.

This function is a context manager.

Example:

with omni.repo.man.change_syspath([]"/my-sitepackages/"]):
    import fooboo

Parameters:

paths (List[str]) – Value to change sys.path to.

omni.repo.man.ci_message(

message_name: str,

single_arg=None,

*,

ci_platforms: List[str] | str | None = None,

**properties,

)

omni.repo.man.ci_message_block(

name=None,

description='',

description_args_indexs=[],

ci_platforms: List[str] | str | None = None,

)

omni.repo.man.clean_folders(folders: List[str])

Clean folders. Use system specific cmd way.

omni.repo.man.clean_logfiles(file_path: Path, last_accessed: int = 900)

Provided a logfile file_path, glob match against it to find potential stale copies of logfiles.

This cleans up after exception_utils._push_logfile_to_teamcity from prior runs.

Parameters:

file_path (Path) – The repo logfile path.

omni.repo.man.close_gitlab_block(name: str)

omni.repo.man.close_teamcity_block(name: str)

omni.repo.man.copyfile(src, dst, *args, **kwargs)

Wrapper for shutil.copyfile that handles windows longpaths

omni.repo.man.discover_set_config_tokens(

config: Dict,

optional_tokens: Dict | None = {},

)

Iterate through tokens defined in the project’s repo.toml or via the CLI –set-token argument.

Parameters:

• config (Dict) – a dictionary representation of our combined toml config files.

• optional_tokens (Optional[Dict], optional) – a dictionary of token overrides, typically acquired via CLI –set-token overrides.

omni.repo.man.enable_logging_to_file(

file_path: str,

level: int = 20,

max_logfile_bytes: int | None = None,

)

Enable “omni.repo” logging to file.

Helper function to setup logging for “omni.repo” logger into the file specified.

Usually application itself configures logging, so this function is just for quick prototyping.

Parameters:

file_path (str) – File to log in.

omni.repo.man.enable_std_logging()

Enable “omni.repo” logging to stdout.

Helper function to setup logging for “omni.repo” logger.

Log level set to INFO.

Usually application itself configures logging, so this function is just for quick prototyping.

omni.repo.man.execute_git(

args: list,

cwd: str | None = None,

timeout: int = 60,

max_tries: int = 5,

retry_delay: int = 5,

silence_errors: bool = False,

) → dict

omni.repo.man.exponential_backoff_retry(

op_name: str,

retry_count: int = 10,

base_delay: float = 1.0,

growth_rate: float = 1.0,

exception_types: Exception = (<class 'OSError'>,

<class 'OSError'>),

) → Any

Exponential backoff decorator. Useful for network and storage scenarios where increasingly longer retry durations may guarantee success.

The formula for the delay is f(x) = base_delay(1 + growth_rate)^x.

Parameters:

• op_name (str) – Human understandable name for the operation that func is trying to accomplish.

• retry_count (int, optional) – Number of times func should be retried. Defaults to 10.

• base_delay (float, optional) – Base delay value in seconds. Defaults to 1.00.

• growth_rate (float, optional) – Rate at which base delay grows exponentially. Defaults to 1.00.

• exception_types (Exception, optional) – Acceptable exception type that will be retried against. Defaults to (OSError, IOError).

Returns:

return whatever the heck func returns.

Return type:

Any

omni.repo.man.extract_archive_to_folder(

archive_path: bytes | str | PathLike,

folder_to_extract_to: bytes | str | PathLike,

)

Extract archive to folder based on file extension.

omni.repo.man.fetch_deps(

deps_dict: Dict,

in_platform: str | None = None,

host_deps_path: str | None = None,

)

Install dependencies using packman.

Example to cover all parameters:

DEPS = {
    "repo_fileutils": {                             # Package name
        "version": "0.1.0",                         # Package version
        "platforms": ["windows-x86_64"]             # (optional) List of platforms to install on. Compared against `in_platform` param.
        "link_path_host": "repo_fileutils",         # (optional) Where to folder link installed package. Relative to `host_deps_path` param.
        "source_path_host": "C:/repo_fileutils",    # (optional) DEVELOPMENT MODE: use this folder as a package source location instead of installing from python.
        "remotes": ["gitlab-repo"],                 # (optional) Remotes to use
        "add_to_sys_path": True                     # (optional) Add package folder to sys.path
    }
}

dep_folders = omni.repo.man.fetch_deps(DEPS)
print(dep_folders["repo_fileutils"])

Parameters:

• deps_dict (Dict) – Dictionary of dependencies to install. Keys are package names, values should contain dict of attributes.

• in_platform (str, optional) – Platform name.

• host_deps_path (str, optional) – Host deps folder path.

Returns:

Installed dependency folders.

Return type:

Dict

omni.repo.man.filtered_ci_platform(

current_platform: str,

ci_platforms: List[str] | str | None = None,

)

omni.repo.man.find_and_extract_package(

path_wildcard: str,

clean: bool = False,

) → Tuple[bytes | str | PathLike, str]

Find and extract a package into short hashed named folder, return path to a folder. If folder already exist do nothing, unless clean is True.

Parameters:

• path_wildcard (str) – Path with wildcards to search package in.

• clean (bool) – Clean folder if already exist.

Returns:

Path to unpacked package and path to the archive.

omni.repo.man.find_git_path()

omni.repo.man.force_vendor_deps()

omni.repo.man.gather_files_from_dict(

config_dict: Dict | List,

keyword: str = 'files',

filters: List[str] = [],

mapping: Dict[str, str] = {},

) → List

Gather and process files from a dictionary using filtering.

This function traverses dictionary in a search for a passed keyword. Found files are processed with mapping substitution and returned. Filters are used to limit the search. Files should be either array of 1 or 2 strings.

This function is used to process packaging, copying, linking configs.

Example:

config = {
    "files" : [
        [ "index.rst", "_build/test"]
    ],
    "windows" : {
        "files" : [
            [ "foo.dll" ]
        ],
        "debug" : {
            "files" : [
                [ "hello.dll" ]
            ],
        }
    },
    "debug" : {
        "files" : [
            [ "bar.lib", "_build/$param"]
        ]
    }
}

args = { "param" : "bar_location" }
files = omni.repo.man.gather_files_from_dict(config, filters=["windows", "debug"], mapping=args)
print(files) # [('index.rst', '_build/test'), 'foo.dll', ('bar.lib', '_build/bar_location'), 'hello.dll']

Parameters:

• config_dict (Dict) – Config dictionary.

• keyword (str, optional) – Keyword to search for.

• filters (List[str], optional) – List of filters. Used as keys in dictionary. Can be nested in any order.

• mapping (Dict, optional) – Mapping to process found files.

Returns:

List of found files.

omni.repo.man.gather_files_from_dict_for_platform(

config_dict: Dict,

root: str,

platform: str,

configs: List[str],

keyword: str = 'files',

extra_filters: List = [],

extra_mapping: Dict = {},

) → List

Gather and process files from a dictionary using filtering.

gather_files_from_dict() is used, but with specific filtering and mapping using platform and config passed.

This function is used to process packaging, copying, linking configs.

Example:

config = {
    "my_package": {
        "windows-x86_64" : {
            "files" : [
                [ "foo$lib_ext" ]
            ],
            "debug" : {
                "files" : [
                    [ "hello.dll" ]
                ],
            }
        },
        "release" : {
            "files" : [
                [ "bar.lib", "_build/$param"]
            ]
        }
    }
}

args = { "param" : "bar_location" }
files = omni.repo.man.gather_files_from_dict_for_platform(config, "my_package", "windows-x86_64", configs = ["debug", "release"], extra_mapping=args)
print(files) # ['foo.dll', 'hello.dll', ('bar.lib', '_build/bar_location')]

Parameters:

• config_dict (Dict) – Config dictionary.

• root (str) – Root key to use. Can be None.

• platform (str) – Platform.

• configs (List[str]) – List of configurations.

• keyword (str, optional) – Keyword to search for.

• extra_filters (List[str], optional) – List of filters. Used as keys in dictionary. Can be nested in any order.

• extra_mapping (Dict, optional) – Mapping to process found files.

Returns:

List of found files.

omni.repo.man.get_abi_platform_translation(platform: str, abi_version: str) → str

Generic entrypoint for transforming platform strings into abi declarative platform strings.

Parameters:

• platform (str) – platform string e.g.: linux_${arch}

• abi_version (str) – abi version, currently only supports glibc in the form of major.minor.

Returns:

abi declarative platform string on linux, returns platform on windows.

Return type:

str

omni.repo.man.get_all_known_configs() → List[str]

Returns all configs known by repoman.

omni.repo.man.get_all_known_platforms() → List[str]

Returns all platforms known by repoman.

omni.repo.man.get_and_validate_host_platform(

supported_platforms: List[str] | None = None,

) → str

Get host platform string (platform-arch, E.g.: “windows-x86_64”) and validate it among supported platforms.

Parameters:

supported_platforms (List[str], optional) – Supported platforms. If None validation step is skipped.

Returns:

Host platform string.

Return type:

str

omni.repo.man.get_arg(argument)

omni.repo.man.get_available_cpu_count(mem_per_core: int = 0) → int

Calculates the available CPU count based on memory per core or cgroup configuration.

This function calculates the maximum number of CPUs that can be utilized based on the specified memory per core requirement and the total memory limit. If no memory per core requirement is specified, it returns the CPU count based on the cgroup configuration.

Parameters:

mem_per_core (int, optional) – The required memory per core in gigabytes. If specified, the function calculates the maximum number of cores that can be supported by the available memory limit.

Returns:

The calculated available CPU count.

Return type:

int

omni.repo.man.get_cgroup_cpu_count() → int

Retrieves the CPU count based on the cgroup configuration.

This function attempts to read the CPU count from the cgroup v2 cpu.max file. If the file contains ‘max’, it tries to fetch the CPU count from the tool’s configuration or defaults to REPO_MAN_DEFAULT_MAX_CORES. If the cgroup v2 file is not found, it falls back to cgroup v1 cpu.cfs_quota_us and cpu.cfs_period_us files to calculate the CPU count. If neither are found, it defaults to the host’s CPU count.

Returns:

The calculated CPU count based on the cgroup configuration or the host’s CPU count.

Return type:

int

omni.repo.man.get_cgroup_memory_limit() → int

Retrieves the memory limit from the cgroup configuration.

This function first attempts to read the memory limit from the cgroup v2 memory.max file. If not found, it tries the cgroup v1 memory/memory.limit_in_bytes file. If neither are found, it falls back to using the available memory from /proc/meminfo.

Returns:

The memory limit in bytes based on the cgroup configuration or the available memory from /proc/meminfo as a fallback

Return type:

int

omni.repo.man.get_ci_platform()

omni.repo.man.get_clang_version()

omni.repo.man.get_compiler_version(compiler, regex, version_arg='-v')

omni.repo.man.get_cpu_count() → int

Return the number of logical cores on the host.

Returns:

The number of logical cores on the host.

Return type:

int

omni.repo.man.get_cross_compile_host()

omni.repo.man.get_default_omni_cache_path() → str

Returns the default OV cache path.

omni.repo.man.get_default_omni_data_path() → str

Returns the default OV data path.

omni.repo.man.get_default_omni_library_path() → str

Returns the default OV library path.

omni.repo.man.get_default_omni_logs_path() → str

Returns the default OV logs path.

omni.repo.man.get_exception_exit_code(exception: Exception) → int

Given an exception attempt to retrieve the associate exit code or default to 1.

Parameters:

exception (Exception) – The Exception that triggered shutdown.

Returns:

The exit code that repo should exit with.

Return type:

int

omni.repo.man.get_gcc_version()

omni.repo.man.get_git_branch(

return_all: bool = False,

silence_errors: bool = True,

) → str | dict

omni.repo.man.get_git_branch_helper(*args, **kwargs)

omni.repo.man.get_git_hash(

cwd: str | None = None,

revision: str = 'HEAD',

hash_length: int = 8,

silence_errors: bool = True,

)

omni.repo.man.get_git_path_hash(

path: str,

gitbranch: str | None = None,

) → Tuple[str, str]

Get the git hash from the path specified. if this is being executed within a merge request, then interfaces_hash_mr will contain the merge request number. if its not a merge request it will have the same value as interfaces_hash.

Returns a tuple (“<hash>”, “<hash>-<mr>”)

omni.repo.man.get_host_platform() → str

Get host platform string (platform-arch, E.g.: “windows-x86_64”)

omni.repo.man.get_memory_count() → int

Return the quantity of free memory on the host.

Returns:

The quantity of free memory on the host, in bytes.

Return type:

int

omni.repo.man.get_merge_request_changed_files_only(

url: str,

max_tries: int = 5,

timeout: int = 5,

) → Tuple[str, list]

Use the gitlab api to get the changed files in a merge request Only returns the changed file names.

Returns list

omni.repo.man.get_omni_cache_path()

Returns OV system path to be used for caching.

omni.repo.man.get_omni_config() → OmniConfig

Returns the singleton instance of the OmniConfig.

omni.repo.man.get_omni_config_path() → str

Returns the path to the dir where OV config (omniverse.toml) is searched for.

omni.repo.man.get_omni_data_path()

Returns OV data user system path to store persistent data.

omni.repo.man.get_omni_logs_path()

Returns OV system path to be used for logs.

omni.repo.man.get_os_details()

omni.repo.man.get_platform_file_mapping(platform: str) → Dict

Returns platform specific string substitution mapping.

>>> get_platform_file_mapping("windows-x86_64")
{'platform': 'windows-x86_64', 'lib_ext': '.dll', 'staticlib_ext': '.lib', 'lib_prefix': '', 'bindings_ext': '.pyd', 'exe_ext': '.exe', 'shell_ext': '.bat', 'path_sep': ';', 'platform_host': 'windows-x86_64'}

omni.repo.man.get_platform_os_and_arch(platform: str) → Tuple[str, str]

Returns platform os and arch separately as a tuple.

>>> get_platform_os_and_arch("linux-x86_64")
('linux', 'x86_64')

omni.repo.man.get_repo_paths(root: str | None = None) → Dict

Get dict of recommended repository path.

If root is None it is automatically derived by assuming that the caller of this function is 2 levels deeper than root.

>>> omni.repo.man.get_repo_paths(".")["build"]
'.\_build'

>>> omni.repo.man.get_repo_paths(".")["host_deps"]
'.\_build\host-deps'

Parameters:

root (str, optional) – Root folder path.

Returns:

Repository paths.

Return type:

Dict

omni.repo.man.get_token(name: str) → str | None

Get token value.

NOTE: this does not include any of the platform values that get_tokens sets. This is a cheap getter for CLI-declared tokens.

Parameters:

name (str) – token name to be retrieved.

Returns:

token value or None if token is not set.

Return type:

Optional[str]

omni.repo.man.get_tokens(platform: str | None = None) → dict

Get all known tokens

omni.repo.man.get_toml_module(force_install=False)

Pip install, import and return toml module

omni.repo.man.get_tool_config() → dict

Get current running tool merged config

omni.repo.man.get_yaml_module(force_install=False)

Pip install, import and return yaml module

omni.repo.man.getframeinfo(frame, context=1)

Get information about a frame or traceback object.

A tuple of five things is returned: the filename, the line number of the current line, the function name, a list of lines of context from the source code, and the index of the current line within that list. The optional second argument specifies the number of lines of context to return, which are centered around the current line.

omni.repo.man.gitlab_api(

url: str,

data=None,

gitlab_url: str = 'https://gitlab-master.nvidia.com/api/v4',

timeout: int = 5,

timeout_increase: int = 5,

timeout_delay: int = 5,

max_tries: int = 5,

) → Tuple[str, dict]

A method to access the gitlab-api. url is required, data(json) is optional, gitlab_url is optional, timeout is optional, max_tries is optional.

If the api you are trying to access needs authentication, you can define a token on the in the system environment - GITLAB_AUTH_TOKEN

Returns: str, dict

omni.repo.man.gitlab_closed_block_format(name: str)

omni.repo.man.gitlab_message(message_name: str, single_arg=None, **properties)

Formats and prints a teamcity-style “service” message but is really just a breadcrumb in gitlab

omni.repo.man.gitlab_message_block(

name=None,

description='',

description_args_indexs=[],

)

omni.repo.man.gitlab_message_format(

message_name: str,

single_arg=None,

**properties,

) → str

Formats a string for printing as an unsupported service message

…but does not print it

omni.repo.man.gitlab_open_block_format(name: str, message: str | None = None)

omni.repo.man.has_options_arg(options, arg)

omni.repo.man.host_and_arch_vendor_dir_path(

tool_name: str,

vendor_txt: Path,

) → Path | None

Provided a tool named tool_name, return a verbose vendor directory path.

This relies on the calling repository to have defined the environment variable OMNI_REPO_ROOT.

Parameters:

• tool_name (str) – The name of the calling tool, such as repo_man or repo_test.

• vendor_txt (Path) – The requirements.txt formatted file containing the desired Python dependencies.

Returns:

The path for the tool’s Python dependencies including the host OS and architecture, Python interpreter version, and the md5 sum of the desired requirements.txt.

Return type:

Path

omni.repo.man.import_tool(module_path: str, tool_name: str)

omni.repo.man.in_corpnet(flagged_url: str)

A function to guess at whether we’re on a given corpnet. Do not use to gate confidential info, just a convenience utility to help with things like testing on/off network, or optimizing calls, changing remotes, avoiding retry-fail-loops, and so forth. Will check for socket on port 80 and 443.

Parameters:

flagged_url – str

Returns: bool

omni.repo.man.in_nvidia(flagged_url: str = 'omnicrashes.nvidia.com')

A function to guess at whether we’re on NVIDIA network. Do not use to gate confidential info, just a convenience utility to help with things like testing on/off network, or optimizing calls, changing remotes, avoiding retry-fail-loops, and so forth. Will check for socket on port 80 and 443.

Parameters:

flagged_url – str

Returns: bool

omni.repo.man.install_dependencies(

vendor_dir: Path,

vendor_txt: Path,

enable_cache: bool = False,

user_cache_lockfile: Path | None = None,

)

Given a vendor directory and requirements.txt pip install with a protecting lockfile to avoid parallel pip installs blowing up pip. Create a breadcrumb next to vendor_dir to indicate a successful pip install.

Parameters:

• vendor_dir (Path) – the pathlib.Path object pointing at our vendor directory.

• vendor_txt (Path) – the pathlib.Path object pointing at our requirements.txt, defining what should be in vendor_directory.

• enable_cache (bool) – optionally enable the user-level pip cache if True.

• user_cache_lockfile (Path) – optional, if set this path will be used to guard the pip install call instead of the generated path based on the vendor_dir.

omni.repo.man.is_background_thread()

omni.repo.man.is_git_status_clean()

Run git status command and check that output is empty.

Returns:

True if git status is clean.

omni.repo.man.is_git_tracked(filepath: str)

param filepath: Path to the file to check return True only when git is found and file is tracked by it.

omni.repo.man.is_linux() → bool

Return True if on Linux.

Returns:

True if on Linux, False otherwise.

Return type:

bool

omni.repo.man.is_macos() → bool

Return True if on MacOS.

Returns:

True if on MacOS, False otherwise.

Return type:

bool

omni.repo.man.is_running_in_gitlab() → bool

omni.repo.man.is_running_in_teamcity() → bool

omni.repo.man.is_running_on_ci()

Check if you’re currently running on CI.

Returns True if you’re running on CI. False otherwise.

omni.repo.man.is_windows() → bool

Return True if on Windows. There are enough Windows-specific logic exceptions to gate.

Returns:

True if on Windows, False otherwise.

Return type:

bool

omni.repo.man.load_toml_config_with_tokens(

repo_root: str,

config_path: str,

platform: str | None = None,

) → Dict

Load toml config file with tokens replacement, used for all repo.toml, repo_tools.toml and other configs

omni.repo.man.lru_cache(maxsize=128, typed=False)

Least-recently-used cache decorator.

If maxsize is set to None, the LRU features are disabled and the cache can grow without bound.

If typed is True, arguments of different types will be cached separately. For example, f(3.0) and f(3) will be treated as distinct calls with distinct results.

Arguments to the cached function must be hashable.

View the cache statistics named tuple (hits, misses, maxsize, currsize) with f.cache_info(). Clear the cache and statistics with f.cache_clear(). Access the underlying function with f.__wrapped__.

See: https://en.wikipedia.org/wiki/Cache_replacement_policies#Least_recently_used_(LRU)

omni.repo.man.main(repo_root: str)

Main function for repo man.

omni.repo.man.manylinux_platform(platform: str, glibc_version: str) → str

Provided a linux platform string of the format linux-${arch}, swap out to create a manylinux_${glibc_version}_${arch} platform string.

Parameters:

• platform (str) – platform string e.g. linux_${arch}

• glibc_version (str) – major.minor formatted version number.

Returns:

manylinux platform string e.g. manylinux_2_35_x86_64

Return type:

str

omni.repo.man.mergedicts(

d1: Dict,

d2: Dict,

enable_array_merge: bool = False,

) → Dict

Merge dict d2 on top of d1 and produce new dict.

Supports “++” syntax for merging lists/tuples if enable_array_merge is set.

Merge is recursive (deep), meaning subdictionaries will also be merged.

omni.repo.man.mergedicts_gen(

d1: Dict,

d2: Dict,

enable_array_merge: bool = False,

)

(generator) Merge dict d2 on top of d1 and produce new dict.

Supports “++” syntax for merging lists/tuples if enable_array_merge is set.

Merge is recursive (deep), meaning subdictionaries will also be merged.

omni.repo.man.normpath(path: str) → str

Normalize path cross-platform. Replace backslashes with forward slashes.

Parameters:

path (str) – Path to normalize.

Returns:

Normalized path.

Return type:

str

omni.repo.man.open_gitlab_block(name: str, message: str = '')

omni.repo.man.open_teamcity_block(name: str, message: str = '')

omni.repo.man.package(

name: str,

version: str,

files: List[Tuple[str, str]],

output_folder: str,

label: str | None = None,

repo_package_version: str = '5.9.9',

)

Package using package maker.

Helper function. Installs repo_package and packages files. The result are a zip and txt files (package and label) in the output dir.

Parameters:

• name (str) – Package name.

• version (str) – Package version.

• files (List[Tuple[str, str]]) – Files to package (wilcards). List of tuples, where first element is source path and second is relative destination in the package.

• output_folder (str) – Path to folder where to produce package and label.

• repo_package_version (str, optional) – Version of repo_package used.

omni.repo.man.package_and_upload(

name: str,

version: str,

files: List[Tuple[str, str]],

output_folder: str,

remotes: List[str] = ['my'],

force: bool = False,

package_only: bool = False,

repo_package_version: str = '5.9.9',

)

Package using package maker and upload resulting package and label.

Helper function.

1. Builds labels like: name@latest.txt, name@MAJOR.latest.txt, name@MAJOR.MINOR.latest.txt

2. Installs repo_package and packages files. The result are a zip and txt files (package and label) in the output dir.

3. Uploads both package and label using provided remote.

Parameters:

• name (str) – Package name.

• version (str) – Package version.

• files (List[Tuple[str, str]]) – Files to package (wilcards). List of tuples, where first element is source path and second is relative destination in the package.

• output_folder (str) – Path to folder where to produce package and label.

• remotes (List[str], optional) – Remotes to use for uploading.

• force (bool, optional) – Force upload (packman –force flag, override existing package). Label is always force uploaded.

• package_only (bool, optional) – If True do not upload resulting package.

• repo_package_version (str, optional) – Version of repo_package used.

Returns:

Package path and labels.

Return type:

tuple[str, str]

omni.repo.man.packman_install_python_dep(

package: str,

version: str,

remotes: List[str] | None = None,

link_path: str | None = None,

)

Install python dependency using packman. It is automatically add to sys.path.

Parameters:

• package (str) – Packman package name.

• version (str) – Packman package version.

• remotes (List[str], optional) – Remotes to use.

• link_path (str, optional) – Folder to link to.

Returns:

Folder package was installed into.

Return type:

str

omni.repo.man.pip_install(

package: str,

folder: str,

version: str | None = None,

module: str | None = None,

force_install: bool = False,

extra_args: List[str] = [],

)

Pip install package into folder (if missing) and import it.

Parameters:

• package (str) – Package name.

• folder (str) – Folder path.

• version (str, optional) – Package version.

• module (str, optional) – Module to import. By default module name equals package name.

• force_install (bool, optional) – Run install command always (even if import succeeds).

• extra_args (List[str], optional) – Extra arguments to pass to pip install command. Can be used to pass additional packages to install or to specify versions.

omni.repo.man.pip_install_multiple(

packages: List[Tuple[str, str]],

folder: str,

force_install: bool = False,

extra_args: List[str] = [],

)

Pip install multiple packages into the folder (if missing) and import it.

Module name is used to import-check before installing. If not provided package name is used.

Package name can contain version.

Example:

omni.repo.man.pip_install_multiple([("flake8-docstrings==1.0.0", "flake8-docstrings"), ("Pillow", "PIL")])

Parameters:

• packages (List[Tuple[str, str]]) – List of (package, module) tuples to install

• folder (str) – Path to folder where packages will be installed

• force_install (bool, optional) – Run install command always (even if import succeeds).

• extra_args (List[str], optional) – Extra arguments to pass to pip install command. Can be used to pass additional packages to install or to specify versions.

omni.repo.man.pip_install_requirements(

requirements: Path,

folder: Path,

quiet: bool = False,

folder_hash: bool = False,

enable_cache: bool = False,

)

Pip install all Python dependencies into folder.

Initially write dependencies out to tempdir and then move that tempdir to the target folder to yield an effectively atomic operation for pip_install_requirements from the perspective of the file system.

Parameters:

• requirements (pathlib.Path) – path to pip requirements.txt

• folder (pathlib.Path) – Folder path for installing dependencies in to

• quiet (bool) – If true capture/mute stdout/stderr coming out of pip on success.

• folder_hash (bool) – optionally hash the output folder as a crude manifest.

• enable_cache (bool) – optionally enable the user-level pip cache if True. If enabled this pip operation must be guarded to prevent a multi-writer scenario otherwise the pip cache will be corrupted.

omni.repo.man.print_log(

message: str,

target_level: int = 20,

use_rich_print: bool = True,

) → None

Emit a message via logger and optionally print to console.

This helper function provides the expected CLI experience of informative messages printed to the console while also enabling full use of the Python logger for verbose logging for debug purposes.

The expectation is that whenever a print(msg) would be used to communicate to the user, we instead use print_log to simultaneously emit to console + logfile. This eliminates the fractured emission of information that involves interleaving stdout + logfile during debug.

Parameters:

• message – a string that needs to be emitted to stdout + logger

• target_level – the logging level that this message should be emitted at.

• use_rich_print – use rich print when set to true, see https://rich.readthedocs.io/

omni.repo.man.print_tokens()

Print all known tokens to stdout and log

omni.repo.man.process_args_to_cmd(args: List[str]) → str

Convert list of arguments passed to subprocess to command line string.

Properly handle quotes and spaces so that resulting string can be copied and pasted into terminal.

Parameters:

args (List[str]) – List of arguments.

Returns:

Command line build of those arguments.

Return type:

str

omni.repo.man.process_custom_tokens(

token_list: List[List[str]],

) → dict

Provided a list of strings, set them in the global namespace and return a dictionary of token keys and values.

Parameters:

token_list – A nested list of strings containing token key:value pairs such as [[‘cupcake:yes’], [‘pie:no’]]

Returns:

A dict of strings containing key:value pairs such as {‘cupcake’: ‘yes’}.

Return type:

tokens

omni.repo.man.read_meminfo() → int

Reads and processes the /proc/meminfo file to extract the available memory.

This function attempts to read the /proc/meminfo file to find the ‘MemAvailable’ entry, which it then processes to return the available memory in bytes.

Returns:

The available memory in bytes, based on the ‘MemAvailable’ entry in /proc/meminfo

Return type:

int

omni.repo.man.replace_env_tokens(s: str, source_debug: str = '') → str

Replace tokens like ${env:PATH} with value of env vars if present, otherwise empty string.

omni.repo.man.replace_file_tokens(s: str, source_debug: str = '') → str

Replace tokens like ${file:} with content of file if present, otherwise empty string.

omni.repo.man.report_error(options: Namespace, exception: Exception)

Perform the teardown and on-exit reporting steps around an encountered Exception-style exit of a tool.

Parameters:

• options (Namespace) – The argparse Namespace for this repo execution.

• exception (Exception) – The encountered Exception, likely some subclass of RepoToolError but possibly an unhandled base Exception

omni.repo.man.resolve_config_interpolations(config: Dict)

Look for ${operator:value} in all dict values and resolve if operator is known.

omni.repo.man.resolve_config_tokens(

config: Dict,

optional_tokens: Dict | None = {},

)

resolve_config_tokens recursively resolves the tokens key within the config dictionary and applies any discovered token pairings to the global token dictionary. optional_tokens are a final dictionary that can overwrite any discovered tokens within config.

Parameters:

• config (Dict) – a dictionary representation of our combined toml config files.

• optional_tokens (Optional[Dict], optional) – a dictionary of token overrides, typically acquired via CLI –set-token overrides.

omni.repo.man.resolve_tokens(

value: str | list | dict | tuple,

platform: str | None = None,

extra_tokens: dict | None = None,

source_debug: str = '',

) → Any

Resolve tokens like ${root} or ${platform} in string, dict, list, tuple recursively.

Unknown tokens are left as is.

If str is passed - new string is returned, otherwise tokens are replaced in-place.

Parameters:

• value (Union[str, list, dict, tuple]) – Value to resolve tokens in.

• platform (str, optional) – Platform to resolve tokens for. If None defaults to host platform.

• extra_tokens (dict, optional) – Extra tokens to resolve. Defaults to None.

• source_debug (str, optional) – Debug info about who called that. For logging and error messages.

Returns:

Resolved string or same object with tokens replaced in-place.

Return type:

Union[str, list, dict, tuple]

omni.repo.man.retry_move(

target: Path,

destination: Path,

folder_hash: bool,

retry_attempts: int,

retry_wait: int,

)

Attempt to move target to destination, optionally deleting destination if it already exists and does not match target. If folder_hash is set also save a md5sum of target next to the destination path.

Parameters:

• target (Path) – the recently populated Python dependency directory

• destination (Path) – the desired path where our dependencies should live

• folder_hash (bool) – md5sum target and save to disk if True

• retry_attempts (int) – number of times to retry this operation

• retry_wait (int) – duration, in seconds, between retry attempts.

omni.repo.man.retry_on_EAGAIN(callable)

Wrapper which makes the input function retry if it receives an EAGAIN error

This error signal means that some non-blocking operation occurred, and that if you try again it might succeed.

EAGAIN is a signal raised when an operation fails in a way that is likely transient. It might apply to disk or network operations or both, but just generally addresses issues where things interfere by acquiring temporary locks (AV, file indexing, etc) It probably affects Windows primarily. Originally from nvteamcity, but moved here in order to apply it to gitlab as well.

omni.repo.man.return_toml_module()

Pass back repo_man’s vendored toml library to avoid setting in globals()

To avoid breaking subtools provide an alternate method and deprecate the old method.

omni.repo.man.return_yaml_module()

Pass back repo_man’s vendored yaml library to avoid setting in globals()

To avoid breaking subtools provide an alternate method and deprecate the old method.

omni.repo.man.rmtree(path, *args, **kwargs)

Wrapper for shutil.rmtree that handles windows longpaths

omni.repo.man.run_process(

args: List,

exit_on_error=False,

timeout=None,

**kwargs,

) → int

Run system process and wait for completion.

Parameters:

• args (List) – List of arguments.

• exit_on_error (bool, optional) – Exit if return code is non zero.

omni.repo.man.run_process_return_output(

args: List,

exit_on_error=False,

print_stdout=True,

print_stderr=True,

return_stderr=True,

quiet=False,

**kwargs,

) → Tuple[int, List[str]]

Run system process and capture stdout/stderr as a list.

Parameters:

• args (List) – List of arguments.

• exit_on_error (bool, optional) – Exit if return code is non zero.

• print_stdout (bool, optional) – Print process stdout to stdout.

• print_stderr (bool, optional) – Print process stderr to stderr. This is only relevant if return_stderr is False. If return_stderr is True, stderr is redirected into stdout. Note that if return_stderr is False and you print both stdout and stderr, all of stdout will be printed before stderr.

• return_stderr (bool, optional) – Return stdout and stderr together.

• quiet (bool, optional) – Print stdout/stderr to stdout on non-zero exit.

Returns:

A tuple with return code and a list of strings containing stdout/stderr from run process.

omni.repo.man.run_script(script_to_run, global_name='__main__')

omni.repo.man.run_script_with_args(script_to_run, args, global_name='__main__')

omni.repo.man.run_script_with_custom_args(

script_to_run,

args,

global_name='__main__',

)

omni.repo.man.run_script_with_sys_args(script_to_run, global_name='__main__')

omni.repo.man.running_in_linbuild()

Check whether we’re running under linbuild (docker).

This checks for the envvar that linbuild specifically sets. It’s possible for users to spoof this, so it’s not a 100% guarantee that we’re actually running under docker.

omni.repo.man.set_cross_compile_host(host)

omni.repo.man.set_niceness()

Increase the process niceness (lower priority).

This will use system calls to reduce the priority of the current process to the minimum value. This is important to set for tools like repo_build, which will use a lot of CPU and IO resources but are non-interactive. On something like repo_build, this will cause the system to give CPU/IO time to processes like your terminal and web browser instead of the build, which will result in the host system appearing to freeze less during a build.

omni.repo.man.set_token(name: str, value: str | None)

Set token value. If value is None, token is removed.

omni.repo.man.short_hash(name: str, length: int = 5) → str

omni.repo.man.show_calling_function(func, *args, **kwargs)

omni.repo.man.spoof_ci_environment(

env: Dict[str, str],

choice: str,

) → Dict[str, str]

omni.repo.man.stack(context=1)

Return a list of records for the stack above the caller’s frame.

omni.repo.man.store_teamcity_statistics(key, value)

omni.repo.man.strtobool(value: str) → bool

omni.repo.man.swap_python_executable(

repo_config: Dict,

options: Namespace,

argv_backup: List[str],

) → None

If the targeted tool or global repo_man settings specify an alternate Python interpreter, exec with the targeted executable.

Optionally if packman_package_name, packman_package_version, and packman_link_path are defined in either config settings, pull the package via Packman.

Parameters:

• repo_config – a dict representing the consolidated [repo] config values from repo.toml

• options – an argparse.Namespace from our sprawling argparse parser.

omni.repo.man.teamcity_api(url: str, data=None) → Tuple[str, dict]

This basically accesses the teamcity rest api. url is required, data (xml only) is optional.

TC_USER and TC_PASS must be defined on the system environment.

Returns: str, xml/dict

omni.repo.man.teamcity_block(*args, **kwargs)

omni.repo.man.teamcity_escape_char(char: str) → str

omni.repo.man.teamcity_escape_value(value: str) → str

omni.repo.man.teamcity_message(message_name: str, single_arg=None, **properties)

Formats and prints a teamcity service message

omni.repo.man.teamcity_message_block(

name=None,

description='',

description_args_indexs=[],

)

omni.repo.man.teamcity_message_format(

message_name: str,

single_arg=None,

**properties,

) → str

Formats a string for printing as a teamcity service message

…but does not print it

omni.repo.man.teamcity_statistics_timer(

key: str | None = None,

description: str = '',

)

omni.repo.man.template_replace(line_input)

omni.repo.man.test()

omni.repo.man.testa(*args, **kwargs)

omni.repo.man.translated_arch() → str

Get host architecture, translate AMD64 -> x86_64.

Returns:

Host CPU architecture

Return type:

str

omni.repo.man.trigger_teamcity_build(

branch_name: str,

commit_id: str,

build_config_id: str,

) → str

Function to trigger teamcity build jobs. The following args are required - branch_name - This is the name of the branch in the repo to build. - commit_id - This is the commit hash of the change to build. - build_config_id - This is the build configuration ID of the job in teamcity. Its found in the general settings for a job under “Build configuration ID”.

Returns the response code from the server.

omni.repo.man.update_directory_md5_hash(

directory: Path,

ignore_files: list | None = None,

ignore_dirs: list | None = None,

)

Hash the directory and write out the file directory.md5 containing the md5 hash.

This will scan through directory while ignoring the files and directories defined, and return the md5sum of the directory.

Parameters:

• directory – the pathlib.Path object pointing at the directory to be hashed.

• ignore_files – a list of strings that are matched and ignored when scanning files in directory

• ignore_dirs – a list of strings that are matched and ignored when scanning sub directories in directory

omni.repo.man.validate_dependencies(

library_base: Path,

tool_name: str,

requirements_file: Path | None = None,

strict_deps: bool | None = False,

) → Path

Consolidation of python dependency bootstrapping to clean up the __init__.py.

This tries to place Python dependencies in ${root}/_repo/deps/${toolname}_deps/ but falls back to the legacy ${root}/omni/repo/${toolname}/_vendor if OMNI_REPO_ROOT envvar is not set.

Parameters:

• library_base (Path) – The path to the library’s root e.g. ${root}/omni/repo/man/

• tool_name (str) – Name of the tool, used for naming the dependency sub directory.

• requirements_file (Path, optional) – The path to a requirements.txt formatted file listing the dependencies.

• strict_deps (bool, optional) – If set to True require dependency hashes in requirements.txt and do not pull inferred Python deps.

Returns:

The vendor_directory that was used.

Return type:

Path

omni.repo.man.validate_platform(

platform: str,

supported_platforms: List[str],

platform_kind: str = 'host',

)

Validate that platform is among supported platforms. Otherwise error and exit.

Parameters:

• platform (str) – Platform to validate.

• supported_platforms (List[str]) – Supported platforms.

• platform_kind (str, optional) – Platform type (host, target, ..). For error message.

omni.repo.man.verify_vendor_integrity(

vendor_base_directory: Path,

vendor_txt: Path,

)

Check the directory to see if its contents match the generated hash.

This is a helper function for Omniverse projects that vendor Python dependencies via repo_man’s pip_install_requirements helper function. At build time we vendor dependencies, and we confirm at runtime that the _vendor directory has not changed. Most of the time a change would mean a mismatched library version or perhaps the _vendor directory of a past install, in the case of local repo_tool development.

This function checks the stashed hash with the current directory. If there is a mismatch, or the expected directory or file does not exist, we then flag for an immediate reinstall of our libraries.

If this is a fresh install we simply hash the vendor directory.

In all cases we then add the vendor directory to sys.path so the tool can access its dependencies.

Parameters:

• vendor_base_directory – the pathlib.Path object pointing at our vendor directory.

• vendor_txt – the pathlib.Path object pointing at our requirements.txt, defining what should be in vendor_directory.

omni.repo.man.windows_get_memory() → int

Using Powershell return the amount of free memory on a Windows host.

This is incredibly slow, but it does not make sense to cache via LRU. Be careful repeatedly calling this. If calling powershell fails in some manner a response of 0 bytes will be returned to mirror read_meminfo’s response.

Returns:

Amount of free memory in bytes.

Return type:

int

omni.repo.man.windows_longpath(path: str | bytes | PathLike)

Add the windows longpath-prefix (\?) if needed

class omni.repo.man.fileutils.TextIOWrapper(

buffer,

encoding=None,

errors=None,

newline=None,

line_buffering=False,

write_through=False,

)

Bases: _TextIOBase

Character and line based layer over a BufferedIOBase object, buffer.

encoding gives the name of the encoding that the stream will be decoded or encoded with. It defaults to locale.getpreferredencoding(False).

errors determines the strictness of encoding and decoding (see help(codecs.Codec) or the documentation for codecs.register) and defaults to “strict”.

newline controls how line endings are handled. It can be None, ‘’, ‘n’, ‘r’, and ‘rn’. It works as follows:

• On input, if newline is None, universal newlines mode is enabled. Lines in the input can end in ‘n’, ‘r’, or ‘rn’, and these are translated into ‘n’ before being returned to the caller. If it is ‘’, universal newline mode is enabled, but line endings are returned to the caller untranslated. If it has any of the other legal values, input lines are only terminated by the given string, and the line ending is returned to the caller untranslated.

• On output, if newline is None, any ‘n’ characters written are translated to the system default line separator, os.linesep. If newline is ‘’ or ‘n’, no translation takes place. If newline is any of the other legal values, any ‘n’ characters written are translated to the given string.

If line_buffering is True, a call to flush is implied when a call to write contains a newline character.

buffer

close()

Flush and close the IO object.

This method has no effect if the file is already closed.

closed

detach()

Separate the underlying buffer from the TextIOBase and return it.

After the underlying buffer has been detached, the TextIO is in an unusable state.

encoding

errors

fileno()

Returns underlying file descriptor if one exists.

OSError is raised if the IO object does not use a file descriptor.

flush()

Flush write buffers, if applicable.

This is not implemented for read-only and non-blocking streams.

isatty()

Return whether this is an ‘interactive’ stream.

Return False if it can’t be determined.

line_buffering

name

newlines

read(size=-1, /)

Read at most n characters from stream.

Read from underlying buffer until we have n characters or we hit EOF. If n is negative or omitted, read until EOF.

readable()

Return whether object was opened for reading.

If False, read() will raise OSError.

readline(size=-1, /)

Read until newline or EOF.

Returns an empty string if EOF is hit immediately.

reconfigure(

*,

encoding=None,

errors=None,

newline=None,

line_buffering=None,

write_through=None,

)

Reconfigure the text stream with new parameters.

This also does an implicit stream flush.

seek(cookie, whence=0, /)

Change stream position.

Change the stream position to the given byte offset. The offset is interpreted relative to the position indicated by whence. Values for whence are:

• 0 – start of stream (the default); offset should be zero or positive

• 1 – current stream position; offset may be negative

• 2 – end of stream; offset is usually negative

Return the new absolute position.

seekable()

Return whether object supports random access.

If False, seek(), tell() and truncate() will raise OSError. This method may need to do a test seek().

tell()

Return current stream position.

truncate(pos=None, /)

Truncate file to size bytes.

File pointer is left unchanged. Size defaults to the current IO position as reported by tell(). Returns the new size.

writable()

Return whether object was opened for writing.

If False, write() will raise OSError.

write(text, /)

Write string to stream. Returns the number of characters written (which is always equal to the length of the string).

write_through

class omni.repo.man.fileutils.ThreadsafeOpen(path, *args, **kwargs)

Bases: object

Class for ensuring that all file operations are atomic, treat initialization like a standard call to ‘open’ that happens to be atomic. This file opener must be used in a “with” block.

omni.repo.man.fileutils.copy2(src, dst, *args, **kwargs)

Wrapper for shutil.copy2 that handles windows longpaths

omni.repo.man.fileutils.copy_and_link_using_dict(

config_dict: Dict,

filters: List[str] = [],

mapping: Dict = {},

allowed_paths: List[str] = [],

)

Copy/link files and folders using config dictionary.

Example:

config = {
    "copy" : [
        [ "index.rst", "_build/test"]
    ],
    "windows" : {
        "copy" : [
            [ "index.rst", "_build/test/b"]
        ],
        "link" : [
            [ "", "_build/test/c"]
        ],
        "debug" : {
            "copy" : [
                [ "index.rst", "_build/test/b"]
            ]
        }
    },
    "debug" : {
        "copy" : [
            [ "index.rst", "_build/test/$param"]
        ]
    }
}

mapping = { "param" : "foo" }
omni.repo.fileutils.copy_and_link_using_dict(config, ["windows", "debug"], mapping)

Parameters:

• config_dict (Dict) – Config dictionary.

• filters (List[str], optional) – List of filters. Used as keys in dictionary. Can be nested in order.

• mapping (Dict, optional) – Substitution mapping.

• allowed_paths (List[str], optional) – List of paths allowed to copy and link into. If empty it is ignored.

omni.repo.man.fileutils.copy_and_link_using_dict_for_platform(

config_dict: Dict,

platform: str,

configs: List[str],

extra_filters: List = [],

extra_mapping: Dict = {},

)

Copy/link files and folders using config dictionary.

Example:

config = {
    "copy" : [
        [ "index.rst", "_build/test"]
    ],
    "windows" : {
        "copy" : [
            [ "index.rst", "_build/test/b"]
        ],
        "link" : [
            [ "", "_build/test/c"]
        ],
        "debug" : {
            "copy" : [
                [ "index.rst", "_build/test/b"]
            ]
        }
    },
    "debug" : {
        "copy" : [
            [ "index.rst", "_build/test/$param"]
        ]
    }
}

mapping = { "param" : "foo" }
omni.repo.fileutils.copy_and_link_using_dict_for_platform(config, "windows", ["release", "debug"], extra_mapping=mapping)

Parameters:

• config_dict (Dict) – Config dictionary.

• platform (str) – Platform.

• configs (List[str]) – List of configurations.

• extra_filters (List[str], optional) – List of filters. Used as keys in dictionary. Can be nested in any order.

• extra_mapping (Dict, optional) – Mapping to process found files.

omni.repo.man.fileutils.copy_files(

src_path,

dst_path,

args={},

allowed_paths: List[Path] = [],

exclude_path_patterns: List[str] = [],

symbol_check: bool = False,

strip_to_first_wildcard: bool = True,

)

Copy files using wildcards.

Parameters:

• src_path (str) – Src path mask.

• dst_path (str) – Destination path mask.

• args (Dict, optional) – Substitution args.

Examples

Copy all python files from one folder to another:

omni.repo.fileutils.copy_files("folder1/*.py", "folder2")

Copy whole folder:

omni.repo.fileutils.copy_files("folder1/**", "folder2")

Substitution:

omni.repo.fileutils.copy_files("folder1/{config}/**", "folder2", args={"config": "debug"})

omni.repo.man.fileutils.copy_files_using_json_config(

json_file: str,

key: str | None = None,

args: Dict = {},

)

Copy files using json config.

Use json keys as source, values as destinations.

Parameters:

• json_file (str) – Json file path.

• key (str, optional) – Json root key to use. Defaults to None.

• args (Dict, optional) – Substitution args.

omni.repo.man.fileutils.copyfile(src, dst, *args, **kwargs)

Wrapper for shutil.copyfile that handles windows longpaths

omni.repo.man.fileutils.gather_files_from_dict(

config_dict: Dict | List,

keyword: str = 'files',

filters: List[str] = [],

mapping: Dict[str, str] = {},

) → List

Gather and process files from a dictionary using filtering.

This function traverses dictionary in a search for a passed keyword. Found files are processed with mapping substitution and returned. Filters are used to limit the search. Files should be either array of 1 or 2 strings.

This function is used to process packaging, copying, linking configs.

Example:

config = {
    "files" : [
        [ "index.rst", "_build/test"]
    ],
    "windows" : {
        "files" : [
            [ "foo.dll" ]
        ],
        "debug" : {
            "files" : [
                [ "hello.dll" ]
            ],
        }
    },
    "debug" : {
        "files" : [
            [ "bar.lib", "_build/$param"]
        ]
    }
}

args = { "param" : "bar_location" }
files = omni.repo.man.gather_files_from_dict(config, filters=["windows", "debug"], mapping=args)
print(files) # [('index.rst', '_build/test'), 'foo.dll', ('bar.lib', '_build/bar_location'), 'hello.dll']

Parameters:

• config_dict (Dict) – Config dictionary.

• keyword (str, optional) – Keyword to search for.

• filters (List[str], optional) – List of filters. Used as keys in dictionary. Can be nested in any order.

• mapping (Dict, optional) – Mapping to process found files.

Returns:

List of found files.

omni.repo.man.fileutils.gather_files_from_dict_for_platform(

config_dict: Dict,

root: str,

platform: str,

configs: List[str],

keyword: str = 'files',

extra_filters: List = [],

extra_mapping: Dict = {},

) → List

Gather and process files from a dictionary using filtering.

gather_files_from_dict() is used, but with specific filtering and mapping using platform and config passed.

This function is used to process packaging, copying, linking configs.

Example:

config = {
    "my_package": {
        "windows-x86_64" : {
            "files" : [
                [ "foo$lib_ext" ]
            ],
            "debug" : {
                "files" : [
                    [ "hello.dll" ]
                ],
            }
        },
        "release" : {
            "files" : [
                [ "bar.lib", "_build/$param"]
            ]
        }
    }
}

args = { "param" : "bar_location" }
files = omni.repo.man.gather_files_from_dict_for_platform(config, "my_package", "windows-x86_64", configs = ["debug", "release"], extra_mapping=args)
print(files) # ['foo.dll', 'hello.dll', ('bar.lib', '_build/bar_location')]

Parameters:

• config_dict (Dict) – Config dictionary.

• root (str) – Root key to use. Can be None.

• platform (str) – Platform.

• configs (List[str]) – List of configurations.

• keyword (str, optional) – Keyword to search for.

• extra_filters (List[str], optional) – List of filters. Used as keys in dictionary. Can be nested in any order.

• extra_mapping (Dict, optional) – Mapping to process found files.

Returns:

List of found files.

omni.repo.man.fileutils.link_folders(

src_path,

dst_path,

args={},

allowed_paths: List[Path] = [],

)

Create folder link pointing to src_path named dst_path.

Parameters:

• src_path (str) – Source folder link points to.

• dst_path (str) – New folder path to become a link. Should not exist.

• args (Dict, optional) – Substitution args.

omni.repo.man.fileutils.link_folders_using_json_config(

json_file: str,

key: str | None = None,

args: Dict = {},

)

Create folder link using json config.

Use json keys what link should point to and json values as folder link names to be created.

Parameters:

• json_file (str) – Json file path.

• key (str, optional) – Json root key to use. Defaults to None.

• args (Dict, optional) – Substitution args.

omni.repo.man.fileutils.lock_file(file_to_lock: TextIOWrapper)

Lock a file for exclusive access

omni.repo.man.fileutils.os_supports_long_paths()

omni.repo.man.fileutils.path_needs_windows_longpath_prefix(path: str | bytes | PathLike)

The longpath-prefix is needed if all 3 of these are true: - We’re on windows - Longpath support is not enabled in the current process - the path is >= 260 characters

omni.repo.man.fileutils.rmtree(path, *args, **kwargs)

Wrapper for shutil.rmtree that handles windows longpaths

omni.repo.man.fileutils.unlock_file(file_to_unlock: TextIOWrapper)

Unlock exclusive access to a file

omni.repo.man.fileutils.warn(message, category=None, stacklevel=1, source=None)

Issue a warning, or maybe ignore it or raise an exception.

omni.repo.man.fileutils.windows_longpath(path: str | bytes | PathLike)

Add the windows longpath-prefix (\?) if needed

omni.repo.man.publish.ci_message(

message_name: str,

single_arg=None,

*,

ci_platforms: List[str] | str | None = None,

**properties,

)

omni.repo.man.publish.get_packages_and_labels(

package_names: Iterable[str] | str,

folder: str,

config: str | None = None,

)

omni.repo.man.publish.setup_repo_tool(

parser: ArgumentParser,

config: Dict,

) → Callable | None

Entry point for ‘repo_publish’ tool

class omni.repo.man.upload.RepoUploadConfig(*args: Any, **kwargs: Any)

Bases: BaseRepoToolConfig

Configuration class for repo_upload.

exec: str = Ellipsis

files: List[Tuple[str, str]] = Ellipsis

output_folder: str = Ellipsis

package_name: str = Ellipsis

package_version: str = Ellipsis

repo_tool_section: ClassVar[str] = 'repo_upload'

omni.repo.man.upload.setup_repo_tool(

parser: ArgumentParser,

config: Dict,

) → Callable | None

Entry point for ‘repo_upload’ tool

omni.repo.man.nspect.change_envvar(name: str, value: str)

Change environment variable for the execution block and then revert it back.

This function is a context manager.

Example:

with omni.repo.man.change_envvar("PYTHONPATH", "C:/hello"):
    print(os.environ.get("PYTHONPATH"))

Parameters:

• name (str) – Env var to change.

• value – new value

omni.repo.man.nspect.get_repo_nspect_id(error_if_not_set: bool = True) → str

Get the nspect id for the repo.

Read from the standard location from repo.toml for now. In the future might change to be some other source. So use that API not to make any assumptions.

omni.repo.man.nspect.get_tool_config() → dict

Get current running tool merged config

omni.repo.man.nspect.install_nspect()

Install nspect from the CT omniverse pypi server. Called only once, second call is a no-op.

omni.repo.man.nspect.lru_cache(maxsize=128, typed=False)

Least-recently-used cache decorator.

If maxsize is set to None, the LRU features are disabled and the cache can grow without bound.

If typed is True, arguments of different types will be cached separately. For example, f(3.0) and f(3) will be treated as distinct calls with distinct results.

Arguments to the cached function must be hashable.

View the cache statistics named tuple (hits, misses, maxsize, currsize) with f.cache_info(). Clear the cache and statistics with f.cache_clear(). Access the underlying function with f.__wrapped__.

See: https://en.wikipedia.org/wiki/Cache_replacement_policies#Least_recently_used_(LRU)

omni.repo.man.nspect.pip_install_multiple(

packages: List[Tuple[str, str]],

folder: str,

force_install: bool = False,

extra_args: List[str] = [],

)

Pip install multiple packages into the folder (if missing) and import it.

Module name is used to import-check before installing. If not provided package name is used.

Package name can contain version.

Example:

omni.repo.man.pip_install_multiple([("flake8-docstrings==1.0.0", "flake8-docstrings"), ("Pillow", "PIL")])

Parameters:

• packages (List[Tuple[str, str]]) – List of (package, module) tuples to install

• folder (str) – Path to folder where packages will be installed

• force_install (bool, optional) – Run install command always (even if import succeeds).

• extra_args (List[str], optional) – Extra arguments to pass to pip install command. Can be used to pass additional packages to install or to specify versions.

omni.repo.man.nspect.resolve_tokens(

value: str | list | dict | tuple,

platform: str | None = None,

extra_tokens: dict | None = None,

source_debug: str = '',

) → Any

Resolve tokens like ${root} or ${platform} in string, dict, list, tuple recursively.

Unknown tokens are left as is.

If str is passed - new string is returned, otherwise tokens are replaced in-place.

Parameters:

• value (Union[str, list, dict, tuple]) – Value to resolve tokens in.

• platform (str, optional) – Platform to resolve tokens for. If None defaults to host platform.

• extra_tokens (dict, optional) – Extra tokens to resolve. Defaults to None.

• source_debug (str, optional) – Debug info about who called that. For logging and error messages.

Returns:

Resolved string or same object with tokens replaced in-place.

Return type:

Union[str, list, dict, tuple]

omni.repo.man.nspect.run_nspect(args: List[str], exit_on_error=False) → int

Run omni-nspect command line tool with the given args.

Parameters:

• args (List[str]) – Arguments to pass to omni-nspect.

• exit_on_error (bool, optional) – If True, exit on error. Defaults to False.

Returns:

exit code

Return type:

int

omni.repo.man.nspect.run_process(

args: List,

exit_on_error=False,

timeout=None,

**kwargs,

) → int

Run system process and wait for completion.

Parameters:

• args (List) – List of arguments.

• exit_on_error (bool, optional) – Exit if return code is non zero.

omni.repo.man.nspect.urlparse(url, scheme='', allow_fragments=True)

Parse a URL into 6 components: <scheme>://<netloc>/<path>;<params>?<query>#<fragment>

The result is a named 6-tuple with fields corresponding to the above. It is either a ParseResult or ParseResultBytes object, depending on the type of the url parameter.

The username, password, hostname, and port sub-components of netloc can also be accessed as attributes of the returned object.

The scheme argument provides the default value of the scheme component when no scheme is found in url.

If allow_fragments is False, no attempt is made to separate the fragment component from the previous component, which can be either path or query.

Note that % escapes are not expanded.