class_docstr: Runs job while capturing environment and logs results.
      
class_docstr_tok: ["Runs","job","while","capturing","environment","and","logs","results","."]
code: def run(self, experiment):
      if isinstance(experiment, six.string_types):
          experiment = self.db.get_experiment(experiment)
      elif not isinstance(experiment, Experiment):
          raise ValueError(
              "Unknown type of experiment: " + str(type(experiment))
          )
  
      self.logger.info("Experiment key: " + experiment.key)
  
      with model.get_db_provider(self.config) as db:
          db.start_experiment(experiment)
  
          """ Override env variables with those inside the queued message
              """
          env = dict(os.environ)
          if "env" in self.config.keys():
              for k, v in six.iteritems(self.config["env"]):
                  if v is not None:
                      env[str(k)] = str(v)
  
          env["PYTHONUNBUFFERED"] = "TRUE"
  
          fs_tracker.setup_experiment(env, experiment, clean=False)
          log_path = fs_tracker.get_artifact_cache(
              "output", experiment.key
          )
  
          self.logger.debug("Child process environment:")
          self.logger.debug(str(env))
  
          sched = BackgroundScheduler()
          sched.start()
  
          with open(log_path, "w") as output_file:
              python = "python"
              if experiment.pythonver[0] == "3":
                  python = "python3"
  
              python = which(python)
  
              cmd = [python, experiment.filename] + experiment.args
              cwd = experiment.artifacts["workspace"].local_path
              container_artifact = experiment.artifacts.get(
                  "_singularity"
              )
              if container_artifact:
                  container = container_artifact.get("local")
                  if not container:
                      container = container_artifact.get("qualified")
  
                  cwd = fs_tracker.get_artifact_cache(
                      "workspace", experiment.key
                  )
  
                  for tag, art in six.iteritems(experiment.artifacts):
                      local_path = art.get("local")
                      if not art["mutable"] and os.path.exists(
                          local_path
                      ):
                          os.symlink(
                              art["local"],
                              os.path.join(os.path.dirname(cwd), tag),
                          )
  
                  if experiment.filename is not None:
                      cmd = [
                          "singularity",
                          "exec",
                          container,
                      ] + cmd
                  else:
                      cmd = ["singularity", "run", container]
  
              self.logger.info(
                  "Running cmd: {0} in {1}".format(cmd, cwd)
              )
  
              p = subprocess.Popen(
                  cmd,
                  stdout=subprocess.PIPE,
                  stderr=subprocess.STDOUT,
                  env=env,
                  cwd=cwd,
                  text=True,
              )
  
              def kill_subprocess():
                  p.kill()
  
              def get_duration(tag: str):
                  value = self.config.get(tag, "0m")
                  return int(str2duration(value).total_seconds() / 60)
  
              def checkpoint():
                  try:
                      db.checkpoint_experiment(experiment)
                  except BaseException as e:
                      self.logger.info(e)
                      check_for_kb_interrupt()
  
              minutes = get_duration("saveWorkspaceFrequency")
              sched.add_job(checkpoint, "interval", minutes=minutes)
  
              metrics_path = fs_tracker.get_artifact_cache(
                  "_metrics", experiment.key
              )
  
              minutes = get_duration("saveMetricsFrequency")
              sched.add_job(
                  lambda: save_metrics(metrics_path),
                  "interval",
                  minutes=minutes,
              )
  
              def kill_if_stopped():
                  try:
                      db_expr = db.get_experiment(
                          experiment.key, getinfo=False
                      )
                  except:
                      check_for_kb_interrupt()
                      db_expr = None
  
                  # Transient issues with getting experiment data might
                  # result in a None value being returned, as result
                  # leave the experiment running because we wont be able to
                  # do anything else even if this experiment is stopped
                  # in any event if the experiment runs too long then it
                  # will exceed its allocated time and stop
                  if db_expr is not None:
                      if db_expr.status == "stopped":
                          kill_subprocess()
                          return
  
                  if (
                      experiment.max_duration is not None
                      and time.time()
                      > experiment.time_started
                      + int(
                          str2duration(
                              experiment.max_duration
                          ).total_seconds()
                      )
                  ):
                      kill_subprocess()
                      return
  
                  # If our tasks queue is signalled inactive
                  # during work process execution, that means we need to drop
                  # current execution and exit
                  if not self.task_queue.is_active():
                      kill_subprocess()
  
              sched.add_job(kill_if_stopped, "interval", seconds=10)
  
              while True:
                  output = p.stdout.readline()
                  if output == "" and p.poll() is not None:
                      break
                  if output:
                      line_out = output.strip()
                      print(line_out)
                      output_file.write(line_out)
  
              try:
                  p.wait()
              finally:
                  save_metrics(metrics_path)
                  sched.shutdown()
                  db.checkpoint_experiment(experiment)
                  db.finish_experiment(experiment)
                  return p.returncode
  
imports: import os
  import sys
  import subprocess
  import argparse
  import json
  import psutil
  import time
  import six
  import signal
  import pdb
  from apscheduler.schedulers.background import BackgroundScheduler
  from studio import fs_tracker, model
  from studio.util import logs
  from studio.queues.local_queue import LocalQueue
  from studio.util.gpu_util import get_available_gpus, get_gpu_mapping, get_gpus_summary
  from studio.artifacts.artifact import Artifact
  from studio.experiments.experiment import Experiment
  from studio.util.util import sixdecode, str2duration, retry, parse_verbosity, check_for_kb_interrupt
label_desc: Use image node_id 2 for calling the LocalExecutor obj's underlying member method code with example usage: obj.run(experiment) and returns: int, p