Remote

swerex.runtime.remote.RemoteRuntime

RemoteRuntime(*, logger: Logger | None = None, **kwargs: Any)

Bases: AbstractRuntime

A runtime that connects to a remote server.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Keyword arguments to pass to the RemoteRuntimeConfig constructor.	{}

Source code in swerex/runtime/remote.py

def __init__(
    self,
    *,
    logger: logging.Logger | None = None,
    **kwargs: Any,
):
    """A runtime that connects to a remote server.

    Args:
        **kwargs: Keyword arguments to pass to the `RemoteRuntimeConfig` constructor.
    """
    self._config = RemoteRuntimeConfig(**kwargs)
    self.logger = logger or get_logger("rex-runtime")
    if not self._config.host.startswith("http"):
        self.logger.warning("Host %s does not start with http, adding http://", self._config.host)
        self._config.host = f"http://{self._config.host}"

logger instance-attribute

logger = logger or get_logger('rex-runtime')

close async

close() -> CloseResponse

Closes the runtime.

Source code in swerex/runtime/remote.py

async def close(self) -> CloseResponse:
    """Closes the runtime."""
    return self._request("close", None, CloseResponse)

close_session async

close_session(request: CloseSessionRequest) -> CloseSessionResponse

Closes a shell session.

Source code in swerex/runtime/remote.py

async def close_session(self, request: CloseSessionRequest) -> CloseSessionResponse:
    """Closes a shell session."""
    return self._request("close_session", request, CloseSessionResponse)

create_session async

create_session(request: CreateSessionRequest) -> CreateSessionResponse

Creates a new session.

Source code in swerex/runtime/remote.py

async def create_session(self, request: CreateSessionRequest) -> CreateSessionResponse:
    """Creates a new session."""
    return self._request("create_session", request, CreateSessionResponse)

execute async

execute(command: Command) -> CommandResponse

Executes a command (independent of any shell session).

Source code in swerex/runtime/remote.py

async def execute(self, command: Command) -> CommandResponse:
    """Executes a command (independent of any shell session)."""
    return self._request("execute", command, CommandResponse)

from_config classmethod

from_config(config: RemoteRuntimeConfig) -> Self

Source code in swerex/runtime/remote.py

@classmethod
def from_config(cls, config: RemoteRuntimeConfig) -> Self:
    return cls(**config.model_dump())

is_alive async

is_alive(*, timeout: float | None = None) -> IsAliveResponse

Checks if the runtime is alive.

Internal server errors are thrown, everything else just has us return False together with the message.

Source code in swerex/runtime/remote.py

async def is_alive(self, *, timeout: float | None = None) -> IsAliveResponse:
    """Checks if the runtime is alive.

    Internal server errors are thrown, everything else just has us return False
    together with the message.
    """
    try:
        response = requests.get(
            f"{self._api_url}/is_alive", headers=self._headers, timeout=self._get_timeout(timeout)
        )
        if response.status_code == 200:
            return IsAliveResponse(**response.json())
        elif response.status_code == 511:
            exc_transfer = _ExceptionTransfer(**response.json()["swerexception"])
            self._handle_transfer_exception(exc_transfer)
        msg = (
            f"Status code {response.status_code} from {self._api_url}/is_alive. "
            f"Message: {response.json().get('detail')}"
        )
        return IsAliveResponse(is_alive=False, message=msg)
    except requests.RequestException:
        msg = f"Failed to connect to {self._config.host}\n"
        msg += traceback.format_exc()
        return IsAliveResponse(is_alive=False, message=msg)
    except Exception:
        msg = f"Failed to connect to {self._config.host}\n"
        msg += traceback.format_exc()
        return IsAliveResponse(is_alive=False, message=msg)

read_file async

read_file(request: ReadFileRequest) -> ReadFileResponse

Reads a file

Source code in swerex/runtime/remote.py

async def read_file(self, request: ReadFileRequest) -> ReadFileResponse:
    """Reads a file"""
    return self._request("read_file", request, ReadFileResponse)

run_in_session async

run_in_session(action: Action) -> Observation

Runs a command in a session.

Source code in swerex/runtime/remote.py

async def run_in_session(self, action: Action) -> Observation:
    """Runs a command in a session."""
    return self._request("run_in_session", action, Observation)

upload async

upload(request: UploadRequest) -> UploadResponse

Uploads a file

Source code in swerex/runtime/remote.py

async def upload(self, request: UploadRequest) -> UploadResponse:
    """Uploads a file"""
    source = Path(request.source_path).resolve()
    self.logger.debug("Uploading file from %s to %s", request.source_path, request.target_path)
    if source.is_dir():
        with tempfile.TemporaryDirectory() as temp_dir:
            zip_path = Path(temp_dir) / "zipped_transfer.zip"
            shutil.make_archive(str(zip_path.with_suffix("")), "zip", source)
            self.logger.debug("Created zip file at %s", zip_path)
            files = {"file": zip_path.open("rb")}
            data = {"target_path": request.target_path, "unzip": "true"}
            response = requests.post(f"{self._api_url}/upload", files=files, data=data, headers=self._headers)
            self._handle_response_errors(response)
            return UploadResponse(**response.json())
    elif source.is_file():
        self.logger.debug("Uploading file from %s to %s", source, request.target_path)
        files = {"file": source.open("rb")}
        data = {"target_path": request.target_path, "unzip": "false"}
        response = requests.post(f"{self._api_url}/upload", files=files, data=data, headers=self._headers)
        self._handle_response_errors(response)
        return UploadResponse(**response.json())
    else:
        msg = f"Source path {source} is not a file or directory"
        raise ValueError(msg)

wait_until_alive async

wait_until_alive(*, timeout: float = 60.0)

Source code in swerex/runtime/remote.py

async def wait_until_alive(self, *, timeout: float = 60.0):
    return await _wait_until_alive(self.is_alive, timeout=timeout)

write_file async

write_file(request: WriteFileRequest) -> WriteFileResponse

Writes a file

Source code in swerex/runtime/remote.py

async def write_file(self, request: WriteFileRequest) -> WriteFileResponse:
    """Writes a file"""
    return self._request("write_file", request, WriteFileResponse)