code: def run_command(
      self,
      command: str,
      env: Optional[Dict] = None,
      timeout: float = 3600.0,
      raise_on_timeout: bool = True,
      pip: Optional[List[str]] = None,
  ) -> float:
      prepare_command_strs = []
      prepare_command_timeouts = []
      # Convert the prepare commands, envs and timeouts into shell-compliant
      # strings that can be passed to the wrapper script
      for (
          prepare_command,
          prepare_env,
          prepare_timeout,
      ) in self.prepare_commands:
          prepare_env = self.get_full_command_env(prepare_env)
          env_str = _get_env_str(prepare_env)
          prepare_command_strs.append(f"{env_str} {prepare_command}")
          prepare_command_timeouts.append(prepare_timeout)
  
      prepare_commands_shell = " ".join(
          shlex.quote(str(x)) for x in prepare_command_strs
      )
      prepare_commands_timeouts_shell = " ".join(
          shlex.quote(str(x)) for x in prepare_command_timeouts
      )
  
      full_env = self.get_full_command_env(env)
  
      no_raise_on_timeout_str = (
          " --test-no-raise-on-timeout" if not raise_on_timeout else ""
      )
      full_command = (
          f"python anyscale_job_wrapper.py '{command}' "
          f"--test-workload-timeout {timeout}{no_raise_on_timeout_str} "
          "--results-cloud-storage-uri "
          f"'{join_cloud_storage_paths(self.upload_path, self._RESULT_OUTPUT_JSON)}' "
          "--metrics-cloud-storage-uri "
          f"'"
          f"{join_cloud_storage_paths(self.upload_path, self._METRICS_OUTPUT_JSON)}' "
          "--output-cloud-storage-uri "
          f"'{join_cloud_storage_paths(self.upload_path, self.output_json)}' "
          f"--upload-cloud-storage-uri '{self.upload_path}' "
          f"--prepare-commands {prepare_commands_shell} "
          f"--prepare-commands-timeouts {prepare_commands_timeouts_shell} "
      )
      if self._artifact_path:
          full_command += f"--artifact-path '{self._artifact_path}' "
  
      timeout = min(
          (self.cluster_manager.maximum_uptime_minutes - 1) * 60,
          # The timeout set here is just for the prepare commands + test workload
          # WITHOUT wait for nodes time included, as that is set separately.
          # Since wait for nodes is a part of prepare_commands, we manually
          # subtract the timeout for it here.
          # We also add 15 mins for upload & metrics collection.
          timeout
          + sum(prepare_command_timeouts)
          - self._wait_for_nodes_timeout
          + 900,
      )
  
      job_status_code, time_taken = self.job_manager.run_and_wait(
          full_command,
          full_env,
          working_dir=".",
          upload_path=self.upload_path,
          timeout=int(timeout),
          pip=pip,
      )
      try:
          error = self.job_manager.last_job_result.state.error
      except AttributeError:
          error = None
  
      self._handle_command_output(
          job_status_code, error, raise_on_timeout=raise_on_timeout
      )
  
      return time_taken
  
imports: import json
  import os
  import re
  import tempfile
  import shlex
  from typing import TYPE_CHECKING, Any, Dict, Optional, List
  from ray_release.cluster_manager.cluster_manager import ClusterManager
  from ray_release.command_runner.job_runner import JobRunner
  from ray_release.exception import TestCommandTimeout, TestCommandError, JobOutOfRetriesError, PrepareCommandError, PrepareCommandTimeout, JobBrokenError, FetchResultError, JobTerminatedBeforeStartError, JobNoLogsError, JobTerminatedError
  from ray_release.file_manager.job_file_manager import JobFileManager
  from ray_release.job_manager import AnyscaleJobManager
  from ray_release.logger import logger
  from ray_release.util import join_cloud_storage_paths, generate_tmp_cloud_storage_path, get_anyscale_sdk, S3_CLOUD_STORAGE
label_desc: Use image node_id 8 for calling the AnyscaleJobRunner obj's underlying member method code with example usage: obj.run_command(command, env, timeout, raise_on_timeout, pip) and returns: time_taken