# Generated from soulfire/bot.proto. DO NOT EDIT.
# Generated by protoc-gen-connectrpc-py v0.11.1 with parameter "protobuf=google".
# ruff: noqa: PGH004
# ruff: noqa
# fmt: off
from __future__ import annotations
from typing import Protocol, TYPE_CHECKING
from connectrpc.client import ConnectClient, ConnectClientSync
from connectrpc.code import Code
from connectrpc.compat import google_protobuf_binary_codec, google_protobuf_codecs
from connectrpc.compression.gzip import GzipCompression
from connectrpc.errors import ConnectError
from connectrpc.method import IdempotencyLevel, MethodInfo
from connectrpc.protocol import ProtocolType
from connectrpc.server import ConnectASGIApplication, ConnectWSGIApplication, Endpoint, EndpointSync
from pyqwest import Client, SyncClient
from .bot_pb2 import BotClickDialogButtonRequest, BotClickDialogButtonResponse, BotCloseContainerRequest, BotCloseContainerResponse, BotCloseDialogRequest, BotCloseDialogResponse, BotContainerButtonClickRequest, BotContainerButtonClickResponse, BotGetDialogRequest, BotGetDialogResponse, BotInfoRequest, BotInfoResponse, BotInventoryClickRequest, BotInventoryClickResponse, BotInventoryStateRequest, BotInventoryStateResponse, BotListRequest, BotListResponse, BotMouseClickRequest, BotMouseClickResponse, BotOpenInventoryRequest, BotOpenInventoryResponse, BotPovFrame, BotRenderPovRequest, BotRenderPovResponse, BotResetMovementRequest, BotResetMovementResponse, BotSetContainerTextRequest, BotSetContainerTextResponse, BotSetHotbarSlotRequest, BotSetHotbarSlotResponse, BotSetMovementStateRequest, BotSetMovementStateResponse, BotSetRotationRequest, BotSetRotationResponse, BotSubmitDialogRequest, BotSubmitDialogResponse, BotUpdateConfigEntryRequest, BotUpdateConfigEntryResponse, BotWatchPovRequest, BotWorldMapRequest, BotWorldMapResponse, RestartBotsRequest, RestartBotsResponse, SetBotsDesiredStateRequest, SetBotsDesiredStateResponse, WatchBotStatusesRequest, WatchBotStatusesResponse
if TYPE_CHECKING:
from collections.abc import AsyncGenerator, AsyncIterator, Iterable, Iterator, Mapping
from connectrpc.codec import Codec
from connectrpc.compression import Compression
from connectrpc.interceptor import Interceptor, InterceptorSync
from connectrpc.request import Headers, RequestContext
_DEFAULT_CODECS = google_protobuf_codecs()
_PROTO_BINARY_CODEC = google_protobuf_binary_codec()
_GZIP_COMPRESSION = GzipCompression()
[docs]
class BotService(Protocol):
[docs]
async def set_bots_desired_state(self, request: SetBotsDesiredStateRequest, ctx: RequestContext[SetBotsDesiredStateRequest, SetBotsDesiredStateResponse]) -> SetBotsDesiredStateResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def restart_bots(self, request: RestartBotsRequest, ctx: RequestContext[RestartBotsRequest, RestartBotsResponse]) -> RestartBotsResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def watch_bot_statuses(self, request: WatchBotStatusesRequest, ctx: RequestContext[WatchBotStatusesRequest, WatchBotStatusesResponse]) -> AsyncIterator[WatchBotStatusesResponse]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def get_bot_list(self, request: BotListRequest, ctx: RequestContext[BotListRequest, BotListResponse]) -> BotListResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def get_bot_info(self, request: BotInfoRequest, ctx: RequestContext[BotInfoRequest, BotInfoResponse]) -> BotInfoResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def update_bot_config_entry(self, request: BotUpdateConfigEntryRequest, ctx: RequestContext[BotUpdateConfigEntryRequest, BotUpdateConfigEntryResponse]) -> BotUpdateConfigEntryResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def render_bot_pov(self, request: BotRenderPovRequest, ctx: RequestContext[BotRenderPovRequest, BotRenderPovResponse]) -> BotRenderPovResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def watch_bot_pov(self, request: BotWatchPovRequest, ctx: RequestContext[BotWatchPovRequest, BotPovFrame]) -> AsyncIterator[BotPovFrame]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def get_bot_world_map(self, request: BotWorldMapRequest, ctx: RequestContext[BotWorldMapRequest, BotWorldMapResponse]) -> BotWorldMapResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def click_inventory_slot(self, request: BotInventoryClickRequest, ctx: RequestContext[BotInventoryClickRequest, BotInventoryClickResponse]) -> BotInventoryClickResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def get_inventory_state(self, request: BotInventoryStateRequest, ctx: RequestContext[BotInventoryStateRequest, BotInventoryStateResponse]) -> BotInventoryStateResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def close_container(self, request: BotCloseContainerRequest, ctx: RequestContext[BotCloseContainerRequest, BotCloseContainerResponse]) -> BotCloseContainerResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def open_inventory(self, request: BotOpenInventoryRequest, ctx: RequestContext[BotOpenInventoryRequest, BotOpenInventoryResponse]) -> BotOpenInventoryResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def mouse_click(self, request: BotMouseClickRequest, ctx: RequestContext[BotMouseClickRequest, BotMouseClickResponse]) -> BotMouseClickResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def set_container_text(self, request: BotSetContainerTextRequest, ctx: RequestContext[BotSetContainerTextRequest, BotSetContainerTextResponse]) -> BotSetContainerTextResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def set_hotbar_slot(self, request: BotSetHotbarSlotRequest, ctx: RequestContext[BotSetHotbarSlotRequest, BotSetHotbarSlotResponse]) -> BotSetHotbarSlotResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def set_movement_state(self, request: BotSetMovementStateRequest, ctx: RequestContext[BotSetMovementStateRequest, BotSetMovementStateResponse]) -> BotSetMovementStateResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def reset_movement(self, request: BotResetMovementRequest, ctx: RequestContext[BotResetMovementRequest, BotResetMovementResponse]) -> BotResetMovementResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def set_rotation(self, request: BotSetRotationRequest, ctx: RequestContext[BotSetRotationRequest, BotSetRotationResponse]) -> BotSetRotationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def get_dialog(self, request: BotGetDialogRequest, ctx: RequestContext[BotGetDialogRequest, BotGetDialogResponse]) -> BotGetDialogResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def submit_dialog(self, request: BotSubmitDialogRequest, ctx: RequestContext[BotSubmitDialogRequest, BotSubmitDialogResponse]) -> BotSubmitDialogResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def close_dialog(self, request: BotCloseDialogRequest, ctx: RequestContext[BotCloseDialogRequest, BotCloseDialogResponse]) -> BotCloseDialogResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
class BotServiceASGIApplication(ConnectASGIApplication[BotService]):
def __init__(
self,
service: BotService | AsyncGenerator[BotService],
*,
interceptors: Iterable[Interceptor] = (),
read_max_bytes: int | None = None,
compressions: Iterable[Compression] | None = None,
codecs: Iterable[Codec] | None = _DEFAULT_CODECS,
) -> None:
super().__init__(
service=service,
endpoints=lambda svc: {
"/soulfire.v1.BotService/SetBotsDesiredState": Endpoint.unary(
method=MethodInfo(
name="SetBotsDesiredState",
service_name="soulfire.v1.BotService",
input=SetBotsDesiredStateRequest,
output=SetBotsDesiredStateResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.set_bots_desired_state,
),
"/soulfire.v1.BotService/RestartBots": Endpoint.unary(
method=MethodInfo(
name="RestartBots",
service_name="soulfire.v1.BotService",
input=RestartBotsRequest,
output=RestartBotsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.restart_bots,
),
"/soulfire.v1.BotService/WatchBotStatuses": Endpoint.server_stream(
method=MethodInfo(
name="WatchBotStatuses",
service_name="soulfire.v1.BotService",
input=WatchBotStatusesRequest,
output=WatchBotStatusesResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.watch_bot_statuses,
),
"/soulfire.v1.BotService/GetBotList": Endpoint.unary(
method=MethodInfo(
name="GetBotList",
service_name="soulfire.v1.BotService",
input=BotListRequest,
output=BotListResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.get_bot_list,
),
"/soulfire.v1.BotService/GetBotInfo": Endpoint.unary(
method=MethodInfo(
name="GetBotInfo",
service_name="soulfire.v1.BotService",
input=BotInfoRequest,
output=BotInfoResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.get_bot_info,
),
"/soulfire.v1.BotService/UpdateBotConfigEntry": Endpoint.unary(
method=MethodInfo(
name="UpdateBotConfigEntry",
service_name="soulfire.v1.BotService",
input=BotUpdateConfigEntryRequest,
output=BotUpdateConfigEntryResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.update_bot_config_entry,
),
"/soulfire.v1.BotService/RenderBotPov": Endpoint.unary(
method=MethodInfo(
name="RenderBotPov",
service_name="soulfire.v1.BotService",
input=BotRenderPovRequest,
output=BotRenderPovResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.render_bot_pov,
),
"/soulfire.v1.BotService/WatchBotPov": Endpoint.server_stream(
method=MethodInfo(
name="WatchBotPov",
service_name="soulfire.v1.BotService",
input=BotWatchPovRequest,
output=BotPovFrame,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.watch_bot_pov,
),
"/soulfire.v1.BotService/GetBotWorldMap": Endpoint.unary(
method=MethodInfo(
name="GetBotWorldMap",
service_name="soulfire.v1.BotService",
input=BotWorldMapRequest,
output=BotWorldMapResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.get_bot_world_map,
),
"/soulfire.v1.BotService/ClickInventorySlot": Endpoint.unary(
method=MethodInfo(
name="ClickInventorySlot",
service_name="soulfire.v1.BotService",
input=BotInventoryClickRequest,
output=BotInventoryClickResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.click_inventory_slot,
),
"/soulfire.v1.BotService/GetInventoryState": Endpoint.unary(
method=MethodInfo(
name="GetInventoryState",
service_name="soulfire.v1.BotService",
input=BotInventoryStateRequest,
output=BotInventoryStateResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.get_inventory_state,
),
"/soulfire.v1.BotService/CloseContainer": Endpoint.unary(
method=MethodInfo(
name="CloseContainer",
service_name="soulfire.v1.BotService",
input=BotCloseContainerRequest,
output=BotCloseContainerResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.close_container,
),
"/soulfire.v1.BotService/OpenInventory": Endpoint.unary(
method=MethodInfo(
name="OpenInventory",
service_name="soulfire.v1.BotService",
input=BotOpenInventoryRequest,
output=BotOpenInventoryResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.open_inventory,
),
"/soulfire.v1.BotService/MouseClick": Endpoint.unary(
method=MethodInfo(
name="MouseClick",
service_name="soulfire.v1.BotService",
input=BotMouseClickRequest,
output=BotMouseClickResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.mouse_click,
),
"/soulfire.v1.BotService/ClickContainerButton": Endpoint.unary(
method=MethodInfo(
name="ClickContainerButton",
service_name="soulfire.v1.BotService",
input=BotContainerButtonClickRequest,
output=BotContainerButtonClickResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.click_container_button,
),
"/soulfire.v1.BotService/SetContainerText": Endpoint.unary(
method=MethodInfo(
name="SetContainerText",
service_name="soulfire.v1.BotService",
input=BotSetContainerTextRequest,
output=BotSetContainerTextResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.set_container_text,
),
"/soulfire.v1.BotService/SetHotbarSlot": Endpoint.unary(
method=MethodInfo(
name="SetHotbarSlot",
service_name="soulfire.v1.BotService",
input=BotSetHotbarSlotRequest,
output=BotSetHotbarSlotResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.set_hotbar_slot,
),
"/soulfire.v1.BotService/SetMovementState": Endpoint.unary(
method=MethodInfo(
name="SetMovementState",
service_name="soulfire.v1.BotService",
input=BotSetMovementStateRequest,
output=BotSetMovementStateResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.set_movement_state,
),
"/soulfire.v1.BotService/ResetMovement": Endpoint.unary(
method=MethodInfo(
name="ResetMovement",
service_name="soulfire.v1.BotService",
input=BotResetMovementRequest,
output=BotResetMovementResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.reset_movement,
),
"/soulfire.v1.BotService/SetRotation": Endpoint.unary(
method=MethodInfo(
name="SetRotation",
service_name="soulfire.v1.BotService",
input=BotSetRotationRequest,
output=BotSetRotationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.set_rotation,
),
"/soulfire.v1.BotService/GetDialog": Endpoint.unary(
method=MethodInfo(
name="GetDialog",
service_name="soulfire.v1.BotService",
input=BotGetDialogRequest,
output=BotGetDialogResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.get_dialog,
),
"/soulfire.v1.BotService/SubmitDialog": Endpoint.unary(
method=MethodInfo(
name="SubmitDialog",
service_name="soulfire.v1.BotService",
input=BotSubmitDialogRequest,
output=BotSubmitDialogResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.submit_dialog,
),
"/soulfire.v1.BotService/ClickDialogButton": Endpoint.unary(
method=MethodInfo(
name="ClickDialogButton",
service_name="soulfire.v1.BotService",
input=BotClickDialogButtonRequest,
output=BotClickDialogButtonResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.click_dialog_button,
),
"/soulfire.v1.BotService/CloseDialog": Endpoint.unary(
method=MethodInfo(
name="CloseDialog",
service_name="soulfire.v1.BotService",
input=BotCloseDialogRequest,
output=BotCloseDialogResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.close_dialog,
),
},
interceptors=interceptors,
read_max_bytes=read_max_bytes,
compressions=compressions,
codecs=codecs,
)
@property
def path(self) -> str:
"""Returns the URL path to mount the application to when serving multiple applications."""
return "/soulfire.v1.BotService"
[docs]
class BotServiceClient(ConnectClient):
def __init__(
self,
address: str,
*,
codec: Codec | None = _PROTO_BINARY_CODEC,
protocol: ProtocolType = ProtocolType.CONNECT,
accept_compression: Iterable[Compression] | None = None,
send_compression: Compression | None = _GZIP_COMPRESSION,
timeout_ms: int | None = None,
read_max_bytes: int | None = None,
interceptors: Iterable[Interceptor] = (),
http_client: Client | None = None,
) -> None:
super().__init__(
address=address,
codec=codec,
protocol=protocol,
accept_compression=accept_compression,
send_compression=send_compression,
timeout_ms=timeout_ms,
read_max_bytes=read_max_bytes,
interceptors=interceptors,
http_client=http_client,
)
[docs]
async def set_bots_desired_state(
self,
request: SetBotsDesiredStateRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> SetBotsDesiredStateResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="SetBotsDesiredState",
service_name="soulfire.v1.BotService",
input=SetBotsDesiredStateRequest,
output=SetBotsDesiredStateResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def restart_bots(
self,
request: RestartBotsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> RestartBotsResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="RestartBots",
service_name="soulfire.v1.BotService",
input=RestartBotsRequest,
output=RestartBotsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def watch_bot_statuses(
self,
request: WatchBotStatusesRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> AsyncIterator[WatchBotStatusesResponse]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="WatchBotStatuses",
service_name="soulfire.v1.BotService",
input=WatchBotStatusesRequest,
output=WatchBotStatusesResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def get_bot_list(
self,
request: BotListRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotListResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GetBotList",
service_name="soulfire.v1.BotService",
input=BotListRequest,
output=BotListResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def get_bot_info(
self,
request: BotInfoRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotInfoResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GetBotInfo",
service_name="soulfire.v1.BotService",
input=BotInfoRequest,
output=BotInfoResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def update_bot_config_entry(
self,
request: BotUpdateConfigEntryRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotUpdateConfigEntryResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="UpdateBotConfigEntry",
service_name="soulfire.v1.BotService",
input=BotUpdateConfigEntryRequest,
output=BotUpdateConfigEntryResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def render_bot_pov(
self,
request: BotRenderPovRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotRenderPovResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="RenderBotPov",
service_name="soulfire.v1.BotService",
input=BotRenderPovRequest,
output=BotRenderPovResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def watch_bot_pov(
self,
request: BotWatchPovRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> AsyncIterator[BotPovFrame]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="WatchBotPov",
service_name="soulfire.v1.BotService",
input=BotWatchPovRequest,
output=BotPovFrame,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def get_bot_world_map(
self,
request: BotWorldMapRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotWorldMapResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GetBotWorldMap",
service_name="soulfire.v1.BotService",
input=BotWorldMapRequest,
output=BotWorldMapResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def click_inventory_slot(
self,
request: BotInventoryClickRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotInventoryClickResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="ClickInventorySlot",
service_name="soulfire.v1.BotService",
input=BotInventoryClickRequest,
output=BotInventoryClickResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def get_inventory_state(
self,
request: BotInventoryStateRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotInventoryStateResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GetInventoryState",
service_name="soulfire.v1.BotService",
input=BotInventoryStateRequest,
output=BotInventoryStateResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def close_container(
self,
request: BotCloseContainerRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotCloseContainerResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="CloseContainer",
service_name="soulfire.v1.BotService",
input=BotCloseContainerRequest,
output=BotCloseContainerResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def open_inventory(
self,
request: BotOpenInventoryRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotOpenInventoryResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="OpenInventory",
service_name="soulfire.v1.BotService",
input=BotOpenInventoryRequest,
output=BotOpenInventoryResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def mouse_click(
self,
request: BotMouseClickRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotMouseClickResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="MouseClick",
service_name="soulfire.v1.BotService",
input=BotMouseClickRequest,
output=BotMouseClickResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def set_container_text(
self,
request: BotSetContainerTextRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotSetContainerTextResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="SetContainerText",
service_name="soulfire.v1.BotService",
input=BotSetContainerTextRequest,
output=BotSetContainerTextResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def set_hotbar_slot(
self,
request: BotSetHotbarSlotRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotSetHotbarSlotResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="SetHotbarSlot",
service_name="soulfire.v1.BotService",
input=BotSetHotbarSlotRequest,
output=BotSetHotbarSlotResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def set_movement_state(
self,
request: BotSetMovementStateRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotSetMovementStateResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="SetMovementState",
service_name="soulfire.v1.BotService",
input=BotSetMovementStateRequest,
output=BotSetMovementStateResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def reset_movement(
self,
request: BotResetMovementRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotResetMovementResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="ResetMovement",
service_name="soulfire.v1.BotService",
input=BotResetMovementRequest,
output=BotResetMovementResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def set_rotation(
self,
request: BotSetRotationRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotSetRotationResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="SetRotation",
service_name="soulfire.v1.BotService",
input=BotSetRotationRequest,
output=BotSetRotationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def get_dialog(
self,
request: BotGetDialogRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotGetDialogResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GetDialog",
service_name="soulfire.v1.BotService",
input=BotGetDialogRequest,
output=BotGetDialogResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def submit_dialog(
self,
request: BotSubmitDialogRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotSubmitDialogResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="SubmitDialog",
service_name="soulfire.v1.BotService",
input=BotSubmitDialogRequest,
output=BotSubmitDialogResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def close_dialog(
self,
request: BotCloseDialogRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotCloseDialogResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="CloseDialog",
service_name="soulfire.v1.BotService",
input=BotCloseDialogRequest,
output=BotCloseDialogResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
class BotServiceSync(Protocol):
[docs]
def set_bots_desired_state(self, request: SetBotsDesiredStateRequest, ctx: RequestContext[SetBotsDesiredStateRequest, SetBotsDesiredStateResponse]) -> SetBotsDesiredStateResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def restart_bots(self, request: RestartBotsRequest, ctx: RequestContext[RestartBotsRequest, RestartBotsResponse]) -> RestartBotsResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def watch_bot_statuses(self, request: WatchBotStatusesRequest, ctx: RequestContext[WatchBotStatusesRequest, WatchBotStatusesResponse]) -> Iterator[WatchBotStatusesResponse]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def get_bot_list(self, request: BotListRequest, ctx: RequestContext[BotListRequest, BotListResponse]) -> BotListResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def get_bot_info(self, request: BotInfoRequest, ctx: RequestContext[BotInfoRequest, BotInfoResponse]) -> BotInfoResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def update_bot_config_entry(self, request: BotUpdateConfigEntryRequest, ctx: RequestContext[BotUpdateConfigEntryRequest, BotUpdateConfigEntryResponse]) -> BotUpdateConfigEntryResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def render_bot_pov(self, request: BotRenderPovRequest, ctx: RequestContext[BotRenderPovRequest, BotRenderPovResponse]) -> BotRenderPovResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def watch_bot_pov(self, request: BotWatchPovRequest, ctx: RequestContext[BotWatchPovRequest, BotPovFrame]) -> Iterator[BotPovFrame]:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def get_bot_world_map(self, request: BotWorldMapRequest, ctx: RequestContext[BotWorldMapRequest, BotWorldMapResponse]) -> BotWorldMapResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def click_inventory_slot(self, request: BotInventoryClickRequest, ctx: RequestContext[BotInventoryClickRequest, BotInventoryClickResponse]) -> BotInventoryClickResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def get_inventory_state(self, request: BotInventoryStateRequest, ctx: RequestContext[BotInventoryStateRequest, BotInventoryStateResponse]) -> BotInventoryStateResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def close_container(self, request: BotCloseContainerRequest, ctx: RequestContext[BotCloseContainerRequest, BotCloseContainerResponse]) -> BotCloseContainerResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def open_inventory(self, request: BotOpenInventoryRequest, ctx: RequestContext[BotOpenInventoryRequest, BotOpenInventoryResponse]) -> BotOpenInventoryResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def mouse_click(self, request: BotMouseClickRequest, ctx: RequestContext[BotMouseClickRequest, BotMouseClickResponse]) -> BotMouseClickResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def set_container_text(self, request: BotSetContainerTextRequest, ctx: RequestContext[BotSetContainerTextRequest, BotSetContainerTextResponse]) -> BotSetContainerTextResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def set_hotbar_slot(self, request: BotSetHotbarSlotRequest, ctx: RequestContext[BotSetHotbarSlotRequest, BotSetHotbarSlotResponse]) -> BotSetHotbarSlotResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def set_movement_state(self, request: BotSetMovementStateRequest, ctx: RequestContext[BotSetMovementStateRequest, BotSetMovementStateResponse]) -> BotSetMovementStateResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def reset_movement(self, request: BotResetMovementRequest, ctx: RequestContext[BotResetMovementRequest, BotResetMovementResponse]) -> BotResetMovementResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def set_rotation(self, request: BotSetRotationRequest, ctx: RequestContext[BotSetRotationRequest, BotSetRotationResponse]) -> BotSetRotationResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def get_dialog(self, request: BotGetDialogRequest, ctx: RequestContext[BotGetDialogRequest, BotGetDialogResponse]) -> BotGetDialogResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def submit_dialog(self, request: BotSubmitDialogRequest, ctx: RequestContext[BotSubmitDialogRequest, BotSubmitDialogResponse]) -> BotSubmitDialogResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def close_dialog(self, request: BotCloseDialogRequest, ctx: RequestContext[BotCloseDialogRequest, BotCloseDialogResponse]) -> BotCloseDialogResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
class BotServiceWSGIApplication(ConnectWSGIApplication):
def __init__(
self,
service: BotServiceSync,
interceptors: Iterable[InterceptorSync] = (),
read_max_bytes: int | None = None,
compressions: Iterable[Compression] | None = None,
codecs: Iterable[Codec] | None = _DEFAULT_CODECS,
) -> None:
super().__init__(
endpoints={
"/soulfire.v1.BotService/SetBotsDesiredState": EndpointSync.unary(
method=MethodInfo(
name="SetBotsDesiredState",
service_name="soulfire.v1.BotService",
input=SetBotsDesiredStateRequest,
output=SetBotsDesiredStateResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.set_bots_desired_state,
),
"/soulfire.v1.BotService/RestartBots": EndpointSync.unary(
method=MethodInfo(
name="RestartBots",
service_name="soulfire.v1.BotService",
input=RestartBotsRequest,
output=RestartBotsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.restart_bots,
),
"/soulfire.v1.BotService/WatchBotStatuses": EndpointSync.server_stream(
method=MethodInfo(
name="WatchBotStatuses",
service_name="soulfire.v1.BotService",
input=WatchBotStatusesRequest,
output=WatchBotStatusesResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.watch_bot_statuses,
),
"/soulfire.v1.BotService/GetBotList": EndpointSync.unary(
method=MethodInfo(
name="GetBotList",
service_name="soulfire.v1.BotService",
input=BotListRequest,
output=BotListResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.get_bot_list,
),
"/soulfire.v1.BotService/GetBotInfo": EndpointSync.unary(
method=MethodInfo(
name="GetBotInfo",
service_name="soulfire.v1.BotService",
input=BotInfoRequest,
output=BotInfoResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.get_bot_info,
),
"/soulfire.v1.BotService/UpdateBotConfigEntry": EndpointSync.unary(
method=MethodInfo(
name="UpdateBotConfigEntry",
service_name="soulfire.v1.BotService",
input=BotUpdateConfigEntryRequest,
output=BotUpdateConfigEntryResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.update_bot_config_entry,
),
"/soulfire.v1.BotService/RenderBotPov": EndpointSync.unary(
method=MethodInfo(
name="RenderBotPov",
service_name="soulfire.v1.BotService",
input=BotRenderPovRequest,
output=BotRenderPovResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.render_bot_pov,
),
"/soulfire.v1.BotService/WatchBotPov": EndpointSync.server_stream(
method=MethodInfo(
name="WatchBotPov",
service_name="soulfire.v1.BotService",
input=BotWatchPovRequest,
output=BotPovFrame,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.watch_bot_pov,
),
"/soulfire.v1.BotService/GetBotWorldMap": EndpointSync.unary(
method=MethodInfo(
name="GetBotWorldMap",
service_name="soulfire.v1.BotService",
input=BotWorldMapRequest,
output=BotWorldMapResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.get_bot_world_map,
),
"/soulfire.v1.BotService/ClickInventorySlot": EndpointSync.unary(
method=MethodInfo(
name="ClickInventorySlot",
service_name="soulfire.v1.BotService",
input=BotInventoryClickRequest,
output=BotInventoryClickResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.click_inventory_slot,
),
"/soulfire.v1.BotService/GetInventoryState": EndpointSync.unary(
method=MethodInfo(
name="GetInventoryState",
service_name="soulfire.v1.BotService",
input=BotInventoryStateRequest,
output=BotInventoryStateResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.get_inventory_state,
),
"/soulfire.v1.BotService/CloseContainer": EndpointSync.unary(
method=MethodInfo(
name="CloseContainer",
service_name="soulfire.v1.BotService",
input=BotCloseContainerRequest,
output=BotCloseContainerResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.close_container,
),
"/soulfire.v1.BotService/OpenInventory": EndpointSync.unary(
method=MethodInfo(
name="OpenInventory",
service_name="soulfire.v1.BotService",
input=BotOpenInventoryRequest,
output=BotOpenInventoryResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.open_inventory,
),
"/soulfire.v1.BotService/MouseClick": EndpointSync.unary(
method=MethodInfo(
name="MouseClick",
service_name="soulfire.v1.BotService",
input=BotMouseClickRequest,
output=BotMouseClickResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.mouse_click,
),
"/soulfire.v1.BotService/ClickContainerButton": EndpointSync.unary(
method=MethodInfo(
name="ClickContainerButton",
service_name="soulfire.v1.BotService",
input=BotContainerButtonClickRequest,
output=BotContainerButtonClickResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.click_container_button,
),
"/soulfire.v1.BotService/SetContainerText": EndpointSync.unary(
method=MethodInfo(
name="SetContainerText",
service_name="soulfire.v1.BotService",
input=BotSetContainerTextRequest,
output=BotSetContainerTextResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.set_container_text,
),
"/soulfire.v1.BotService/SetHotbarSlot": EndpointSync.unary(
method=MethodInfo(
name="SetHotbarSlot",
service_name="soulfire.v1.BotService",
input=BotSetHotbarSlotRequest,
output=BotSetHotbarSlotResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.set_hotbar_slot,
),
"/soulfire.v1.BotService/SetMovementState": EndpointSync.unary(
method=MethodInfo(
name="SetMovementState",
service_name="soulfire.v1.BotService",
input=BotSetMovementStateRequest,
output=BotSetMovementStateResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.set_movement_state,
),
"/soulfire.v1.BotService/ResetMovement": EndpointSync.unary(
method=MethodInfo(
name="ResetMovement",
service_name="soulfire.v1.BotService",
input=BotResetMovementRequest,
output=BotResetMovementResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.reset_movement,
),
"/soulfire.v1.BotService/SetRotation": EndpointSync.unary(
method=MethodInfo(
name="SetRotation",
service_name="soulfire.v1.BotService",
input=BotSetRotationRequest,
output=BotSetRotationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.set_rotation,
),
"/soulfire.v1.BotService/GetDialog": EndpointSync.unary(
method=MethodInfo(
name="GetDialog",
service_name="soulfire.v1.BotService",
input=BotGetDialogRequest,
output=BotGetDialogResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.get_dialog,
),
"/soulfire.v1.BotService/SubmitDialog": EndpointSync.unary(
method=MethodInfo(
name="SubmitDialog",
service_name="soulfire.v1.BotService",
input=BotSubmitDialogRequest,
output=BotSubmitDialogResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.submit_dialog,
),
"/soulfire.v1.BotService/ClickDialogButton": EndpointSync.unary(
method=MethodInfo(
name="ClickDialogButton",
service_name="soulfire.v1.BotService",
input=BotClickDialogButtonRequest,
output=BotClickDialogButtonResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.click_dialog_button,
),
"/soulfire.v1.BotService/CloseDialog": EndpointSync.unary(
method=MethodInfo(
name="CloseDialog",
service_name="soulfire.v1.BotService",
input=BotCloseDialogRequest,
output=BotCloseDialogResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.close_dialog,
),
},
interceptors=interceptors,
read_max_bytes=read_max_bytes,
compressions=compressions,
codecs=codecs,
)
@property
def path(self) -> str:
"""Returns the URL path to mount the application to when serving multiple applications."""
return "/soulfire.v1.BotService"
[docs]
class BotServiceClientSync(ConnectClientSync):
def __init__(
self,
address: str,
*,
codec: Codec | None = _PROTO_BINARY_CODEC,
protocol: ProtocolType = ProtocolType.CONNECT,
accept_compression: Iterable[Compression] | None = None,
send_compression: Compression | None = _GZIP_COMPRESSION,
timeout_ms: int | None = None,
read_max_bytes: int | None = None,
interceptors: Iterable[InterceptorSync] = (),
http_client: SyncClient | None = None,
) -> None:
super().__init__(
address=address,
codec=codec,
protocol=protocol,
accept_compression=accept_compression,
send_compression=send_compression,
timeout_ms=timeout_ms,
read_max_bytes=read_max_bytes,
interceptors=interceptors,
http_client=http_client,
)
[docs]
def set_bots_desired_state(
self,
request: SetBotsDesiredStateRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> SetBotsDesiredStateResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="SetBotsDesiredState",
service_name="soulfire.v1.BotService",
input=SetBotsDesiredStateRequest,
output=SetBotsDesiredStateResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def restart_bots(
self,
request: RestartBotsRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> RestartBotsResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="RestartBots",
service_name="soulfire.v1.BotService",
input=RestartBotsRequest,
output=RestartBotsResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def watch_bot_statuses(
self,
request: WatchBotStatusesRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> Iterator[WatchBotStatusesResponse]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="WatchBotStatuses",
service_name="soulfire.v1.BotService",
input=WatchBotStatusesRequest,
output=WatchBotStatusesResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def get_bot_list(
self,
request: BotListRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotListResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GetBotList",
service_name="soulfire.v1.BotService",
input=BotListRequest,
output=BotListResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def get_bot_info(
self,
request: BotInfoRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotInfoResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GetBotInfo",
service_name="soulfire.v1.BotService",
input=BotInfoRequest,
output=BotInfoResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def update_bot_config_entry(
self,
request: BotUpdateConfigEntryRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotUpdateConfigEntryResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="UpdateBotConfigEntry",
service_name="soulfire.v1.BotService",
input=BotUpdateConfigEntryRequest,
output=BotUpdateConfigEntryResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def render_bot_pov(
self,
request: BotRenderPovRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotRenderPovResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="RenderBotPov",
service_name="soulfire.v1.BotService",
input=BotRenderPovRequest,
output=BotRenderPovResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def watch_bot_pov(
self,
request: BotWatchPovRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> Iterator[BotPovFrame]:
return self.execute_server_stream(
request=request,
method=MethodInfo(
name="WatchBotPov",
service_name="soulfire.v1.BotService",
input=BotWatchPovRequest,
output=BotPovFrame,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def get_bot_world_map(
self,
request: BotWorldMapRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotWorldMapResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GetBotWorldMap",
service_name="soulfire.v1.BotService",
input=BotWorldMapRequest,
output=BotWorldMapResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def click_inventory_slot(
self,
request: BotInventoryClickRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotInventoryClickResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="ClickInventorySlot",
service_name="soulfire.v1.BotService",
input=BotInventoryClickRequest,
output=BotInventoryClickResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def get_inventory_state(
self,
request: BotInventoryStateRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotInventoryStateResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GetInventoryState",
service_name="soulfire.v1.BotService",
input=BotInventoryStateRequest,
output=BotInventoryStateResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def close_container(
self,
request: BotCloseContainerRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotCloseContainerResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="CloseContainer",
service_name="soulfire.v1.BotService",
input=BotCloseContainerRequest,
output=BotCloseContainerResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def open_inventory(
self,
request: BotOpenInventoryRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotOpenInventoryResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="OpenInventory",
service_name="soulfire.v1.BotService",
input=BotOpenInventoryRequest,
output=BotOpenInventoryResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def mouse_click(
self,
request: BotMouseClickRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotMouseClickResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="MouseClick",
service_name="soulfire.v1.BotService",
input=BotMouseClickRequest,
output=BotMouseClickResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def set_container_text(
self,
request: BotSetContainerTextRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotSetContainerTextResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="SetContainerText",
service_name="soulfire.v1.BotService",
input=BotSetContainerTextRequest,
output=BotSetContainerTextResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def set_hotbar_slot(
self,
request: BotSetHotbarSlotRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotSetHotbarSlotResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="SetHotbarSlot",
service_name="soulfire.v1.BotService",
input=BotSetHotbarSlotRequest,
output=BotSetHotbarSlotResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def set_movement_state(
self,
request: BotSetMovementStateRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotSetMovementStateResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="SetMovementState",
service_name="soulfire.v1.BotService",
input=BotSetMovementStateRequest,
output=BotSetMovementStateResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def reset_movement(
self,
request: BotResetMovementRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotResetMovementResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="ResetMovement",
service_name="soulfire.v1.BotService",
input=BotResetMovementRequest,
output=BotResetMovementResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def set_rotation(
self,
request: BotSetRotationRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotSetRotationResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="SetRotation",
service_name="soulfire.v1.BotService",
input=BotSetRotationRequest,
output=BotSetRotationResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def get_dialog(
self,
request: BotGetDialogRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotGetDialogResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GetDialog",
service_name="soulfire.v1.BotService",
input=BotGetDialogRequest,
output=BotGetDialogResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def submit_dialog(
self,
request: BotSubmitDialogRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotSubmitDialogResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="SubmitDialog",
service_name="soulfire.v1.BotService",
input=BotSubmitDialogRequest,
output=BotSubmitDialogResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def close_dialog(
self,
request: BotCloseDialogRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> BotCloseDialogResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="CloseDialog",
service_name="soulfire.v1.BotService",
input=BotCloseDialogRequest,
output=BotCloseDialogResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)