API Reference

OpenPlaceRecognition package.

opr.const

Definition of package-level constants.

opr.testing

Testing functions implementation.

opr.testing.get_recalls(query_embs: ndarray, db_embs: ndarray, dist_matrix: ndarray, dist_thresh: float = 25.0, at_n: int = 25) → Tuple[ndarray, float, float | None][source]

Calculate Recall@N, Recall@1% and mean top-1 distance for the given query and db embeddings.

Parameters:

  • query_embs (np.ndarray) – Query embeddings array.

  • db_embs (np.ndarray) – Database embeddings array.

  • dist_matrix (np.ndarray) – Distance matrix of shape (query_len, db_len).

  • dist_thresh (float) – Distance threshold for positive match. Defaults to 25.0.

  • at_n (int) – The maximum N value for the Recall@N metric. Defaults to 25.

Returns:

(Recall@N, Recall@1%, mean top-1 distance).

The ‘mean top-1 distance’ metric may be None if Recall@1 = 0.

Return type:

Tuple[np.ndarray, float, Optional[float]]

opr.testing.test(model: Module, dataloader: DataLoader, distance_threshold: float = 25.0, device: str | int | device = 'cuda') → Tuple[ndarray, float, float][source]

Evaluates the model on the test set.

Parameters:

  • model (nn.Module) – The model to test.

  • dataloader (DataLoader) – The data loader for the test set.

  • distance_threshold (float) – The distance threshold for a correct match. Defaults to 25.0.

  • device (Union[str, int, torch.device]) – Device (“cpu” or “cuda”). Defaults to “cuda”.

Returns:

Array of AverageRecall@N (N from 1 to 25), AverageRecall@1%

and mean top-1 distance.

Return type:

Tuple[np.ndarray, float, float]

Raises:

ValueError – If the required coordinate columns are not found in the dataset.

opr.utils

Package-level utility functions.

opr.utils.accumulate_dict(dst_dict: Dict[str, Any], src_dict: Dict[str, Any]) → Dict[str, Any][source]

Updates dst_dict with values from src_dict.

Recursively traverses the src_dict dictionary and appends values to lists in dst_dict. If a key does not exist in dst_dict, it is created with a new list containing the corresponding value.

Parameters:

  • dst_dict (Dict[str, Any]) – A dictionary representing statistics for an epoch.

  • src_dict (Dict[str, Any]) – A dictionary representing statistics for a step within an epoch.

Returns:

A dictionary representing updated statistics for an epoch.

Return type:

Dict[str, Any]

Example usage:
>>> dst_dict = {}
>>> src_dict_1 = {"train": {"total": 0.1, "image": {"loss": 0.1}}}
>>> src_dict_2 = {"train": {"total": 0.2, "image": {"loss": 0.2}}}
>>> dst_dict = accumulate_dict(dst_dict, src_dict_1)
>>> print(dst_dict)
{'train': {'total': [0.1], 'image': {'loss': [0.1]}}}
>>> dst_dict = accumulate_dict(dst_dict, src_dict_2)
>>> print(dst_dict)
{'train': {'total': [0.1, 0.2], 'image': {'loss': [0.1, 0.2]}}}
opr.utils.cartesian_to_spherical(points: ndarray, dataset_name: str) → ndarray[source]

Converts cartesian coordinates to spherical coordinates.

opr.utils.compute_epoch_stats_mean(epoch_stats: Dict[str, Any]) → Dict[str, Any][source]

Computes the mean value of each list in epoch_stats.

Recursively traverses the epoch_stats dictionary and computes the mean value of each list using np.mean(). If a key does not contain a list, its value is returned as is.

Parameters:

epoch_stats (Dict[str, Any]) – A dictionary representing statistics for an epoch.

Returns:

A dictionary representing the mean value of each list in epoch_stats.

Return type:

Dict[str, Any]

Example usage:
>>> epoch_stats = {'train': {'total': [0.1, 0.2], 'image': {'loss': [0.1, 0.2]}}}
>>> epoch_stats_mean = compute_epoch_stats_mean(epoch_stats)
>>> print(epoch_stats_mean)
{'train': {'total': 0.15000000000000002, 'image': {'loss': 0.15000000000000002}}}
opr.utils.distribute_batch_size(global_batch_size: int, num_replicas: int) → List[int][source]

Distributes the global batch size over the replicas.

Parameters:

  • global_batch_size (int) – The global batch size.

  • num_replicas (int) – The number of replicas.

Returns:

A list of batch sizes for each replica.

Return type:

List[int]

Examples

>>> print(distribute_batch_size(4096, 6))
[43, 43, 43, 43, 42, 42]
opr.utils.flatten_dict(nested_dict: Dict[str, Any], parent_key: str = '', sep: str = '/') → Dict[str, Any][source]

Flatten a nested dictionary with keys separated by sep.

Parameters:

  • nested_dict (Dict[str, Any]) – A nested dictionary to flatten.

  • parent_key (str) – The string of parent key (used for recursion).

  • sep (str) – The separator to use between keys in the flattened dictionary.

Returns:

A flattened dictionary with keys separated by sep.

Return type:

Dict[str, Any]

opr.utils.get_local_batch_size(global_batch_size: int, num_replicas: int, rank: int) → int[source]

Gets the local batch size on the given rank in the global batch.

Parameters:

  • global_batch_size (int) – The global batch size.

  • num_replicas (int) – The number of replicas.

  • rank (int) – The rank of the replica.

Returns:

The local batch size on the given rank in the global batch.

Return type:

int

opr.utils.get_start_end_indices_of_local_batch(global_batch_size: int, num_replicas: int, rank: int) → Tuple[int, int][source]

Gets the start and end indices of the local batch on the given rank in the global batch.

Parameters:

  • global_batch_size (int) – The global batch size.

  • num_replicas (int) – The number of replicas.

  • rank (int) – The rank of the replica.

Returns:

A list of tuples containing the start and end indices

of the local batch in the global batch.

Return type:

List[Tuple[int, int]]

Examples

>>> print(get_start_end_indices_in_global_batch(256, 6, 1))
(43, 86)
opr.utils.in_sorted_array(e: int, array: Tensor) → bool[source]

Checks whether the given value e is in sorted array.

Code adopted from repository: https://github.com/jac99/MinkLocMultimodal, MIT License

Parameters:

  • e (int) – Value to search for.

  • array (Tensor) – Sorted array to look from.

Returns:

Whether the given value e is in sorted array.

Return type:

bool

opr.utils.init_model(model: Module, weights_path: str | PathLike | None, device: str | int | device) → Module[source]

Transfers the model to the device, loads the weights and sets the model to eval mode.

Parameters:

  • model (nn.Module) – Model.

  • weights_path (Union[str, PathLike]) – Path to the model weights. If None, the weights are not loaded.

  • device (Union[str, int, torch.device]) – Device to use.

Returns:

Model in eval mode.

Return type:

nn.Module

opr.utils.merge_nested_dicts(dict1: Dict[str, Any], dict2: Dict[str, Any]) → Dict[str, Any][source]

Recursively merge two nested dictionaries that have overlapping outer keys.

Parameters:

  • dict1 (Dict[str, Any]) – First dictionary object.

  • dict2 (Dict[str, Any]) – Second dictionary object.

Returns:

Merged dictionary.

Return type:

Dict[str, Any]

opr.utils.parse_device(device: str | int | device) → device[source]

Parse given device argument and return torch.device object.

Parameters:

device (Union[str, int, torch.device]) – Device argument.

Returns:

Device object.

Return type:

torch.device

Raises:

ValueError – If device is not a string, integer or torch.device object.

opr.utils.set_seed(seed: int = 0, make_deterministic: bool = False) → None[source]

Set the random seed for the random, numpy, and torch libraries and enables deterministic mode.

Parameters:

  • seed (int) – The random seed to use. Defaults to 0.

  • make_deterministic (bool) – Whether to make PyTorch deterministic. If True, disables PyTorch’s benchmark mode and enables its deterministic mode. If False, leaves PyTorch’s settings unchanged. Defaults to True.