Prediction: refactor records to new records

wei07283091 · May 9, 2020 · e1cac6e · e1cac6e
1 parent 01062c8
commit e1cac6e
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 33 deletions.
diff --git a/modules/prediction/common/message_process.cc b/modules/prediction/common/message_process.cc
@@ -17,7 +17,8 @@
 #include "modules/prediction/common/message_process.h"
 
 #include "cyber/common/file.h"
-#include "cyber/record/file/record_file_writer.h"
+#include "cyber/record/record_reader.h"
+#include "cyber/record/record_writer.h"
 
 #include "modules/common/adapters/adapter_gflags.h"
 #include "modules/prediction/common/feature_output.h"
@@ -41,7 +42,7 @@ namespace prediction {
 using apollo::common::adapter::AdapterConfig;
 using apollo::cyber::record::RecordMessage;
 using apollo::cyber::record::RecordReader;
-using apollo::cyber::record::RecordFileWriter;
+using apollo::cyber::record::RecordWriter;
 using apollo::cyber::proto::SingleMessage;
 using apollo::localization::LocalizationEstimate;
 using apollo::perception::PerceptionObstacle;
@@ -284,34 +285,36 @@ void MessageProcess::OnStoryTelling(const Stories& story) {
          << "].";
 }
 
-void MessageProcess::ProcessOfflineData(const std::string& record_filename) {
-  RecordReader reader(record_filename);
+void MessageProcess::ProcessOfflineData(const std::string& record_filepath) {
+  RecordReader reader(record_filepath);
   RecordMessage message;
-  RecordFileWriter writer;
-  writer.Open(record_filename + ".new_prediction");
+  RecordWriter writer;
+  if (FLAGS_prediction_offline_mode == PredictionConstants::kDumpRecord) {
+    writer.Open(record_filepath + ".new_prediction");
+  }
   while (reader.ReadMessage(&message)) {
-    if (FLAGS_prediction_offline_mode == PredictionConstants::kDumpRecord &&
-        message.channel_name != FLAGS_prediction_topic) {
-      writer.WriteMessage(RecordMessageToSingleMessage(message));
-    }
     if (message.channel_name == FLAGS_perception_obstacle_topic) {
       PerceptionObstacles perception_obstacles;
       if (perception_obstacles.ParseFromString(message.content)) {
+        if (FLAGS_prediction_offline_mode == PredictionConstants::kDumpRecord) {
+          writer.WriteMessage<PerceptionObstacles>(message.channel_name,
+              perception_obstacles, message.time);
+        }
         PredictionObstacles prediction_obstacles;
         OnPerception(perception_obstacles, &prediction_obstacles);
         if (FLAGS_prediction_offline_mode == PredictionConstants::kDumpRecord) {
-          SingleMessage single_message;
-          std::string content = "";
-          prediction_obstacles.SerializeToString(&content);
-          single_message.set_content(content);
-          single_message.set_time(message.time);
-          single_message.set_channel_name(FLAGS_prediction_topic);
-          writer.WriteMessage(RecordMessageToSingleMessage(message));
+          writer.WriteMessage<PredictionObstacles>(FLAGS_prediction_topic,
+              prediction_obstacles, message.time);
+          AINFO << "Generated a new prediction message.";
         }
       }
     } else if (message.channel_name == FLAGS_localization_topic) {
       LocalizationEstimate localization;
       if (localization.ParseFromString(message.content)) {
+        if (FLAGS_prediction_offline_mode == PredictionConstants::kDumpRecord) {
+          writer.WriteMessage<LocalizationEstimate>(message.channel_name,
+              localization, message.time);
+        }
         OnLocalization(localization);
       }
     } else if (message.channel_name == FLAGS_planning_trajectory_topic) {
@@ -321,16 +324,9 @@ void MessageProcess::ProcessOfflineData(const std::string& record_filename) {
       }
     }
   }
-  writer.Close();
-}
-
-SingleMessage MessageProcess::RecordMessageToSingleMessage(
-    const RecordMessage& record_message) {
-  SingleMessage single_message;
-  single_message.set_channel_name(record_message.channel_name);
-  single_message.set_time(record_message.time);
-  single_message.set_content(record_message.content);
-  return single_message;
+  if (FLAGS_prediction_offline_mode == PredictionConstants::kDumpRecord) {
+    writer.Close();
+  }
 }
 
 }  // namespace prediction

diff --git a/modules/prediction/common/message_process.h b/modules/prediction/common/message_process.h
@@ -21,7 +21,6 @@
 #pragma once
 
 #include <string>
-#include "cyber/record/record_reader.h"
 #include "cyber/proto/record.pb.h"
 
 #include "modules/localization/proto/localization.pb.h"
@@ -58,10 +57,7 @@ class MessageProcess {
 
   static void OnStoryTelling(const storytelling::Stories &story);
 
-  static void ProcessOfflineData(const std::string &record_filename);
-
-  static apollo::cyber::proto::SingleMessage RecordMessageToSingleMessage(
-      const apollo::cyber::record::RecordMessage& record_message);
+  static void ProcessOfflineData(const std::string &record_filepath);
 };
 
 }  // namespace prediction

diff --git a/modules/tools/prediction/data_pipelines/scripts/records_to_dump_records.sh b/modules/tools/prediction/data_pipelines/scripts/records_to_dump_records.sh
@@ -38,4 +38,5 @@ fi
     --map_dir=/apollo/modules/map/data/${MAP_DIR} \
     --prediction_offline_mode=6 \
     --prediction_offline_bags=${SRC_DIR} \
-    --noenable_multi_thread
+    --noenable_multi_thread \
+    --noenable_async_draw_base_image