Add basic data structures (#4)

* WIP: Add data structures

* histogram, graph and ngram

* Ngram, graph and sequence functions

* Tree

* Pull request small fixes

* Multiset

* Finite state automata

* Fixing classmethod

* Update sturctures docs

* Fix pathspec error

* Format Python code

* Fix functional import

* Format Python code

Co-authored-by: verges <pverges8@gmail.com>
Co-authored-by: formatting <mikeheddes@users.noreply.github.com>

Author: mikeheddes

.github/workflows/format.yml

@@ -33,6 +33,7 @@ jobs:
             git config --global user.name 'formatting'
             git config --global user.email 'mikeheddes@users.noreply.github.com'
             git remote set-url origin https://x-access-token:${{ secrets.GITHUB_TOKEN }}@github.com/$GITHUB_REPOSITORY
+            git fetch
             git checkout $GITHUB_HEAD_REF
             git commit -am "Format Python code"
             git push

dev-requirements.txt

@@ -6,4 +6,5 @@ numpy
 sphinx
 sphinx-rtd-theme
 flake8
-pytest
+pytest
+black

docs/_templates/class.rst

@@ -6,4 +6,5 @@
 {{ name | underline}}
 
 .. autoclass:: {{ name }}
-    :members:
+    :members:
+    :special-members:

docs/index.rst

@@ -9,6 +9,7 @@ Welcome to the Torchhd documentation!
 
    functional
    embeddings
+   structures
    datasets
 
 

docs/structures.rst

@@ -0,0 +1,15 @@
+Structures
+==================
+
+.. currentmodule:: torchhd.structures
+
+.. autosummary::
+    :toctree: generated/
+    :template: class.rst
+
+    Memory
+    Multiset
+    Sequence
+    Graph
+    Tree
+    FiniteStateAutomata

torchhd/__init__.py

@@ -1,9 +1,11 @@
 import torchhd.functional as functional
 import torchhd.embeddings as embeddings
+import torchhd.structures as structures
 import torchhd.datasets as datasets
 
 __all__ = [
     "functional",
     "embeddings",
+    "structures",
     "datasets",
 ]

torchhd/functional.py

@@ -12,7 +12,7 @@
     "circular_hv",
     "bind",
     "bundle",
-    "batch_bundle",
+    "multiset",
     "permute",
     "hard_quantize",
     "soft_quantize",
@@ -333,7 +333,7 @@ def bundle(input: torch.Tensor, other: torch.Tensor, *, out=None) -> torch.Tenso
     return torch.add(input, other, out=out)
 
 
-def batch_bundle(
+def multiset(
     input: torch.Tensor,
     *,
     dim=-2,
@@ -449,6 +449,21 @@ def hamming_similarity(input: torch.Tensor, others: torch.Tensor) -> torch.Tenso
     return torch.sum(input == others, dim=-1, dtype=input.dtype)
 
 
+def ngrams(input: torch.Tensor, n=3):
+    for i in range(0, n):
+        if i == (n - 1):
+            last_sample = None
+        else:
+            last_sample = -(n - i - 1)
+        sample = permute(input[:, i:last_sample], shifts=n - i - 1)
+        if n is None:
+            n_gram = sample
+        else:
+            n_gram = bind(n, sample)
+
+    return multiset(n_gram)
+
+
 def map_range(
     input: torch.Tensor,
     in_min: float,

torchhd/structures.py

@@ -0,0 +1,197 @@
+from typing import Any, List, Optional, Tuple
+import torch
+
+import torchhd.functional as functional
+
+
+class Memory:
+    """Associative memory"""
+
+    def __init__(self, threshold=0.5):
+        self.threshold = threshold
+        self.keys: List[torch.Tensor] = []
+        self.values: List[Any] = []
+
+    def __len__(self) -> int:
+        """Returns the number of items in memory"""
+        return len(self.values)
+
+    def add(self, key: torch.Tensor, value: Any) -> None:
+        """Adds one (key, value) pair to memory"""
+        self.keys.append(key)
+        self.values.append(value)
+
+    def _get_index(self, key: torch.Tensor) -> int:
+        key_stack = torch.stack(self.keys, dim=0)
+        sim = functional.cosine_similarity(key, key_stack)
+        value, index = torch.max(sim, 0)
+
+        if value.item() < self.threshold:
+            raise IndexError()
+
+        return index
+
+    def __getitem__(self, key: torch.Tensor) -> Tuple[torch.Tensor, Any]:
+        """Get the (key, value) pair with an approximate key"""
+        index = self._get_index(key)
+        return self.keys[index], self.values[index]
+
+    def __setitem__(self, key: torch.Tensor, value: Any) -> None:
+        """Set the value of an (key, value) pair with an approximate key"""
+        index = self._get_index(key)
+        self.values[index] = value
+
+    def __delitem__(self, key: torch.Tensor) -> None:
+        """Delete the (key, value) pair with an approximate key"""
+        index = self._get_index(key)
+        del self.keys[index]
+        del self.values[index]
+
+
+class Multiset:
+    def __init__(self, dimensions, threshold=0.5, device=None, dtype=None):
+        self.threshold = threshold
+        self.cardinality = 0
+        dtype = dtype if dtype is not None else torch.get_default_dtype()
+        self.value = torch.zeros(dimensions, dtype=dtype, device=device)
+
+    def add(self, input: torch.Tensor) -> None:
+        self.value = functional.bundle(self.value, input)
+        self.cardinality += 1
+
+    def remove(self, input: torch.Tensor) -> None:
+        if input not in self:
+            return
+        self.value = functional.bundle(self.value, -input)
+        self.cardinality -= 1
+
+    def __contains__(self, input: torch.Tensor):
+        sim = functional.cosine_similarity(input, self.values.unsqueeze(0))
+        return sim.item() > self.threshold
+
+    def __len__(self) -> int:
+        return self.cardinality
+
+    @classmethod
+    def from_ngrams(cls, input: torch.Tensor, n=3, threshold=0.5):
+        instance = cls(input.size(-1), threshold, input.device, input.dtype)
+        instance.value = functional.ngrams(input, n)
+        return instance
+
+    @classmethod
+    def from_tensors(cls, input: torch.Tensor, dim=-2, threshold=0.5):
+        instance = cls(input.size(-1), threshold, input.device, input.dtype)
+        instance.value = functional.multiset(input=input, dim=dim)
+        return instance
+
+
+class Sequence:
+    def __init__(self, dimensions, threshold=0.5, device=None, dtype=None):
+        self.length = 0
+        self.threshold = threshold
+        dtype = dtype if dtype is not None else torch.get_default_dtype()
+        self.value = torch.zeros(dimensions, dtype=dtype, device=device)
+
+    def append(self, input: torch.Tensor) -> None:
+        rotated_value = functional.permute(self.value, shifts=1)
+        self.value = functional.bundle(input, rotated_value)
+
+    def appendleft(self, input: torch.Tensor) -> None:
+        rotated_input = functional.permute(input, shifts=len(self))
+        self.value = functional.bundle(self.value, rotated_input)
+
+    def pop(self, input: torch.Tensor) -> Optional[torch.Tensor]:
+        self.value = functional.bundle(self.value, -input)
+        self.value = functional.permute(self.value, shifts=-1)
+        self.length -= 1
+
+    def popleft(self, input: torch.Tensor) -> None:
+        rotated_input = functional.permute(input, shifts=len(self) + 1)
+        self.value = functional.bundle(self.value, -rotated_input)
+        self.length -= 1
+
+    def __getitem__(self, index: int) -> torch.Tensor:
+        rotated_value = functional.permute(self.value, shifts=-index)
+        return rotated_value
+
+    def __len__(self) -> int:
+        return self.length
+
+
+class Graph:
+    def __init__(
+        self, dimensions, threshold=0.5, directed=False, device=None, dtype=None
+    ):
+        self.length = 0
+        self.threshold = threshold
+        self.dtype = dtype if dtype is not None else torch.get_default_dtype()
+        self.value = torch.zeros(dimensions, dtype=dtype, device=device)
+        self.directed = directed
+
+    def add_edge(self, node1: torch.Tensor, node2: torch.Tensor):
+        if self.directed:
+            edge = functional.bind(node1, node2)
+        else:
+            edge = functional.bind(node1, functional.permute(node2))
+        self.value = functional.bundle(self.value, edge)
+
+    def edge_exists(self, node1: torch.Tensor, node2: torch.Tensor):
+        if self.directed:
+            edge = functional.bind(node1, node2)
+        else:
+            edge = functional.bind(node1, functional.permute(node2))
+        return edge in self
+
+    def node_neighbours(self, input: torch.Tensor):
+        return functional.bind(self.value, input)
+
+    def __contains__(self, input: torch.Tensor):
+        sim = functional.cosine_similarity(input, self.value.unsqueeze(0))
+        return sim.item() > self.threshold
+
+
+class Tree:
+    def __init__(self, dimensions, device=None, dtype=None):
+        self.dtype = dtype if dtype is not None else torch.get_default_dtype()
+        self.value = torch.zeros(dimensions, dtype=dtype, device=device)
+        self.l_r = functional.random_hv(2, dimensions)
+
+    def add_leaf(self, value, path):
+        for i in path:
+            if i == "l":
+                value = functional.bind(value, self.left)
+            else:
+                value = functional.bind(value, self.right)
+        self.value = functional.bundle(self.value, value)
+
+    @property
+    def left(self):
+        return self.l_r[0]
+
+    @property
+    def right(self):
+        return self.l_r[1]
+
+
+class FiniteStateAutomata:
+    def __init__(self, dimensions, device=None, dtype=None):
+        self.dtype = dtype if dtype is not None else torch.get_default_dtype()
+        self.value = torch.zeros(dimensions, dtype=dtype, device=device)
+
+    def add_transition(
+        self,
+        token: torch.Tensor,
+        initial_state: torch.Tensor,
+        final_state: torch.Tensor,
+    ):
+        transition_edge = functional.bind(
+            initial_state, functional.permute(final_state)
+        )
+        transition = functional.bind(token, transition_edge)
+        self.value = functional.bundle(self.value, transition)
+
+    def change_state(self, token: torch.Tensor, current_state: torch.Tensor):
+        # Returns the next state + some noise
+        next_state = functional.bind(self.value, current_state)
+        next_state = functional.bind(next_state, token)
+        return functional.permute(next_state, shifts=-1)