1# Copyright 2021 The Pigweed Authors 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); you may not 4# use this file except in compliance with the License. You may obtain a copy of 5# the License at 6# 7# https://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 12# License for the specific language governing permissions and limitations under 13# the License. 14"""Exceptions for RPC-related errors.""" 15 16from typing import Optional 17 18from pw_status import Status 19 20from pw_rpc.client import PendingRpc 21 22 23class RpcTimeout(Exception): 24 def __init__(self, rpc: PendingRpc, timeout: Optional[float]): 25 super().__init__( 26 f'No response received for {rpc.method} after {timeout} s' 27 ) 28 self.rpc = rpc 29 self.timeout = timeout 30 31 32class RpcError(Exception): 33 def __init__(self, rpc: PendingRpc, status: Status): 34 """Raised when there is an RPC-layer error.""" 35 if status is Status.NOT_FOUND: 36 msg = ': the RPC server does not support this RPC' 37 elif status is Status.DATA_LOSS: 38 msg = ': an error occurred while decoding the RPC payload' 39 else: 40 msg = '' 41 42 super().__init__(f'{rpc.method} failed with error {status}{msg}') 43 self.rpc = rpc 44 self.status = status 45