feat(autoware_debug_tools): add frequent_log_checker.py

Signed-off-by: Takayuki Murooka <[email protected]>

common/autoware_debug_tools/README.md

@@ -119,3 +119,20 @@ ros2 run autoware_debug_tools rosout_log_reconstructor
 ```
 
 ![rosout_log_example](images/rosout_log_example.png)
+
+## Frequent Log Checker
+
+This script shows the frequent log from the `launch.log`.
+In detail, the log which appears more than the threshold times during the duration will be shown.
+
+### Usage
+
+```bash
+ros2 run autoware_debug_tools frequent_log_checker <file-path>
+```
+
+The command options are
+
+- `-d`: duration to check the frequent log
+- `-c`: threshold of the log count during the duration
+- `-f`: log format. Several log formats are pre-defined in the script.

common/autoware_debug_tools/autoware_debug_tools/frequent_log_checker.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+
+# Copyright 2024 TIER IV, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import argparse
+
+
+class Log:
+    def __init__(self, node_name, timestamp, content, file_and_line, full_message):
+        self.node_name = node_name
+        self.timestamp = timestamp
+        self.content = content
+        self.file_and_line = file_and_line
+        self.full_message = full_message
+
+    def is_same(self, log):
+        return self.node_name == log.node_name and self.file_and_line == log.file_and_line
+
+    def is_in(self, log_list):
+        for target_log in log_list:
+            if self.is_same(target_log):
+                return True
+        return False
+
+
+def check(log_file, duration_to_check, log_count_threshold, log_format):
+    recent_log_list = []
+    unique_frequent_log_list = []
+
+    # with open("autoware.log", "r") as f:
+    with open(log_file, "r") as f:
+        for full_message in f.readlines():
+            try:
+                if log_format == "1":
+                    node_name = full_message.split("[")[4].split("]")[0]
+                    timestamp = float(full_message.split("[")[0][:-1])
+                    content = full_message.split("]")[3].split(" at ")[0]
+                    file_and_line = full_message.split("]")[3].split(" at ")[1]
+                    recent_log = Log(node_name, timestamp, content, file_and_line, full_message)
+                elif log_format == "2":
+                    node_name = full_message.split("]")[0][1:]
+                    timestamp = float(full_message.split("]")[1].split(" ")[2])
+                    content = full_message.split("]")[3].split(" at ")[0]
+                    file_and_line = full_message.split("]")[3].split(" at ")[1]
+                    recent_log = Log(node_name, timestamp, content, file_and_line, full_message)
+                else:
+                    continue
+            except IndexError:
+                continue
+            except ValueError:
+                continue
+
+            # skip if the log is already considered as frequent
+            if recent_log.is_in(unique_frequent_log_list):
+                continue
+            recent_log_list.append(recent_log)
+
+            # remove obsolete or already frequent log
+            for log in recent_log_list[:]:
+                duration = timestamp - log.timestamp
+                if duration_to_check < duration:
+                    recent_log_list.remove(log)
+
+            # extract duplicated (= frequent) log
+            for i in range(len(recent_log_list)):
+                log_count = 0
+                if recent_log_list[i].is_in(unique_frequent_log_list):
+                    continue
+
+                for j in range(len(recent_log_list)):
+                    if i <= j:
+                        continue
+                    if recent_log_list[i].is_same(recent_log_list[j]):
+                        log_count += 1
+
+                if log_count_threshold <= log_count:
+                    unique_frequent_log_list.append(recent_log_list[i])
+
+    for frequent_log in unique_frequent_log_list:
+        print(frequent_log.full_message)
+
+
+def main():
+    parser = argparse.ArgumentParser(description="frequent log checker")
+    parser.add_argument("log_file", help="launch log file")
+    parser.add_argument("-d", "--log-duration", default=1.0, help="duration to check log")
+    parser.add_argument("-c", "--log-count", default=2, help="log count threshold")
+    parser.add_argument("-f", "--log-format", default="1", help="log format")
+    args = parser.parse_args()
+
+    check(args.log_file, args.log_duration, args.log_count, args.log_format)
+
+
+if __name__ == "__main__":
+    main()

common/autoware_debug_tools/setup.py

@@ -28,6 +28,7 @@
             "topic_connection_checker = autoware_debug_tools.topic_connection_checker.topic_connection_checker_node:main",
             "topic_localizer = autoware_debug_tools.topic_connection_checker.localize_topic:main",
             "rosout_log_reconstructor = autoware_debug_tools.rosout_log_reconstructor:main",
+            "frequent_log_checker = autoware_debug_tools.frequent_log_checker:main",
         ],
     },
 )