talksik/mediapipe
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Project import generated by Copybara.

Browse Source 
GitOrigin-RevId: 1138530ad1578c5d6615b3e3d041775c75d310c4
This commit is contained in:
MediaPipe Team
2019-09-11 14:29:38 -07:00
committed by jqtang
parent 423c21b454
commit 61bc4556af
173 changed files with 1806 additions and 1824 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mediapipe/framework/stream_handler/BUILD
+4 -4
View File
@@ -54,7 +54,7 @@ mediapipe_cc_proto_library(
name = "default_input_stream_handler_cc_proto",
srcs = ["default_input_stream_handler.proto"],
cc_deps = ["//mediapipe/framework:mediapipe_options_cc_proto"],
visibility = ["//mediapipe/framework:__subpackages__"],
visibility = ["//visibility:public"],
deps = [":default_input_stream_handler_proto"],
)
@@ -62,7 +62,7 @@ mediapipe_cc_proto_library(
name = "fixed_size_input_stream_handler_cc_proto",
srcs = ["fixed_size_input_stream_handler.proto"],
cc_deps = ["//mediapipe/framework:mediapipe_options_cc_proto"],
visibility = ["//mediapipe/framework:__subpackages__"],
visibility = ["//visibility:public"],
deps = [":fixed_size_input_stream_handler_proto"],
)
@@ -70,7 +70,7 @@ mediapipe_cc_proto_library(
name = "sync_set_input_stream_handler_cc_proto",
srcs = ["sync_set_input_stream_handler.proto"],
cc_deps = ["//mediapipe/framework:mediapipe_options_cc_proto"],
visibility = ["//mediapipe/framework:__subpackages__"],
visibility = ["//visibility:public"],
deps = [":sync_set_input_stream_handler_proto"],
)
@@ -78,7 +78,7 @@ mediapipe_cc_proto_library(
name = "timestamp_align_input_stream_handler_cc_proto",
srcs = ["timestamp_align_input_stream_handler.proto"],
cc_deps = ["//mediapipe/framework:mediapipe_options_cc_proto"],
visibility = ["//mediapipe/framework:__subpackages__"],
visibility = ["//visibility:public"],
deps = [":timestamp_align_input_stream_handler_proto"],
)
mediapipe/framework/stream_handler/barrier_input_stream_handler_test.cc
+3 -3
View File
@@ -66,7 +66,7 @@ class BarrierInputStreamHandlerTest : public ::testing::Test {
id < input_tag_map->EndId(); ++id) {
const std::string& stream_name = names[id.value()];
name_to_id_[stream_name] = id;
MEDIAPIPE_ASSERT_OK(input_stream_managers_[id.value()].Initialize(
MP_ASSERT_OK(input_stream_managers_[id.value()].Initialize(
stream_name, &packet_type_, /*back_edge=*/false));
}
SetupInputStreamHandler(input_tag_map);
@@ -89,9 +89,9 @@ class BarrierInputStreamHandlerTest : public ::testing::Test {
/*calculator_run_in_parallel=*/false);
ASSERT_TRUE(status_or_handler.ok());
input_stream_handler_ = std::move(status_or_handler.ValueOrDie());
MEDIAPIPE_ASSERT_OK(input_stream_handler_->InitializeInputStreamManagers(
MP_ASSERT_OK(input_stream_handler_->InitializeInputStreamManagers(
input_stream_managers_.get()));
MEDIAPIPE_ASSERT_OK(
MP_ASSERT_OK(
calculator_context_manager_.PrepareForRun(setup_shards_callback_));
input_stream_handler_->PrepareForRun(headers_ready_callback_,
notification_callback_,
mediapipe/framework/stream_handler/default_input_stream_handler_test.cc
+47 -47
View File
@@ -53,34 +53,34 @@ TEST(DefaultInputStreamHandlerTest, NoBatchingWorks) {
tool::AddVectorSink("output1", &config, &sink_1);
CalculatorGraph graph;
MEDIAPIPE_ASSERT_OK(graph.Initialize(config));
MEDIAPIPE_ASSERT_OK(graph.StartRun({}));
MP_ASSERT_OK(graph.Initialize(config));
MP_ASSERT_OK(graph.StartRun({}));
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input0", Adopt(new int(1)).At(Timestamp(1))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// No packets expected as the second stream is not ready to be processed.
EXPECT_EQ(0, sink_0.size());
EXPECT_EQ(0, sink_1.size());
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input1", Adopt(new int(2)).At(Timestamp(2))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// First stream can produce output because the timestamp bound of the second
// stream is higher.
EXPECT_EQ(1, sink_0.size());
EXPECT_EQ(0, sink_1.size());
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input0", Adopt(new int(2)).At(Timestamp(2))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// Both streams have packets at the same timestamp, therefore both can produce
// packets.
EXPECT_EQ(2, sink_0.size());
EXPECT_EQ(1, sink_1.size());
MEDIAPIPE_ASSERT_OK(graph.CloseAllInputStreams());
MEDIAPIPE_ASSERT_OK(graph.WaitUntilDone());
MP_ASSERT_OK(graph.CloseAllInputStreams());
MP_ASSERT_OK(graph.WaitUntilDone());
}
// This test shows the effect of batching on the DefaultInputStreamHandler.
@@ -107,43 +107,43 @@ TEST(DefaultInputStreamHandlerTest, Batches) {
tool::AddVectorSink("output0", &config, &sink);
CalculatorGraph graph;
MEDIAPIPE_ASSERT_OK(graph.Initialize(config));
MEDIAPIPE_ASSERT_OK(graph.StartRun({}));
MP_ASSERT_OK(graph.Initialize(config));
MP_ASSERT_OK(graph.StartRun({}));
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input0", Adopt(new int(1)).At(Timestamp(1))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// There shouldn't be any outputs until a set of two packets is batched.
EXPECT_TRUE(sink.empty());
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input0", Adopt(new int(2)).At(Timestamp(2))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// There should be two packets, processed during a single invocation.
ASSERT_EQ(2, sink.size());
EXPECT_THAT(std::vector<int>({sink[0].Get<int>(), sink[1].Get<int>()}),
testing::ElementsAre(1, 2));
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input0", Adopt(new int(3)).At(Timestamp(3))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// There shouldn't be any outputs until another set of two packets is batched.
EXPECT_EQ(2, sink.size());
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input0", Adopt(new int(4)).At(Timestamp(4))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// The new batch was complete. There should be two more output packets.
ASSERT_EQ(4, sink.size());
EXPECT_THAT(std::vector<int>({sink[0].Get<int>(), sink[1].Get<int>(),
sink[2].Get<int>(), sink[3].Get<int>()}),
testing::ElementsAre(1, 2, 3, 4));
MEDIAPIPE_ASSERT_OK(graph.CloseAllInputStreams());
MEDIAPIPE_ASSERT_OK(graph.WaitUntilDone());
MP_ASSERT_OK(graph.CloseAllInputStreams());
MP_ASSERT_OK(graph.WaitUntilDone());
}
// This test shows that any packets get flushed (outputted) when the input
@@ -171,32 +171,32 @@ TEST(DefaultInputStreamHandlerTest, BatchIsFlushedWhenClosing) {
tool::AddVectorSink("output0", &config, &sink);
CalculatorGraph graph;
MEDIAPIPE_ASSERT_OK(graph.Initialize(config));
MEDIAPIPE_ASSERT_OK(graph.StartRun({}));
MP_ASSERT_OK(graph.Initialize(config));
MP_ASSERT_OK(graph.StartRun({}));
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input0", Adopt(new int(1)).At(Timestamp(1))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// There shouldn't be any outputs until a set of two packets is batched.
EXPECT_TRUE(sink.empty());
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input0", Adopt(new int(2)).At(Timestamp(2))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// There should be two packets, processed during a single invocation.
ASSERT_EQ(2, sink.size());
EXPECT_THAT(std::vector<int>({sink[0].Get<int>(), sink[1].Get<int>()}),
testing::ElementsAre(1, 2));
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input0", Adopt(new int(3)).At(Timestamp(3))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// There shouldn't be any outputs until another set of two packets is batched.
EXPECT_EQ(2, sink.size());
MEDIAPIPE_ASSERT_OK(graph.CloseAllInputStreams());
MEDIAPIPE_ASSERT_OK(graph.WaitUntilDone());
MP_ASSERT_OK(graph.CloseAllInputStreams());
MP_ASSERT_OK(graph.WaitUntilDone());
// When the streams are done, the packets currently being batched should be
// flushed out.
@@ -237,45 +237,45 @@ TEST(DefaultInputStreamHandlerTest, DoesntPropagateTimestampWhenBatching) {
std::vector<Packet> sink;
tool::AddVectorSink("output", &config, &sink);
CalculatorGraph graph;
MEDIAPIPE_ASSERT_OK(graph.Initialize(config));
MEDIAPIPE_ASSERT_OK(graph.StartRun({}));
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.Initialize(config));
MP_ASSERT_OK(graph.StartRun({}));
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input0", Adopt(new int(0)).At(Timestamp(0))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
EXPECT_TRUE(sink.empty());
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input0", Adopt(new int(1)).At(Timestamp(1))));
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input1", Adopt(new int(1)).At(Timestamp(1))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// Both calculators have packet 1. First node is currently batching and it
// propagates the first input timestamp in the batch. Therefore, the
// second node should produce output for the packet at 0.
EXPECT_EQ(1, sink.size());
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input0", Adopt(new int(2)).At(Timestamp(2))));
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input1", Adopt(new int(2)).At(Timestamp(2))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// Due to batching on the first node, timestamp is not propagated for the
// packet at timestamp 2. Therefore, the second node cannot process the packet
// at timestamp 1.
EXPECT_EQ(1, sink.size());
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input0", Adopt(new int(3)).At(Timestamp(3))));
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input1", Adopt(new int(3)).At(Timestamp(3))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// Batching is complete on the first node. It produced outputs at timestamp 1,
// 2, and 3. The first node can now process the input packets at timestamps 1,
// 2, and 3 as well.
EXPECT_EQ(4, sink.size());
MEDIAPIPE_ASSERT_OK(graph.CloseAllInputStreams());
MEDIAPIPE_ASSERT_OK(graph.WaitUntilDone());
MP_ASSERT_OK(graph.CloseAllInputStreams());
MP_ASSERT_OK(graph.WaitUntilDone());
EXPECT_EQ(4, sink.size());
}
mediapipe/framework/stream_handler/fixed_size_input_stream_handler_test.cc
+18 -18
View File
@@ -165,8 +165,8 @@ TEST_P(FixedSizeInputStreamHandlerTest, DropsPackets) {
std::vector<Packet> output_packets;
tool::AddVectorSink("output_packets", &graph_config, &output_packets);
CalculatorGraph graph;
MEDIAPIPE_ASSERT_OK(graph.Initialize(graph_config));
MEDIAPIPE_ASSERT_OK(graph.Run());
MP_ASSERT_OK(graph.Initialize(graph_config));
MP_ASSERT_OK(graph.Run());
// The TestSlowCalculator consumes one packet after every tenth packet
// is sent. All other packets are dropped by the FixedSizeInputStreamHandler.
@@ -209,10 +209,10 @@ TEST_P(FixedSizeInputStreamHandlerTest, DropsPacketsInFullStream) {
std::vector<Packet> output_packets;
tool::AddVectorSink("output_packets", &graph_config, &output_packets);
CalculatorGraph graph;
MEDIAPIPE_ASSERT_OK(
MP_ASSERT_OK(
graph.Initialize(graph_config, {{"max_count", MakePacket<int>(10)},
{"batch_size", MakePacket<int>(10)}}));
MEDIAPIPE_ASSERT_OK(graph.Run());
MP_ASSERT_OK(graph.Run());
}
// Tests FixedSizeInputStreamHandler with several input streams running
@@ -249,8 +249,8 @@ TEST_P(FixedSizeInputStreamHandlerTest, ParallelWriteAndRead) {
&output_packets[i]);
}
CalculatorGraph graph;
MEDIAPIPE_ASSERT_OK(graph.Initialize(graph_config, {}));
MEDIAPIPE_ASSERT_OK(graph.StartRun({}));
MP_ASSERT_OK(graph.Initialize(graph_config, {}));
MP_ASSERT_OK(graph.StartRun({}));
{
::mediapipe::ThreadPool pool(3);
@@ -262,15 +262,15 @@ TEST_P(FixedSizeInputStreamHandlerTest, ParallelWriteAndRead) {
std::string stream_name = absl::StrCat("in_", w);
for (int i = 0; i < 50; ++i) {
Packet p = MakePacket<int>(i).At(Timestamp(i));
MEDIAPIPE_EXPECT_OK(graph.AddPacketToInputStream(stream_name, p));
MP_EXPECT_OK(graph.AddPacketToInputStream(stream_name, p));
absl::SleepFor(absl::Microseconds(100));
}
});
}
}
MEDIAPIPE_ASSERT_OK(graph.CloseAllInputStreams());
MEDIAPIPE_ASSERT_OK(graph.WaitUntilDone());
MP_ASSERT_OK(graph.CloseAllInputStreams());
MP_ASSERT_OK(graph.WaitUntilDone());
for (int i = 0; i < 3; ++i) {
EXPECT_EQ(output_packets[i].size(), output_packets[0].size());
for (int j = 0; j < output_packets[i].size(); j++) {
@@ -320,27 +320,27 @@ TEST_P(FixedSizeInputStreamHandlerTest, LateArrivalDrop) {
&output_packets[i]);
}
CalculatorGraph graph;
MEDIAPIPE_ASSERT_OK(graph.Initialize(graph_config, {}));
MEDIAPIPE_ASSERT_OK(graph.StartRun({}));
MP_ASSERT_OK(graph.Initialize(graph_config, {}));
MP_ASSERT_OK(graph.StartRun({}));
for (int i = 1; i <= 6; i++) {
MEDIAPIPE_EXPECT_OK(graph.AddPacketToInputStream(
MP_EXPECT_OK(graph.AddPacketToInputStream(
in_streams[0], MakePacket<int>(i).At(Timestamp(i))));
}
for (int i = 3; i <= 7; i++) {
MEDIAPIPE_EXPECT_OK(graph.AddPacketToInputStream(
MP_EXPECT_OK(graph.AddPacketToInputStream(
in_streams[1], MakePacket<int>(i).At(Timestamp(i))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
}
// At this point everything before ts 5 should be dropped.
for (int i = 4; i <= 7; i++) {
MEDIAPIPE_EXPECT_OK(graph.AddPacketToInputStream(
MP_EXPECT_OK(graph.AddPacketToInputStream(
in_streams[2], MakePacket<int>(i).At(Timestamp(i))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
}
MEDIAPIPE_ASSERT_OK(graph.CloseAllInputStreams());
MEDIAPIPE_ASSERT_OK(graph.WaitUntilDone());
MP_ASSERT_OK(graph.CloseAllInputStreams());
MP_ASSERT_OK(graph.WaitUntilDone());
if (GetParam()) {
EXPECT_THAT(TimestampValues(output_packets[0]),
mediapipe/framework/stream_handler/immediate_input_stream_handler_test.cc
+2 -2
View File
@@ -89,9 +89,9 @@ class ImmediateInputStreamHandlerTest : public ::testing::Test {
/*calculator_run_in_parallel=*/false);
ASSERT_TRUE(status_or_handler.ok());
input_stream_handler_ = std::move(status_or_handler.ValueOrDie());
MEDIAPIPE_ASSERT_OK(input_stream_handler_->InitializeInputStreamManagers(
MP_ASSERT_OK(input_stream_handler_->InitializeInputStreamManagers(
input_stream_managers_.get()));
MEDIAPIPE_ASSERT_OK(cc_manager_.PrepareForRun(setup_shards_callback_));
MP_ASSERT_OK(cc_manager_.PrepareForRun(setup_shards_callback_));
input_stream_handler_->PrepareForRun(headers_ready_callback_,
notification_callback_,
schedule_callback_, error_callback_);
mediapipe/framework/stream_handler/mux_input_stream_handler_test.cc
+5 -5
View File
@@ -77,14 +77,14 @@ TEST(MuxInputStreamHandlerTest, AtomicAccessToControlAndDataStreams) {
input_stream_handler { input_stream_handler: "MuxInputStreamHandler" }
})");
CalculatorGraph graph;
MEDIAPIPE_ASSERT_OK(graph.Initialize(config));
MEDIAPIPE_ASSERT_OK(graph.StartRun({}));
MP_ASSERT_OK(graph.Initialize(config));
MP_ASSERT_OK(graph.StartRun({}));
for (int i = 0; i < 2000; ++i) {
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input", Adopt(new int(i)).At(Timestamp(i))));
}
MEDIAPIPE_ASSERT_OK(graph.CloseAllInputStreams());
MEDIAPIPE_ASSERT_OK(graph.WaitUntilDone());
MP_ASSERT_OK(graph.CloseAllInputStreams());
MP_ASSERT_OK(graph.WaitUntilDone());
}
} // namespace
mediapipe/framework/stream_handler/set_input_stream_handler_test.cc
+10 -10
View File
@@ -84,14 +84,14 @@ TEST(MuxInputStreamHandlerTest, AtomicAccessToControlAndDataStreams) {
# MuxInputStreamHandler set in GetContract().
})");
CalculatorGraph graph;
MEDIAPIPE_ASSERT_OK(graph.Initialize(config));
MEDIAPIPE_ASSERT_OK(graph.StartRun({}));
MP_ASSERT_OK(graph.Initialize(config));
MP_ASSERT_OK(graph.StartRun({}));
for (int i = 0; i < 2000; ++i) {
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input", Adopt(new int(i)).At(Timestamp(i))));
}
MEDIAPIPE_ASSERT_OK(graph.CloseAllInputStreams());
MEDIAPIPE_ASSERT_OK(graph.WaitUntilDone());
MP_ASSERT_OK(graph.CloseAllInputStreams());
MP_ASSERT_OK(graph.WaitUntilDone());
}
// Copied from pass_through_calculator.cc, and modified to specify
@@ -218,8 +218,8 @@ TEST(FixedSizeInputStreamHandlerTest, ParallelWriteAndRead) {
&output_packets[i]);
}
CalculatorGraph graph;
MEDIAPIPE_ASSERT_OK(graph.Initialize(graph_config, {}));
MEDIAPIPE_ASSERT_OK(graph.StartRun({}));
MP_ASSERT_OK(graph.Initialize(graph_config, {}));
MP_ASSERT_OK(graph.StartRun({}));
{
::mediapipe::ThreadPool pool(NUM_STREAMS);
@@ -231,15 +231,15 @@ TEST(FixedSizeInputStreamHandlerTest, ParallelWriteAndRead) {
std::string stream_name = absl::StrCat("in_", w);
for (int i = 0; i < 50; ++i) {
Packet p = MakePacket<int>(i).At(Timestamp(i));
MEDIAPIPE_EXPECT_OK(graph.AddPacketToInputStream(stream_name, p));
MP_EXPECT_OK(graph.AddPacketToInputStream(stream_name, p));
absl::SleepFor(absl::Microseconds(100));
}
});
}
}
MEDIAPIPE_ASSERT_OK(graph.CloseAllInputStreams());
MEDIAPIPE_ASSERT_OK(graph.WaitUntilDone());
MP_ASSERT_OK(graph.CloseAllInputStreams());
MP_ASSERT_OK(graph.WaitUntilDone());
for (int i = 0; i < NUM_STREAMS; ++i) {
EXPECT_EQ(output_packets[i].size(), output_packets[0].size());
for (int j = 0; j < output_packets[i].size(); j++) {
mediapipe/framework/stream_handler/sync_set_input_stream_handler_test.cc
+7 -7
View File
@@ -266,16 +266,16 @@ TEST(SyncSetInputStreamHandlerTest, OrdinaryOperation) {
VLOG(2) << "Modified configuration: " << modified_config.DebugString();
// Setup and run the graph.
MEDIAPIPE_ASSERT_OK(graph.Initialize(
MP_ASSERT_OK(graph.Initialize(
modified_config,
{{"lambda", MakePacket<ProcessFunction>(InputsToDebugString)}}));
std::deque<Packet> outputs;
MEDIAPIPE_ASSERT_OK(
MP_ASSERT_OK(
graph.ObserveOutputStream("output", [&outputs](const Packet& packet) {
outputs.push_back(packet);
return ::mediapipe::OkStatus();
}));
MEDIAPIPE_ASSERT_OK(graph.StartRun({}));
MP_ASSERT_OK(graph.StartRun({}));
for (int command_index = 0; command_index < shuffled_commands.size();
/* command_index is incremented by the inner loop. */) {
int initial_command_index = command_index;
@@ -295,14 +295,14 @@ TEST(SyncSetInputStreamHandlerTest, OrdinaryOperation) {
VLOG(1) << "Adding (" << stream_name << ", Timestamp: " << timestamp
<< ")";
if (timestamp == Timestamp::Done()) {
MEDIAPIPE_ASSERT_OK(graph.CloseInputStream(stream_name));
MP_ASSERT_OK(graph.CloseInputStream(stream_name));
} else {
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
stream_name, MakePacket<int>(0).At(timestamp)));
}
}
// Ensure that we produce all packets which we can.
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// Check the output strings (ignoring order, since calculator may
// have run in parallel).
@@ -319,7 +319,7 @@ TEST(SyncSetInputStreamHandlerTest, OrdinaryOperation) {
EXPECT_THAT(actual_strings,
testing::UnorderedElementsAreArray(expected_strings));
}
MEDIAPIPE_ASSERT_OK(graph.WaitUntilDone());
MP_ASSERT_OK(graph.WaitUntilDone());
}
}
mediapipe/framework/stream_handler/timestamp_align_input_stream_handler_test.cc
+29 -29
View File
@@ -49,38 +49,38 @@ TEST(TimestampAlignInputStreamHandlerTest, Initialization) {
tool::AddVectorSink("output_camera", &config, &sink_camera);
CalculatorGraph graph;
MEDIAPIPE_ASSERT_OK(graph.Initialize(config));
MEDIAPIPE_ASSERT_OK(graph.StartRun({}));
MP_ASSERT_OK(graph.Initialize(config));
MP_ASSERT_OK(graph.StartRun({}));
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input_camera", Adopt(new int(1)).At(Timestamp(101))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// The timestamp base stream's packet is output immediately.
EXPECT_EQ(0, sink_video.size());
ASSERT_EQ(1, sink_camera.size());
EXPECT_EQ(1, sink_camera[0].Get<int>());
EXPECT_EQ(Timestamp(101), sink_camera[0].Timestamp());
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input_camera", Adopt(new int(2)).At(Timestamp(102))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// The timestamp base stream's packet is output immediately.
EXPECT_EQ(0, sink_video.size());
ASSERT_EQ(2, sink_camera.size());
EXPECT_EQ(2, sink_camera[1].Get<int>());
EXPECT_EQ(Timestamp(102), sink_camera[1].Timestamp());
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input_video", Adopt(new int(1)).At(Timestamp(1))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// No packet is output. The packet added to input_video is buffered in the
// input stream.
EXPECT_EQ(0, sink_video.size());
EXPECT_EQ(2, sink_camera.size());
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input_camera", Adopt(new int(3)).At(Timestamp(103))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// Both input streams have a packet. The following equivalence of timestamps
// is established:
// input_video input_camera
@@ -96,23 +96,23 @@ TEST(TimestampAlignInputStreamHandlerTest, Initialization) {
EXPECT_EQ(3, sink_camera[2].Get<int>());
EXPECT_EQ(Timestamp(103), sink_camera[2].Timestamp());
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input_camera", Adopt(new int(4)).At(Timestamp(104))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// The timestamp base stream does not receive special treatment now.
EXPECT_EQ(1, sink_video.size());
EXPECT_EQ(3, sink_camera.size());
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input_video", Adopt(new int(4)).At(Timestamp(4))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
EXPECT_EQ(1, sink_video.size());
ASSERT_EQ(4, sink_camera.size());
EXPECT_EQ(4, sink_camera[3].Get<int>());
EXPECT_EQ(Timestamp(104), sink_camera[3].Timestamp());
MEDIAPIPE_ASSERT_OK(graph.CloseAllInputStreams());
MEDIAPIPE_ASSERT_OK(graph.WaitUntilDone());
MP_ASSERT_OK(graph.CloseAllInputStreams());
MP_ASSERT_OK(graph.WaitUntilDone());
ASSERT_EQ(2, sink_video.size());
EXPECT_EQ(4, sink_camera.size());
EXPECT_EQ(4, sink_video[1].Get<int>());
@@ -145,21 +145,21 @@ TEST(TimestampAlignInputStreamHandlerTest, TickRate) {
tool::AddVectorSink("output_camera", &config, &sink_camera);
CalculatorGraph graph;
MEDIAPIPE_ASSERT_OK(graph.Initialize(config));
MEDIAPIPE_ASSERT_OK(graph.StartRun({}));
MP_ASSERT_OK(graph.Initialize(config));
MP_ASSERT_OK(graph.StartRun({}));
// Video timestamps start from 0 seconds. Video frame rate is 2 fps.
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input_video", Adopt(new int(0)).At(Timestamp(0))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// No packets expected as the timestamp base stream has not seen any packet.
EXPECT_EQ(0, sink_video.size());
EXPECT_EQ(0, sink_camera.size());
// Camera timestamps start from 100 seconds. Camera frame rate is 1 fps.
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input_camera", Adopt(new int(0)).At(Timestamp(100000000))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
// Both input streams have a packet. The following equivalence of timestamps
// is established:
// input_video input_camera
@@ -171,17 +171,17 @@ TEST(TimestampAlignInputStreamHandlerTest, TickRate) {
EXPECT_EQ(0, sink_camera[0].Get<int>());
EXPECT_EQ(Timestamp(100000000), sink_camera[0].Timestamp());
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input_video", Adopt(new int(1)).At(Timestamp(500000))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
EXPECT_EQ(1, sink_video.size());
EXPECT_EQ(1, sink_camera.size());
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input_video", Adopt(new int(2)).At(Timestamp(1000000))));
MEDIAPIPE_ASSERT_OK(graph.AddPacketToInputStream(
MP_ASSERT_OK(graph.AddPacketToInputStream(
"input_camera", Adopt(new int(1)).At(Timestamp(101000000))));
MEDIAPIPE_ASSERT_OK(graph.WaitUntilIdle());
MP_ASSERT_OK(graph.WaitUntilIdle());
ASSERT_EQ(3, sink_video.size());
ASSERT_EQ(3, sink_camera.size());
EXPECT_EQ(1, sink_video[1].Get<int>());
@@ -193,8 +193,8 @@ TEST(TimestampAlignInputStreamHandlerTest, TickRate) {
EXPECT_EQ(1, sink_camera[2].Get<int>());
EXPECT_EQ(Timestamp(101000000), sink_camera[2].Timestamp());
MEDIAPIPE_ASSERT_OK(graph.CloseAllInputStreams());
MEDIAPIPE_ASSERT_OK(graph.WaitUntilDone());
MP_ASSERT_OK(graph.CloseAllInputStreams());
MP_ASSERT_OK(graph.WaitUntilDone());
ASSERT_EQ(3, sink_video.size());
ASSERT_EQ(3, sink_camera.size());
}