docstr: Rewrite a batch of lines to reduce duplicate log messages.
  
  Args:
      batch: The batch of lines from a single source.
  
  Returns:
      List of batches from this and possibly other previous sources to print.
docstr_tok: ["Rewrite","a","batch","of","lines","to","reduce","duplicate","log","messages",".","Args",":","batch",":","The","batch","of","lines","from","a","single","source",".","Returns",":","List","of","batches","from","this","and","possibly","other","previous","sources","to","print","."]
code: def deduplicate(self, batch: LogBatch) -> List[LogBatch]:
      """Rewrite a batch of lines to reduce duplicate log messages.
  
      Args:
          batch: The batch of lines from a single source.
  
      Returns:
          List of batches from this and possibly other previous sources to print.
      """
      if not RAY_DEDUP_LOGS:
          return [batch]
  
      now = self.timesource()
      metadata = batch.copy()
      del metadata["lines"]
      source = (metadata.get("ip"), metadata.get("pid"))
      output: List[LogBatch] = [dict(**metadata, lines=[])]
  
      # Decide which lines to emit from the input batch. Put the outputs in the
      # first output log batch (output[0]).
      for line in batch["lines"]:
          if RAY_TQDM_MAGIC in line or (
              self.allow_re and self.allow_re.search(line)
          ):
              output[0]["lines"].append(line)
              continue
          elif self.skip_re and self.skip_re.search(line):
              continue
          dedup_key = _canonicalise_log_line(line)
  
          if dedup_key in self.recent:
              sources = self.recent[dedup_key].sources
              sources.add(source)
              # We deduplicate the warnings/errorm essages from
              # raylet by default.
              if len(sources) > 1 or batch["pid"] == "raylet":
                  state = self.recent[dedup_key]
                  self.recent[dedup_key] = DedupState(
                      state.timestamp,
                      state.count + 1,
                      line,
                      metadata,
                      sources,
                  )
              else:
                  # Don't dedup messages from the same source, just print.
                  output[0]["lines"].append(line)
          else:
              self.recent[dedup_key] = DedupState(
                  now, 0, line, metadata, {source}
              )
              output[0]["lines"].append(line)
  
      # Flush patterns from the buffer that are older than the aggregation window.
      while self.recent:
          if (
              now - next(iter(self.recent.values())).timestamp
              < self.agg_window_s
          ):
              break
          dedup_key = next(iter(self.recent))
          state = self.recent.pop(dedup_key)
          # we already logged an instance of this line immediately when received,
          # so don't log for count == 0
          if state.count > 1:
              # (Actor pid=xxxx) [repeated 2x across cluster] ...
              output.append(
                  dict(**state.metadata, lines=[state.formatted()])
              )
              # Continue aggregating for this key but reset timestamp and count.
              state.timestamp = now
              state.count = 0
              self.recent[dedup_key] = state
          elif state.count > 0:
              # Aggregation wasn't fruitful, print the line and stop aggregating.
              output.append(dict(state.metadata, lines=[state.line]))
  
      return output
  
imports: import colorama
  from dataclasses import dataclass
  import logging
  import os
  import re
  import sys
  import threading
  import time
  from typing import Callable, Dict, List, Set, Tuple, Any, Optional
  import ray
  from ray.experimental.tqdm_ray import RAY_TQDM_MAGIC
  from ray._private.ray_constants import RAY_DEDUP_LOGS, RAY_DEDUP_LOGS_AGG_WINDOW_S, RAY_DEDUP_LOGS_ALLOW_REGEX, RAY_DEDUP_LOGS_SKIP_REGEX
  from ray.util.debug import log_once
label_desc: Use image node_id 2 for calling the LogDeduplicator obj's underlying member method code with example usage: obj.deduplicate(batch) and returns: output, list