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/exception/win32_exception.h>
  36 #include <common/concurrency/future_util.h>
  37 #include <common/diagnostics/graph.h>
  38
  39 #include <core/consumer/frame_consumer.h>
  40 #include <core/mixer/audio/audio_util.h>
  41 #include <core/video_format.h>
  42 #include <core/parameters/parameters.h>
  43 #include <core/mixer/read_frame.h>
  44
  45 namespace caspar { namespace portaudio {
  46
  47 #define PA_CHECK(err) if (err != paNoError) \
  48         BOOST_THROW_EXCEPTION(caspar_exception() \
  49                         << msg_info(std::string("PortAudio error: ") \
  50                         + Pa_GetErrorText(err)))
  51
  52 typedef std::vector<int16_t, tbb::cache_aligned_allocator<int16_t>> audio_buffer_16;
  53
  54 int callback(
  55                 const void*, // input
  56                 void* output,
  57                 unsigned long sample_frames_per_buffer,
  58                 const PaStreamCallbackTimeInfo* time_info,
  59                 PaStreamCallbackFlags status_flags,
  60                 void* user_data);
  61
  62 struct portaudio_initializer
  63 {
  64         portaudio_initializer()
  65         {
  66                 PA_CHECK(Pa_Initialize());
  67         }
  68
  69         ~portaudio_initializer()
  70         {
  71                 try
  72                 {
  73                         PA_CHECK(Pa_Terminate());
  74                 }
  75                 catch (...)
  76                 {
  77                         CASPAR_LOG_CURRENT_EXCEPTION();
  78                 }
  79         }
  80 };
  81
  82 struct portaudio_consumer : public core::frame_consumer
  83 {
  84         safe_ptr<diagnostics::graph>                                                                    graph_;
  85         boost::timer                                                                                                    tick_timer_;
  86         int                                                                                                                             channel_index_;
  87
  88         tbb::atomic<bool>                                                                                               started_;
  89         tbb::concurrent_bounded_queue<std::shared_ptr<audio_buffer_16>> frames_in_queue_;
  90         std::pair<std::shared_ptr<audio_buffer_16>, unsigned int>               frame_being_consumed_;
  91         std::shared_ptr<audio_buffer_16>                                                                to_deallocate_in_output_thread_;
  92
  93         core::video_format_desc                                                                                 format_desc_;
  94         core::channel_layout                                                                                    channel_layout_;
  95
  96         std::shared_ptr<PaStream>                                                                               stream_;
  97 public:
  98         portaudio_consumer()
  99                 : channel_index_(-1)
 100                 , channel_layout_(
 101                                 core::default_channel_layout_repository().get_by_name(L"STEREO"))
 102         {
 103                 started_ = false;
 104                 frames_in_queue_.set_capacity(2);
 105
 106                 graph_->set_color("dropped-frame", diagnostics::color(0.3f, 0.6f, 0.3f));
 107                 graph_->set_color("late-frame", diagnostics::color(0.6f, 0.3f, 0.3f));
 108
 109                 diagnostics::register_graph(graph_);
 110         }
 111
 112         bool promote_frame()
 113         {
 114                 to_deallocate_in_output_thread_.swap(frame_being_consumed_.first);
 115                 frame_being_consumed_.first.reset();
 116                 frame_being_consumed_.second = 0;
 117
 118                 return frames_in_queue_.try_pop(frame_being_consumed_.first);
 119         }
 120
 121         void write_samples(
 122                         int16_t* output,
 123                         unsigned int sample_frames_per_buffer,
 124                         PaStreamCallbackFlags status_flags)
 125         {
 126                 if (!frame_being_consumed_.first)
 127                 {
 128                         promote_frame();
 129                 }
 130
 131                 auto needed = sample_frames_per_buffer * channel_layout_.num_channels;
 132
 133                 while (needed > 0 && frame_being_consumed_.first)
 134                 {
 135                         auto start_index = frame_being_consumed_.second;
 136                         auto available = frame_being_consumed_.first->size() - start_index;
 137                         auto to_provide = std::min(needed, available);
 138
 139                         std::memcpy(
 140                                         output,
 141                                         frame_being_consumed_.first->data() + start_index,
 142                                         to_provide * sizeof(int16_t));
 143
 144                         output += to_provide;
 145                         needed -= to_provide;
 146                         available -= to_provide;
 147                         frame_being_consumed_.second += to_provide;
 148
 149                         if (available == 0 && !promote_frame())
 150                                 break;
 151                 }
 152
 153                 if (needed > 0)
 154                 {
 155                         std::memset(output, 0, needed * sizeof(int16_t));
 156
 157                         CASPAR_LOG(trace) << print() << L"late-frame: Inserted "
 158                                         << needed << L" zero-samples";
 159                         graph_->set_tag("late-frame");
 160                 }
 161         }
 162
 163         ~portaudio_consumer()
 164         {
 165                 if (stream_ && started_)
 166                         PA_CHECK(Pa_StopStream(stream_.get()));
 167
 168                 CASPAR_LOG(info) << print() << L" Successfully Uninitialized.";
 169         }
 170
 171         virtual void initialize(const core::video_format_desc& format_desc, int channel_index) override
 172         {
 173                 format_desc_    = format_desc;
 174                 channel_index_  = channel_index;
 175                 graph_->set_text(print());
 176
 177                 static portaudio_initializer init;
 178
 179                 auto device_index = Pa_GetDefaultOutputDevice();
 180
 181                 if (device_index == paNoDevice)
 182                         BOOST_THROW_EXCEPTION(caspar_exception() << msg_info("No port audio device detected"));
 183
 184                 auto device = Pa_GetDeviceInfo(device_index);
 185
 186                 PaStreamParameters parameters;
 187                 parameters.channelCount = channel_layout_.num_channels;
 188                 parameters.device = device_index;
 189                 parameters.hostApiSpecificStreamInfo = nullptr;
 190                 parameters.sampleFormat = paInt16;
 191                 parameters.suggestedLatency = device->defaultLowOutputLatency;
 192
 193                 PaStream* stream;
 194
 195                 PA_CHECK(Pa_OpenStream(
 196                                 &stream,
 197                                 nullptr, // input config
 198                                 &parameters,
 199                                 format_desc.audio_sample_rate,
 200                                 *std::min_element(format_desc.audio_cadence.begin(), format_desc.audio_cadence.end()),
 201                                 0,
 202                                 callback,
 203                                 this));
 204                 stream_.reset(stream, [](PaStream* s) { Pa_CloseStream(s); });
 205
 206                 CASPAR_LOG(info) << print() << " Sucessfully Initialized.";
 207         }
 208
 209         virtual int64_t presentation_frame_age_millis() const override
 210         {
 211                 auto info = Pa_GetStreamInfo(stream_.get());
 212
 213                 if (info == nullptr)
 214                         return 0;
 215
 216                 return static_cast<int64_t>(info->outputLatency * 1000.0);
 217         }
 218
 219         virtual boost::unique_future<bool> send(const safe_ptr<core::read_frame>& frame) override
 220         {
 221                 to_deallocate_in_output_thread_.reset();
 222                 std::shared_ptr<audio_buffer_16> buffer;
 223
 224                 if (core::needs_rearranging(
 225                                 frame->multichannel_view(),
 226                                 channel_layout_,
 227                                 channel_layout_.num_channels))
 228                 {
 229                         core::audio_buffer downmixed;
 230                         downmixed.resize(
 231                                         frame->multichannel_view().num_samples()
 232                                                         * channel_layout_.num_channels,
 233                                         0);
 234
 235                         auto dest_view = core::make_multichannel_view<int32_t>(
 236                                         downmixed.begin(), downmixed.end(), channel_layout_);
 237
 238                         core::rearrange_or_rearrange_and_mix(
 239                                         frame->multichannel_view(),
 240                                         dest_view,
 241                                         core::default_mix_config_repository());
 242
 243                         buffer = std::make_shared<audio_buffer_16>(
 244                                         core::audio_32_to_16(downmixed));
 245                 }
 246                 else
 247                 {
 248                         buffer = std::make_shared<audio_buffer_16>(
 249                                         core::audio_32_to_16(frame->audio_data()));
 250                 }
 251
 252                 if (!frames_in_queue_.try_push(buffer))
 253                         graph_->set_tag("dropped-frame");
 254
 255                 if (!started_)
 256                 {
 257                         PA_CHECK(Pa_StartStream(stream_.get()));
 258                         started_ = true;
 259                 }
 260
 261                 return wrap_as_future(true);
 262         }
 263
 264         virtual std::wstring print() const override
 265         {
 266                 return L"portaudio[" + boost::lexical_cast<std::wstring>(channel_index_) + L"|" + format_desc_.name + L"]";
 267         }
 268
 269         virtual bool has_synchronization_clock() const override
 270         {
 271                 return false;
 272         }
 273
 274         virtual boost::property_tree::wptree info() const override
 275         {
 276                 boost::property_tree::wptree info;
 277                 info.add(L"type", L"portaudio-consumer");
 278                 return info;
 279         }
 280
 281         virtual size_t buffer_depth() const override
 282         {
 283                 return 3;
 284         }
 285
 286         virtual int index() const override
 287         {
 288                 return 530;
 289         }
 290 };
 291
 292 int callback(
 293                 const void*, // input
 294                 void* output,
 295                 unsigned long sample_frames_per_buffer,
 296                 const PaStreamCallbackTimeInfo* time_info,
 297                 PaStreamCallbackFlags status_flags,
 298                 void* user_data)
 299 {
 300         win32_exception::ensure_handler_installed_for_thread(
 301                         "portaudio-callback-thread");
 302         auto consumer = static_cast<portaudio_consumer*>(user_data);
 303
 304         consumer->write_samples(
 305                         static_cast<int16_t*>(output),
 306                         sample_frames_per_buffer,
 307                         status_flags);
 308
 309         return 0;
 310 }
 311
 312 safe_ptr<core::frame_consumer> create_consumer(const core::parameters& params)
 313 {
 314         if(params.size() < 1 || params[0] != L"AUDIO")
 315                 return core::frame_consumer::empty();
 316
 317         return make_safe<portaudio_consumer>();
 318 }
 319
 320 safe_ptr<core::frame_consumer> create_consumer()
 321 {
 322         return make_safe<portaudio_consumer>();
 323 }
 324
 325 }}