gflownet.utils.common ===================== .. py:module:: gflownet.utils.common Functions --------- .. autoapisummary:: gflownet.utils.common.set_device gflownet.utils.common.set_float_precision gflownet.utils.common.set_int_precision gflownet.utils.common.torch2np gflownet.utils.common.download_file_if_not_exists gflownet.utils.common.resolve_path gflownet.utils.common.find_latest_checkpoint gflownet.utils.common.read_hydra_config gflownet.utils.common.gflownet_from_config gflownet.utils.common.load_gflownet_from_rundir gflownet.utils.common.batch_with_rest gflownet.utils.common.tfloat gflownet.utils.common.tlong gflownet.utils.common.tint gflownet.utils.common.tbool gflownet.utils.common.concat_items gflownet.utils.common.extend gflownet.utils.common.copy gflownet.utils.common.bootstrap_samples gflownet.utils.common.example_documented_function gflownet.utils.common.select_indices Module Contents --------------- .. py:function:: set_device(device) Get `torch` device from device. .. admonition:: Examples >>> set_device("cuda") device(type='cuda') >>> set_device("cpu") device(type='cpu') >>> set_device(torch.device("cuda")) device(type='cuda') :param device: Device. :type device: Union[str, torch.device] :returns: *torch.device* -- `torch` device. .. py:function:: set_float_precision(precision) Get `torch` float type from precision. .. admonition:: Examples >>> set_float_precision(32) torch.float32 >>> set_float_precision(torch.float32) torch.float32 :param precision: Precision. :type precision: Union[int, torch.dtype] :returns: *torch.dtype* -- `torch` float type. :raises ValueError: If precision is not one of [16, 32, 64]. .. py:function:: set_int_precision(precision) Get `torch` integer type from `int` precision. .. admonition:: Examples >>> set_int_precision(32) torch.int32 >>> set_int_precision(torch.int32) torch.int32 :param precision: Integer precision. :type precision: Union[int, torch.dtype] :returns: *torch.dtype* -- `torch` integer type. :raises ValueError: If precision is not one of [16, 32, 64]. .. py:function:: torch2np(x) Convert a torch tensor to a numpy array. :param x: Data to be converted. :type x: Union[torch.Tensor, np.ndarray, list] :returns: *np.ndarray* -- Converted data. .. py:function:: download_file_if_not_exists(path, url) Download a file from google drive if path doestn't exist. url should be in the format: https://drive.google.com/uc?id=FILE_ID .. py:function:: resolve_path(path) Resolve a path by expanding environment variables, user home directory, and making it absolute. .. admonition:: Examples >>> resolve_path("~/scratch/$SLURM_JOB_ID/data") Path("/home/user/scratch/12345/data") :param path: Path to be resolved. :type path: Union[str, Path] :returns: *Path* -- Resolved path. .. py:function:: find_latest_checkpoint(ckpt_dir) Find the latest checkpoint in the input directory. If the directory contains a checkpoint file with the name "final", that checkpoint is returned. Otherwise, the latest checkpoint is returned based on the iteration number set in the file names. :param ckpt_dir: Directory in which to search for the checkpoints. :type ckpt_dir: Union[str, Path] :returns: *Path* -- Path to the latest checkpoint. :raises ValueError: If no checkpoint files are found in the input directory. .. py:function:: read_hydra_config(rundir=None, config_name='config') .. py:function:: gflownet_from_config(config, env=None) Create GFlowNet from a Hydra OmegaConf config. :param config: Config. :type config: DictConfig :param env: Optional environment instance to be used in the initialization. :type env: GFlowNetEnv :returns: *GFN* -- GFlowNet. .. py:function:: load_gflownet_from_rundir(rundir, no_wandb=True, print_config=False, device=None, load_last_checkpoint=True, is_resumed = False) Load GFlowNet from a run path (directory with a `.hydra` directory inside). :param rundir: Path to the run directory. Must contain a `.hydra` directory. :type rundir: Union[str, Path] :param no_wandb: Whether to disable wandb in the GFN init, by default True. :type no_wandb: bool, optional :param print_config: Whether to print the loaded config, by default False. :type print_config: bool, optional :param device: Device to which the models should be moved. If None (default), take the device from the loaded config. :type device: str, optional :param load_last_checkpoint: Whether to load the final models, by default True. :type load_last_checkpoint: bool, optional :param is_resumed: Whether the GFlowNet is loaded to resume training. :type is_resumed: bool, optional :returns: *Tuple[GFN, DictConfig]* -- Loaded GFlowNet and the loaded config. :raises ValueError: If no checkpoints are found in the directory. .. py:function:: batch_with_rest(start, stop, step, tensor=False) Yields batches of indices from start to stop with step size. The last batch may be smaller than step. :param start: Start index :type start: int :param stop: End index (exclusive) :type stop: int :param step: Step size :type step: int :param tensor: Whether to return a `torch` tensor of indices instead of a `numpy` array, by default False. :type tensor: bool, optional :Yields: *Union[np.ndarray, torch.Tensor]* -- Batch of indices .. py:function:: tfloat(x, device, float_type) Convert input to a float tensor. If the input is a list of tensors, the tensors are stacked along the first dimension. The resulting tensor is moved to the specified device. :param x: :type x: Union[List[torch.Tensor], torch.Tensor, List[Union[int, float]], Union[int, :param float]]: Input to be converted to a float tensor. :param device: Device to which the tensor should be moved. :type device: torch.device :param float_type: Float type to which the tensor should be converted. :type float_type: torch.dtype :returns: *Union[torch.Tensor, List[torch.Tensor]]* -- Float tensor. .. py:function:: tlong(x, device) Convert input to a long tensor. If the input is a list of tensors, the tensors are stacked along the first dimension. The resulting tensor is moved to the specified device. :param x: :type x: Union[List[torch.Tensor], torch.Tensor, List[Union[int, float]], Union[int, :param float]]: Input to be converted to a long tensor. :param device: Device to which the tensor should be moved. :type device: torch.device :returns: *Union[torch.Tensor, List[torch.Tensor]]* -- Long tensor. .. py:function:: tint(x, device, int_type) Convert input to an integer tensor. If the input is a list of tensors, the tensors are stacked along the first dimension. The resulting tensor is moved to the specified device. :param x: :type x: Union[List[torch.Tensor], torch.Tensor, List[Union[int, float]], Union[int, :param float]]: Input to be converted to an integer tensor. :param device: Device to which the tensor should be moved. :type device: torch.device :param int_type: Integer type to which the tensor should be converted. :type int_type: torch.dtype :returns: *Union[torch.Tensor, List[torch.Tensor]]* -- Integer tensor. .. py:function:: tbool(x, device) Convert input to a boolean tensor. If the input is a list of tensors, the tensors are stacked along the first dimension. The resulting tensor is moved to the specified device. :param x: :type x: Union[List[torch.Tensor], torch.Tensor, List[Union[int, float]], Union[int, :param float]]: Input to be converted to a boolean tensor. :param device: Device to which the tensor should be moved. :type device: torch.device :returns: *Union[torch.Tensor, List[torch.Tensor]]* -- Boolean tensor. .. py:function:: concat_items(list_of_items, indices=None) Concatenates a list of items into a single tensor or array. :param list_of_items: List of items to be concatenated, i.e. list of arrays or list of tensors. :param indices: Indices to select in the resulting concatenated tensor or array, by default None. :type indices: Union[List[np.ndarray], List[torch.Tensor]], optional :returns: *Union[np.ndarray, torch.Tensor]* -- Concatenated tensor or array, with optional selection of indices. :raises NotImplementedError: If the input type is not supported, i.e., not a list of arrays or a list of tensors. .. py:function:: extend(orig, new) Extends the original list or tensor with the new list or tensor. :returns: *Union[List, TensorType["..."]]* -- Extended list or tensor. :raises NotImplementedError: If the input type is not supported, i.e., not a list or a tensor. .. py:function:: copy(x) Makes copy of the input tensor or list. A tensor is cloned and detached from the computational graph. :param x: Input tensor or list to be copied. :type x: Union[List, TensorType["..."]] :returns: *Union[List, TensorType["..."]]* -- Copy of the input tensor or list. .. py:function:: bootstrap_samples(tensor, num_samples) Bootstraps tensor along the last dimention returns tensor of the shape [initial_shape, num_samples] .. py:function:: example_documented_function(arg1, arg2) Summary line: this function is not used anywhere, it's just an example. Extended description of function from the docstrings tutorial :ref:`write docstrings-extended`. Refer to * functions with :py:func:`gflownet.utils.common.set_device` * classes with :py:class:`gflownet.gflownet.GFlowNetAgent` * methods with :py:meth:`gflownet.envs.base.GFlowNetEnv.get_action_space` * constants with :py:const:`gflownet.envs.base.CMAP` Prepenend with ``~`` to refer to the name of the object only instead of the full path -> :py:func:`~gflownet.utils.common.set_device` will display as ``set_device`` instead of the full path. Great maths: .. math:: \int_0^1 x^2 dx = \frac{1}{3} .. important:: A docstring with **math** MUST be a raw Python string (a string prepended with an ``r``: ``r"raw"``) to avoid backslashes being treated as escape characters. Alternatively, you can use double backslashes. .. warning:: Display a warning. See :ref:`learn by example`. (<-- this is a cross reference, learn about it `here `_) .. admonition:: Examples >>> function(1, 'a') True >>> function(1, 2) True >>> function(1, 1) Traceback (most recent call last): ... .. admonition:: Notes This block uses ``$ ... $`` for inline maths -> $e^{\frac{x}{2}}$. Or ``$$ ... $$`` for block math instead of the ``.. math:`` directive above. $$\int_0^1 x^2 dx = \frac{1}{3}$$ :param arg1: Description of arg1 :type arg1: int :param arg2: Description of arg2 :type arg2: str :returns: *bool* -- Description of return value .. py:function:: select_indices(iterable, indices = None) Select elements form iterable :param iterable: An iterable to select elements from. It can have multiple dimensions and selection is always preformed over the first dimension :type iterable: list, tuple, tensor or np.ndarray :param indices: 1-dimentional sequence of indecies for selecting elements, optional. If None, the iterable will be returned as is. Default None :type indices: list, tuple, tensor or np.ndarray :returns: *list, tuple, tensor or np.ndarray* -- A sequence of selected elements. The type of the returned sequence is the same as the type of the input iterable