ENCODERS & DECODERS

flyhash

Encodes dense vector data as Sparse Distributed Representations (SDRs) using the classic flyhash algorithm.

Details

Encoder plugin for the input component.
Encodes dense, real-valued vectors to SDRs.
Maps similar vectors to similar SDRs.

Flyhash encoding

The following circuit encodes dense vectors as SDRs:

{ "hyperparameters": {"default": [1000, 10]},
  "dataflow": [
    {"component": "input", "plugin": "flyhash", "send": [1]},
    {"component": "output", "receive": [1]}
  ]}

Source code

def flyhash(config):
    hyperparameters = config.get("hyperparameters", [0, 0])
    n, p = hyperparameters[0], hyperparameters[1]
    key = (n, p)

    if key not in circuit_flyhash_matrices:
        circuit_flyhash_matrices[key] = {}
        circuit_flyhash_resting[key] = {}

    R = circuit_flyhash_matrices[key]
    restingpotential = circuit_flyhash_resting[key]

    sparsity = config.get("sparsity", 0.1)

    if config.get("component") == "output":
        raise ValueError("flyhash cannot be used as a decoder.")

    def f(lst):
        if not isinstance(lst, list):
            lst = list(lst)
        d = len(lst)
        if p <= 0 or d == 0:
            return []

        arr = np.array(lst)
        centeredlist = arr - np.mean(arr)
        tolerance = 1e-10

        if d not in R:
            c = max(1, round(d * sparsity))

            rows = np.repeat(np.arange(n), c)
            cols = np.concatenate(
                [rng.choice(d, size=c, replace=False) for _ in range(n)])
            vals = np.ones(n * c)

            R[d] = sp.csr_matrix((vals, (rows, cols)), shape=(n, d))
            restingpotential[d] = rng.uniform(0, tolerance * 0.01, size=n)

        vec = R[d].dot(centeredlist) + restingpotential[d]

        min_val, max_val = np.min(vec), np.max(vec)
        span = max_val - min_val
        scale = max(abs(min_val), abs(max_val))

        if scale == 0 or span <= tolerance * scale:
            return []

        k = min(p, len(vec))
        top_k_indices = np.argsort(vec)[-k:]
        return sorted((top_k_indices + 1).tolist())

    def clear():
        R.clear()
        restingpotential.clear()

    return {
        "label": "FLY",
        "function": f,
        "clear": clear
    }