1  import os
       2  import time
       3  
       4  from test.support import MS_WINDOWS
       5  from .results import TestResults
       6  from .runtests import RunTests
       7  from .utils import print_warning
       8  
       9  if MS_WINDOWS:
      10      from .win_utils import WindowsLoadTracker
      11  
      12  
      13  class ESC[4;38;5;81mLogger:
      14      def __init__(self, results: TestResults, quiet: bool, pgo: bool):
      15          self.start_time = time.perf_counter()
      16          self.test_count_text = ''
      17          self.test_count_width = 3
      18          self.win_load_tracker: WindowsLoadTracker | None = None
      19          self._results: TestResults = results
      20          self._quiet: bool = quiet
      21          self._pgo: bool = pgo
      22  
      23      def log(self, line: str = '') -> None:
      24          empty = not line
      25  
      26          # add the system load prefix: "load avg: 1.80 "
      27          load_avg = self.get_load_avg()
      28          if load_avg is not None:
      29              line = f"load avg: {load_avg:.2f} {line}"
      30  
      31          # add the timestamp prefix:  "0:01:05 "
      32          log_time = time.perf_counter() - self.start_time
      33  
      34          mins, secs = divmod(int(log_time), 60)
      35          hours, mins = divmod(mins, 60)
      36          formatted_log_time = "%d:%02d:%02d" % (hours, mins, secs)
      37  
      38          line = f"{formatted_log_time} {line}"
      39          if empty:
      40              line = line[:-1]
      41  
      42          print(line, flush=True)
      43  
      44      def get_load_avg(self) -> float | None:
      45          if hasattr(os, 'getloadavg'):
      46              return os.getloadavg()[0]
      47          if self.win_load_tracker is not None:
      48              return self.win_load_tracker.getloadavg()
      49          return None
      50  
      51      def display_progress(self, test_index: int, text: str) -> None:
      52          if self._quiet:
      53              return
      54          results = self._results
      55  
      56          # "[ 51/405/1] test_tcl passed"
      57          line = f"{test_index:{self.test_count_width}}{self.test_count_text}"
      58          fails = len(results.bad) + len(results.env_changed)
      59          if fails and not self._pgo:
      60              line = f"{line}/{fails}"
      61          self.log(f"[{line}] {text}")
      62  
      63      def set_tests(self, runtests: RunTests) -> None:
      64          if runtests.forever:
      65              self.test_count_text = ''
      66              self.test_count_width = 3
      67          else:
      68              self.test_count_text = '/{}'.format(len(runtests.tests))
      69              self.test_count_width = len(self.test_count_text) - 1
      70  
      71      def start_load_tracker(self) -> None:
      72          if not MS_WINDOWS:
      73              return
      74  
      75          try:
      76              self.win_load_tracker = WindowsLoadTracker()
      77          except PermissionError as error:
      78              # Standard accounts may not have access to the performance
      79              # counters.
      80              print_warning(f'Failed to create WindowsLoadTracker: {error}')
      81  
      82      def stop_load_tracker(self) -> None:
      83          if self.win_load_tracker is None:
      84              return
      85          self.win_load_tracker.close()
      86          self.win_load_tracker = None