# Generated from soulfire/world.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 .world_pb2 import CanSeeBlockRequest, CanSeeBlockResponse, EstimateDigTimeRequest, EstimateDigTimeResponse, EstimateExplosionDamageRequest, EstimateExplosionDamageResponse, GetPlayerSnapshotRequest, GetPlayerSnapshotResponse, GetWorldBlockRequest, GetWorldBlockResponse, GetWorldEntityRequest, GetWorldEntityResponse, QueryBlocksRequest, QueryBlocksResponse, QueryEntitiesRequest, QueryEntitiesResponse, RaycastRequest, RaycastResponse
if TYPE_CHECKING:
from collections.abc import AsyncGenerator, Iterable, 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 WorldService(Protocol):
[docs]
async def get_player_snapshot(self, request: GetPlayerSnapshotRequest, ctx: RequestContext[GetPlayerSnapshotRequest, GetPlayerSnapshotResponse]) -> GetPlayerSnapshotResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def get_world_block(self, request: GetWorldBlockRequest, ctx: RequestContext[GetWorldBlockRequest, GetWorldBlockResponse]) -> GetWorldBlockResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def query_blocks(self, request: QueryBlocksRequest, ctx: RequestContext[QueryBlocksRequest, QueryBlocksResponse]) -> QueryBlocksResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def get_world_entity(self, request: GetWorldEntityRequest, ctx: RequestContext[GetWorldEntityRequest, GetWorldEntityResponse]) -> GetWorldEntityResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def query_entities(self, request: QueryEntitiesRequest, ctx: RequestContext[QueryEntitiesRequest, QueryEntitiesResponse]) -> QueryEntitiesResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def raycast(self, request: RaycastRequest, ctx: RequestContext[RaycastRequest, RaycastResponse]) -> RaycastResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def can_see_block(self, request: CanSeeBlockRequest, ctx: RequestContext[CanSeeBlockRequest, CanSeeBlockResponse]) -> CanSeeBlockResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def estimate_dig_time(self, request: EstimateDigTimeRequest, ctx: RequestContext[EstimateDigTimeRequest, EstimateDigTimeResponse]) -> EstimateDigTimeResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
async def estimate_explosion_damage(self, request: EstimateExplosionDamageRequest, ctx: RequestContext[EstimateExplosionDamageRequest, EstimateExplosionDamageResponse]) -> EstimateExplosionDamageResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
class WorldServiceASGIApplication(ConnectASGIApplication[WorldService]):
def __init__(
self,
service: WorldService | AsyncGenerator[WorldService],
*,
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.WorldService/GetPlayerSnapshot": Endpoint.unary(
method=MethodInfo(
name="GetPlayerSnapshot",
service_name="soulfire.v1.WorldService",
input=GetPlayerSnapshotRequest,
output=GetPlayerSnapshotResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.get_player_snapshot,
),
"/soulfire.v1.WorldService/GetWorldBlock": Endpoint.unary(
method=MethodInfo(
name="GetWorldBlock",
service_name="soulfire.v1.WorldService",
input=GetWorldBlockRequest,
output=GetWorldBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.get_world_block,
),
"/soulfire.v1.WorldService/QueryBlocks": Endpoint.unary(
method=MethodInfo(
name="QueryBlocks",
service_name="soulfire.v1.WorldService",
input=QueryBlocksRequest,
output=QueryBlocksResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.query_blocks,
),
"/soulfire.v1.WorldService/GetWorldEntity": Endpoint.unary(
method=MethodInfo(
name="GetWorldEntity",
service_name="soulfire.v1.WorldService",
input=GetWorldEntityRequest,
output=GetWorldEntityResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.get_world_entity,
),
"/soulfire.v1.WorldService/QueryEntities": Endpoint.unary(
method=MethodInfo(
name="QueryEntities",
service_name="soulfire.v1.WorldService",
input=QueryEntitiesRequest,
output=QueryEntitiesResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.query_entities,
),
"/soulfire.v1.WorldService/Raycast": Endpoint.unary(
method=MethodInfo(
name="Raycast",
service_name="soulfire.v1.WorldService",
input=RaycastRequest,
output=RaycastResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.raycast,
),
"/soulfire.v1.WorldService/CanSeeBlock": Endpoint.unary(
method=MethodInfo(
name="CanSeeBlock",
service_name="soulfire.v1.WorldService",
input=CanSeeBlockRequest,
output=CanSeeBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.can_see_block,
),
"/soulfire.v1.WorldService/EstimateDigTime": Endpoint.unary(
method=MethodInfo(
name="EstimateDigTime",
service_name="soulfire.v1.WorldService",
input=EstimateDigTimeRequest,
output=EstimateDigTimeResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.estimate_dig_time,
),
"/soulfire.v1.WorldService/EstimateExplosionDamage": Endpoint.unary(
method=MethodInfo(
name="EstimateExplosionDamage",
service_name="soulfire.v1.WorldService",
input=EstimateExplosionDamageRequest,
output=EstimateExplosionDamageResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=svc.estimate_explosion_damage,
),
},
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.WorldService"
[docs]
class WorldServiceClient(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 get_player_snapshot(
self,
request: GetPlayerSnapshotRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetPlayerSnapshotResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GetPlayerSnapshot",
service_name="soulfire.v1.WorldService",
input=GetPlayerSnapshotRequest,
output=GetPlayerSnapshotResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def get_world_block(
self,
request: GetWorldBlockRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetWorldBlockResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GetWorldBlock",
service_name="soulfire.v1.WorldService",
input=GetWorldBlockRequest,
output=GetWorldBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def query_blocks(
self,
request: QueryBlocksRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> QueryBlocksResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="QueryBlocks",
service_name="soulfire.v1.WorldService",
input=QueryBlocksRequest,
output=QueryBlocksResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def get_world_entity(
self,
request: GetWorldEntityRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetWorldEntityResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="GetWorldEntity",
service_name="soulfire.v1.WorldService",
input=GetWorldEntityRequest,
output=GetWorldEntityResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def query_entities(
self,
request: QueryEntitiesRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> QueryEntitiesResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="QueryEntities",
service_name="soulfire.v1.WorldService",
input=QueryEntitiesRequest,
output=QueryEntitiesResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def raycast(
self,
request: RaycastRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> RaycastResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="Raycast",
service_name="soulfire.v1.WorldService",
input=RaycastRequest,
output=RaycastResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def can_see_block(
self,
request: CanSeeBlockRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> CanSeeBlockResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="CanSeeBlock",
service_name="soulfire.v1.WorldService",
input=CanSeeBlockRequest,
output=CanSeeBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def estimate_dig_time(
self,
request: EstimateDigTimeRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> EstimateDigTimeResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="EstimateDigTime",
service_name="soulfire.v1.WorldService",
input=EstimateDigTimeRequest,
output=EstimateDigTimeResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
async def estimate_explosion_damage(
self,
request: EstimateExplosionDamageRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> EstimateExplosionDamageResponse:
return await self.execute_unary(
request=request,
method=MethodInfo(
name="EstimateExplosionDamage",
service_name="soulfire.v1.WorldService",
input=EstimateExplosionDamageRequest,
output=EstimateExplosionDamageResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
class WorldServiceSync(Protocol):
[docs]
def get_player_snapshot(self, request: GetPlayerSnapshotRequest, ctx: RequestContext[GetPlayerSnapshotRequest, GetPlayerSnapshotResponse]) -> GetPlayerSnapshotResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def get_world_block(self, request: GetWorldBlockRequest, ctx: RequestContext[GetWorldBlockRequest, GetWorldBlockResponse]) -> GetWorldBlockResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def query_blocks(self, request: QueryBlocksRequest, ctx: RequestContext[QueryBlocksRequest, QueryBlocksResponse]) -> QueryBlocksResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def get_world_entity(self, request: GetWorldEntityRequest, ctx: RequestContext[GetWorldEntityRequest, GetWorldEntityResponse]) -> GetWorldEntityResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def query_entities(self, request: QueryEntitiesRequest, ctx: RequestContext[QueryEntitiesRequest, QueryEntitiesResponse]) -> QueryEntitiesResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def raycast(self, request: RaycastRequest, ctx: RequestContext[RaycastRequest, RaycastResponse]) -> RaycastResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def can_see_block(self, request: CanSeeBlockRequest, ctx: RequestContext[CanSeeBlockRequest, CanSeeBlockResponse]) -> CanSeeBlockResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def estimate_dig_time(self, request: EstimateDigTimeRequest, ctx: RequestContext[EstimateDigTimeRequest, EstimateDigTimeResponse]) -> EstimateDigTimeResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
def estimate_explosion_damage(self, request: EstimateExplosionDamageRequest, ctx: RequestContext[EstimateExplosionDamageRequest, EstimateExplosionDamageResponse]) -> EstimateExplosionDamageResponse:
raise ConnectError(Code.UNIMPLEMENTED, 'Not implemented')
[docs]
class WorldServiceWSGIApplication(ConnectWSGIApplication):
def __init__(
self,
service: WorldServiceSync,
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.WorldService/GetPlayerSnapshot": EndpointSync.unary(
method=MethodInfo(
name="GetPlayerSnapshot",
service_name="soulfire.v1.WorldService",
input=GetPlayerSnapshotRequest,
output=GetPlayerSnapshotResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.get_player_snapshot,
),
"/soulfire.v1.WorldService/GetWorldBlock": EndpointSync.unary(
method=MethodInfo(
name="GetWorldBlock",
service_name="soulfire.v1.WorldService",
input=GetWorldBlockRequest,
output=GetWorldBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.get_world_block,
),
"/soulfire.v1.WorldService/QueryBlocks": EndpointSync.unary(
method=MethodInfo(
name="QueryBlocks",
service_name="soulfire.v1.WorldService",
input=QueryBlocksRequest,
output=QueryBlocksResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.query_blocks,
),
"/soulfire.v1.WorldService/GetWorldEntity": EndpointSync.unary(
method=MethodInfo(
name="GetWorldEntity",
service_name="soulfire.v1.WorldService",
input=GetWorldEntityRequest,
output=GetWorldEntityResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.get_world_entity,
),
"/soulfire.v1.WorldService/QueryEntities": EndpointSync.unary(
method=MethodInfo(
name="QueryEntities",
service_name="soulfire.v1.WorldService",
input=QueryEntitiesRequest,
output=QueryEntitiesResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.query_entities,
),
"/soulfire.v1.WorldService/Raycast": EndpointSync.unary(
method=MethodInfo(
name="Raycast",
service_name="soulfire.v1.WorldService",
input=RaycastRequest,
output=RaycastResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.raycast,
),
"/soulfire.v1.WorldService/CanSeeBlock": EndpointSync.unary(
method=MethodInfo(
name="CanSeeBlock",
service_name="soulfire.v1.WorldService",
input=CanSeeBlockRequest,
output=CanSeeBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.can_see_block,
),
"/soulfire.v1.WorldService/EstimateDigTime": EndpointSync.unary(
method=MethodInfo(
name="EstimateDigTime",
service_name="soulfire.v1.WorldService",
input=EstimateDigTimeRequest,
output=EstimateDigTimeResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.estimate_dig_time,
),
"/soulfire.v1.WorldService/EstimateExplosionDamage": EndpointSync.unary(
method=MethodInfo(
name="EstimateExplosionDamage",
service_name="soulfire.v1.WorldService",
input=EstimateExplosionDamageRequest,
output=EstimateExplosionDamageResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
function=service.estimate_explosion_damage,
),
},
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.WorldService"
[docs]
class WorldServiceClientSync(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 get_player_snapshot(
self,
request: GetPlayerSnapshotRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetPlayerSnapshotResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GetPlayerSnapshot",
service_name="soulfire.v1.WorldService",
input=GetPlayerSnapshotRequest,
output=GetPlayerSnapshotResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def get_world_block(
self,
request: GetWorldBlockRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetWorldBlockResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GetWorldBlock",
service_name="soulfire.v1.WorldService",
input=GetWorldBlockRequest,
output=GetWorldBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def query_blocks(
self,
request: QueryBlocksRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> QueryBlocksResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="QueryBlocks",
service_name="soulfire.v1.WorldService",
input=QueryBlocksRequest,
output=QueryBlocksResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def get_world_entity(
self,
request: GetWorldEntityRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> GetWorldEntityResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="GetWorldEntity",
service_name="soulfire.v1.WorldService",
input=GetWorldEntityRequest,
output=GetWorldEntityResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def query_entities(
self,
request: QueryEntitiesRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> QueryEntitiesResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="QueryEntities",
service_name="soulfire.v1.WorldService",
input=QueryEntitiesRequest,
output=QueryEntitiesResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def raycast(
self,
request: RaycastRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> RaycastResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="Raycast",
service_name="soulfire.v1.WorldService",
input=RaycastRequest,
output=RaycastResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def can_see_block(
self,
request: CanSeeBlockRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> CanSeeBlockResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="CanSeeBlock",
service_name="soulfire.v1.WorldService",
input=CanSeeBlockRequest,
output=CanSeeBlockResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def estimate_dig_time(
self,
request: EstimateDigTimeRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> EstimateDigTimeResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="EstimateDigTime",
service_name="soulfire.v1.WorldService",
input=EstimateDigTimeRequest,
output=EstimateDigTimeResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)
[docs]
def estimate_explosion_damage(
self,
request: EstimateExplosionDamageRequest,
*,
headers: Headers | Mapping[str, str] | None = None,
timeout_ms: int | None = None,
) -> EstimateExplosionDamageResponse:
return self.execute_unary(
request=request,
method=MethodInfo(
name="EstimateExplosionDamage",
service_name="soulfire.v1.WorldService",
input=EstimateExplosionDamageRequest,
output=EstimateExplosionDamageResponse,
idempotency_level=IdempotencyLevel.UNKNOWN,
),
headers=headers,
timeout_ms=timeout_ms,
)