in pedalboard/io/AudioStream.h [702:990]
inline void init_audio_stream(py::module &m) {
py::class_<AudioStream, std::shared_ptr<AudioStream>>(m, "AudioStream",
R"(
A class that allows interacting with live streams of audio from an input
audio device (i.e.: a microphone, audio interface, etc) and/or to an
output device (speaker, headphones), allowing access to the audio
stream from within Python code.
Use :py:meth:`AudioStream.play` to play audio data to your speakers::
# Play a 10-second chunk of an audio file:
with AudioFile("my_audio_file.mp3") as f:
chunk = f.read(f.samplerate * 10)
AudioStream.play(chunk, f.samplerate)
Or use :py:meth:`AudioStream.write` to stream audio in chunks::
# Play an audio file by looping through it in chunks:
with AudioStream(output_device="default") as stream:
with AudioFile("my_audio_file.mp3") as f:
while f.tell() < f.frames:
# Decode and play 512 samples at a time:
stream.write(f.read(512))
:class:`AudioStream` may also be used to pass live audio through a :class:`Pedalboard`::
# Pass both an input and output device name to connect both ends:
input_device_name = AudioStream.default_input_device_name
output_device_name = AudioStream.default_output_device_name
with AudioStream(input_device_name, output_device_name) as stream:
# In this block, audio is streaming through `stream`!
# Audio will be coming out of your speakers at this point.
# Add plugins to the live audio stream:
reverb = Reverb()
stream.plugins.append(reverb)
# Change plugin properties as the stream is running:
reverb.wet_level = 1.0
# Delete plugins:
del stream.plugins[0]
# Or use AudioStream synchronously:
stream = AudioStream(input_device_name, output_device_name)
stream.plugins.append(Reverb(wet_level=1.0))
stream.run() # Run the stream until Ctrl-C is received
.. warning::
The :class:`AudioStream` class implements a context manager interface
to ensure that audio streams are never left "dangling" (i.e.: running in
the background without being stopped).
While it is possible to call the :meth:`__enter__` method directly to run an
audio stream in the background, this can have some nasty side effects. If the
:class:`AudioStream` object is no longer reachable (not bound to a variable,
not in scope, etc), the audio stream will continue to run forever, and
won't stop until the Python interpreter exits.
To run an :class:`AudioStream` in the background, use Python's
:py:mod:`threading` module to run the stream object method on a
background thread, allowing for easier cleanup.
*Introduced in v0.7.0 for macOS and Windows. Linux support introduced in v0.9.14.*
:py:meth:`read` *and* :py:meth:`write` *methods introduced in v0.9.12.*
)")
.def(py::init([](std::optional<std::string> inputDeviceName,
std::optional<std::string> outputDeviceName,
std::optional<std::shared_ptr<Chain>> pedalboard,
std::optional<double> sampleRate,
std::optional<int> bufferSize, bool allowFeedback,
int numInputChannels, int numOutputChannels) {
return std::make_shared<AudioStream>(
inputDeviceName, outputDeviceName, pedalboard, sampleRate,
bufferSize, allowFeedback, numInputChannels,
numOutputChannels);
}),
py::arg("input_device_name") = py::none(),
py::arg("output_device_name") = py::none(),
py::arg("plugins") = py::none(), py::arg("sample_rate") = py::none(),
py::arg("buffer_size") = py::none(),
py::arg("allow_feedback") = false, py::arg("num_input_channels") = 1,
py::arg("num_output_channels") = 2)
.def("run", &AudioStream::stream,
"Start streaming audio from input to output, passing the audio "
"stream through the :py:attr:`plugins` on this AudioStream object. "
"This call will block the current thread until a "
":py:exc:`KeyboardInterrupt` (``Ctrl-C``) is received.")
.def_property_readonly("running", &AudioStream::getIsRunning,
":py:const:`True` if this stream is currently "
"streaming live "
"audio, :py:const:`False` otherwise.")
.def("__enter__", &AudioStream::enter,
"Use this :class:`AudioStream` as a context manager. Entering the "
"context manager will immediately start the audio stream, sending "
"audio through to the output device.")
.def("__exit__", &AudioStream::exit,
"Exit the context manager, ending the audio stream. Once called, "
"the audio stream will be stopped (i.e.: :py:attr:`running` will "
"be "
":py:const:`False`).")
.def("__repr__",
[](const AudioStream &stream) {
std::ostringstream ss;
ss << "<pedalboard.io.AudioStream";
#ifdef JUCE_MODULE_AVAILABLE_juce_audio_devices
auto audioDeviceSetup = stream.getAudioDeviceSetup();
if (stream.getNumInputChannels() > 0) {
ss << " input_device_name=\""
<< audioDeviceSetup.inputDeviceName.toStdString() << "\"";
} else {
ss << " input_device_name=None";
}
if (stream.getNumOutputChannels() > 0) {
ss << " output_device_name=\""
<< audioDeviceSetup.outputDeviceName.toStdString() << "\"";
} else {
ss << " output_device_name=None";
}
ss << " sample_rate="
<< juce::String(audioDeviceSetup.sampleRate, 2).toStdString();
ss << " buffer_size=" << audioDeviceSetup.bufferSize;
if (stream.getIsRunning()) {
ss << " running";
} else {
ss << " not running";
}
#endif
ss << " at " << &stream;
ss << ">";
return ss.str();
})
.def_property_readonly(
"buffer_size",
[](AudioStream &stream) {
#ifdef JUCE_MODULE_AVAILABLE_juce_audio_devices
return stream.getAudioDeviceSetup().bufferSize;
#else
return 0;
#endif
},
"The size (in frames) of the buffer used between the audio "
"hardware "
"and Python.")
.def_property("plugins", &AudioStream::getPedalboard,
&AudioStream::setPedalboard,
"The Pedalboard object that this AudioStream will use to "
"process audio.")
.def_property_readonly(
"dropped_input_frame_count", &AudioStream::getDroppedInputFrameCount,
"The number of frames of audio that were dropped since the last "
"call "
"to :py:meth:`read`. To prevent audio from being dropped during "
"recording, ensure that you call :py:meth:`read` as often as "
"possible or increase your buffer size.")
.def_property_readonly(
"sample_rate", &AudioStream::getSampleRate,
"The sample rate that this stream is operating at.")
.def_property_readonly("num_input_channels",
&AudioStream::getNumInputChannels,
"The number of input channels on the input "
"device. Will be ``0`` if "
"no input device is connected.")
.def_property_readonly(
"num_output_channels", &AudioStream::getNumOutputChannels,
"The number of output channels on the output "
"device. Will be ``0`` if no output device is connected.")
.def_property(
"ignore_dropped_input", &AudioStream::getIgnoreDroppedInput,
&AudioStream::setIgnoreDroppedInput,
"Controls the behavior of the :py:meth:`read` method when audio "
"data "
"is dropped. If this property is false (the default), the "
":py:meth:`read` method will raise a :py:exc:`RuntimeError` if any "
"audio data is dropped in between calls. If this property is true, "
"the :py:meth:`read` method will return the most recent audio data "
"available, even if some audio data was dropped.\n\n.. note::\n "
"The :py:attr:`dropped_input_frame_count` property is unaffected "
"by "
"this setting.")
.def(
"write",
[](AudioStream &stream,
const py::array_t<float, py::array::c_style> inputArray,
float sampleRate) {
if (sampleRate != stream.getSampleRate()) {
throw std::runtime_error(
"The sample rate provided to `write` (" +
std::to_string(sampleRate) +
" Hz) does not match the output device's sample rate (" +
std::to_string(stream.getSampleRate()) +
" Hz). To write audio data to the output device, the "
"sample "
"rate of the audio must match the output device's sample "
"rate.");
}
stream.write(copyPyArrayIntoJuceBuffer(inputArray));
},
py::arg("audio"), py::arg("sample_rate"),
"Write (play) audio data to the output device. This method will "
"block until the provided audio buffer is played in its "
"entirety.\n\nIf the provided sample rate does not match the "
"output "
"device's sample rate, an error will be thrown. In this case, you "
"can use :py:class:`StreamResampler` to resample the audio before "
"calling :py:meth:`write`.")
.def(
"read",
[](AudioStream &stream, int numSamples) {
return copyJuceBufferIntoPyArray(stream.read(numSamples),
ChannelLayout::NotInterleaved, 0);
},
py::arg("num_samples") = 0,
"Read (record) audio data from the input device. When called with "
"no "
"arguments, this method returns all of the available audio data in "
"the buffer. If `num_samples` is provided, this method will block "
"until the desired number of samples have been received. The audio "
"is recorded at the :py:attr:`sample_rate` of this stream.\n\n.. "
"warning::\n Recording audio is a **real-time** operation, so "
"if "
"your code doesn't call :py:meth:`read` quickly enough, some audio "
"will be lost. To warn about this, :py:meth:`read` will throw an "
"exception if audio data is dropped. This behavior can be disabled "
"by setting :py:attr:`ignore_dropped_input` to :py:const:`True`. "
"The "
"number of dropped samples since the last call to :py:meth:`read` "
"can be retrieved by accessing the "
":py:attr:`dropped_input_frame_count` property.")
.def_property_readonly(
"buffered_input_sample_count",
&AudioStream::getNumBufferedInputFrames,
"The number of frames of audio that are currently in the input "
"buffer. This number will change rapidly, and will be "
":py:const:`None` if no input device is connected.")
.def("close", &AudioStream::close,
"Close the audio stream, stopping the audio device and releasing "
"any resources. After calling close, this AudioStream object is no "
"longer usable.")
.def_static(
"play",
[](const py::array_t<float, py::array::c_style> audio,
float sampleRate, std::optional<std::string> outputDeviceName) {
juce::AudioBuffer<float> buffer = copyPyArrayIntoJuceBuffer(audio);
AudioStream(std::nullopt, outputDeviceName ? *outputDeviceName : "",
std::nullopt, {sampleRate}, std::nullopt, false, 0,
buffer.getNumChannels())
.write(buffer);
},
py::arg("audio"), py::arg("sample_rate"),
py::arg("output_device_name") = py::none(),
"Play audio data to the speaker, headphones, or other output "
"device. "
"This method will block until the audio is finished playing.")
.def_property_readonly_static(
"input_device_names",
[](py::object *obj) -> std::vector<std::string> {
return AudioStream::getDeviceNames(true);
},
"The input devices (i.e.: microphones, audio interfaces, etc.) "
"currently available on the current machine.")
.def_property_readonly_static(
"output_device_names",
[](py::object *obj) -> std::vector<std::string> {
return AudioStream::getDeviceNames(false);
},
"The output devices (i.e.: speakers, headphones, etc.) currently "
"available on the current machine.")
.def_property_readonly_static(
"default_input_device_name",
[](py::object *obj) -> std::optional<std::string> {
return AudioStream::getDefaultDeviceName(true, 1);
},
"The name of the default input device (i.e.: microphone, audio "
"interface, etc.) currently available on the current machine. May "
"be :py:const:`None` if no input devices are present.")
.def_property_readonly_static(
"default_output_device_name",
[](py::object *obj) -> std::optional<std::string> {
return AudioStream::getDefaultDeviceName(false, 2);
},
"The name of the default output device (i.e.: speakers, "
"headphones, etc.) currently available on the current machine. May "
"be :py:const:`None` if no output devices are present.");
}