API Reference

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

Bases: omni.repo.man.config._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: argparse.Namespace)

apply_overrides(config: Dict)

consume_argv(argv: List[str]) → List[str]

Returns a new argv list with the processes cmd args:

consumes everything after – argument as “extra arguments”
consumes config override arguments (–/foo=2) removed

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

Bases: omni.repo.man.exceptions.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: omni.repo.man.exceptions.RepoToolError

exit_code = 2

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

Bases: omni.repo.man.exceptions.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: omni.repo.man.exceptions.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: omni.repo.man.exceptions.RepoToolError

A Network error has occurred.

exception omni.repo.man.PipError(message: str, emit_stack: bool = False): Bases: omni.repo.man.exceptions.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: omni.repo.man.exceptions.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: omni.repo.man.config._BaseConfig

Section for repo_man config

folders: omni.repo.man.config.RepoManFolders = Ellipsis

logging: str = Ellipsis

repo_tool_section: ClassVar[str] = 'repo'

tokens: Dict[str, str] = Ellipsis

class omni.repo.man.RepoTelemetry(tool_name: str, tool_version: Optional[str], merged_repo_config: dict, index: str = 'omni-devplat-metrics-repo-tools', index_version: str = 'v1', payload_id: str = 'default')

Bases: object

gather_data()

post_to_nvdf(timeout: float = 0.25)

Post the contents of self.payloads to the configured NVDF index.

Parameters: timeout – A float representing the duration in seconds before giving up on posting to NVDF. This is set intentionally short as the default payload is very small and posting to NVDF while on-prem/in CI takes 10s of milliseconds. Setting a long default timeout adds a pause to all repo tool shutdowns executed when not connected to the VPN, which is visually awkward. Larger payloads, e.g. repo_build’s compiler metrics, may require a longer timeout duration.

set_exit_code(exit_code: Union[str, int, float], payload_id='default')

Set the exit code.

Parameters: exit_code – The status or exit code from the repo tool being run

set_payload_value(name: str, value, payload_id: str = 'default', _type='undefined')

Parameters

name – The key name to use in NVDataFlow to identify the data.
value – Typically a string or numerical value, but can be any JSON compmatible type.
_type – NVDataFlow encourages defining data types in their indices by prepending certain characters.
payload_id – A unique identifier for a given payload so that multiple datasets can be posted in a single request.

Returns

None

timer(payload_id)

Parameters: payload_id (str) – A unique ID to be used for this timer loop. It should be descriptive such that it can easily be understood, such as repo_man_initialization, or repo_test_duration.
Yields: None – this exists strictly as a timer loop.

wait_for_gather_data(): Wait for the gather data thread to complete.

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: argparse.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: omni.repo.man.exceptions.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: omni.repo.man.exceptions.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: omni.repo.man.exceptions.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 collections.abc.MutableMapping

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

Bases: omni.repo.man.exceptions.RepoToolError

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

omni.repo.man.add_config_arg(tool_parser: argparse.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: omni.repo.man.entry._ArgumentParserWithHelpOnError, tools: List[omni.repo.man.entry.RepoTool]) → omni.repo.man.entry._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: argparse.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: Union[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.are_paths_equal(path1, path2)

omni.repo.man.assemble_config(repo_root: str, cmd_preprocessor: Optional[omni.repo.man.cmd_config_overrider.CmdPreprocessor] = 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() → argparse.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() → omni.repo.man.entry._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: Optional[str] = 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: Optional[Union[List[str], str]] = None, **properties)

omni.repo.man.ci_message_block(name=None, description='', description_args_indexs=[], ci_platforms: Optional[Union[List[str], str]] = None)

omni.repo.man.clean_folders(folders: List[str]): Clean folders. Use system specific cmd way.

omni.repo.man.clean_logfiles(file_path: pathlib.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: Optional[Dict] = {})

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: Optional[int] = 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: Optional[str] = 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: Union[bytes, str, os.PathLike], folder_to_extract_to: Union[bytes, str, os.PathLike]): Extract archive to folder based on file extension.

omni.repo.man.fetch_deps(deps_dict: Dict, in_platform: Optional[str] = None, host_deps_path: Optional[str] = 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: Optional[Union[List[str], str]] = None)

omni.repo.man.find_and_extract_package(path_wildcard: str, clean: bool = False) → Tuple[Union[bytes, str, os.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: Union[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_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: Optional[List[str]] = 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_cross_compile_host()

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) → Union[str, dict]

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

omni.repo.man.get_git_hash(cwd: Optional[str] = None, revision: str = 'HEAD', hash_length: int = 8, silence_errors: bool = True)

omni.repo.man.get_git_path_hash(path: str, gitbranch: Optional[str] = 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_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_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': ';'}

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: Optional[str] = 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_sentry_sdk(force_install=False): Pip install, import and return sentry module

omni.repo.man.get_tokens(platform: Optional[str] = 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: Optional[str] = None)

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

omni.repo.man.host_and_arch_vendor_dir_path(tool_name: str, vendor_txt: pathlib.Path) → Optional[pathlib.Path]

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: pathlib.Path, vendor_txt: pathlib.Path)

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.

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_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: Optional[str] = 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.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: Optional[str] = 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.

Builds labels like: name@latest.txt, name@MAJOR.latest.txt, name@MAJOR.MINOR.latest.txt
Installs repo_package and packages files. The result are a zip and txt files (package and label) in the output dir.
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: Optional[List[str]] = None, link_path: Optional[str] = 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: Optional[str] = None, module: Optional[str] = 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: pathlib.Path, folder: pathlib.Path, quiet: bool = False, folder_hash: 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.

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: argparse.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: Optional[Dict] = {})

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: Union[str, list, dict, tuple], platform: Optional[str] = None, extra_tokens: Optional[dict] = 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: pathlib.Path, destination: pathlib.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_sentry_sdk_module()

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

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

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: Optional[str]): 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: argparse.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: Optional[str] = None, description: str = '')

omni.repo.man.template_replace(line_input)

omni.repo.man.test()

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

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: pathlib.Path, ignore_files: Optional[list] = None, ignore_dirs: Optional[list] = 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: pathlib.Path, tool_name: str, requirements_file: Optional[pathlib.Path] = None, strict_deps: Optional[bool] = False) → pathlib.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: pathlib.Path, vendor_txt: pathlib.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_longpath(path: Union[str, bytes, os.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: _io._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[pathlib.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: Optional[str] = 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: Union[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[pathlib.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: Optional[str] = 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: _io.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: Union[str, bytes, os.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: _io.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: Union[str, bytes, os.PathLike]): Add the windows longpath-prefix (\?) if needed

omni.repo.man.publish.ci_message(message_name: str, single_arg=None, *, ci_platforms: Optional[Union[List[str], str]] = None, **properties)

omni.repo.man.publish.get_packages_and_labels(package_names: Union[Iterable[str], str], folder: str, config: Optional[str] = None)

omni.repo.man.publish.setup_repo_tool(parser: argparse.ArgumentParser, config: Dict) → Optional[Callable]: Entry point for ‘repo_publish’ tool

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

Bases: omni.repo.man.config.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: argparse.ArgumentParser, config: Dict) → Optional[Callable]: 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: Union[str, list, dict, tuple], platform: Optional[str] = None, extra_tokens: Optional[dict] = 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.