Add initial tests for Granulator

source/include/signalflow/buffer/buffer.h

@@ -329,6 +329,7 @@ class EnvelopeBuffer : public Buffer
       *------------------------------------------------------------------------*/
     EnvelopeBuffer(std::string shape, int num_frames = SIGNALFLOW_DEFAULT_ENVELOPE_BUFFER_LENGTH);
     EnvelopeBuffer(const std::function<float(float)> f);
+    EnvelopeBuffer(std::vector<float> samples);
 
     /**------------------------------------------------------------------------
      * @param position An envelope position between [0, 1].

source/src/buffer/buffer.cpp

@@ -541,6 +541,9 @@ EnvelopeBuffer::EnvelopeBuffer(const std::function<float(float)> f)
     this->fill(f);
 }
 
+EnvelopeBuffer::EnvelopeBuffer(std::vector<float> samples)
+    : Buffer(samples) {}
+
 double EnvelopeBuffer::offset_to_frame(double offset)
 {
     return signalflow_scale_lin_lin(offset, 0, 1, 0, this->num_frames - 1);

source/src/node/buffer/granulation/grain.cpp

@@ -34,7 +34,10 @@ void Grain::step()
 
 double Grain::get_progress()
 {
-    return (double) this->samples_processed / this->length_samples;
+    /*------------------------------------------------------------------------
+     * Subtract 1 to ensure that the progress output is 0..1 inclusive
+     *-----------------------------------------------------------------------*/
+    return (double) this->samples_processed / (this->length_samples - 1);
 }
 
 bool Grain::is_finished()

source/src/node/buffer/granulation/grainsegments.cpp

@@ -98,8 +98,8 @@ void SegmentedGranulator::process(Buffer &out, int num_frames)
                 /*------------------------------------------------------------------------
                  * Step forward in the grain and apply envelope.
                  *-----------------------------------------------------------------------*/
-                grain->step();
                 float amp = this->envelope->get(0, grain->get_progress());
+                grain->step();
 
                 for (int channel = 0; channel < this->num_output_channels; channel++)
                 {

source/src/node/buffer/granulation/granulator.cpp

@@ -88,9 +88,12 @@ void Granulator::process(Buffer &out, int num_frames)
             {
                 /*------------------------------------------------------------------------
                  * Step forward in the grain and apply envelope.
+                 * Query the amplitude before stepping forward in the grain so that we
+                 * obtain the amplitude sample for the correct index (e.g, phase = 0
+                 * for the first sample).
                  *-----------------------------------------------------------------------*/
-                grain->step();
                 float amp = this->envelope->get(0, grain->get_progress());
+                grain->step();
 
                 /*------------------------------------------------------------------------
                  * Calculate pan.

tests/test_nodes_granulator.py

@@ -0,0 +1,52 @@
+import signalflow as sf
+from . import graph
+
+import numpy as np
+import pytest
+
+
+@pytest.fixture(scope="function")
+def granulator(graph):
+    graph.sample_rate = 1000
+
+    buf = sf.Buffer(1, 5)
+    buf.fill(1.0)
+    granulator = sf.Granulator(buf, duration=buf.duration)
+    granulator.set_buffer("envelope", sf.EnvelopeBuffer("rectangular"))
+
+    return granulator
+
+
+def test_nodes_granulator(graph, granulator):
+    # Without any triggers, output is zero
+    graph.render_subgraph(granulator)
+    assert np.all(granulator.output_buffer[0] == 0.0)
+
+
+def test_nodes_granulator_intervals(graph, granulator):
+    # Trigger once every 10 samples.
+    clock = sf.Impulse(100)
+    granulator.clock = clock
+    graph.render_subgraph(granulator)
+    assert np.all(granulator.output_buffer[0][:1000] == np.tile([0.5, 0.5, 0.5, 0.5, 0.5,
+                                                                 0.0, 0.0, 0.0, 0.0, 0.0], 100))
+
+
+def test_nodes_granulator_overlap(graph, granulator):
+    # Trigger once every 4 samples, check that overlap is applied properly
+    clock = sf.Impulse(250)
+    granulator.clock = clock
+    graph.render_subgraph(granulator)
+    assert np.all(granulator.output_buffer[0][:10] == [0.5, 0.5, 0.5, 0.5, 1.0,
+                                                       0.5, 0.5, 0.5, 1.0, 0.5])
+
+
+def test_nodes_granulator_envelope(graph, granulator):
+    # Trigger once every 4 samples, check that overlap is applied properly
+    clock = sf.Impulse(0)
+    envelope = sf.EnvelopeBuffer([0.25, 0.5, 1.0, 0.5, 0.25])
+    granulator.clock = clock
+    granulator.set_buffer("envelope", envelope)
+    graph.render_subgraph(granulator)
+    assert np.all(granulator.output_buffer[0][:10] == [0.125, 0.25, 0.5, 0.25, 0.125,
+                                                       0, 0, 0, 0, 0])