git.sesse.net Git - casparcg/blob - modules/portaudio/consumer/portaudio_consumer.cpp

   1 /*
   2 * Copyright 2013 Sveriges Television AB http://casparcg.com/
   3 *
   4 * This file is part of CasparCG (www.casparcg.com).
   5 *
   6 * CasparCG is free software: you can redistribute it and/or modify
   7 * it under the terms of the GNU General Public License as published by
   8 * the Free Software Foundation, either version 3 of the License, or
   9 * (at your option) any later version.
  10 *
  11 * CasparCG is distributed in the hope that it will be useful,
  12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 * GNU General Public License for more details.
  15 *
  16 * You should have received a copy of the GNU General Public License
  17 * along with CasparCG. If not, see <http://www.gnu.org/licenses/>.
  18 *
  19 * Author: Helge Norberg, helge.norberg@svt.se
  20 */
  21
  22 #include "portaudio_consumer.h"
  23
  24 #include <cmath>
  25
  26 #include <tbb/atomic.h>
  27 #include <tbb/concurrent_queue.h>
  28
  29 #include <boost/timer.hpp>
  30
  31 #include <portaudio.h>
  32
  33 #include <common/log/log.h>
  34 #include <common/exception/exceptions.h>
  35 #include <common/concurrency/future_util.h>
  36 #include <common/diagnostics/graph.h>
  37
  38 #include <core/consumer/frame_consumer.h>
  39 #include <core/mixer/audio/audio_util.h>
  40 #include <core/video_format.h>
  41 #include <core/parameters/parameters.h>
  42 #include <core/mixer/read_frame.h>
  43
  44 namespace caspar { namespace portaudio {
  45
  46 #define PA_CHECK(err) if (err != paNoError) \
  47         BOOST_THROW_EXCEPTION(caspar_exception() \
  48                         << msg_info(std::string("PortAudio error: ") \
  49                         + Pa_GetErrorText(err)))
  50
  51 typedef std::vector<int16_t, tbb::cache_aligned_allocator<int16_t>> audio_buffer_16;
  52
  53 int callback(
  54                 const void*, // input
  55                 void* output,
  56                 unsigned long sample_frames_per_buffer,
  57                 const PaStreamCallbackTimeInfo* time_info,
  58                 PaStreamCallbackFlags status_flags,
  59                 void* user_data);
  60
  61 struct portaudio_initializer
  62 {
  63         portaudio_initializer()
  64         {
  65                 PA_CHECK(Pa_Initialize());
  66         }
  67
  68         ~portaudio_initializer()
  69         {
  70                 try
  71                 {
  72                         PA_CHECK(Pa_Terminate());
  73                 }
  74                 catch (...)
  75                 {
  76                         CASPAR_LOG_CURRENT_EXCEPTION();
  77                 }
  78         }
  79 };
  80
  81 struct portaudio_consumer : public core::frame_consumer
  82 {
  83         safe_ptr<diagnostics::graph>                                                                    graph_;
  84         boost::timer                                                                                                    tick_timer_;
  85         int                                                                                                                             channel_index_;
  86
  87         tbb::atomic<bool>                                                                                               started_;
  88         tbb::concurrent_bounded_queue<std::shared_ptr<audio_buffer_16>> frames_in_queue_;
  89         std::pair<std::shared_ptr<audio_buffer_16>, unsigned int>               frame_being_consumed_;
  90         std::shared_ptr<audio_buffer_16>                                                                to_deallocate_in_output_thread_;
  91
  92         core::video_format_desc                                                                                 format_desc_;
  93         core::channel_layout                                                                                    channel_layout_;
  94
  95         std::shared_ptr<PaStream>                                                                               stream_;
  96 public:
  97         portaudio_consumer()
  98                 : channel_index_(-1)
  99                 , channel_layout_(
 100                                 core::default_channel_layout_repository().get_by_name(L"STEREO"))
 101         {
 102                 started_ = false;
 103                 frames_in_queue_.set_capacity(2);
 104
 105                 graph_->set_color("dropped-frame", diagnostics::color(0.3f, 0.6f, 0.3f));
 106                 graph_->set_color("late-frame", diagnostics::color(0.6f, 0.3f, 0.3f));
 107
 108                 diagnostics::register_graph(graph_);
 109         }
 110
 111         bool promote_frame()
 112         {
 113                 to_deallocate_in_output_thread_.swap(frame_being_consumed_.first);
 114                 frame_being_consumed_.first.reset();
 115                 frame_being_consumed_.second = 0;
 116
 117                 return frames_in_queue_.try_pop(frame_being_consumed_.first);
 118         }
 119
 120         void write_samples(
 121                         int16_t* output,
 122                         unsigned int sample_frames_per_buffer,
 123                         PaStreamCallbackFlags status_flags)
 124         {
 125                 if (!frame_being_consumed_.first)
 126                 {
 127                         promote_frame();
 128                 }
 129
 130                 auto needed = sample_frames_per_buffer * channel_layout_.num_channels;
 131
 132                 while (needed > 0 && frame_being_consumed_.first)
 133                 {
 134                         auto start_index = frame_being_consumed_.second;
 135                         auto available = frame_being_consumed_.first->size() - start_index;
 136                         auto to_provide = std::min(needed, available);
 137
 138                         std::memcpy(
 139                                         output,
 140                                         frame_being_consumed_.first->data() + start_index,
 141                                         to_provide * sizeof(int16_t));
 142
 143                         output += to_provide;
 144                         needed -= to_provide;
 145                         available -= to_provide;
 146                         frame_being_consumed_.second += to_provide;
 147
 148                         if (available == 0 && !promote_frame())
 149                                 break;
 150                 }
 151
 152                 if (needed > 0)
 153                 {
 154                         std::memset(output, 0, needed * sizeof(int16_t));
 155
 156                         CASPAR_LOG(trace) << print() << L"late-frame: Inserted "
 157                                         << needed << L" zero-samples";
 158                         graph_->set_tag("late-frame");
 159                 }
 160         }
 161
 162         ~portaudio_consumer()
 163         {
 164                 if (stream_ && started_)
 165                         PA_CHECK(Pa_StopStream(stream_.get()));
 166
 167                 CASPAR_LOG(info) << print() << L" Successfully Uninitialized.";
 168         }
 169
 170         virtual void initialize(const core::video_format_desc& format_desc, int channel_index) override
 171         {
 172                 format_desc_    = format_desc;
 173                 channel_index_  = channel_index;
 174                 graph_->set_text(print());
 175
 176                 static portaudio_initializer init;
 177
 178                 auto device_index = Pa_GetDefaultOutputDevice();
 179
 180                 if (device_index == paNoDevice)
 181                         BOOST_THROW_EXCEPTION(caspar_exception() << msg_info("No port audio device detected"));
 182
 183                 auto device = Pa_GetDeviceInfo(device_index);
 184
 185                 PaStreamParameters parameters;
 186                 parameters.channelCount = channel_layout_.num_channels;
 187                 parameters.device = device_index;
 188                 parameters.hostApiSpecificStreamInfo = nullptr;
 189                 parameters.sampleFormat = paInt16;
 190                 parameters.suggestedLatency = device->defaultLowOutputLatency;
 191
 192                 PaStream* stream;
 193
 194                 PA_CHECK(Pa_OpenStream(
 195                                 &stream,
 196                                 nullptr, // input config
 197                                 &parameters,
 198                                 format_desc.audio_sample_rate,
 199                                 *std::min_element(format_desc.audio_cadence.begin(), format_desc.audio_cadence.end()),
 200                                 0,
 201                                 callback,
 202                                 this));
 203                 stream_.reset(stream, [](PaStream* s) { Pa_CloseStream(s); });
 204
 205                 CASPAR_LOG(info) << print() << " Sucessfully Initialized.";
 206         }
 207
 208         virtual int64_t presentation_frame_age_millis() const override
 209         {
 210                 auto info = Pa_GetStreamInfo(stream_.get());
 211
 212                 if (info == nullptr)
 213                         return 0;
 214
 215                 return static_cast<int64_t>(info->outputLatency * 1000.0);
 216         }
 217
 218         virtual boost::unique_future<bool> send(const safe_ptr<core::read_frame>& frame) override
 219         {
 220                 to_deallocate_in_output_thread_.reset();
 221                 std::shared_ptr<audio_buffer_16> buffer;
 222
 223                 if (core::needs_rearranging(
 224                                 frame->multichannel_view(),
 225                                 channel_layout_,
 226                                 channel_layout_.num_channels))
 227                 {
 228                         core::audio_buffer downmixed;
 229                         downmixed.resize(
 230                                         frame->multichannel_view().num_samples()
 231                                                         * channel_layout_.num_channels,
 232                                         0);
 233
 234                         auto dest_view = core::make_multichannel_view<int32_t>(
 235                                         downmixed.begin(), downmixed.end(), channel_layout_);
 236
 237                         core::rearrange_or_rearrange_and_mix(
 238                                         frame->multichannel_view(),
 239                                         dest_view,
 240                                         core::default_mix_config_repository());
 241
 242                         buffer = std::make_shared<audio_buffer_16>(
 243                                         core::audio_32_to_16(downmixed));
 244                 }
 245                 else
 246                 {
 247                         buffer = std::make_shared<audio_buffer_16>(
 248                                         core::audio_32_to_16(frame->audio_data()));
 249                 }
 250
 251                 if (!frames_in_queue_.try_push(buffer))
 252                         graph_->set_tag("dropped-frame");
 253
 254                 if (!started_)
 255                 {
 256                         PA_CHECK(Pa_StartStream(stream_.get()));
 257                         started_ = true;
 258                 }
 259
 260                 return wrap_as_future(true);
 261         }
 262
 263         virtual std::wstring print() const override
 264         {
 265                 return L"portaudio[" + boost::lexical_cast<std::wstring>(channel_index_) + L"|" + format_desc_.name + L"]";
 266         }
 267
 268         virtual bool has_synchronization_clock() const override
 269         {
 270                 return false;
 271         }
 272
 273         virtual boost::property_tree::wptree info() const override
 274         {
 275                 boost::property_tree::wptree info;
 276                 info.add(L"type", L"portaudio-consumer");
 277                 return info;
 278         }
 279
 280         virtual size_t buffer_depth() const override
 281         {
 282                 return 3;
 283         }
 284
 285         virtual int index() const override
 286         {
 287                 return 530;
 288         }
 289 };
 290
 291 int callback(
 292                 const void*, // input
 293                 void* output,
 294                 unsigned long sample_frames_per_buffer,
 295                 const PaStreamCallbackTimeInfo* time_info,
 296                 PaStreamCallbackFlags status_flags,
 297                 void* user_data)
 298 {
 299         auto consumer = static_cast<portaudio_consumer*>(user_data);
 300
 301         consumer->write_samples(
 302                         static_cast<int16_t*>(output),
 303                         sample_frames_per_buffer,
 304                         status_flags);
 305
 306         return 0;
 307 }
 308
 309 safe_ptr<core::frame_consumer> create_consumer(const core::parameters& params)
 310 {
 311         if(params.size() < 1 || params[0] != L"AUDIO")
 312                 return core::frame_consumer::empty();
 313
 314         return make_safe<portaudio_consumer>();
 315 }
 316
 317 safe_ptr<core::frame_consumer> create_consumer()
 318 {
 319         return make_safe<portaudio_consumer>();
 320 }
 321
 322 }}