gRPC Helpers

polestar_api.grpc

Minimal gRPC call helpers using grpclib.

unary_unary async

unary_unary(channel: Channel, method: str, request_data: bytes, *, metadata: dict[str, str] | None = None, retries: int = 2) -> bytes

Make a unary-unary gRPC call with retry on transient errors.

unary_stream async

unary_stream(channel: Channel, method: str, request_data: bytes, *, metadata: dict[str, str] | None = None, retries: int = 2) -> AsyncIterator[bytes]

Make a server-streaming gRPC call with retry on transient errors.

Chronos streaming endpoints are subscriptions: the server sends one or more messages then keeps the stream open for live updates. The server may also send GOAWAY/UNAVAILABLE before delivering any data; the Android app handles this via its RetryPolicy. We retry only when no messages have been yielded yet.