MCPcopy Create free account
hub / github.com/agentscope-ai/Trinity-RFT / monitor

Method monitor

trinity/manager/log_manager.py:236–274  ·  view source on GitHub ↗

Main monitoring loop

(self)

Source from the content-addressed store, hash-verified

234 print(f"{Colors.YELLOW}[INFO] Stopped tracking: {filepath}{Colors.RESET}")
235
236 def monitor(self):
237 """Main monitoring loop"""
238 self.running = True
239 last_scan = 0
240
241 print(f"{Colors.CYAN}{'=' * 60}{Colors.RESET}")
242 print(f"{Colors.CYAN}Log monitoring started{Colors.RESET}")
243 print(f"{Colors.CYAN}Monitoring directory: {self.log_dir}{Colors.RESET}")
244 print(f"{Colors.CYAN}Keyword: {self.keyword}{Colors.RESET}")
245 print(f"{Colors.CYAN}Minimum level: {self.min_level_name}{Colors.RESET}")
246 print(f"{Colors.CYAN}{'=' * 60}{Colors.RESET}\n")
247
248 # Initial scan
249 self.scan_new_files()
250
251 try:
252 while self.running:
253 current_time = time.time()
254
255 # Periodically scan for new files (every 5 seconds)
256 if current_time - last_scan > 5:
257 self.scan_new_files()
258 last_scan = current_time
259
260 # Read new content from all files
261 with self.lock:
262 for tracker in list(self.trackers.values()):
263 lines = tracker.read_new_lines()
264 for line in lines:
265 if tracker.should_display(line):
266 output = tracker.format_output(line)
267 print(output)
268
269 time.sleep(self.scan_interval)
270
271 except KeyboardInterrupt:
272 print(f"\n{Colors.YELLOW}[INFO] Received stop signal, exiting...{Colors.RESET}")
273 finally:
274 self.stop()
275
276 def stop(self):
277 """Stop monitoring"""

Callers 1

log_commandFunction · 0.95

Calls 6

scan_new_filesMethod · 0.95
stopMethod · 0.95
read_new_linesMethod · 0.80
should_displayMethod · 0.80
format_outputMethod · 0.80
sleepMethod · 0.80

Tested by

no test coverage detected