# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

import numpy as np
import random


class SequenceEncoder(object):
    """
    Helper class for encoding / decoding number sequences
    """
    def encode(self, C, num_rows):
        x = np.zeros((num_rows, 10))
        for i, c in enumerate(C):
            x[i, c] = 1
        return x

    def decode(self, x, calc_argmax=True):
        if calc_argmax:
            x = x.argmax(axis=-1)
        return x
    
    def encode_position(self, num_rows, index):
        x = np.zeros([num_rows])
        x[index] = 1
        return x
    
    def decode_position(self,position_sequence):
        return position_sequence.argmax()
    
    def return_subsequence(self, 
                           input_sequence, 
                           start_position_sequence, 
                           end_position_sequence):
        start = self.decode_position(start_position_sequence)
        end = self.decode_position(end_position_sequence)
        decoded_input = self.decode(input_seq)
        return decoded_input[start:end]


sequence_encoder = SequenceEncoder()

sequence = [1, 2, 3]
max_len = 20

input_seq = sequence_encoder.encode([1,2,3], max_len)

start = sequence_encoder.encode_position(max_len, 0)

end = sequence_encoder.encode_position(max_len, 3)

sequence_encoder.return_subsequence(input_seq, start, end)


# +

def generate_low_high_low_sequence(length, min_seglen=5, max_seglen=10):
    seq_before = [(random.randint(1,5)) for x in range(random.randint(min_seglen, max_seglen))]
    seq_during = [(random.randint(6,9)) for x in range(random.randint(min_seglen, max_seglen))]
    seq_after = [random.randint(1,5) for x in range(random.randint(min_seglen, max_seglen))]
    seq = seq_before + seq_during + seq_after

    # Pad it up to max len with 0's
    seq = seq + ([0] * (length - len(seq)))
    return [seq, len(seq_before), len(seq_before) + len(seq_during)-1]


# -

x ,start, end = generate_low_high_low_sequence(60)

x[start]

x[end]


def generate_batch(batch_size, sequence_gen_fun, input_len=60, min_seg=5, max_seg=10):
    sequences = []
    start_indices = []
    end_indices = []
    seq_encoder = SequenceEncoder()
    for i in range(batch_size):
        seq, start, end = sequence_gen_fun(input_len, min_seg, max_seg)
        sequences.append(sequence_encoder.encode(seq, input_len))
        start_indices.append(sequence_encoder.encode_position(input_len, start))
        end_indices.append(sequence_encoder.encode_position(input_len, end))
        
    
    return np.stack(sequences), np.stack([np.stack(start_indices), np.stack(end_indices)], axis=-1)

x_batch, y_batch = generate_batch(45000,generate_low_high_low_sequence)

x_batch.shape

y_batch.shape

from keras.models import Model, Sequential
from keras.layers import LSTM, TimeDistributed, RepeatVector, Dropout, Dense, Activation, Input, Bidirectional
from keras.optimizers import Adam, SGD




num_indices = 2
input_len = 60
enc_input = Input(shape=(input_len, 10))
enc = LSTM(20, return_sequences=True)(enc_input)
dec = LSTM(20, return_sequences=True, activation='tanh')(enc)
dec = Dense(units=2, activation='softmax')(dec)
model = Model(inputs=[enc_input], outputs=[dec])
model.summary()

model.compile(optimizer='adam', loss='mse', metrics=['accuracy'])

iterations = 3
for i in range(iterations):
    model.fit(x_batch, y_batch, epochs=1, batch_size=128)
    x_test, y_test = generate_batch(10, generate_low_high_low_sequence, 60, 3, 5)
    y_pred = model.predict(x_test)
    truth = np.argmax(y_test, axis=1)
    pred = np.argmax(y_pred, axis=1)
    for j, val in enumerate(x_test):
        print("sequence: ")
        print(sequence_encoder.decode(val))
        print("actual: ")
        print(truth[j])
        print("prediction:")
        print(pred[j])
        print("-"*60)
        
    print("*" * 60)

sequence_encoder.decode(x_batch[0])

u = model.predict(x_batch[:1]).transpose()

np.argmax(u[1])

i, j = sequence_encoder.encode_position(60, start), sequence_encoder.encode_position(60, end)

seq, start, end = generate_low_high_low_sequence(60)

test = np.stack([sequence_encoder.encode(seq, 60)])

np.argmax(model.predict(test), axis=1)

pred = model.predict(test)

pred



StarcoderdataJupyterScriptsDedupFiltered