Utilities
Utilities for PyKEEN.
- class Bias(dim: int)[source]
A module wrapper for adding a bias.
Initialize the module.
- Parameters:
dim (int) – >0 The dimension of the input.
- class ExtraReprMixin[source]
A mixin for modules with hierarchical extra_repr.
It takes up the
torch.nn.Module.extra_repr()idea, and additionally provides a simple composable way to generate the components ofextra_repr()viaiter_extra_repr().If combined with torch.nn.Module, make sure to put
ExtraReprMixinbehindtorch.nn.Moduleto prefer the latter’s__repr__()implementation.- iter_extra_repr() Iterable[str][source]
Iterate over the components of the
extra_repr().This method is typically overridden. A common pattern would be
def iter_extra_repr(self) -> Iterable[str]: yield from super().iter_extra_repr() yield "<key1>=<value1>" yield "<key2>=<value2>"
- Returns:
an iterable over individual components of the
extra_repr()- Return type:
- class Result[source]
A superclass of results that can be saved to a directory.
- abstractmethod save_to_directory(directory: str, **kwargs) None[source]
Save the results to the directory.
- Parameters:
directory (str)
- Return type:
None
- abstractmethod save_to_ftp(directory: str, ftp: FTP) None[source]
Save the results to the directory in an FTP server.
- abstractmethod save_to_s3(directory: str, bucket: str, s3=None) None[source]
Save all artifacts to the given directory in an S3 Bucket.
- Parameters:
directory (str) – The directory in the S3 bucket
bucket (str) – The name of the S3 bucket
s3 – A client from
boto3.client(), if already instantiated
- Return type:
None
- all_in_bounds(x: Tensor, low: float | None = None, high: float | None = None, a_tol: float = 0.0) bool[source]
Check if tensor values respect lower and upper bound.
- batched_dot(a: Tensor, b: Tensor) Tensor[source]
Compute “element-wise” dot-product between batched vectors.
- broadcast_index_shapes(shapes: Iterable[tuple[int, ...]]) tuple[int, ...][source]
Determine the common shape of the given index shapes.
- Parameters:
shapes (Iterable[tuple[int, ...]]) – the shapes of the index tensors; they must have the same number of dimensions, cf.
pad_trailing_dims()- Returns:
the broadcasted shape
- Raises:
ValueError – if the shapes are not broadcastable
- Return type:
- broadcast_upgrade_to_sequences(*xs: X | Sequence[X]) Sequence[Sequence[X]][source]
Apply upgrade_to_sequence to each input, and afterwards repeat singletons to match the maximum length.
- Parameters:
xs (X | Sequence[X]) – length: m the inputs.
- Returns:
a sequence of length m, where each element is a sequence and all elements have the same length.
- Raises:
ValueError – if there is a non-singleton sequence input with length different from the maximum sequence length.
- Return type:
>>> broadcast_upgrade_to_sequences(1) ((1,),) >>> broadcast_upgrade_to_sequences(1, 2) ((1,), (2,)) >>> broadcast_upgrade_to_sequences(1, (2, 3)) ((1, 1), (2, 3))
- calculate_broadcasted_elementwise_result_shape(first: tuple[int, ...], second: tuple[int, ...]) tuple[int, ...][source]
Determine the return shape of a broadcasted elementwise operation.
Follows the same (right-aligned) semantics as
torch.broadcast_shapes(): the shorter shape is padded with leading singleton dimensions, and a dimension of size 1 broadcasts to the other size (including 0).
- check_shapes(*x: tuple[Tensor | tuple[int, ...], str], raise_on_errors: bool = True) bool[source]
Verify that a sequence of tensors are of matching shapes.
- Parameters:
x (tuple[Tensor | tuple[int, ...], str]) – A tuple (t, s), where t is a tensor, or an actual shape of a tensor (a tuple of integers), and s is a string, where each character corresponds to a (named) dimension. If the shapes of different tensors share a character, the corresponding dimensions are expected to be of equal size.
raise_on_errors (bool) – Whether to raise an exception in case of a mismatch.
- Returns:
Whether the shapes matched.
- Raises:
ValueError – If the shapes mismatch and raise_on_error is True.
- Return type:
Examples: >>> check_shapes(((10, 20), “bd”), ((10, 20, 20), “bdd”)) True >>> check_shapes(((10, 20), “bd”), ((10, 30, 20), “bdd”), raise_on_errors=False) False
- clamp_norm(x: Tensor, maxnorm: float, p: str | int = 'fro', dim: int | Iterable[int] | None = None) Tensor[source]
Ensure that a tensor’s norm does not exceeds some threshold.
- combine_complex(x_re: Tensor, x_im: Tensor) Tensor[source]
Combine a complex tensor from real and imaginary part.
- compact_mapping(mapping: Mapping[X, int]) tuple[Mapping[X, int], Mapping[int, int]][source]
Update a mapping (key -> id) such that the IDs range from 0 to (number of unique IDs) - 1.
The relative order of IDs is preserved, i.e., if
old_i < old_jthennew_i < new_j. The mapping does not need to be injective; keys sharing the same old ID will share the same new ID.
- complex_normalize(x: Tensor) Tensor[source]
Normalize a vector of complex numbers such that each element is of unit-length.
Let \(x \in \mathbb{C}^d\) denote a complex vector. Then, the operation computes
\[x_i' = \frac{x_i}{|x_i|}\]where \(|x_i| = \sqrt{Re(x_i)^2 + Im(x_i)^2}\) is the modulus of complex number
- class compose(*operations: Callable[[X], X], name: str)[source]
A class representing the composition of several functions.
Initialize the composition with a sequence of operations.
- Parameters:
operations (Callable[[X], X]) – unary operations that will be applied in succession
name (str) – The name of the composed function.
- create_relation_to_entity_set_mapping(triples: Iterable[tuple[int, int, int]]) tuple[Mapping[int, set[int]], Mapping[int, set[int]]][source]
Create mappings from relation IDs to the set of their head / tail entities.
- einsum(*args, **kwargs)[source]
Compute an Einstein summation, using
opt_einsumas a backend if it is installed.
- ensure_complex(*xs: Tensor) Iterable[Tensor][source]
Ensure that all tensors are of complex dtype.
Real tensors of shape
(*, 2 * d)are interpreted asdinterleaved pairs of real and imaginary parts, and converted to complex tensors of shape(*, d).- Parameters:
xs (Tensor) – the tensors
- Yields:
complex tensors.
- Raises:
ValueError – if a real tensor’s last dimension is not even
TypeError – if a real tensor’s dtype is not supported, cf.
view_complex_native()
- Return type:
- ensure_ftp_directory(*, ftp: FTP, directory: Path | str) None[source]
Ensure the directory exists on the FTP server.
- ensure_torch_random_state(random_state: None | int | Generator) Generator[source]
Prepare a random state for PyTorch.
- ensure_tuple(*x: X | Sequence[X]) Sequence[Sequence[X]][source]
Ensure that all elements in the sequence are upgraded to sequences.
- Parameters:
x (X | Sequence[X]) – A sequence of sequences or literals
- Returns:
An upgraded sequence of sequences
- Return type:
>>> ensure_tuple(1, (1,), (1, 2)) ((1,), (1,), (1, 2))
- estimate_cost_of_sequence(shape: tuple[int, ...], *other_shapes: tuple[int, ...]) int[source]
Cost of a sequence of broadcasted element-wise operations of tensors, given their shapes.
- extend_batch(batch: Tensor, max_id: int, dim: int, ids: Tensor | None = None) Tensor[source]
Extend batch for 1-to-all scoring by explicit enumeration.
- Parameters:
- Returns:
shape: (batch_size * num_choices, 3) A large batch, where every pair from the original batch is combined with every ID.
- Return type:
- fix_dataclass_init_docs(cls: type[X]) type[X][source]
Fix the
__init__documentation for adataclasses.dataclass.
- flatten_dictionary(dictionary: Mapping[str, Any], prefix: str | None = None, sep: str = '.') dict[str, Any][source]
Flatten a nested dictionary.
- get_batchnorm_modules(module: Module) list[Module][source]
Return all submodules which are batch normalization layers.
- get_connected_components(pairs: Iterable[tuple[X, X]]) Collection[Collection[X]][source]
Calculate the connected components for a graph given as edge list.
The implementation uses a union-find data structure with path compression.
- Parameters:
pairs (Iterable[tuple[X, X]]) – the edge list, i.e., pairs of node ids.
- Returns:
a collection of connected components, i.e., a collection of disjoint collections of node ids.
- Return type:
Collection[Collection[X]]
- get_devices(module: Module) Collection[device][source]
Return the device(s) from each components of the model.
- Parameters:
module (Module)
- Return type:
- get_dropout_modules(module: Module) list[Module][source]
Return all submodules which are dropout layers.
- get_edge_index(*, triples_factory: Any | None = None, mapped_triples: Tensor | None = None, edge_index: Tensor | None = None) Tensor[source]
Get the edge index from a number of different sources.
- Parameters:
- Raises:
ValueError – if none of the source was different from None
- Returns:
shape: (2, m) the edge index
- Return type:
- get_expected_norm(p: int | float | str, d: int) float[source]
Compute the expected value of the L_p norm.
\[E[\|x\|_p] = d^{1/p} E[|x_1|^p]^{1/p}\]under the assumption that \(x_i \sim N(0, 1)\), i.e.
\[E[|x_1|^p] = 2^{p/2} \cdot \Gamma(\frac{p+1}{2} \cdot \pi^{-1/2}\]
- get_num_ids(x: Tensor | ndarray) int[source]
Return the number of ids values, i.e., the maximum ID plus one, or zero for an empty tensor / array.
- get_optimal_sequence(*shapes: tuple[int, ...]) tuple[int, tuple[int, ...]][source]
Find the optimal sequence in which to combine tensors elementwise based on the shapes.
- get_preferred_device(module: Module, allow_ambiguity: bool = True) device[source]
Return the preferred device.
- get_until_first_blank(s: str) str[source]
Recapitulate all lines in the string until the first blank line.
Lines consisting only of whitespace count as blank. Leading blank lines are skipped, such that for docstrings starting with a line break, the first paragraph is returned. Each line is stripped of surrounding whitespace, and the lines are joined by single spaces.
- invert_mapping(mapping: Mapping[K, V]) Mapping[V, K][source]
Invert a mapping.
- Parameters:
mapping (Mapping[K, V]) – The mapping, key -> value.
- Returns:
The inverse mapping, value -> key.
- Raises:
ValueError – if the mapping is not bijective
- Return type:
Mapping[V, K]
- is_triple_tensor_subset(a: Tensor, b: Tensor) bool[source]
Check whether one tensor of triples is a subset of another one.
- isin_many_dim(elements: Tensor, test_elements: Tensor, dim: int = 0) Tensor[source]
Return whether elements are contained in test elements.
- Parameters:
- Returns:
shape: (elements.shape[dim],) a boolean mask indicating for each element whether it occurs in test_elements. Duplicates in either input are handled correctly.
- Return type:
- logcumsumexp(a: ndarray, axis: int | None = None) ndarray[source]
Compute
log(cumsum(exp(a), axis=axis))in a numerically stable way.The computation uses
numpy.logaddexpvianumpy.ufunc.accumulate(), i.e., each prefix is reduced stably on its own. Hence, there is no underflow for prefixes whose values are much smaller than the global maximum, and-infentries (exp(-inf) = 0) are handled correctly.The
axissemantics follownumpy.cumsum().- Parameters:
a (ndarray) – shape: s the array
axis (int | None) – the axis along which to accumulate. If
None(the default), the array is flattened first, like fornumpy.cumsum().
- Returns:
shape:
(prod(s),)ifaxisisNone, elsesthe log-cumsum-exp of the array- Return type:
See also
- lp_norm(x: Tensor, p: float, dim: int | None, normalize: bool) Tensor[source]
Return the \(L_p\) norm.
- max_value(x: Tensor | ndarray) int | None[source]
Return the maximum value, or None if the tensor / array is empty.
- merge_kwargs(kwargs: Sequence[Mapping[str, Any] | None], **extra_kwargs: Any | None) Sequence[Mapping[str, Any] | None][source]
- merge_kwargs(kwargs: Mapping[str, Any] | None, **extra_kwargs: Any | None) Mapping[str, Any] | None
Add extra fields to parameters, skipping None entries.
If a list is provided, extra parameters are added to each.
- negative_norm(x: Tensor, p: str | int | float = 2, power_norm: bool = False) Tensor[source]
Evaluate negative norm of a vector.
- Parameters:
x (Tensor) – shape: (batch_size, num_heads, num_relations, num_tails, dim) The vectors.
p (str | int | float) – The p for the norm. cf.
torch.linalg.vector_norm().power_norm (bool) – Whether to return \(|x-y|_p^p\), cf. https://github.com/pytorch/pytorch/issues/28119
- Returns:
shape: (batch_size, num_heads, num_relations, num_tails) The scores.
- Return type:
- negative_norm_of_sum(*x: Tensor, p: str | int | float = 2, power_norm: bool = False) Tensor[source]
Evaluate negative norm of a sum of vectors on already broadcasted representations.
- Parameters:
x (Tensor) – shape: (batch_size, num_heads, num_relations, num_tails, dim) The representations.
p (str | int | float) – The p for the norm. cf.
torch.linalg.vector_norm().power_norm (bool) – Whether to return \(|x-y|_p^p\), cf. https://github.com/pytorch/pytorch/issues/28119
- Returns:
shape: (batch_size, num_heads, num_relations, num_tails) The scores.
- Return type:
- nested_get(d: Mapping[str, Any], *key: str, default=None) Any[source]
Get from a nested dictionary.
- normalize_path(path: str | Path | IO[str] | IO[bytes] | None, *other: str | Path, mkdir: bool = False, is_file: bool = False, default: str | Path | IO[str] | IO[bytes] | None = None) Path[source]
Normalize a path.
- Parameters:
path (str | Path | IO[str] | IO[bytes] | None) – the path in either of the valid forms.
mkdir (bool) – whether to ensure that the path refers to an existing directory by creating it if necessary
is_file (bool) – whether the path is intended to be a file - only relevant for creating directories
default (str | Path | IO[str] | IO[bytes] | None) – the default to use if path is None
- Raises:
TypeError – if path is of unsuitable type, or a file handle without a usable file name, e.g.,
io.StringIOValueError – if path and default are both None
- Returns:
the absolute and resolved path
- Return type:
- pad_trailing_dims(x: Tensor, ndim: int) Tensor[source]
Append singleton dimensions until the tensor has the given number of dimensions.
This is useful for tensors which are aligned from the left, e.g., index tensors whose batch dimensions come first, since
torchbroadcasts from the right.- Parameters:
- Raises:
ValueError – if the tensor already has more than the desired number of dimensions
- Returns:
the tensor with trailing singleton dimensions appended
- Return type:
- parallel_prefix_unsqueeze(x: Sequence[Tensor], ndim: int) Sequence[Tensor][source]
- parallel_prefix_unsqueeze(x: Tensor, ndim: int) Tensor
Prepend the given number of singleton dimensions to all representations.
- powersum_norm(x: Tensor, p: float, dim: int | None, normalize: bool) Tensor[source]
Return the power sum norm.
- prefix_unsqueeze_target(target: Literal['head', 'relation', 'tail'], ndim: int, h: HeadRepresentation, r: RelationRepresentation, t: TailRepresentation) tuple[HeadRepresentation, RelationRepresentation, TailRepresentation][source]
Prepend batch dimensions to the target’s representations.
When the same candidates are scored for each batch element, the target’s representations are looked up once, with shape
(num, *dims). They need the batch dimensions prepended to broadcast against the other two positions, which have shape(*batch_shape, 1, *dims).- Parameters:
target (Literal['head', 'relation', 'tail']) – the target position
ndim (int) – the number of batch dimensions to prepend
h (HeadRepresentation) – the head representations
r (RelationRepresentation) – the relation representations
t (TailRepresentation) – the tail representations
- Raises:
ValueError – if the target is invalid
- Returns:
the representations, with the target’s ones unsqueezed
- Return type:
tuple[HeadRepresentation, RelationRepresentation, TailRepresentation]
- prepare_filter_triples(mapped_triples: Tensor, additional_filter_triples: Tensor | list[Tensor] | None = None, warn: bool = True) Tensor[source]
Prepare the filter triples from the evaluation triples, and additional filter triples.
- project_entity(e: Tensor, e_p: Tensor, r_p: Tensor) Tensor[source]
Project entity relation-specific.
\[e_{\bot} = M_{re} e = (r_p e_p^T + I^{d_r \times d_e}) e = r_p e_p^T e + I^{d_r \times d_e} e = r_p (e_p^T e) + e'\]and additionally enforces
\[\|e_{\bot}\|_2 \leq 1\]
- rate_limited(xs: Iterable[X], min_avg_time: float = 1.0) Iterable[X][source]
Iterate over iterable with rate limit.
- resolve_device(device: str | device | None = None) device[source]
Resolve a torch.device given a desired device (string).
- set_random_seed(seed: int) tuple[None, Generator, None][source]
Set the random seed on numpy, torch, and python.
- Parameters:
seed (int) – The seed that will be used in
np.random.seed(),torch.manual_seed(), andrandom.seed().- Returns:
A three tuple with None, the torch generator, and None.
- Return type:
- split_complex(x: Tensor) tuple[Tensor, Tensor][source]
Split a complex tensor into real and imaginary part.
- tensor_product(*tensors: Tensor) Tensor[source]
Compute element-wise product of tensors in broadcastable shape.
- tensor_sum(*tensors: Tensor) Tensor[source]
Compute element-wise sum of tensors in broadcastable shape.
- triple_tensor_to_set(tensor: Tensor) set[tuple[int, ...]][source]
Convert a tensor of triples to a set of int-tuples.
- unpack_singletons(*xs: tuple[X]) Sequence[X | tuple[X]][source]
Unpack sequences of length one.
- Parameters:
xs (tuple[X]) – A sequence of tuples of length 1 or more
- Returns:
An unpacked sequence of sequences
- Return type:
>>> unpack_singletons((1,), (1, 2), (1, 2, 3)) (1, (1, 2), (1, 2, 3))
- upgrade_to_sequence(x: X | Sequence[X]) Sequence[X][source]
Ensure that the input is a sequence.
Note
While strings are technically also a sequence, i.e.,
isinstance("test", typing.Sequence) is True
this may lead to unexpected behaviour when calling upgrade_to_sequence(“test”). We thus handle strings as non-sequences. To recover the other behavior, the following may be used:
upgrade_to_sequence(tuple("test"))
- Parameters:
x (X | Sequence[X]) – A literal or sequence of literals
- Returns:
If a literal was given, a one element tuple with it in it. Otherwise, return the given value.
- Return type:
Sequence[X]
>>> upgrade_to_sequence(1) (1,) >>> upgrade_to_sequence((1, 2, 3)) (1, 2, 3) >>> upgrade_to_sequence("test") ('test',) >>> upgrade_to_sequence(tuple("test")) ('t', 'e', 's', 't')
- view_complex(x: Tensor) Tensor[source]
Convert a PyKEEN complex tensor representation into a torch one.
PyKEEN stores complex tensors of shape
(*, d)as real tensors of shape(*, 2 * d), where the last dimension contains interleaved pairs of real and imaginary part, i.e., the layout oftorch.view_as_real()after flattening the last two dimensions.- Parameters:
x (Tensor) – shape:
(*, 2 * d)the real-valued tensor. If it is already complex, it is returned unchanged.- Returns:
shape:
(*, d)the complex tensor. It shares memory withxwhenever the strides permit it; otherwise, it is a copy.- Raises:
ValueError – if the last dimension of a real-valued input is not even.
TypeError – if a real-valued input’s dtype is not supported, cf.
view_complex_native().
- Return type:
- view_complex_native(x: Tensor) Tensor[source]
Convert a PyKEEN complex tensor representation into a torch one using
torch.view_as_complex().In contrast to
view_complex(), the input has to be real-valued.- Parameters:
x (Tensor) – shape:
(*, 2 * d)the real-valued tensor, where the last dimension containsdinterleaved pairs of real and imaginary parts.- Returns:
shape:
(*, d)the complex tensor. It shares memory withxwhenever the strides permit it; otherwise, e.g., for some non-contiguous inputs, it is a copy.- Raises:
TypeError – if the dtype of
xis not one offloat16,float32, orfloat64, which are the dtypes supported bytorch.view_as_complex(). In particular, complex inputs are rejected.ValueError – if the last dimension of
xis not even.
- Return type:
- env(file=None)[source]
Print the env or output as HTML if in Jupyter.
- Parameters:
file – The file to print to if not in a Jupyter setting. Defaults to sys.stdout
- Returns:
A
IPython.display.HTMLif in a Jupyter notebook setting, otherwise none.
Version information for PyKEEN.
- get_git_branch() str | None[source]
Get the PyKEEN branch, if installed from git in editable mode.
- Returns:
Returns the name of the current branch, or None if not installed in development mode.
- Return type:
str | None