Fix TM static inputs issue.

[ctrl-z-9000-times]

Explanation and discussion thread:

https://discourse.numenta.org/t/the-static-input-problem/10447

TODO:

• Make a unit test for this

[dkeeney]

Looks good

[dkeeney]

Thank you Dave. I also like your write up.

[ctrl-z-9000-times]

Thanks!

I know this is a real corner case that should very rarely occur in the wild,
but the bug happens for square waves which are a pretty common artificial test case so I thought I'd fix it.

And also the analysis of the issue turned out to be more interesting than I thought it would be!

[ctrl-z-9000-times]

For posterity, here is the code I used to generate the figures on the discourse thread:

import numpy as np
import matplotlib.pyplot as plt
from htm import SDR
from htm.algorithms import TemporalMemory
from htm.encoders.scalar_encoder import ScalarEncoder, ScalarEncoderParameters

pct_overlap = lambda a, b: a.getOverlap(b) / max(a.getSum(), b.getSum())

# NOTE: This bug only happens when the sequence jumps to a static pattern. When
# it moves slowly: the cells at the front of the pattern that burst, they
# associate with all of the older activating cells, and those older cells will
# remain active for a few more time-steps so the new bursting cells will keep
# their winners and also form synapses to the new bursting cells. When it moves
# slowly it can correctly form a clique of cells to represent the static
# pattern.
#         So a sine wave will never trigger this and a square wave will always.

if True:
    print("Static Pattern Test ...")
    minicolumns = SDR(1000).randomize(.05)
    tm = TemporalMemory(minicolumns.dimensions)
    # Fill the TM with some random sequence data.
    for _ in range(1000): tm.compute(minicolumns.randomize(.05))
    # Run a static pattern through the TM.
    anom = []
    nact = []
    for t in range(2500):
        tm.compute(minicolumns)
        active = tm.getActiveCells()
        nact.append(active.getSum() / minicolumns.getSum() / 32)
        anom.append(tm.anomaly)

    print("Percent Active", nact[-1])
    # 
    plt.figure("Initial Static Pattern Test")
    plt.title("Static Pattern Test\nInitial Behavior")
    plt.xlabel('Time Step')
    plt.ylabel('Raw Anomaly Score')
    init = anom[:40]
    plt.plot(np.arange(len(init)), init, 'k', label="Raw Anomaly Score")

    plt.figure("Final Static Pattern Test")
    plt.title("Static Pattern Test\nLong Run Behavior")
    plt.xlabel('Time Step')
    plt.ylabel('Raw Anomaly Score')
    plt.plot(np.arange(len(anom)), anom, 'k', label="Raw Anomaly Score")

if True:
    # Demonstrate that **normally** the TM does not care about the speed of the sequence.
    print("Speed Invariance Test ...")
    # Create the HTM system.
    enc_params = ScalarEncoderParameters()
    enc_params.radius = 2
    enc_params.activeBits = 40
    enc_params.minimum = -1
    enc_params.maximum = +21
    enc = ScalarEncoder(enc_params)
    tm = TemporalMemory([enc.size])
    print("Num Minicolumns", enc.size)
    # Fill the TM with some random sequence data.
    for _ in range(1000):
        tm.compute(SDR(enc.size).randomize(enc_params.sparsity))

    tau = 2 * 3.14159
    # First train on the sequence
    reps = 10
    for x in np.linspace(0, reps * tau, reps * 100):
        inp = -np.cos(x) * 10 + 10
        tm.compute(enc.encode(inp))

    wave1 = []
    for x in np.linspace(0, tau, 100):
        inp = -np.cos(x) * 10 + 10
        tm.compute(enc.encode(inp))
        wave1.append(inp)
    wave1_id = tm.getActiveCells()
    assert tm.anomaly == 0, "wave1_id anomaly, pattern not learned"

    wave2 = []
    for x in np.linspace(0, tau, 200):
        inp = -np.cos(x) * 10 + 10
        tm.compute(enc.encode(inp))
        wave2.append(inp)
    wave2_id = tm.getActiveCells()
    assert tm.anomaly == 0, "wave2_id anomaly, pattern not learned"

    wave3 = []
    wave3_x = list(np.linspace(0, tau, 100))
    for pause in range(20):
        wave3_x.insert(70, wave3_x[70])
    for pause in range(20):
        wave3_x.insert(13, wave3_x[13])
    for x in wave3_x:
        inp = -np.cos(x) * 10 + 10
        tm.compute(enc.encode(inp))
        wave3.append(inp)
    wave3_id = tm.getActiveCells()
    assert tm.anomaly == 0, "wave3_id anomaly, pattern not learned"

    assert pct_overlap(wave1_id, wave2_id) == 1
    assert pct_overlap(wave1_id, wave3_id) == 1

    for x in np.linspace(0, tau / 2, 100):
        inp = -np.cos(x) * 10 + 10
        tm.compute(enc.encode(inp))
    halfway = tm.getActiveCells()
    assert pct_overlap(wave1_id, halfway) < .1

    plt.figure("Speed Invariance")
    plt.title("Identical Sequences to the Temporal Memory")
    plt.plot(np.arange(len(wave1)), wave1)
    plt.plot(np.arange(len(wave2)), wave2)
    plt.plot(np.arange(len(wave3)), wave3)
    plt.xlabel("Time Steps")
    plt.ylabel("Input Value")

plt.show()