From 6db418d2ec48c241758f851cb6e5bb923f099afc Mon Sep 17 00:00:00 2001 From: Nimra Aftab Date: Tue, 14 Mar 2023 09:11:47 -0400 Subject: [PATCH 1/8] create sample filter functions for testing --- extensions/lgpilot/filters/IIR_filter.py | 20 ++++++++++ .../lgpilot/filters/butterworth_filter.py | 37 +++++++++++++++++++ 2 files changed, 57 insertions(+) create mode 100644 extensions/lgpilot/filters/IIR_filter.py create mode 100644 extensions/lgpilot/filters/butterworth_filter.py diff --git a/extensions/lgpilot/filters/IIR_filter.py b/extensions/lgpilot/filters/IIR_filter.py new file mode 100644 index 00000000..dbd3c041 --- /dev/null +++ b/extensions/lgpilot/filters/IIR_filter.py @@ -0,0 +1,20 @@ +import numpy as np + +def filter_function(x, a, b): + """ A function to apply an IRR filter to a signal. + Input is , and filter parameters are and . + The output is a signal that has been processed with the filter. + """ + z = np.zeros(max(len(b), len(a))) + y = np.zeros_like(x) + for n in range(len(x)): + y[n] = b[0] * x[n] + for m in range(1, len(b)): + if n-m >= 0: + y[n] += b[m] * x[n-m] + for m in range(1, len(a)): + if n-m >= 0: + y[n] -= a[m] * y[n-m] + z[0] = x[n] + z[1:] = z[:-1] + return y diff --git a/extensions/lgpilot/filters/butterworth_filter.py b/extensions/lgpilot/filters/butterworth_filter.py new file mode 100644 index 00000000..7680e4cd --- /dev/null +++ b/extensions/lgpilot/filters/butterworth_filter.py @@ -0,0 +1,37 @@ +import numpy as np + +def filter_function(x, cutoff_freq, sample_rate, order=4): + """ + A function to apply a Butterworth filter to a signal. + + Args: + x (numpy.ndarray): The input signal to be filtered. + cutoff_freq (float): The cutoff frequency of the filter. + sample_rate (float): The sample rate of the signal. + order (int): The order of the filter. Defaults to 4. + + Returns: + numpy.ndarray: The filtered signal. + """ + nyquist_freq = sample_rate / 2 + normal_cutoff = cutoff_freq / nyquist_freq + b = np.zeros(order+1) + a = np.zeros(order+1) + + for i in range(order+1): + binomial_coefficient = 1 + for j in range(order): + if j != i: + binomial_coefficient *= (normal_cutoff - np.exp(1j*np.pi*(j-i)/order)) / (1 - np.exp(1j*np.pi*(j-i)/order)) + b[i] = np.real(binomial_coefficient) + a[i] = np.imag(binomial_coefficient) + + gain = 1 / np.abs(np.polyval(b, 1j*normal_cutoff) / np.polyval(a, 1j*normal_cutoff)) + b *= gain + + y = np.zeros_like(x) + + for i in range(len(x)): + y[i] = np.sum(b * x[max(0,i-len(b)+1):i+1]) - np.sum(a[1:] * x[max(0,i-len(a)):i]) + + return y From 7a9b5ddf191ee917e3745d243f3951281f815058 Mon Sep 17 00:00:00 2001 From: Nimra Aftab Date: Tue, 14 Mar 2023 09:13:16 -0400 Subject: [PATCH 2/8] Add function to support converting filters into nodes --- extensions/lgpilot/codeToNode.py | 39 ++++++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) create mode 100644 extensions/lgpilot/codeToNode.py diff --git a/extensions/lgpilot/codeToNode.py b/extensions/lgpilot/codeToNode.py new file mode 100644 index 00000000..0efbf80d --- /dev/null +++ b/extensions/lgpilot/codeToNode.py @@ -0,0 +1,39 @@ +import sys + +def create_setup(writer): + writer.write(f"\nclass FilterNode(lg.Node):\n") + writer.write("\tINPUT = lg.Topic(InputMessage)\n\tOUTPUT = lg.Topic(OutputMessage)\n") + writer.write("\tdef setup(self):\n\t\tself.func = filter_function\n") + return + +def create_filter(writer): + writer.write("\t@lg.subscriber(INPUT)\n\t@lg.publisher(OUTPUT)\n") + writer.write("\n\tdef filter(self, message: MessageInput):\n") + writer.write("\t\ty = self.func(message.x)\n\t\tyield self.OUTPUT, y") + + +def code_to_node(filename): + """ Take a python file containing a filter function and + output a file named node.py containing labgraph node. + + """ + with open(filename, 'r') as reader: + with open("node.py", 'w') as writer: + filter_function = False + filter_type = "" + for line in reader: + if "import" in line: # copy all import statements + writer.write(line) + if "filter_function" in line: # copy the filter function + filter_function = True + if filter_function: + writer.write(line) + if "return" in line: + filter_function = False + create_setup(writer) + create_filter(writer) + + + +if __name__ == "__main__": + code_to_node(sys.argv[1]) \ No newline at end of file From 001fd48b202a43784d02ea0055b2685458608973 Mon Sep 17 00:00:00 2001 From: Nimra Aftab Date: Fri, 31 Mar 2023 13:24:25 -0400 Subject: [PATCH 3/8] Support converting sklearn functions to nodes --- extensions/lgpilot/codeToNode.py | 56 +++++++++++++++++++------------- 1 file changed, 33 insertions(+), 23 deletions(-) diff --git a/extensions/lgpilot/codeToNode.py b/extensions/lgpilot/codeToNode.py index 0efbf80d..728a204c 100644 --- a/extensions/lgpilot/codeToNode.py +++ b/extensions/lgpilot/codeToNode.py @@ -1,38 +1,48 @@ import sys +import re -def create_setup(writer): - writer.write(f"\nclass FilterNode(lg.Node):\n") - writer.write("\tINPUT = lg.Topic(InputMessage)\n\tOUTPUT = lg.Topic(OutputMessage)\n") - writer.write("\tdef setup(self):\n\t\tself.func = filter_function\n") - return +def create_func(writer, library_name, function_name, function_args): + writer.write(f"from {library_name} import {function_name}\n") + writer.write(f"def {function_name}({function_args}):\n\t") + writer.write(f"y = {function_name}({function_args})\n\t") + writer.write("return y\n") -def create_filter(writer): - writer.write("\t@lg.subscriber(INPUT)\n\t@lg.publisher(OUTPUT)\n") - writer.write("\n\tdef filter(self, message: MessageInput):\n") - writer.write("\t\ty = self.func(message.x)\n\t\tyield self.OUTPUT, y") +def create_setup(writer, function_name): + writer.write(f"\nclass {function_name}Node(lg.Node):\n\t") + writer.write("INPUT = lg.Topic(InputMessage)\n\tOUTPUT = lg.Topic(OutputMessage)\n\n\t") + writer.write(f"def setup(self):\n\t\tself.func = {function_name}\n\n\t") + +def create_feature(writer, function_name, function_args): + writer.write("@lg.subscriber(INPUT)\n\t@lg.publisher(OUTPUT)\n\n\t") + writer.write(f"def {function_name}_feature(self, message: InputMessage):\n\t\t") + # turn a string containing a list of function args (ex: 'x, y, z') into InputMessage attributes (ex: message.x, message.y, message.z) + params = [f"message.{i.strip()}" for i in function_args.split(",")] + # turn a list of parameters (ex: ["message.x", "message.y", "message.z"]) into a string of arguments + # ex: y = self.func(message.x, message.y, message.z) + writer.write("y = self.func(" + ", ".join(params) + ")\n\t\tyield self.OUTPUT, y") def code_to_node(filename): - """ Take a python file containing a filter function and + """ Take a python file containing a function and output a file named node.py containing labgraph node. """ + library_name, function_name, function_args = "", "", "" with open(filename, 'r') as reader: with open("node.py", 'w') as writer: - filter_function = False - filter_type = "" for line in reader: - if "import" in line: # copy all import statements - writer.write(line) - if "filter_function" in line: # copy the filter function - filter_function = True - if filter_function: - writer.write(line) - if "return" in line: - filter_function = False - create_setup(writer) - create_filter(writer) - + # first check if line contains library_name and function name + result = re.search("from (.*) import (.*)", line) + if result is not None: # a match was found + library_name, function_name = result.group(1), result.group(2) + # next check if line contains function arguments + result = re.search(f"[a-zA-z]* = {function_name}\((.*)\)", line) + if result is not None: + function_args = result.group(1) + + create_func(writer, library_name, function_name, function_args) + create_setup(writer, function_name) + create_feature(writer, function_name, function_args) if __name__ == "__main__": From a92e3f08130ed362901a632f206c97777b1ba64e Mon Sep 17 00:00:00 2001 From: Nimra Aftab Date: Fri, 31 Mar 2023 13:24:58 -0400 Subject: [PATCH 4/8] remove code for filters --- extensions/lgpilot/filters/IIR_filter.py | 20 ---------- .../lgpilot/filters/butterworth_filter.py | 37 ------------------- 2 files changed, 57 deletions(-) delete mode 100644 extensions/lgpilot/filters/IIR_filter.py delete mode 100644 extensions/lgpilot/filters/butterworth_filter.py diff --git a/extensions/lgpilot/filters/IIR_filter.py b/extensions/lgpilot/filters/IIR_filter.py deleted file mode 100644 index dbd3c041..00000000 --- a/extensions/lgpilot/filters/IIR_filter.py +++ /dev/null @@ -1,20 +0,0 @@ -import numpy as np - -def filter_function(x, a, b): - """ A function to apply an IRR filter to a signal. - Input is , and filter parameters are and . - The output is a signal that has been processed with the filter. - """ - z = np.zeros(max(len(b), len(a))) - y = np.zeros_like(x) - for n in range(len(x)): - y[n] = b[0] * x[n] - for m in range(1, len(b)): - if n-m >= 0: - y[n] += b[m] * x[n-m] - for m in range(1, len(a)): - if n-m >= 0: - y[n] -= a[m] * y[n-m] - z[0] = x[n] - z[1:] = z[:-1] - return y diff --git a/extensions/lgpilot/filters/butterworth_filter.py b/extensions/lgpilot/filters/butterworth_filter.py deleted file mode 100644 index 7680e4cd..00000000 --- a/extensions/lgpilot/filters/butterworth_filter.py +++ /dev/null @@ -1,37 +0,0 @@ -import numpy as np - -def filter_function(x, cutoff_freq, sample_rate, order=4): - """ - A function to apply a Butterworth filter to a signal. - - Args: - x (numpy.ndarray): The input signal to be filtered. - cutoff_freq (float): The cutoff frequency of the filter. - sample_rate (float): The sample rate of the signal. - order (int): The order of the filter. Defaults to 4. - - Returns: - numpy.ndarray: The filtered signal. - """ - nyquist_freq = sample_rate / 2 - normal_cutoff = cutoff_freq / nyquist_freq - b = np.zeros(order+1) - a = np.zeros(order+1) - - for i in range(order+1): - binomial_coefficient = 1 - for j in range(order): - if j != i: - binomial_coefficient *= (normal_cutoff - np.exp(1j*np.pi*(j-i)/order)) / (1 - np.exp(1j*np.pi*(j-i)/order)) - b[i] = np.real(binomial_coefficient) - a[i] = np.imag(binomial_coefficient) - - gain = 1 / np.abs(np.polyval(b, 1j*normal_cutoff) / np.polyval(a, 1j*normal_cutoff)) - b *= gain - - y = np.zeros_like(x) - - for i in range(len(x)): - y[i] = np.sum(b * x[max(0,i-len(b)+1):i+1]) - np.sum(a[1:] * x[max(0,i-len(a)):i]) - - return y From cd6941dd3b71d66a99e9512fb4ab3762e2c8c9d8 Mon Sep 17 00:00:00 2001 From: Nimra Aftab Date: Wed, 5 Apr 2023 13:54:37 -0400 Subject: [PATCH 5/8] Create and test convolution node --- extensions/lgpilot/convolve.py | 12 +++++ extensions/lgpilot/node.py | 18 +++++++ extensions/lgpilot/test.py | 92 ++++++++++++++++++++++++++++++++++ 3 files changed, 122 insertions(+) create mode 100644 extensions/lgpilot/convolve.py create mode 100644 extensions/lgpilot/node.py create mode 100644 extensions/lgpilot/test.py diff --git a/extensions/lgpilot/convolve.py b/extensions/lgpilot/convolve.py new file mode 100644 index 00000000..bf30158f --- /dev/null +++ b/extensions/lgpilot/convolve.py @@ -0,0 +1,12 @@ +import numpy as np +from scipy.signal import convolve + +# Create two arrays +x = np.array([1, 2, 3, 4]) +h = np.array([1, 2, 3]) + +# Perform convolution +y = convolve(x, h) + +# Print result +print(y) \ No newline at end of file diff --git a/extensions/lgpilot/node.py b/extensions/lgpilot/node.py new file mode 100644 index 00000000..7740f1c3 --- /dev/null +++ b/extensions/lgpilot/node.py @@ -0,0 +1,18 @@ +from scipy.signal import convolve +def convolve(x, h): + y = convolve(x, h) + return y + +class convolveNode(lg.Node): + INPUT = lg.Topic(InputMessage) + OUTPUT = lg.Topic(OutputMessage) + + def setup(self): + self.func = convolve + + @lg.subscriber(INPUT) + @lg.publisher(OUTPUT) + + def convolve_feature(self, message: InputMessage): + y = self.func(message.x, message.h) + yield self.OUTPUT, y \ No newline at end of file diff --git a/extensions/lgpilot/test.py b/extensions/lgpilot/test.py new file mode 100644 index 00000000..2197643a --- /dev/null +++ b/extensions/lgpilot/test.py @@ -0,0 +1,92 @@ + +# Import labgraph +import labgraph as lg +# Imports required for this example +from scipy.signal import convolve +import numpy as np + +# A data type used in streaming, see docs: Messages +class InputMessage(lg.Message): + x: np.ndarray + h: np.ndarray + +class OutputMessage(lg.Message): + data: np.ndarray + +# ================================= NOISE GENERATOR ==================================== +# Configuration for NoiseGenerator, see docs: Lifecycles and Configuration +class NoiseGeneratorConfig(lg.Config): + #TODO: Is this needed? + sample_rate: float # Rate at which to generate noise + num_features: int # Number of features to generate + + +# A data source node that generates random noise to a single output topic +class NoiseGenerator(lg.Node): + OUTPUT = lg.Topic(InputMessage) + config: NoiseGeneratorConfig #TODO: Is this needed? + + # A publisher method that produces data on a single topic + @lg.publisher(OUTPUT) + def generate_noise(self): + yield self.OUTPUT, InputMessage(x=np.array([1, 2, 3, 4]), h=np.array([1, 2, 3])) + + +# ================================= CONVOLUTION =================================== + + +def convolve(x, h): + y = convolve(x, h) + return y + +class convolveNode(lg.Node): + INPUT = lg.Topic(InputMessage) + OUTPUT = lg.Topic(OutputMessage) + + def setup(self): + self.func = convolve + + @lg.subscriber(INPUT) + @lg.publisher(OUTPUT) + + def convolve_feature(self, message: InputMessage): + y = self.func(message.x, message.h) + yield self.OUTPUT, y + +# ================================== CONVOLVED NOISE ==================================== + + +# Configuration for ConvolvedNoise +class ConvolvedNoiseConfig(lg.Config): + x: np.ndarray + h: np.ndarray + +# A group that combines noise generation and rolling averaging. The output topic +# contains averaged noise. We could just put all three nodes in a graph below, but we +# add this group to demonstrate the grouping functionality. +class AveragedNoise(lg.Group): + OUTPUT = lg.Topic(OutputMessage) + + #TODO: Is this needed? + config: AveragedNoiseConfig + GENERATOR: NoiseGenerator + CONVOLVENODE: convolveNode + + def connections(self) -> lg.Connections: + # To produce convolved noise, we connect the noise generator to the convolver + # Then we "expose" the averager's output as an output of this group + return ( + (self.GENERATOR.OUTPUT, self.CONVOLVENODE.INPUT), + (self.CONVOLVENODE.OUTPUT, self.OUTPUT), + ) + + #TODO: What should happen here? + def setup(self) -> None: + # Cascade this group's configuration to its contained nodes + self.GENERATOR.configure( + NoiseGeneratorConfig( + sample_rate=self.config.sample_rate, + num_features=self.config.num_features, + ) + ) + self.ROLLING_AVERAGER.configure(RollingConfig(window=self.config.window)) From 2ab61cb891d4e40a6df2c0b1a505fcc6125748b5 Mon Sep 17 00:00:00 2001 From: Nimra Aftab Date: Fri, 7 Apr 2023 19:07:03 -0400 Subject: [PATCH 6/8] Update NoiseGeneratorConfig --- extensions/lgpilot/test.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/extensions/lgpilot/test.py b/extensions/lgpilot/test.py index 2197643a..74678e9e 100644 --- a/extensions/lgpilot/test.py +++ b/extensions/lgpilot/test.py @@ -16,15 +16,14 @@ class OutputMessage(lg.Message): # ================================= NOISE GENERATOR ==================================== # Configuration for NoiseGenerator, see docs: Lifecycles and Configuration class NoiseGeneratorConfig(lg.Config): - #TODO: Is this needed? - sample_rate: float # Rate at which to generate noise - num_features: int # Number of features to generate + x: np.ndarray + h: np.ndarray # A data source node that generates random noise to a single output topic class NoiseGenerator(lg.Node): OUTPUT = lg.Topic(InputMessage) - config: NoiseGeneratorConfig #TODO: Is this needed? + config: NoiseGeneratorConfig # A publisher method that produces data on a single topic @lg.publisher(OUTPUT) From 6d46a7470586b367ad559ad4cadf8ec6430c2e7d Mon Sep 17 00:00:00 2001 From: Nimra Aftab Date: Fri, 7 Apr 2023 19:10:42 -0400 Subject: [PATCH 7/8] Replace unnecessary code with import statement --- extensions/lgpilot/test.py | 22 +--------------------- 1 file changed, 1 insertion(+), 21 deletions(-) diff --git a/extensions/lgpilot/test.py b/extensions/lgpilot/test.py index 74678e9e..1d59c52d 100644 --- a/extensions/lgpilot/test.py +++ b/extensions/lgpilot/test.py @@ -4,6 +4,7 @@ # Imports required for this example from scipy.signal import convolve import numpy as np +from node import convolve, convolveNode # A data type used in streaming, see docs: Messages class InputMessage(lg.Message): @@ -31,27 +32,6 @@ def generate_noise(self): yield self.OUTPUT, InputMessage(x=np.array([1, 2, 3, 4]), h=np.array([1, 2, 3])) -# ================================= CONVOLUTION =================================== - - -def convolve(x, h): - y = convolve(x, h) - return y - -class convolveNode(lg.Node): - INPUT = lg.Topic(InputMessage) - OUTPUT = lg.Topic(OutputMessage) - - def setup(self): - self.func = convolve - - @lg.subscriber(INPUT) - @lg.publisher(OUTPUT) - - def convolve_feature(self, message: InputMessage): - y = self.func(message.x, message.h) - yield self.OUTPUT, y - # ================================== CONVOLVED NOISE ==================================== From 12fe18128f889e81bb50186b0329d3ad04cb7fd8 Mon Sep 17 00:00:00 2001 From: Nimra Aftab Date: Wed, 12 Apr 2023 19:28:05 -0400 Subject: [PATCH 8/8] update test for convolution node --- extensions/lgpilot/test.py | 153 ++++++++++++++++++++++++++----------- 1 file changed, 109 insertions(+), 44 deletions(-) diff --git a/extensions/lgpilot/test.py b/extensions/lgpilot/test.py index 1d59c52d..88efd559 100644 --- a/extensions/lgpilot/test.py +++ b/extensions/lgpilot/test.py @@ -1,10 +1,21 @@ - # Import labgraph import labgraph as lg # Imports required for this example from scipy.signal import convolve import numpy as np -from node import convolve, convolveNode + +import labgraph as lg +import numpy as np +import pytest +from ...generators.sine_wave_generator import ( + SineWaveChannelConfig, + SineWaveGenerator, +) + +from ..mixer_one_input_node import MixerOneInputConfig, MixerOneInputNode +from ..signal_capture_node import SignalCaptureConfig, SignalCaptureNode +from ..signal_generator_node import SignalGeneratorNode + # A data type used in streaming, see docs: Messages class InputMessage(lg.Message): @@ -14,58 +25,112 @@ class InputMessage(lg.Message): class OutputMessage(lg.Message): data: np.ndarray -# ================================= NOISE GENERATOR ==================================== -# Configuration for NoiseGenerator, see docs: Lifecycles and Configuration -class NoiseGeneratorConfig(lg.Config): - x: np.ndarray - h: np.ndarray +# ================================= CONVOLUTION =================================== -# A data source node that generates random noise to a single output topic -class NoiseGenerator(lg.Node): - OUTPUT = lg.Topic(InputMessage) - config: NoiseGeneratorConfig - # A publisher method that produces data on a single topic - @lg.publisher(OUTPUT) - def generate_noise(self): - yield self.OUTPUT, InputMessage(x=np.array([1, 2, 3, 4]), h=np.array([1, 2, 3])) - +def convolve(x, h): + y = convolve(x, h) + return y -# ================================== CONVOLVED NOISE ==================================== +class ConvolveNode(lg.Node): + INPUT = lg.Topic(InputMessage) + OUTPUT = lg.Topic(OutputMessage) + def setup(self): + self.func = convolve -# Configuration for ConvolvedNoise -class ConvolvedNoiseConfig(lg.Config): - x: np.ndarray - h: np.ndarray + @lg.subscriber(INPUT) + @lg.publisher(OUTPUT) + + def convolve_feature(self, message: InputMessage): + y = self.func(message.x, message.h) + yield self.OUTPUT, y + +# ====================================================================== + + +# class MixerOneInputConfig(lg.Config): +# # This is an NxM matrix (for M inputs, N outputs) +# weights: np.ndarray + +class ConvolveInputConfig(lg.Config): + array: np.ndarray + kernel: np.ndarray + + +class MyGraphConfig(lg.Config): + sine_wave_channel_config: SineWaveChannelConfig + convolve_config: ConvolveInputConfig + capture_config: SignalCaptureConfig -# A group that combines noise generation and rolling averaging. The output topic -# contains averaged noise. We could just put all three nodes in a graph below, but we -# add this group to demonstrate the grouping functionality. -class AveragedNoise(lg.Group): - OUTPUT = lg.Topic(OutputMessage) - #TODO: Is this needed? - config: AveragedNoiseConfig - GENERATOR: NoiseGenerator - CONVOLVENODE: convolveNode +class MyGraph(lg.Graph): + + sample_source: SignalGeneratorNode + convolve_node: ConvolveNode + capture_node: SignalCaptureNode + + def setup(self) -> None: + self.capture_node.configure(self.config.capture_config) + self.sample_source.set_generator( + SineWaveGenerator(self.config.sine_wave_channel_config) + ) + self.convolve_node.configure(self.config.convolve_config) def connections(self) -> lg.Connections: - # To produce convolved noise, we connect the noise generator to the convolver - # Then we "expose" the averager's output as an output of this group return ( - (self.GENERATOR.OUTPUT, self.CONVOLVENODE.INPUT), - (self.CONVOLVENODE.OUTPUT, self.OUTPUT), + (self.convolve_node.INPUT, self.sample_source.SAMPLE_TOPIC), + (self.capture_node.SAMPLE_TOPIC, self.mixer_node.OUTPUT), ) - #TODO: What should happen here? - def setup(self) -> None: - # Cascade this group's configuration to its contained nodes - self.GENERATOR.configure( - NoiseGeneratorConfig( - sample_rate=self.config.sample_rate, - num_features=self.config.num_features, - ) - ) - self.ROLLING_AVERAGER.configure(RollingConfig(window=self.config.window)) + +def test_convolve_input_node() -> None: + """ + Tests that node convolves correctly, uses numpy arrays and kernel sizes as input + """ + + sample_rate = 1 # Hz + test_duration = 10 # sec + + # Test configurations + shape = (2,) + amplitudes = np.array([5.0, 3.0]) + frequencies = np.array([5, 10]) + phase_shifts = np.array([1.0, 5.0]) + midlines = np.array([3.0, -2.5]) + + test_array = [1, 2, 3] + test_kernel = [2] + + # Generate expected values + + expected = convolve(test_array, test_kernel) # use the convolve from the library to generate the expected values + + # Create the graph + generator_config = SineWaveChannelConfig( + shape, amplitudes, frequencies, phase_shifts, midlines, sample_rate + ) + capture_config = SignalCaptureConfig(int(test_duration / sample_rate)) + + # mixer_weights = np.identity(2) + # mixer_config = MixerOneInputConfig(mixer_weights) + + convolve_input_array = [1, 2, 3] + convolve_input_kernel = [2] + + convolve_config = ConvolveInputConfig(convolve_input_array, convolve_input_kernel) + + my_graph_config = MyGraphConfig(generator_config, convolve_config, capture_config) + + graph = MyGraph() + graph.configure(my_graph_config) + + runner = lg.LocalRunner(module=graph) + runner.run() + received = np.array(graph.capture_node.samples).T + np.testing.assert_almost_equal(received, expected) + +# 1. test the convolve function +# 2. create the graph and run it +# 3. repeat the same thing for other APIs -- just need to create simple test cases \ No newline at end of file