class_docstr: TaskController inherits from BaseController and manages business logic associated with tasks
  within the project.
  
  Attributes
  ----------
  environment : datmo.core.controller.environment.environment.EnvironmentController
      used to create environment if new definition file
  snapshot : datmo.core.controller.snapshot.SnapshotController
      used to create snapshots before and after tasks
  
  Methods
  -------
  create(dictionary)
      creates a Task object with the permanent parameters
  _run_helper(environment_id, log_filepath, options)
      helper for run to start environment and run with the appropriate parameters
  run(self, id, dictionary=None)
      runs the task and tracks the run, logs, inputs and outputs
  list(sort_key=None, sort_order=None)
      lists all tasks within the project given filters
  delete(id)
      deletes the specified task from the project
docstr: Run a task with parameters. If dictionary specified, create a new task with new run parameters.
  Snapshot objects are created before and after the task to keep track of the state. During the run,
  you can access task outputs using environment variable DATMO_TASK_DIR or `/task` which points to
  location for the task files. Create config.json, stats.json and any weights or any file such
  as graphs and visualizations within that directory for quick access
  
  Parameters
  ----------
  task_id : str
      id for the task you would like to run
  snapshot_dict : dict
      set of parameters to create a snapshot (see SnapshotController for details.
      default is None, which means dictionary with `visible` False will be added to
      hide auto-generated snapshot) NOTE: `visible` False will always be False regardless
      of whether the user provides another value for `visible`.
  task_dict : dict
      set of parameters to characterize the task run
      (default is None, which translate to {}, see datmo.core.entity.task.Task for more details on inputs)
  
  Returns
  -------
  Task
      the Task object which completed its run with updated parameters
  
  Raises
  ------
  TaskRunError
      If there is any error in creating files for the task or downstream errors
docstr_tok: ["Run","a","task","with","parameters",".","If","dictionary","specified",",","create","a","new","task","with","new","run","parameters",".","Snapshot","objects","are","created","before","and","after","the","task","to","keep","track","of","the","state",".","During","the","run",",","you","can","access","task","outputs","using","environment","variable","DATMO_TASK_DIR","or","`","\/task","`","which","points","to","location","for","the","task","files",".","Create","config.json",",","stats.json","and","any","weights","or","any","file","such","as","graphs","and","visualizations","within","that","directory","for","quick","access","Parameters","--","--","--","--","--","task_id",":","str","id","for","the","task","you","would","like","to","run","snapshot_dict",":","dict","set","of","parameters","to","create","a","snapshot","(","see","SnapshotController","for","details",".","default","is","None",",","which","means","dictionary","with","`","visible","`","False","will","be","added","to","hide","auto-generated","snapshot",")","NOTE",":","`","visible","`","False","will","always","be","False","regardless","of","whether","the","user","provides","another","value","for","`","visible","`",".","task_dict",":","dict","set","of","parameters","to","characterize","the","task","run","(","default","is","None",",","which","translate","to","{","}",",","see","datmo.core.entity.task.Task","for","more","details","on","inputs",")","Returns","--","--","--","-","Task","the","Task","object","which","completed","its","run","with","updated","parameters","Raises","--","--","--","TaskRunError","If","there","is","any","error","in","creating","files","for","the","task","or","downstream","errors"]
code: def run(self, task_id, snapshot_dict=None, task_dict=None):
      """Run a task with parameters. If dictionary specified, create a new task with new run parameters.
      Snapshot objects are created before and after the task to keep track of the state. During the run,
      you can access task outputs using environment variable DATMO_TASK_DIR or `/task` which points to
      location for the task files. Create config.json, stats.json and any weights or any file such
      as graphs and visualizations within that directory for quick access
  
      Parameters
      ----------
      task_id : str
          id for the task you would like to run
      snapshot_dict : dict
          set of parameters to create a snapshot (see SnapshotController for details.
          default is None, which means dictionary with `visible` False will be added to
          hide auto-generated snapshot) NOTE: `visible` False will always be False regardless
          of whether the user provides another value for `visible`.
      task_dict : dict
          set of parameters to characterize the task run
          (default is None, which translate to {}, see datmo.core.entity.task.Task for more details on inputs)
  
      Returns
      -------
      Task
          the Task object which completed its run with updated parameters
  
      Raises
      ------
      TaskRunError
          If there is any error in creating files for the task or downstream errors
      """
      # Ensure visible=False is present in the snapshot dictionary
      if not snapshot_dict:
          snapshot_dict = {"visible": False}
      else:
          snapshot_dict["visible"] = False
  
      if not task_dict:
          task_dict = {}
      # Obtain Task to run
      task_obj = self.dal.task.get_by_id(task_id)
  
      # Ensure that at least 1 of command, command_list,  or interactive is present in task_dict
      important_task_args = ["command", "command_list", "interactive"]
      if (
          not task_dict.get("command", task_obj.command)
          and not task_dict.get("command_list", task_obj.command_list)
          and not task_dict.get("interactive", task_obj.interactive)
      ):
          raise RequiredArgumentMissing(
              __(
                  "error",
                  "controller.task.run.arg",
                  " or ".join(important_task_args),
              )
          )
  
      if task_obj.status is None:
          task_obj.status = "RUNNING"
      else:
          raise TaskRunError(
              __("error", "cli.run.run.already_running", task_obj.id)
          )
      # Create Task directory for user during run
      task_dirpath = os.path.join(".datmo", "tasks", task_obj.id)
      try:
          _ = self.file_driver.create(task_dirpath, directory=True)
      except Exception:
          raise TaskRunError(
              __("error", "controller.task.run", task_dirpath)
          )
      # Create the before snapshot prior to execution
      before_snapshot_dict = snapshot_dict.copy()
      before_snapshot_dict["message"] = (
          "autogenerated snapshot created before task %s is run"
          % task_obj.id
      )
      before_snapshot_obj = self.snapshot.create(before_snapshot_dict)
      # Update the task with pre-execution parameters, prefer list first then look for string command
      # List command will overwrite a string command if given
      if task_dict.get("command_list", task_obj.command_list):
          task_dict["command"] = " ".join(
              task_dict.get("command_list", task_obj.command_list)
          )
      else:
          if task_dict.get("command", task_obj.command):
              task_dict["command_list"] = shlex.split(
                  task_dict.get("command", task_obj.command)
              )
          elif not task_dict.get("interactive", task_obj.interactive):
              # If it's not interactive then there is not expected task
              raise TaskNoCommandGiven()
  
      validate("create_task", task_dict)
      task_obj = self.dal.task.update(
          {
              "id": task_obj.id,
              "before_snapshot_id": task_dict.get(
                  "before_snapshot_id", before_snapshot_obj.id
              ),
              "command": task_dict.get("command", task_obj.command),
              "command_list": task_dict.get(
                  "command_list", task_obj.command_list
              ),
              "gpu": task_dict.get("gpu", False),
              "mem_limit": task_dict.get("mem_limit", None),
              "workspace": task_dict.get("workspace", None),
              "data_file_path_map": task_dict.get(
                  "data_file_path_map", task_obj.data_file_path_map
              ),
              "data_directory_path_map": task_dict.get(
                  "data_directory_path_map",
                  task_obj.data_directory_path_map,
              ),
              "interactive": task_dict.get(
                  "interactive", task_obj.interactive
              ),
              "detach": task_dict.get("detach", task_obj.detach),
              "ports": task_dict.get("ports", task_obj.ports),
              "task_dirpath": task_dict.get(
                  "task_dirpath", task_dirpath
              ),
              "log_filepath": task_dict.get(
                  "log_filepath", os.path.join(task_dirpath, "task.log")
              ),
              "start_time": task_dict.get(
                  "start_time", datetime.utcnow()
              ),
              "status": task_obj.status,
          }
      )
  
      # Copy over files from the before_snapshot file collection to task dir
      file_collection_obj = self.dal.file_collection.get_by_id(
          before_snapshot_obj.file_collection_id
      )
      self.file_driver.copytree(
          os.path.join(self.home, file_collection_obj.path),
          os.path.join(self.home, task_obj.task_dirpath),
      )
  
      return_code, run_id, logs = 0, None, None
  
      try:
          # Set the parameters set in the task
          if task_obj.detach and task_obj.interactive:
              raise TaskInteractiveDetachError(
                  __(
                      "error",
                      "controller.task.run.args.detach.interactive",
                  )
              )
  
          environment_run_options = {
              "command": task_obj.command_list,
              "ports": [] if task_obj.ports is None else task_obj.ports,
              "name": "datmo-task-" + self.model.id + "-" + task_obj.id,
              "volumes": {},
              "mem_limit": task_obj.mem_limit,
              "workspace": task_obj.workspace,
              "gpu": task_obj.gpu,
              "detach": task_obj.detach,
              "stdin_open": task_obj.interactive,
              "tty": task_obj.interactive,
              "api": False,
          }
          # mount the data volume
          environment_run_options = self._update_environment_run_options(
              environment_run_options,
              data_file_path_map=task_obj.data_file_path_map,
              data_directory_path_map=task_obj.data_directory_path_map,
          )
          # mount the project and task folder
          environment_run_options["volumes"].update(
              {
                  os.path.join(self.home, task_obj.task_dirpath): {
                      "bind": "/task/",
                      "mode": "rw",
                  },
                  self.home: {"bind": "/home/", "mode": "rw"},
              }
          )
          # Run environment via the helper function
          return_code, run_id, logs = self._run_helper(
              before_snapshot_obj.environment_id,
              environment_run_options,
              os.path.join(self.home, task_obj.log_filepath),
          )
  
      except Exception as e:
          return_code = 1
          logs += "Error running task: %" % e.message
      finally:
          # Create the after snapshot after execution is completed with new paths
          after_snapshot_dict = snapshot_dict.copy()
          after_snapshot_dict["message"] = (
              "autogenerated snapshot created after task %s is run"
              % task_obj.id
          )
  
          # Add in absolute paths from running task directory
          absolute_task_dir_path = os.path.join(
              self.home, task_obj.task_dirpath
          )
          absolute_paths = []
          for item in os.listdir(absolute_task_dir_path):
              path = os.path.join(absolute_task_dir_path, item)
              if os.path.isfile(path) or os.path.isdir(path):
                  absolute_paths.append(path)
          after_snapshot_dict.update(
              {
                  "paths": absolute_paths,
                  "environment_id": before_snapshot_obj.environment_id,
              }
          )
          after_snapshot_obj = self.snapshot.create(after_snapshot_dict)
  
          # (optional) Remove temporary task directory path
          # Update the task with post-execution parameters
          end_time = datetime.utcnow()
          duration = (end_time - task_obj.start_time).total_seconds()
          update_task_dict = {
              "id": task_obj.id,
              "after_snapshot_id": after_snapshot_obj.id,
              "logs": logs,
              "status": "SUCCESS" if return_code == 0 else "FAILED",
              # "results": task_obj.results, # TODO: update during run
              "end_time": end_time,
              "duration": duration,
          }
          if logs is not None:
              update_task_dict[
                  "results"
              ] = self._parse_logs_for_results(logs)
              if update_task_dict["results"] is not None:
                  snapshot_update_dict = dict(
                      {"id": after_snapshot_obj.id}
                  )
                  if after_snapshot_obj.stats:
                      after_snapshot_obj.stats.update(
                          update_task_dict["results"]
                      )
                  else:
                      after_snapshot_obj.stats = update_task_dict[
                          "results"
                      ]
                  snapshot_update_dict[
                      "stats"
                  ] = after_snapshot_obj.stats.copy()
                  self.dal.snapshot.update(snapshot_update_dict)
          if run_id is not None:
              update_task_dict["run_id"] = run_id
          return self.dal.task.update(update_task_dict)
  
imports: import os
  import time
  import shlex
  import threading
  import webbrowser
  from datetime import datetime
  from datmo.core.controller.base import BaseController
  from datmo.core.controller.snapshot import SnapshotController
  from datmo.core.controller.environment.environment import EnvironmentController
  from datmo.core.entity.task import Task
  from datmo.core.util.validation import validate
  from datmo.core.util.spinner import Spinner
  from datmo.core.util.i18n import get
  from datmo.core.util.exceptions import TaskRunError, RequiredArgumentMissing, ProjectNotInitialized, PathDoesNotExist, TaskInteractiveDetachError, TooManyArgumentsFound, EntityNotFound, DoesNotExist, TaskNoCommandGiven
label_desc: Use image node_id 7 for calling the TaskController obj's underlying member method code with example usage: obj.run(task_id, snapshot_dict, task_dict) and returns: self