Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add file reset capabilities #37

Merged
merged 1 commit into from
Nov 10, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions data_tamer_cpp/include/data_tamer/sinks/mcap_sink.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,11 @@ class MCAPSink : public DataSinkBase
/// WARNING: this can consume a large amount of disk space very quickly.
void setMaxTimeBeforeReset(std::chrono::seconds reset_time);

/// When resetting the MCAP recording (see `setMaxTimeBeforeReset`),
/// if `create_new_file` is true then the filename will be incremented
/// and then saved instead of overwriting the previous file.
void setCreateNewFileOnReset(bool create_new_file);

/// Stop recording and save the file
void stopRecording();

Expand All @@ -54,6 +59,7 @@ class MCAPSink : public DataSinkBase
*
* @param filepath file path of the new file (should be ".mcap" extension)
* @param do_compression if true, compress the data on the fly.
* WARNING: if this is called with the same filename as previously, the file counter will be reset, too.
*/
void restartRecording(std::string const& filepath, bool do_compression = false);

Expand All @@ -65,13 +71,20 @@ class MCAPSink : public DataSinkBase
std::unordered_map<uint64_t, uint16_t> hash_to_channel_id_;
std::unordered_map<std::string, Schema> schemas_;

// file reset variables
bool create_file_on_reset_ = false;
std::string original_filepath_;
size_t file_reset_counter_ = 1;

std::chrono::seconds reset_time_ = std::chrono::seconds(60 * 10);
std::chrono::system_clock::time_point start_time_;

bool forced_stop_recording_ = false;
std::recursive_mutex mutex_;

void openFile(std::string const& filepath);
void restartRecordingImpl(std::string const& filepath, bool do_compression,
bool new_file);
};

} // namespace DataTamer
28 changes: 26 additions & 2 deletions data_tamer_cpp/src/sinks/mcap_sink.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
#include <chrono>
#include <sstream>
#include <mutex>
#include <string>

#ifndef USING_ROS2
#define MCAP_IMPLEMENTATION
Expand Down Expand Up @@ -35,7 +36,7 @@ namespace DataTamer
static constexpr char const* kDataTamer = "data_tamer";

MCAPSink::MCAPSink(const std::string& filepath, bool do_compression)
: filepath_(filepath), compression_(do_compression)
: filepath_(filepath), compression_(do_compression), original_filepath_(filepath)
{
openFile(filepath_);
}
Expand Down Expand Up @@ -121,7 +122,13 @@ bool MCAPSink::storeSnapshot(const Snapshot& snapshot)
auto const now = std::chrono::system_clock::now();
if(reset_time_ != std::chrono::seconds(0) && now - start_time_ > reset_time_)
{
restartRecording(filepath_, compression_);
if(create_file_on_reset_)
{
// change the current filepath to the original with "_[# resets]"" appended
filepath_ = original_filepath_ + "_" + std::to_string(file_reset_counter_);
++file_reset_counter_;
}
restartRecordingImpl(filepath_, compression_, false);
}
return true;
}
Expand All @@ -131,6 +138,11 @@ void MCAPSink::setMaxTimeBeforeReset(std::chrono::seconds reset_time)
reset_time_ = reset_time;
}

void MCAPSink::setCreateNewFileOnReset(bool create_file_on_reset)
{
create_file_on_reset_ = create_file_on_reset;
}

void MCAPSink::stopRecording()
{
std::scoped_lock lk(mutex_);
Expand All @@ -140,8 +152,20 @@ void MCAPSink::stopRecording()
}

void MCAPSink::restartRecording(const std::string& filepath, bool do_compression)
{
restartRecordingImpl(filepath, do_compression, true);
}

void MCAPSink::restartRecordingImpl(const std::string& filepath, bool do_compression,
bool new_file)
{
std::scoped_lock lk(mutex_);
if(new_file)
{
// if this was called by a user, we need to change the filepath that we will increment when reset
file_reset_counter_ = 1;
original_filepath_ = filepath;
}
filepath_ = filepath;
compression_ = do_compression;
openFile(filepath_);
Expand Down
Loading