zmqaudio...

lsldert/Makefile

@@ -10,7 +10,7 @@ LDLIBS+=-lzmq -lboost_program_options -lpigpio -llsl32
 #INCLUDE+=RS-232
 CXXFLAGS+=-Wall -std=c++11
 
-all: zmq_trigger playtone msgqueue_pubsub lsldert_proxy zmq_trigger_subscriber playsound lsldertc hwserver hwserver2 hwserver3 zmqaudio
+all: zmq_trigger playtone msgqueue_pubsub lsldert_proxy zmq_trigger_subscriber playsound lsldertc hwserver hwserver2 hwserver3 zmqaudio zmqaudio_pa
 
 lsldertc: lsldertc.c
 hwserver: hwserver.c
@@ -27,9 +27,11 @@ msgqueue_pubsub: msgqueue_pubsub.cc
 lsldert_proxy: lsldert_proxy.cc
 lsldert_proxy: LDLIBS+=-lpthread
 zmq_trigger_subscriber: zmq_trigger_subscriber.cc
-zmqaudio: LDLIBS=-lzmq
-zmqaudio: zmqaudio.cc zmqaudio.h
+zmqaudio_pa: LDLIBS=-lzmq
+zmqaudio_pa: zmqaudio_pa.cc zmqaudio_pa.h
+zmqaudio: LDLIBS=-lzmq -lm -lasound -lportaudio
+zmqaudio: zmqaudio.cc zmqaudio.h pcm_async.cc
 paex_sine: paex_sine.c
-zmqaudio paex_sine: LDLIBS+=-lm -lportaudio
+zmqaudio_pa paex_sine: LDLIBS+=-lm -lportaudio
 sine: sine.cc
 sine: LDLIBS+=-lportaudio -lportaudiocpp

lsldert/pcm_async.cc

@@ -2,6 +2,7 @@
  *  This small demo sends a simple sinusoidal wave to your speakers.
  */
 
+#include "zmqaudio.h"
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -28,6 +29,73 @@ static snd_pcm_sframes_t buffer_size;
 static snd_pcm_sframes_t period_size;
 static snd_output_t *output = NULL;
 
+static zmq_audiobuffer *zbuf = nullptr;   // TODO, hack
+
+
+static void play_from_buffer(const snd_pcm_channel_area_t *areas, 
+			  snd_pcm_uframes_t offset,
+			  int count, zmq_audiobuffer *zbuf)
+{
+	static double max_phase = 2. * M_PI;
+	double step = max_phase*freq/(double)rate;
+	unsigned char *samples[channels];
+	int steps[channels];
+	unsigned int chn;
+	int format_bits = snd_pcm_format_width(format);
+	unsigned int maxval = (1 << (format_bits - 1)) - 1;
+	int bps = format_bits / 8;  /* bytes per sample */
+	int phys_bps = snd_pcm_format_physical_width(format) / 8;
+	int big_endian = snd_pcm_format_big_endian(format) == 1;
+	int to_unsigned = snd_pcm_format_unsigned(format) == 1;
+	int is_float = (format == SND_PCM_FORMAT_FLOAT_LE ||
+			format == SND_PCM_FORMAT_FLOAT_BE);
+
+	/* verify and prepare the contents of areas */
+	for (chn = 0; chn < channels; chn++) {
+		if ((areas[chn].first % 8) != 0) {
+			printf("areas[%u].first == %u, aborting...\n", chn, areas[chn].first);
+			exit(EXIT_FAILURE);
+		}
+		samples[chn] = /*(signed short *)*/(((unsigned char *)areas[chn].addr) + (areas[chn].first / 8));
+		if ((areas[chn].step % 16) != 0) {
+			printf("areas[%u].step == %u, aborting...\n", chn, areas[chn].step);
+			exit(EXIT_FAILURE);
+		}
+		steps[chn] = areas[chn].step / 8;
+		samples[chn] += offset * steps[chn];
+	}
+	/* fill the channel areas */
+	while (count-- > 0) {
+		union {
+			float f;
+			int i;
+		} fval;
+		int res, i;
+		if (is_float) {
+			//TODO fval.f = sin(phase);
+			res = fval.i;
+		} else
+			//TODO res = sin(phase) * maxval;
+		if (to_unsigned)
+			res ^= 1U << (format_bits - 1);
+		for (chn = 0; chn < channels; chn++) {
+			/* Generate data in native endian format */
+			if (big_endian) {
+				for (i = 0; i < bps; i++)
+					*(samples[chn] + phys_bps - 1 - i) = (res >> i * 8) & 0xff;
+			} else {
+				for (i = 0; i < bps; i++)
+					*(samples[chn] + i) = (res >>  i * 8) & 0xff;
+			}
+			samples[chn] += steps[chn];
+		}
+		//TODO phase += step;
+		//TODO if (phase >= max_phase)
+			//TODO phase -= max_phase;
+	}
+	//TODO *_phase = phase;
+}
+
 static void generate_sine(const snd_pcm_channel_area_t *areas, 
 			  snd_pcm_uframes_t offset,
 			  int count, double *_phase)
@@ -386,6 +454,7 @@ struct async_private_data {
 	signed short *samples;
 	snd_pcm_channel_area_t *areas;
 	double phase;
+        zmq_audiobuffer *zbuf;
 };
 
 static void async_callback(snd_async_handler_t *ahandler)
@@ -399,7 +468,10 @@ static void async_callback(snd_async_handler_t *ahandler)
 	
 	avail = snd_pcm_avail_update(handle);
 	while (avail >= period_size) {
-		generate_sine(areas, 0, period_size, &data->phase);
+                if (data->zbuf)
+			play_from_buffer(areas, 0, period_size, data->zbuf);
+                else
+			generate_sine(areas, 0, period_size, &data->phase);
 		err = snd_pcm_writei(handle, samples, period_size);
 		if (err < 0) {
 			printf("Write error: %s\n", snd_strerror(err));
@@ -424,6 +496,7 @@ static int async_loop(snd_pcm_t *handle,
 	data.samples = samples;
 	data.areas = areas;
 	data.phase = 0;
+        data.zbuf = zbuf;
 	err = snd_async_add_pcm_handler(&ahandler, handle, async_callback, &data);
 	if (err < 0) {
 		printf("Unable to register async handler\n");
@@ -761,7 +834,7 @@ static void help(void)
 	printf("\n");
 }
 
-int play()
+int play(zmq_audiobuffer *_zbuf)
 {
 	snd_pcm_t *handle;
 	int err;
@@ -775,8 +848,11 @@ int play()
 	snd_pcm_hw_params_alloca(&hwparams);
 	snd_pcm_sw_params_alloca(&swparams);
 
+        zbuf = _zbuf;
+        rate = zbuf->fsamp;
 	rate = rate < 4000 ? 4000 : rate;
 	rate = rate > 196000 ? 196000 : rate;
+        channels = zbuf->nchan;
 	channels = channels < 1 ? 1 : channels;
 	channels = channels > 1024 ? 1024 : channels;
 	buffer_time = buffer_time < 1000 ? 1000 : buffer_time;
@@ -846,7 +922,7 @@ int play()
 	return 0;
 }
 
-int main(int argc, char *argv[])
+int nomore_main(int argc, char *argv[])
 {
 	struct option long_option[] =
 	{

lsldert/zmqaudio.cc

@@ -40,7 +40,6 @@
 // license above.
 //
 #include "zmqaudio.h"
-#include "portaudio.h"
 #include "zhelpers.hpp"
 #include <iostream>
 #include <math.h>

lsldert/zmqaudio_pa.cc

+//
+// zmqaudio_pa.cc -- play audio samples received over zeromq sockets
+// Copyright (c) 2020 Günter Windau
+// based on the example in paex_sine.c
+//
+
+//
+// This program uses the PortAudio Portable Audio Library.
+// For more information see: http://www.portaudio.com/
+// Copyright (c) 1999-2000 Ross Bencina and Phil Burk
+//
+// Permission is hereby granted, free of charge, to any person obtaining
+// a copy of this software and associated documentation files
+// (the "Software"), to deal in the Software without restriction,
+// including without limitation the rights to use, copy, modify, merge,
+// publish, distribute, sublicense, and/or sell copies of the Software,
+// and to permit persons to whom the Software is furnished to do so,
+// subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be
+// included in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+// EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+// IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR
+// ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF
+// CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+// WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+//
+
+//
+// The text above constitutes the entire PortAudio license; however,
+// the PortAudio community also makes the following non-binding requests:
+//
+// Any person wishing to distribute modifications to the Software is
+// requested to send the modifications to the original developer so that
+// they can be incorporated into the canonical version. It is also
+// requested that these non-binding requests be included along with the
+// license above.
+//
+#include "zmqaudio.h"
+#include "portaudio.h"
+#include "zhelpers.hpp"
+#include <iostream>
+#include <math.h>
+#include <stdio.h>
+#include <sched.h>
+
+#define NUM_SECONDS (5)
+#define SAMPLE_RATE (44100)
+
+#ifndef M_PI
+#define M_PI (3.14159265)
+#endif
+
+#define TABLE_SIZE (200)
+typedef struct {
+    float sine[TABLE_SIZE];
+    int left_phase;
+    int right_phase;
+    char message[20];
+} paTestData;
+
+zmq_audiobuffer *zmq_audiobuffer::_playing_now = nullptr;
+
+int verbose = 1;
+const int nrbuffers = 22;
+zmq_audiobuffer *buffer[nrbuffers];
+
+PaError print_error(PaError e) {
+    if (e != paNoError)
+        Pa_Terminate();
+    fprintf(stderr, "An error occured while using the portaudio stream\n");
+    fprintf(stderr, "Error number: %d\n", e);
+    fprintf(stderr, "Error message: %s\n", Pa_GetErrorText(e));
+    return e;
+}
+
+zmq_audiobuffer::zmq_audiobuffer(std::uint32_t _fsamp, std::uint32_t _nsamp,
+                                 std::uint32_t _nchan)
+    : fsamp{_fsamp}, nsamp{_nsamp}, nchan{_nchan},
+      data{new float[_nsamp * _nchan]}, idata{0},
+      frames_per_buffer{paFramesPerBufferUnspecified}, playing{false}, stream{nullptr} {
+    assert(sizeof(float) == 4); // need 32 bit IEEE 754 floats
+    frames_per_buffer=64;
+}
+
+zmq_audiobuffer::~zmq_audiobuffer() {
+    delete[] data;
+    Pa_StopStream(stream);
+    Pa_CloseStream(stream);
+}
+
+// This routine will be called by the PortAudio engine when audio is needed.
+// It may called at interrupt level on some machines so don't do anything
+// that could mess up the system like calling malloc() or free().
+//
+int zmq_audiobuffer::stream_callback(
+    const void *, void *outputBuffer,
+    unsigned long framesPerBuffer,
+    const PaStreamCallbackTimeInfo *,
+    PaStreamCallbackFlags, void *userData) {
+    std::cerr << "<callback>" << std::flush;
+    zmq_audiobuffer *pbuf = (zmq_audiobuffer *)userData;
+    float *out = (float *)outputBuffer;
+    unsigned long i;
+
+    int ret = paContinue;
+    for (i = 0; i < framesPerBuffer; i++) {
+        if (pbuf->idata < (pbuf->nsamp * pbuf->nchan)) {
+            for (int j = 0; j < pbuf->nchan; j++)
+               *out++ = pbuf->data[pbuf->idata++];
+        }
+        else {
+            for (int j = 0; j < pbuf->nchan; j++)
+               *out++ = 0;
+            ret = paComplete;
+            //std::cerr << "</callback>" << std::endl;
+            //return ret;
+        }
+    }
+    std::cerr << "idata=" << pbuf->idata << " fpb=" << framesPerBuffer << "  nsamp=" << pbuf->nsamp << std::flush;
+    std::cerr << "</callback>" << std::endl;
+    return ret;
+}
+
+//
+// This routine is called by portaudio when playback is done.
+//
+void zmq_audiobuffer::stream_finished(void *userData) {
+    zmq_audiobuffer *pbuf = (zmq_audiobuffer *)userData;
+    pbuf->playing = false;
+    zmq_audiobuffer::_playing_now = nullptr;
+    printf("Stream Completed: %s\n", pbuf->message);
+}
+
+int zmq_audiobuffer::play() {
+
+    if (zmq_audiobuffer * p = zmq_audiobuffer::_playing_now) {
+        std::cerr
+          << "warning: zmq_audiobuffer::play: a buffer is already playing, stopping it"
+          << std::endl;
+        p->stop();
+    }
+    std::cerr << "Start!" << std::endl;
+    zmq_audiobuffer::_playing_now = this;
+
+    idata = 0;
+
+    PaError err = Pa_Initialize();
+    if (err != paNoError)
+        return print_error(err);
+
+    PaStreamParameters outputParameters;
+    outputParameters.device =
+        Pa_GetDefaultOutputDevice(); // default output device
+    if (outputParameters.device == paNoDevice) {
+        fprintf(stderr, "Error: No default output device.\n");
+        return paNoDevice;
+    }
+    outputParameters.channelCount = nchan; // stereo output
+    outputParameters.sampleFormat =
+        paFloat32; // 32 bit floating point output
+    outputParameters.suggestedLatency =
+        Pa_GetDeviceInfo(outputParameters.device)->defaultLowOutputLatency;
+    outputParameters.hostApiSpecificStreamInfo = NULL;
+
+    //PaError err;
+    err = Pa_OpenStream(&stream, NULL, // no input
+                        &outputParameters, fsamp, frames_per_buffer,
+                        paClipOff, // we won't output out of range samples so
+                                   // don't bother clipping them
+                        zmq_audiobuffer::stream_callback, this);
+    sprintf(message, "No Message");
+    err =
+        Pa_SetStreamFinishedCallback(stream, &zmq_audiobuffer::stream_finished);
+    if (err != paNoError) {
+        print_error(err);
+        return err;
+    }
+
+    err = Pa_StartStream(stream);
+    if (err != paNoError) {
+        print_error(err);
+        return err;
+    }
+    
+    playing = true;
+
+    //printf("Play for %d seconds.\n", NUM_SECONDS);
+    //Pa_Sleep((unsigned int)(duration() * 1000));
+    return paNoError;
+}
+
+void zmq_audiobuffer::abort() {
+    std::cerr << "Abort!" << std::endl;
+    Pa_AbortStream(stream);
+    Pa_CloseStream(stream);
+}
+
+void zmq_audiobuffer::stop() {
+    std::cerr << "Stop!" << std::endl;
+    Pa_StopStream(stream);
+    Pa_CloseStream(stream);
+    Pa_Terminate();
+}
+
+int zmq_audiobuffer::fill() {
+    // initialise sinusoidal wavetable
+    const float fl{440};
+    const float fr{880};
+    const float vol{0.1};
+    for (int i = 0; i < nsamp - 1; i++) {
+        data[2 * i] = vol * sin((2 * M_PI * fl * i) / fsamp);
+        data[2 * i + 1] = vol * sin((2 * M_PI * fr * i) / fsamp);
+    }
+    return 0;
+}
+
+int zmq_audiobuffer::fill(zmq::message_t &msg) {
+    uint32_t ndata = msg.size()/sizeof(float);
+    if (ndata != (nsamp*nchan)) {
+        std::cerr <<
+           "in int zmq_audiobuffer::fill(zmq::message_t &msg): audio data size mismatch" <<
+           std::endl <<
+           "ndata=" << ndata << std::endl <<
+           "nsamp=" << nsamp << ", nchan=" << nchan << " total=" << nsamp*nchan <<
+           std::endl;
+        return 1;
+    }
+    std::memcpy(data, msg.data(), msg.size());
+    return 0;
+}
+
+void zmq_audiobuffer::dump(int nsamples)
+{
+    std::cerr << "in void zmq_audiobuffer::dump(int nsamples):" << std::endl;
+    std::cerr << "fsamp=" << fsamp << std::endl;
+    std::cerr << "nsamp=" << nsamp << std::endl;
+    std::cerr << "nchan=" << nchan << std::endl;
+    std::cerr << "frames_per_buffer=" << frames_per_buffer << std::endl;
+    std::cerr << "idata=" << idata << std::endl;
+    for (int i=0; i<nsamples; i++) {
+        std::cerr << "data[" << i << "]=" << data[i] << std::endl;
+    }
+}
+
+
+int zmq_recv_multi(zmq::socket_t &socket, zmq::message_t parts[], int nmax) {
+    int64_t more;
+    size_t more_size = sizeof more;
+    int n = 0;
+    try {
+        do {
+            if (n < nmax) {
+                socket.recv(&parts[n]);
+                if (verbose)
+                    std::cout << "recv part " << n << std::endl;
+                n++;
+            }
+            else {
+                zmq::message_t discard;
+                socket.recv(&discard);
+                if (verbose)
+                    std::cout << "recv part " << n << " (discarded)."
+                              << std::endl;
+            }
+            socket.getsockopt(ZMQ_RCVMORE, &more, &more_size);
+        } while (more);
+    }
+    catch (std::exception &error) {
+        std::cout << "zmq_rev_multi while receiving frame " << n << ": "
+                  << error.what() << std::endl;
+    }
+    return n;
+}
+
+int rtpriority(int n) {
+return 0;
+    int sched_policy = SCHED_FIFO;
+    struct sched_param sched;
+
+    memset(&sched, 0, sizeof(sched));
+
+    if (n > sched_get_priority_max(sched_policy))
+        sched.sched_priority = sched_get_priority_max(sched_policy);
+    else
+        sched.sched_priority = n;
+
+    return sched_setscheduler(0, sched_policy, &sched);
+}
+
+int main(void) {
+
+    int prio1 = 40;
+    int prio2 = 50;
+
+    rtpriority(prio2);
+    //PaError err = Pa_Initialize();
+    //if (err != paNoError)
+    //    return print_error(err);
+
+    rtpriority(prio1);
+    //  Prepare our context and socket
+    zmq::context_t context(1);
+
+    zmq::socket_t ssub(context, ZMQ_SUB);
+    ssub.connect("tcp://raspi6.local:5557");
+    //ssub.connect("tcp://lsldert00.local:5557");
+    ssub.setsockopt(ZMQ_SUBSCRIBE, "A", 1);
+
+    rtpriority(prio2);
+    buffer[0] = new zmq_audiobuffer(44100, 5 * 48000, 2);
+    buffer[0]->fill();
+
+    int run = 1;
+    while (run) {
+        const int maxmsg = 3;
+        zmq::message_t msg[maxmsg];
+        int nmsg = 0;
+
+        nmsg = zmq_recv_multi(ssub, msg, maxmsg);
+        zmq::message_t &request = msg[0];
+        std::string str((char *)request.data());
+
+        std::uint32_t *audio_header = nullptr;
+        if (nmsg > 2) {
+            audio_header = new std::uint32_t[msg[1].size()/sizeof(std::uint32_t)];
+            std::memcpy(audio_header, msg[1].data(), msg[1].size());
+        }
+
+        if (verbose) {
+            std::cout << "Received: '" << str << "' (" << request.size()
+                      << " bytes)" << std::endl;
+            if (nmsg > 1) {
+                for (int i = 1; i < nmsg; i++)
+                    std::cout << "  ...one more message, " << msg[i].size()
+                              << " bytes" << std::endl;
+            }
+        }
+
+        // parse the command and data in the message
+        std::istringstream s(str);
+        std::string cmd;
+        unsigned int ibuf;
+        s >> cmd >> ibuf;
+
+        if (verbose)
+            std::cout << "cmd='" << cmd << "' ibuf=" << ibuf << std::endl;
+
+        if (ibuf >= nrbuffers) {
+            std::cerr << "buffer index out of range" << std::endl;
+            continue;
+        }
+
+        if (cmd == "AF") { // Fill
+            delete buffer[ibuf];
+            
+            if (audio_header) {
+               std::uint32_t Fs = audio_header[0];
+               std::uint32_t nchan = audio_header[1];
+               std::uint32_t nsamp = audio_header[2];
+               buffer[ibuf] = new zmq_audiobuffer(Fs, nsamp, nchan);
+               buffer[ibuf]->fill(msg[2]);
+               if (verbose)
+                  buffer[ibuf]->dump(50);
+            }
+            else
+               buffer[ibuf]=nullptr;
+        }
+        else if (cmd == "AP") { // Play
+            if (buffer[ibuf]) {
+               PaError err=buffer[ibuf]->play();
+               if (err != paNoError) {
+                   print_error(err);
+                   return err;
+               }
+             }
+        }
+        else if (cmd == "AS") { // Stop
+            if (buffer[ibuf]) {
+               buffer[ibuf]->stop();
+               if (verbose)
+                  buffer[ibuf]->dump(4);
+            }
+        }
+
+        delete[] audio_header;
+        //delete[] audio_data;
+
+        //buffer[0] = new zmq_audiobuffer(44100, 5 * 48000, 2);
+        //zmq_audiobuffer *data = buffer[0];
+        //data->fill();
+        //printf("PortAudio Test: output sine wave. SR = %d, BufSize = %d\n",
+               //SAMPLE_RATE, data->frames_per_buffer);
+
+        //err = data->play();
+        //if (err != paNoError)
+            //return print_error(err);
+    }
+//    Pa_Terminate();
+    printf("Test finished.\n");
+
+    return 0;
+    //return err;
+}

lsldert/zmqaudio_pa.h

+#include "portaudio.h"
+#include <cassert>
+#include <cstdint>
+#include <zmq.hpp>
+
+class zmq_audiobuffer {
+
+  public:
+    std::uint32_t fsamp;
+    std::uint32_t nsamp;
+    std::uint32_t nchan;
+    float *data;
+    std::uint32_t idata;
+    char message[20];
+    int frames_per_buffer;
+
+    zmq_audiobuffer(std::uint32_t _fsamp, std::uint32_t _nsamp,
+                    std::uint32_t _nchan);
+    ~zmq_audiobuffer();
+
+    int fill();
+    int fill(zmq::message_t&);
+    int play();
+    void stop();
+    void abort();
+    double duration() {
+        return double(nsamp)/fsamp;
+    }
+
+    void dump(int nsamples);
+
+    const zmq_audiobuffer *playing_now() {
+       return zmq_audiobuffer::_playing_now;
+    }
+protected:
+    bool playing;  // if this buffer is playing