process()

void ContiguousAccessProcessor::process ( const std::shared_ptr< SignalSourceContainer > &  container )

overridevirtual

Process the current region or block of data.

Advances the read position if auto-advance is enabled.

Parameters

container

The SignalSourceContainer to process.

Implements MayaFlux::Kakshya::DataProcessor.

Definition at line 133 of file ContiguousAccessProcessor.cpp.

{
    if (!m_prepared) {
        MF_RT_ERROR(Journal::Component::Kakshya, Journal::Context::ContainerProcessing,
            "ContiguousAccessProcessor not prepared for processing");
        return;
    }
 
    auto source_container = m_source_container_weak.lock();
    if (!source_container || source_container.get() != container.get()) {
        MF_RT_ERROR(Journal::Component::Kakshya, Journal::Context::ContainerProcessing,
            "ContiguousAccessProcessor: Source container mismatch or expired");
        return;
    }
 
    m_is_processing = true;
    m_last_process_time = std::chrono::steady_clock::now();
 
    try {
        uint64_t min_frame = *std::ranges::min_element(m_current_position);
        std::vector<uint64_t> region_coords = { min_frame, 0 };
 
        Region output_region = calculate_output_region(region_coords, m_output_shape);
 
        auto region_data = container->get_region_data(output_region);
        auto& processed_data_vector = container->get_processed_data();
 
        if (m_structure.organization == OrganizationStrategy::INTERLEAVED) {
            processed_data_vector.resize(1);
            if (!region_data.empty()) {
                safe_copy_data_variant(region_data[0], processed_data_vector[0]);
            }
        } else {
            uint64_t channels_to_process = std::min(m_output_shape[1], static_cast<uint64_t>(region_data.size()));
            processed_data_vector.resize(channels_to_process);
 
            for (size_t ch = 0; ch < channels_to_process; ++ch) {
                safe_copy_data_variant(region_data[ch], processed_data_vector[ch]);
            }
        }
 
        if (m_auto_advance) {
            uint64_t frames_to_advance = m_output_shape[0];
 
            m_current_position = advance_position(
                m_current_position,
                frames_to_advance,
                m_structure,
                m_looping_enabled,
                m_loop_region);
 
            if (auto stream = std::dynamic_pointer_cast<StreamContainer>(container)) {
                stream->set_read_position(m_current_position);
            }
        }
 
        container->update_processing_state(ProcessingState::PROCESSED);
    } catch (const std::exception& e) {
        MF_RT_ERROR(Journal::Component::Kakshya, Journal::Context::ContainerProcessing,
            "Error during ContiguousAccessProcessor processing: {}",
            e.what());
        container->update_processing_state(ProcessingState::ERROR);
    }
 
    m_is_processing = false;
}

References MayaFlux::Kakshya::advance_position(), MayaFlux::Kakshya::calculate_output_region(), MayaFlux::Journal::ContainerProcessing, MayaFlux::Kakshya::ERROR, MayaFlux::Kakshya::INTERLEAVED, MayaFlux::Journal::Kakshya, m_auto_advance, m_current_position, m_is_processing, m_last_process_time, m_loop_region, m_looping_enabled, m_output_shape, m_prepared, m_source_container_weak, m_structure, MF_RT_ERROR, MayaFlux::Kakshya::ContainerDataStructure::organization, MayaFlux::Kakshya::PROCESSED, and MayaFlux::Kakshya::safe_copy_data_variant().

Here is the call graph for this function: