text: __all__ = ["Monitor", "ResultsWriter", "get_monitor_files", "load_results"]
  import csv
  import json
  import os
  import time
  from glob import glob
  from typing import Dict, List, Optional, Tuple, Union
  
  import gym
  import numpy as np
  import pandas
  
  from stable_baselines3.common.type_aliases import GymObs, GymStepReturn
  
  
  class Monitor(gym.Wrapper):
      EXT = "monitor.csv"
  
      def __init__(
          self,
          env: gym.Env,
          filename: Optional[str] = None,
          allow_early_resets: bool = True,
          reset_keywords: Tuple[str, ...] = (),
          info_keywords: Tuple[str, ...] = (),
      ):
          super(Monitor, self).__init__(env=env)
          self.t_start = time.time()
          if filename is not None:
              self.results_writer = ResultsWriter(
                  filename,
                  header={"t_start": self.t_start, "env_id": env.spec and env.spec.id},
                  extra_keys=reset_keywords + info_keywords,
              )
          else:
              self.results_writer = None
          self.reset_keywords = reset_keywords
          self.info_keywords = info_keywords
          self.allow_early_resets = allow_early_resets
          self.rewards = None
          self.needs_reset = True
          self.episode_returns = []
          self.episode_lengths = []
          self.episode_times = []
          self.total_steps = 0
          self.current_reset_info = {}
  
      def reset(self, **kwargs) -> GymObs:
          if not self.allow_early_resets and not self.needs_reset:
              raise RuntimeError(
                  "Tried to reset an environment before done. If you want to allow early resets, "
                  "wrap your env with Monitor(env, path, allow_early_resets=True)"
              )
          self.rewards = []
          self.needs_reset = False
          for key in self.reset_keywords:
              value = kwargs.get(key)
              if value is None:
                  raise ValueError(
                      f"Expected you to pass keyword argument {key} into reset"
                  )
              self.current_reset_info[key] = value
          return self.env.reset(**kwargs)
  
      def step(self, action: Union[np.ndarray, int]) -> GymStepReturn:
          if self.needs_reset:
              raise RuntimeError("Tried to step environment that needs reset")
          observation, reward, done, info = self.env.step(action)
          self.rewards.append(reward)
          if done:
              self.needs_reset = True
              ep_rew = sum(self.rewards)
              ep_len = len(self.rewards)
              ep_info = {
                  "r": round(ep_rew, 6),
                  "l": ep_len,
                  "t": round(time.time() - self.t_start, 6),
              }
              for key in self.info_keywords:
                  ep_info[key] = info[key]
              self.episode_returns.append(ep_rew)
              self.episode_lengths.append(ep_len)
              self.episode_times.append(time.time() - self.t_start)
              ep_info.update(self.current_reset_info)
              if self.results_writer:
                  self.results_writer.write_row(ep_info)
              info["episode"] = ep_info
          self.total_steps += 1
          return observation, reward, done, info
  
      def close(self) -> None:
          super(Monitor, self).close()
          if self.results_writer is not None:
              self.results_writer.close()
  
      def get_total_steps(self) -> int:
          return self.total_steps
  
      def get_episode_rewards(self) -> List[float]:
          return self.episode_returns
  
      def get_episode_lengths(self) -> List[int]:
          return self.episode_lengths
  
      def get_episode_times(self) -> List[float]:
          return self.episode_times
  
  
  class LoadMonitorResultsError(Exception):
      pass
  
  
  class ResultsWriter:
      def __init__(
          self,
          filename: str = "",
          header: Dict[str, Union[float, str]] = None,
          extra_keys: Tuple[str, ...] = (),
      ):
          if header is None:
              header = {}
          if not filename.endswith(Monitor.EXT):
              if os.path.isdir(filename):
                  filename = os.path.join(filename, Monitor.EXT)
              else:
                  filename = filename + "." + Monitor.EXT
          self.file_handler = open(filename, "wt")
          self.file_handler.write("#%s\n" % json.dumps(header))
          self.logger = csv.DictWriter(
              self.file_handler, fieldnames=("r", "l", "t") + extra_keys
          )
          self.logger.writeheader()
          self.file_handler.flush()
  
      def write_row(self, epinfo: Dict[str, Union[float, int]]) -> None:
          if self.logger:
              self.logger.writerow(epinfo)
              self.file_handler.flush()
  
      def close(self) -> None:
          self.file_handler.close()
  
  
  def get_monitor_files(path: str) -> List[str]:
      return glob(os.path.join(path, "*" + Monitor.EXT))
  
  
  def load_results(path: str) -> pandas.DataFrame:
      monitor_files = get_monitor_files(path)
      if len(monitor_files) == 0:
          raise LoadMonitorResultsError(
              f"No monitor files of the form *{Monitor.EXT} found in {path}"
          )
      data_frames, headers = [], []
      for file_name in monitor_files:
          with open(file_name, "rt") as file_handler:
              first_line = file_handler.readline()
              assert first_line[0] == "#"
              header = json.loads(first_line[1:])
              data_frame = pandas.read_csv(file_handler, index_col=None)
              headers.append(header)
              data_frame["t"] += header["t_start"]
          data_frames.append(data_frame)
      data_frame = pandas.concat(data_frames)
      data_frame.sort_values("t", inplace=True)
      data_frame.reset_index(inplace=True)
      data_frame["t"] -= min(header["t_start"] for header in headers)
      return data_frame
  