code: def run_command(
      self,
      command: str,
      env: Optional[Dict] = None,
      timeout: float = 3600.0,
      raise_on_timeout: bool = True,
  ) -> float:
      full_env = self.get_full_command_env(env)
  
      if full_env:
          env_str = (
              " ".join(f"{k}={v}" for k, v in full_env.items()) + " "
          )
      else:
          env_str = ""
  
      full_command = f"{env_str}{command}"
      logger.info(
          f"Running command in cluster {self.cluster_manager.cluster_name}: "
          f"{full_command}"
      )
  
      logger.info(
          f"Link to cluster: "
          f"{format_link(self.cluster_manager.get_cluster_url())}"
      )
  
      status_code, time_taken = self.job_manager.run_and_wait(
          full_command, full_env, working_dir=".", timeout=int(timeout)
      )
  
      if status_code != 0:
          raise CommandError(
              f"Command returned non-success status: {status_code}"
          )
  
      return time_taken
  
imports: import json
  import os
  import tempfile
  from typing import TYPE_CHECKING, Any, Dict, Optional
  from ray_release.cluster_manager.cluster_manager import ClusterManager
  from ray_release.command_runner.command_runner import CommandRunner
  from ray_release.exception import ClusterNodesWaitTimeout, CommandError, CommandTimeout, LocalEnvSetupError, LogsError, FetchResultError
  from ray_release.file_manager.file_manager import FileManager
  from ray_release.job_manager import JobManager
  from ray_release.logger import logger
  from ray_release.util import format_link, get_anyscale_sdk
  from ray_release.wheels import install_matching_ray_locally
label_desc: Use image node_id 7 for calling the JobRunner obj's underlying member method code with example usage: obj.run_command(command, env, timeout, raise_on_timeout) and returns: time_taken