timeflux.helpers.testing


A set of tools to facilitate code testing

testing

class timeflux.helpers.testing.DummyData(num_rows=1000, num_cols=5, cols=None, rate=10, jitter=0.05, start_date='2018-01-01', seed=42, round=6)[source]

Generate dummy data.

Initialize the dataframe.

Parameters
  • num_rows (int) – Number of rows

  • num_cols (int) – Number of columns

  • cols (list) – List of column names

  • rate (float) – Frequency, in Hertz

  • jitter (float) – Amount of jitter, relative to rate

  • start_date (string) – Start date

  • seed (int) – Seed for random number generation

  • round (int) – Number of decimals for random numbers

next(self, num_rows=10)[source]

Get the next chunk of data.

Parameters

num_rows (int) – Number of rows to fetch

reset(self)[source]

Reset the cursor.

class timeflux.helpers.testing.DummyXArray(num_time=1000, num_space=5, rate=10, jitter=0.05, start_date='2018-01-01', seed=42, round=6)[source]

Generate dummy data of type XArray.

Initialize the dataframe.

Parameters
  • num_time (int) – Number of rows

  • num_space (int) – Number of columns

  • rate (float) – Frequency, in Hertz

  • jitter (float) – Amount of jitter, relative to rate

  • start_date (string) – Start date

  • seed (int) – Seed for random number generation

  • round (int) – Number of decimals for random numbers

next(self, num_rows=10)[source]

Get the next chunk of data.

Parameters

num_rows (int) – Number of rows to fetch

reset(self)[source]

Reset the cursor.

class timeflux.helpers.testing.ReadData(data)[source]

Generate custom data.

Initialize the dataframe.

Parameters

data (DataFrame) – custom data to stream.

next(self, num_rows=10)[source]

Get the next chunk of data.

Parameters

num_rows (int) – Number of rows to fetch

reset(self)[source]

Reset the cursor.

class timeflux.helpers.testing.Looper(generator, node, input_port='i', output_port='o')[source]

Mimics the scheduler behavior to allow testing the output of a node offline.

Initialize the helper :param generator (Node): timeflux node to test :param data (Object): data generator object with a method next and reset

run(self, chunk_size=None)[source]

Loop across chunks of a generator, update the node and return data and meta. :param chunk_size (int): number of samples per chunk :return: output_data (DataFrame): concatenated output data output_meta: list of meta