util module

Miscellaneous utility methods. Code adopted from https://github.com/HumanCompatibleAI/imitation.git

util.docstring_parameter(*args, **kwargs)

Treats the docstring as a format string, substituting in the arguments.

util.endless_iter(iterable: Iterable[T]) → Iterator[T]

Generator that endlessly yields elements from iterable.

>>> x = range(2)
>>> it = endless_iter(x)
>>> next(it)
0
>>> next(it)
1
>>> next(it)
0

Args:

iterable: The object to endlessly iterate over.

Returns:

An iterator that repeats the elements in iterable forever.

Raises:

ValueError: iterable is empty – the first call it to returns no elements.

util.make_unique_timestamp() → str

Timestamp, with random uuid added to avoid collisions.

util.make_vec_env(env_name: str, n_envs: int = 8, seed: int = 0, parallel: bool = False, log_dir: Optional[str] = None, max_episode_steps: Optional[int] = None, post_wrappers: Optional[Sequence[Callable[[gym.core.Env, int], gym.core.Env]]] = None, env_make_kwargs: Optional[Mapping[str, Any]] = None) → stable_baselines3.common.vec_env.base_vec_env.VecEnv

Makes a vectorized environment.

Args:

env_name: The Env’s string id in Gym. n_envs: The number of duplicate environments. seed: The environment seed. parallel: If True, uses SubprocVecEnv; otherwise, DummyVecEnv. log_dir: If specified, saves Monitor output to this directory. max_episode_steps: If specified, wraps each env in a TimeLimit wrapper

with this episode length. If not specified and max_episode_steps exists for this env_name in the Gym registry, uses the registry max_episode_steps for every TimeLimit wrapper (this automatic wrapper is the default behavior when calling gym.make). Otherwise the environments are passed into the VecEnv unwrapped.

post_wrappers: If specified, iteratively wraps each environment with each

of the wrappers specified in the sequence. The argument should be a Callable accepting two arguments, the Env to be wrapped and the environment index, and returning the wrapped Env.

env_make_kwargs: The kwargs passed to spec.make.

Returns:

A VecEnv initialized with n_envs environments.

util.oric(x: numpy.ndarray) → numpy.ndarray

Optimal rounding under integer constraints.

Given a vector of real numbers such that the sum is an integer, returns a vector of rounded integers that preserves the sum and which minimizes the Lp-norm of the difference between the rounded and original vectors for all p >= 1. Algorithm from https://arxiv.org/abs/1501.00014. Runs in O(n log n) time.

Args:

x: A 1D vector of real numbers that sum to an integer.

Returns:

A 1D vector of rounded integers, preserving the sum.

util.safe_to_tensor(numpy_array: numpy.ndarray, **kwargs) → torch.Tensor

Converts a NumPy array to a PyTorch tensor.

The data is copied in the case where the array is non-writable. Unfortunately if you just use th.as_tensor for this, an ugly warning is logged and there’s undefined behavior if you try to write to the tensor.

Args:

numpy_array: The numpy array to convert to a PyTorch tensor. kwargs: Additional keyword arguments to pass to th.as_tensor.

Returns:

A PyTorch tensor with the same content as numpy_array.

util.tensor_iter_norm(tensor_iter: Iterable[torch.Tensor], ord: Union[int, float] = 2) → torch.Tensor

Compute the norm of a big vector that is produced one tensor chunk at a time.

Args:

tensor_iter: an iterable that yields tensors. ord: order of the p-norm (can be any int or float except 0 and NaN).

Returns:

Norm of the concatenated tensors.

Raises:

ValueError: ord is 0 (unsupported).