2
\$\begingroup\$

Given a class BasicDataFeed whose purpose is to feed questions and answers into an artificial neural network, which is tested in a non-negotiable manner as follows:

import numpy as np
import matplotlib.pyplot as plt


dt = 0.001
dims = 4
t_len = 0.1
pause = 0.01
n_items = 4

cor = np.eye(n_items)


def dataset_func(idx, t):
    return cor[idx] * (t*10)


df = BasicDataFeed(dataset_func, np.eye(n_items), t_len, dims, n_items, pause)

t_steps = list(np.arange(0, 2*n_items*(t_len+pause), dt))
df_out = []
ans_out = []

for tt in t_steps:
    df_out.append(df.feed(tt))
    ans_out.append(df.get_answer(tt))

plt.plot(df_out)

plt.gca().set_prop_cycle(None)

plt.plot(ans_out)
plt.show()

With the desired output:

basic_desired_output

How do I write my state-machine in a more elegant manner? I feel like I'm keeping track of time (and thus the state transitions) incorrectly.

import numpy as np

from random import shuffle

dt = 0.001

class BasicDataFeed(object):

    def __init__(self, dataset, correct, t_len: float, dims: int, n_items: int, pause: float):
        self.data_index = 0
        self.paused = False

        self.time = 0.0
        self.sig_time = 0

        self.pause_time = pause
        self.q_duration = t_len
        self.ans_duration = self.q_duration + self.pause_time

        self.correct = correct
        self.qs = dataset
        self.num_items = n_items
        self.dims = dims
        self.indices = list(range(self.num_items))

    def get_answer(self, t):
        """Signal for correct answer"""
        if self.pause_time < self.time < self.ans_duration:
            return self.correct[self.indices[self.data_index]]
        else:
            return np.zeros(self.num_items)

    def feed(self, t):
        """Feed the question into the network
        this is the main state machine of the network"""
        self.time += dt

        if self.time > self.pause_time and self.sig_time > self.q_duration:

            if self.data_index < self.num_items - 1:
                self.data_index += 1
            else:
                shuffle(self.indices)
                self.data_index = 0

            self.time = 0.0
            self.sig_time = 0.0

        elif self.time > self.pause_time:
            self.paused = False

            q_idx = self.indices[self.data_index]
            return_val = self.qs(q_idx, self.sig_time)
            self.sig_time += dt
            return return_val

        else:
            self.paused = True

        return np.zeros(self.dims)

Note: This is a continuation of my previous question about time-dependent state machine. Also, don't worry about the type annotations, they're just there to help me debug and act as documentation for the code.

\$\endgroup\$
  • \$\begingroup\$ I would split the behavior of the state machine with the input of events. In other words, use a context that does all the time related stuff, and feed events at the right moment to a queue. Let the state machine listen to the queue and process events whenever they are triggered. \$\endgroup\$ – dfhwze May 21 at 17:34

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.