diff --git a/src/rai_core/rai/aggregators/ros2/aggregators.py b/src/rai_core/rai/aggregators/ros2/aggregators.py index 9e1d63398..55f270e3a 100644 --- a/src/rai_core/rai/aggregators/ros2/aggregators.py +++ b/src/rai_core/rai/aggregators/ros2/aggregators.py @@ -33,21 +33,33 @@ class ROS2LogsAggregator(BaseAggregator[Log]): def get(self) -> HumanMessage: msgs = self.get_buffer() - buffer = [] - prev_parsed = None + buffer: List[str] = [] + prev_parsed: str | None = None counter = 0 + + def flush_run() -> None: + nonlocal counter, prev_parsed + if prev_parsed is None: + return + buffer.append(prev_parsed) + if counter > 0: + buffer.append(f"Log above repeated {counter} times") + counter = 0 + for log in msgs: level = self.levels[log.level] parsed = f"[{log.name}] [{level}] [{log.function}] {log.msg}" + if prev_parsed is None: + prev_parsed = parsed + counter = 0 + continue if parsed == prev_parsed: counter += 1 continue - else: - if counter != 0: - parsed = f"Log above repeated {counter} times" - buffer.append(parsed) - counter = 0 + flush_run() prev_parsed = parsed + flush_run() + result = f"Logs summary: {list(dict.fromkeys(buffer))}" self.clear_buffer() return HumanMessage(content=result) diff --git a/tests/aggregators/test_ros2_logs_aggregator.py b/tests/aggregators/test_ros2_logs_aggregator.py index 8837b00bb..f7838236c 100644 --- a/tests/aggregators/test_ros2_logs_aggregator.py +++ b/tests/aggregators/test_ros2_logs_aggregator.py @@ -58,11 +58,61 @@ def test_ros2_logs_aggregator_deduplicates_and_clears_buffer(): assert isinstance(summary, HumanMessage) assert ( summary.content - == "Logs summary: ['[demo_node] [WARNING] [do_work] System warming up', 'Log above repeated 1 times']" + == "Logs summary: ['[demo_node] [WARNING] [do_work] System warming up', " + "'Log above repeated 1 times', " + "'[demo_node] [ERROR] [do_work] System failure detected']" ) assert aggregator.get_buffer() == [] +def test_ros2_logs_aggregator_trailing_run_flushes_repeat_count(): + aggregator = ROS2LogsAggregator() + for _ in range(3): + aggregator( + DummyLog( + level=30, + name="demo_node", + function="do_work", + msg="System warming up", + ) + ) + + summary = aggregator.get() + assert summary.content == ( + "Logs summary: ['[demo_node] [WARNING] [do_work] System warming up', " + "'Log above repeated 2 times']" + ) + assert aggregator.get_buffer() == [] + + +def test_ros2_logs_aggregator_trailing_second_run(): + aggregator = ROS2LogsAggregator() + aggregator( + DummyLog(level=30, name="demo_node", function="do_work", msg="warm") + ) + for _ in range(3): + aggregator( + DummyLog(level=40, name="demo_node", function="do_work", msg="fail") + ) + + summary = aggregator.get() + assert summary.content == ( + "Logs summary: ['[demo_node] [WARNING] [do_work] warm', " + "'[demo_node] [ERROR] [do_work] fail', " + "'Log above repeated 2 times']" + ) + + +def test_ros2_logs_aggregator_single_log_has_no_repeat_line(): + aggregator = ROS2LogsAggregator() + aggregator( + DummyLog(level=20, name="n", function="f", msg="once") + ) + summary = aggregator.get() + assert summary.content == "Logs summary: ['[n] [INFO] [f] once']" + assert "repeated" not in summary.content + + def test_ros2_logs_aggregator_str(): aggregator = ROS2LogsAggregator() assert str(aggregator) == "ROS2LogsAggregator(len=0)"