<commit_before>// License: Apache 2.0. See LICENSE file in root directory.
// Copyright(c) 2016 Intel Corporation. All Rights Reserved.

#include <stddef.h>
#include <assert.h>
#include <tuple>
#include "disk_write.h"
#include "include/file.h"
#include "rs_sdk_version.h"
#include "rs/utils/log_utils.h"

using namespace rs::core;

namespace rs
{
    namespace record
    {
        static const uint32_t MAX_MEMORY_CONSUMPTION_PER_STREAM = 100e6;

        disk_write::disk_write(void):
            m_is_configured(false),
            m_paused(false),
            m_stop_writing(true),
            m_min_fps(0)
        {

        }

        disk_write::~disk_write(void)
        {
            stop();
        }

        uint32_t disk_write::get_min_fps(const std::map<rs_stream, core::file_types::stream_profile>& stream_profiles)
        {
            uint32_t rv = 0xffffffff;
            for(auto it = stream_profiles.begin(); it != stream_profiles.end(); ++it)
            {
                if(static_cast<uint32_t>(it->second.info.framerate) < rv)
                    rv = it->second.info.framerate;
            }
            if(rv == 0)
                throw std::runtime_error("illegal frame rate value");
            if(rv == 0xffffffff)
                throw std::runtime_error("no streams were enabled before start streaming");
            return rv;
        }

        bool disk_write::allow_sample(std::shared_ptr<rs::core::file_types::sample> &sample)
        {
            if(sample->info.type != file_types::sample_type::st_image) return true;
            auto frame = std::dynamic_pointer_cast<file_types::frame_sample>(sample);
            if (frame)
            {
                auto size = (double)(frame->finfo.stride * frame->finfo.height);
                auto max_samples = (double)(MAX_MEMORY_CONSUMPTION_PER_STREAM) / (double)size * frame->finfo.framerate / m_min_fps;
                if(m_samples_count[frame->finfo.stream] > max_samples) return false;
                m_samples_count[frame->finfo.stream]++;
                return true;
            }
            return false;
        }

        void disk_write::record_sample(std::shared_ptr<file_types::sample> &sample)
        {
            LOG_FUNC_SCOPE();
            if (m_paused)
            {
                return;//device is still streaming but samples are not recorded
            }
            bool insert_samples = false;
            {
                std::lock_guard<std::mutex> guard(m_main_mutex);
                insert_samples = allow_sample(sample);
                if (insert_samples)//it is ok that sample queue size may exceed MAX_CACHED_SAMPLES by few samples
                {
                    m_samples_queue.push(sample);
                }
                else
                {
                    LOG_WARN("sample drop, sample type - " << sample->info.type << " ,capture time - " << sample->info.capture_time);
                }
            }
            if(insert_samples)
            {
                std::lock_guard<std::mutex> guard(m_notify_write_thread_mutex);
                m_notify_write_thread_cv.notify_one();
            }
        }

        bool disk_write::start()
        {
            LOG_FUNC_SCOPE();
            if(!m_is_configured) return false;
            m_stop_writing = false;//protection is not required before the thread is started
            assert(!m_thread.joinable());//we don't expect the thread to be active on start
            m_thread = std::thread(&disk_write::write_thread, this);
            return true;
        }

        void disk_write::stop()
        {
            LOG_FUNC_SCOPE();

            std::unique_lock<std::mutex> guard(m_main_mutex);
            m_stop_writing = true;
            guard.unlock();

            m_notify_write_thread_cv.notify_one();

            if (m_thread.joinable())
            {
                m_thread.join();
            }

            guard.lock();
            if(m_file)
                m_file->close();
            guard.unlock();
        }

        void disk_write::set_pause(bool pause)
        {
            std::lock_guard<std::mutex> guard(m_main_mutex);
            m_paused = pause;
        }

        status disk_write::configure(const configuration& config)
        {
            std::lock_guard<std::mutex> guard(m_main_mutex);
            if(m_is_configured) return status::status_exec_aborted;
            m_file = std::unique_ptr<rs::core::file>(new rs::core::file());
            status sts = m_file->open(config.m_file_path, (open_file_option)(open_file_option::write));

            if (sts != status::status_no_error)
                throw std::runtime_error("failed to open file for recording, file path - " + config.m_file_path);

            init_encoder(config);
            m_min_fps = get_min_fps(config.m_stream_profiles);
            write_header(static_cast<uint8_t>(config.m_stream_profiles.size()), config.m_coordinate_system, config.m_capture_mode);
            write_camera_info(config.m_camera_info);
            write_sw_info();
            write_capabilities(config.m_capabilities);
            write_motion_intrinsics(config.m_motion_intrinsics);
            write_stream_info(config.m_stream_profiles);
            write_properties(config.m_options);
            write_first_frame_offset();
            m_is_configured = true;
            return sts;
        }

        void disk_write::init_encoder(const configuration& config)
        {
            uint32_t buffer_size = 0;
            m_encoder.reset(new compression::encoder());
            for(auto profile : config.m_stream_profiles)
            {
                rs_stream stream = profile.second.info.stream;
                rs_format format = profile.second.info.format;
                uint32_t size = profile.second.info.width * profile.second.info.height;
                buffer_size = size > buffer_size ? size : buffer_size;
                if(config.m_compression_config.find(profile.first) != (config.m_compression_config.end()))
                {
                    m_encoder->add_codec(stream, format, config.m_compression_config.at(profile.first));
                }
                else
                {
                    m_encoder->add_codec(stream, format, record::compression_level::high);
                }
            }
            m_encoded_data = std::vector<uint8_t>(buffer_size * 4);//stride is not available, taking worst case.
        }

        void disk_write::write_to_file(const void* data, unsigned int numberOfBytesToWrite, unsigned int& numberOfBytesWritten)
        {
            auto sts = m_file->write_bytes(data, numberOfBytesToWrite, numberOfBytesWritten);
            if(sts != status::status_no_error)
            {
                m_file->close();
                LOG_ERROR("failed writing to file");
                throw std::runtime_error("failed writing to file");
            }
        }

        void disk_write::write_thread(void)
        {
            LOG_FUNC_SCOPE();
            while (!m_stop_writing)
            {
                std::unique_lock<std::mutex> guard(m_notify_write_thread_mutex);
                m_notify_write_thread_cv.wait(guard);
                guard.unlock();

                LOG_VERBOSE("queue contains " << m_samples_queue.size() << " samples")
<commit_msg>fix compression level disabled<commit_after>// License: Apache 2.0. See LICENSE file in root directory.
// Copyright(c) 2016 Intel Corporation. All Rights Reserved.

#include <stddef.h>
#include <assert.h>
#include <tuple>
#include "disk_write.h"
#include "include/file.h"
#include "rs_sdk_version.h"
#include "rs/utils/log_utils.h"

using namespace rs::core;

namespace rs
{
    namespace record
    {
        static const uint32_t MAX_MEMORY_CONSUMPTION_PER_STREAM = 100e6;

        disk_write::disk_write(void):
            m_is_configured(false),
            m_paused(false),
            m_stop_writing(true),
            m_min_fps(0)
        {

        }

        disk_write::~disk_write(void)
        {
            stop();
        }

        uint32_t disk_write::get_min_fps(const std::map<rs_stream, core::file_types::stream_profile>& stream_profiles)
        {
            uint32_t rv = 0xffffffff;
            for(auto it = stream_profiles.begin(); it != stream_profiles.end(); ++it)
            {
                if(static_cast<uint32_t>(it->second.info.framerate) < rv)
                    rv = it->second.info.framerate;
            }
            if(rv == 0)
                throw std::runtime_error("illegal frame rate value");
            if(rv == 0xffffffff)
                throw std::runtime_error("no streams were enabled before start streaming");
            return rv;
        }

        bool disk_write::allow_sample(std::shared_ptr<rs::core::file_types::sample> &sample)
        {
            if(sample->info.type != file_types::sample_type::st_image) return true;
            auto frame = std::dynamic_pointer_cast<file_types::frame_sample>(sample);
            if (frame)
            {
                auto size = (double)(frame->finfo.stride * frame->finfo.height);
                auto max_samples = (double)(MAX_MEMORY_CONSUMPTION_PER_STREAM) / (double)size * frame->finfo.framerate / m_min_fps;
                if(m_samples_count[frame->finfo.stream] > max_samples) return false;
                m_samples_count[frame->finfo.stream]++;
                return true;
            }
            return false;
        }

        void disk_write::record_sample(std::shared_ptr<file_types::sample> &sample)
        {
            LOG_FUNC_SCOPE();
            if (m_paused)
            {
                return;//device is still streaming but samples are not recorded
            }
            bool insert_samples = false;
            {
                std::lock_guard<std::mutex> guard(m_main_mutex);
                insert_samples = allow_sample(sample);
                if (insert_samples)//it is ok that sample queue size may exceed MAX_CACHED_SAMPLES by few samples
                {
                    m_samples_queue.push(sample);
                }
                else
                {
                    LOG_WARN("sample drop, sample type - " << sample->info.type << " ,capture time - " << sample->info.capture_time);
                }
            }
            if(insert_samples)
            {
                std::lock_guard<std::mutex> guard(m_notify_write_thread_mutex);
                m_notify_write_thread_cv.notify_one();
            }
        }

        bool disk_write::start()
        {
            LOG_FUNC_SCOPE();
            if(!m_is_configured) return false;
            m_stop_writing = false;//protection is not required before the thread is started
            assert(!m_thread.joinable());//we don't expect the thread to be active on start
            m_thread = std::thread(&disk_write::write_thread, this);
            return true;
        }

        void disk_write::stop()
        {
            LOG_FUNC_SCOPE();

            std::unique_lock<std::mutex> guard(m_main_mutex);
            m_stop_writing = true;
            guard.unlock();

            m_notify_write_thread_cv.notify_one();

            if (m_thread.joinable())
            {
                m_thread.join();
            }

            guard.lock();
            if(m_file)
                m_file->close();
            guard.unlock();
        }

        void disk_write::set_pause(bool pause)
        {
            std::lock_guard<std::mutex> guard(m_main_mutex);
            m_paused = pause;
        }

        status disk_write::configure(const configuration& config)
        {
            std::lock_guard<std::mutex> guard(m_main_mutex);
            if(m_is_configured) return status::status_exec_aborted;
            m_file = std::unique_ptr<rs::core::file>(new rs::core::file());
            status sts = m_file->open(config.m_file_path, (open_file_option)(open_file_option::write));

            if (sts != status::status_no_error)
                throw std::runtime_error("failed to open file for recording, file path - " + config.m_file_path);

            init_encoder(config);
            m_min_fps = get_min_fps(config.m_stream_profiles);
            write_header(static_cast<uint8_t>(config.m_stream_profiles.size()), config.m_coordinate_system, config.m_capture_mode);
            write_camera_info(config.m_camera_info);
            write_sw_info();
            write_capabilities(config.m_capabilities);
            write_motion_intrinsics(config.m_motion_intrinsics);
            write_stream_info(config.m_stream_profiles);
            write_properties(config.m_options);
            write_first_frame_offset();
            m_is_configured = true;
            return sts;
        }

        void disk_write::init_encoder(const configuration& config)
        {
            uint32_t buffer_size = 0;
            m_encoder.reset(new compression::encoder());
            for(auto profile : config.m_stream_profiles)
            {
                rs_stream stream = profile.second.info.stream;
                rs_format format = profile.second.info.format;
                uint32_t size = profile.second.info.width * profile.second.info.height;
                buffer_size = size > buffer_size ? size : buffer_size;
                auto compression_level = config.m_compression_config.at(profile.first);
                if(config.m_compression_config.find(profile.first) != (config.m_compression_config.end()))
                {
                    if(compression_level != record::compression_level::disabled)
                        m_encoder->add_codec(stream, format, compression_level);
                }
                else
                {
                    m_encoder->add_codec(stream, format, record::compression_level::high);
                }
            }
            m_encoded_data = std::vector<uint8_t>(buffer_size * 4);//stride is not available, taking worst case.
        }

        void disk_write::write_to_file(const void* data, unsigned int numberOfBytesToWrite, unsigned int& numberOfBytesWritten)
        {
            auto sts = m_file->write_bytes(data, numberOfBytesToWrite, numberOfBytesWritten);
            if(sts != status::status_no_error)
            {
                m_file->close();
                LOG_ERROR("failed writing to file");
                throw std::runtime_error("failed writing to file");
            }
        }

        void disk_write::write_thread(void)
        {
            LOG_FUNC_SCOPE();
            while (!m_stop_writing)
            {
                std::unique_lock<std::mutex> guard(m_notify_write_thread_mutex);
                m_notify_write_thread_cv.wait(guard);
                guard.unlock();

                LOG_VERBOSE("queue contains " << m_samples_queue.size() << " samples")

{'fraction_non_alphanumeric': 0.10242914979757085, 'fraction_numerical': 0.004048582995951417, 'mean_word_length': 1.8896471046987717, 'pattern_counts': {'":': 0, '<': 59, '<?xml version=': 0, '>': 73, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 4, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2872610', 'n_tokens_mistral': 4305, 'n_tokens_neox': 4110, 'n_words': 714}