process()

void MayaFlux::Kakshya::AudioOutputAccessProcessor::process ( const std::shared_ptr< SignalSourceContainer > &  container )

overridevirtual

Pull engine snapshot, write m_processed_data, append to m_data.

No-op if the backend service is unavailable or the snapshot is empty. Logs a trace-level message in those cases rather than erroring, since the first few calls during engine startup may arrive before any cycle has completed.

Implements MayaFlux::Kakshya::DataProcessor.

Definition at line 55 of file AudioOutputAccessProcessor.cpp.

{
    if (m_is_processing.exchange(true, std::memory_order_acq_rel))
        return;
 
    auto ac = std::dynamic_pointer_cast<AudioOutputContainer>(container);
    if (!ac) {
        MF_RT_ERROR(Journal::Component::Kakshya, Journal::Context::ContainerProcessing,
            "AudioOutputAccessProcessor requires an AudioOutputContainer");
        m_is_processing.store(false, std::memory_order_release);
        return;
    }
 
    if (!m_backend_service) {
        MF_RT_ERROR(Journal::Component::Kakshya, Journal::Context::ContainerProcessing,
            "AudioBackendService unavailable");
        m_is_processing.store(false, std::memory_order_release);
        return;
    }
 
    const auto snap = m_backend_service->get_output_snapshot();
    if (snap.empty()) {
        MF_RT_WARN(Journal::Component::Kakshya, Journal::Context::ContainerProcessing,
            "AudioOutputAccessProcessor: snapshot empty, no cycle completed yet");
        m_is_processing.store(false, std::memory_order_release);
        return;
    }
 
    ac->update_processing_state(ProcessingState::PROCESSING);
 
    thread_local std::vector<std::vector<double>> tl_channels;
 
    tl_channels.resize(m_channel_count);
    for (auto& ch : tl_channels)
        ch.resize(m_buffer_size);
 
    if (m_organization == OrganizationStrategy::PLANAR) {
        for (uint32_t ch = 0; ch < m_channel_count; ++ch) {
            for (uint32_t f = 0; f < m_buffer_size; ++f) {
                tl_channels[ch][f] = snap[f * m_channel_count + ch];
            }
        }
    } else {
        tl_channels[0].assign(snap.begin(), snap.end());
    }
 
    {
 
        std::unique_lock lock(ac->m_data_mutex);
        auto& pd = ac->get_processed_data();
        if (m_organization == OrganizationStrategy::PLANAR) {
            pd.resize(m_channel_count);
            for (uint32_t ch = 0; ch < m_channel_count; ++ch)
                pd[ch] = DataVariant(tl_channels[ch]);
        } else {
            pd.resize(1);
            pd[0] = DataVariant(tl_channels[0]);
        }
    }
 
    const uint64_t write_head = ac->get_num_frames();
 
    {
        std::unique_lock lock(ac->m_data_mutex);
        if (m_organization == OrganizationStrategy::PLANAR) {
            if (ac->m_data.size() < m_channel_count)
                ac->m_data.resize(m_channel_count, DataVariant(std::vector<double> {}));
            for (uint32_t ch = 0; ch < m_channel_count; ++ch) {
                auto& vec = std::get<std::vector<double>>(ac->m_data[ch]);
                vec.insert(vec.end(), tl_channels[ch].begin(), tl_channels[ch].end());
            }
        } else {
            if (ac->m_data.empty())
                ac->m_data.resize(1, DataVariant(std::vector<double> {}));
            auto& vec = std::get<std::vector<double>>(ac->m_data[0]);
            vec.insert(vec.end(), tl_channels[0].begin(), tl_channels[0].end());
        }
        ac->m_num_frames += m_buffer_size;
        ac->setup_dimensions();
        ac->invalidate_span_cache();
        ac->m_double_extraction_dirty.store(true, std::memory_order_release);
    }
 
    ac->update_processing_state(ProcessingState::PROCESSED);
    m_is_processing.store(false, std::memory_order_release);
}

References MayaFlux::Journal::ContainerProcessing, MayaFlux::Registry::Service::AudioBackendService::get_output_snapshot, MayaFlux::Journal::Kakshya, m_backend_service, m_buffer_size, m_channel_count, m_is_processing, m_organization, MF_RT_ERROR, MF_RT_WARN, MayaFlux::Kakshya::PLANAR, MayaFlux::Kakshya::PROCESSED, and MayaFlux::Kakshya::PROCESSING.