Repository URL to install this package:
|
Version:
4.1.0 ▾
|
# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
"""Client and server classes corresponding to protobuf-defined services."""
import grpc
from bosdyn.api import estop_pb2 as bosdyn_dot_api_dot_estop__pb2
class EstopServiceStub(object):
"""The software robot E-Stop system:
1. Uses challenge-style communication to enforce end user (aka "originators") connection
for Authority to Operate (ATO).
2. Offers the ability to issue a direct denial of ATO.
The EstopService provides a service interface for the robot EStop/Authority to operate the
system.
"""
def __init__(self, channel):
"""Constructor.
Args:
channel: A grpc.Channel.
"""
self.RegisterEstopEndpoint = channel.unary_unary(
'/bosdyn.api.EstopService/RegisterEstopEndpoint',
request_serializer=bosdyn_dot_api_dot_estop__pb2.RegisterEstopEndpointRequest.SerializeToString,
response_deserializer=bosdyn_dot_api_dot_estop__pb2.RegisterEstopEndpointResponse.FromString,
)
self.DeregisterEstopEndpoint = channel.unary_unary(
'/bosdyn.api.EstopService/DeregisterEstopEndpoint',
request_serializer=bosdyn_dot_api_dot_estop__pb2.DeregisterEstopEndpointRequest.SerializeToString,
response_deserializer=bosdyn_dot_api_dot_estop__pb2.DeregisterEstopEndpointResponse.FromString,
)
self.EstopCheckIn = channel.unary_unary(
'/bosdyn.api.EstopService/EstopCheckIn',
request_serializer=bosdyn_dot_api_dot_estop__pb2.EstopCheckInRequest.SerializeToString,
response_deserializer=bosdyn_dot_api_dot_estop__pb2.EstopCheckInResponse.FromString,
)
self.GetEstopConfig = channel.unary_unary(
'/bosdyn.api.EstopService/GetEstopConfig',
request_serializer=bosdyn_dot_api_dot_estop__pb2.GetEstopConfigRequest.SerializeToString,
response_deserializer=bosdyn_dot_api_dot_estop__pb2.GetEstopConfigResponse.FromString,
)
self.SetEstopConfig = channel.unary_unary(
'/bosdyn.api.EstopService/SetEstopConfig',
request_serializer=bosdyn_dot_api_dot_estop__pb2.SetEstopConfigRequest.SerializeToString,
response_deserializer=bosdyn_dot_api_dot_estop__pb2.SetEstopConfigResponse.FromString,
)
self.GetEstopSystemStatus = channel.unary_unary(
'/bosdyn.api.EstopService/GetEstopSystemStatus',
request_serializer=bosdyn_dot_api_dot_estop__pb2.GetEstopSystemStatusRequest.SerializeToString,
response_deserializer=bosdyn_dot_api_dot_estop__pb2.GetEstopSystemStatusResponse.FromString,
)
class EstopServiceServicer(object):
"""The software robot E-Stop system:
1. Uses challenge-style communication to enforce end user (aka "originators") connection
for Authority to Operate (ATO).
2. Offers the ability to issue a direct denial of ATO.
The EstopService provides a service interface for the robot EStop/Authority to operate the
system.
"""
def RegisterEstopEndpoint(self, request, context):
"""Register an Estop "originator" or "endpoint".
This may be a replacement for another active endpoint.
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def DeregisterEstopEndpoint(self, request, context):
"""Deregister the requested estop endpoint.
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def EstopCheckIn(self, request, context):
"""Answer challenge from previous response (unless this is the first call), and request
a stop level.
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def GetEstopConfig(self, request, context):
"""Request the current EstopConfig, describing the expected set of endpoints.
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def SetEstopConfig(self, request, context):
"""Set a new active EstopConfig.
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def GetEstopSystemStatus(self, request, context):
"""Ask for the current status of the estop system.
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def add_EstopServiceServicer_to_server(servicer, server):
rpc_method_handlers = {
'RegisterEstopEndpoint': grpc.unary_unary_rpc_method_handler(
servicer.RegisterEstopEndpoint,
request_deserializer=bosdyn_dot_api_dot_estop__pb2.RegisterEstopEndpointRequest.FromString,
response_serializer=bosdyn_dot_api_dot_estop__pb2.RegisterEstopEndpointResponse.SerializeToString,
),
'DeregisterEstopEndpoint': grpc.unary_unary_rpc_method_handler(
servicer.DeregisterEstopEndpoint,
request_deserializer=bosdyn_dot_api_dot_estop__pb2.DeregisterEstopEndpointRequest.FromString,
response_serializer=bosdyn_dot_api_dot_estop__pb2.DeregisterEstopEndpointResponse.SerializeToString,
),
'EstopCheckIn': grpc.unary_unary_rpc_method_handler(
servicer.EstopCheckIn,
request_deserializer=bosdyn_dot_api_dot_estop__pb2.EstopCheckInRequest.FromString,
response_serializer=bosdyn_dot_api_dot_estop__pb2.EstopCheckInResponse.SerializeToString,
),
'GetEstopConfig': grpc.unary_unary_rpc_method_handler(
servicer.GetEstopConfig,
request_deserializer=bosdyn_dot_api_dot_estop__pb2.GetEstopConfigRequest.FromString,
response_serializer=bosdyn_dot_api_dot_estop__pb2.GetEstopConfigResponse.SerializeToString,
),
'SetEstopConfig': grpc.unary_unary_rpc_method_handler(
servicer.SetEstopConfig,
request_deserializer=bosdyn_dot_api_dot_estop__pb2.SetEstopConfigRequest.FromString,
response_serializer=bosdyn_dot_api_dot_estop__pb2.SetEstopConfigResponse.SerializeToString,
),
'GetEstopSystemStatus': grpc.unary_unary_rpc_method_handler(
servicer.GetEstopSystemStatus,
request_deserializer=bosdyn_dot_api_dot_estop__pb2.GetEstopSystemStatusRequest.FromString,
response_serializer=bosdyn_dot_api_dot_estop__pb2.GetEstopSystemStatusResponse.SerializeToString,
),
}
generic_handler = grpc.method_handlers_generic_handler(
'bosdyn.api.EstopService', rpc_method_handlers)
server.add_generic_rpc_handlers((generic_handler,))
# This class is part of an EXPERIMENTAL API.
class EstopService(object):
"""The software robot E-Stop system:
1. Uses challenge-style communication to enforce end user (aka "originators") connection
for Authority to Operate (ATO).
2. Offers the ability to issue a direct denial of ATO.
The EstopService provides a service interface for the robot EStop/Authority to operate the
system.
"""
@staticmethod
def RegisterEstopEndpoint(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/bosdyn.api.EstopService/RegisterEstopEndpoint',
bosdyn_dot_api_dot_estop__pb2.RegisterEstopEndpointRequest.SerializeToString,
bosdyn_dot_api_dot_estop__pb2.RegisterEstopEndpointResponse.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def DeregisterEstopEndpoint(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/bosdyn.api.EstopService/DeregisterEstopEndpoint',
bosdyn_dot_api_dot_estop__pb2.DeregisterEstopEndpointRequest.SerializeToString,
bosdyn_dot_api_dot_estop__pb2.DeregisterEstopEndpointResponse.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def EstopCheckIn(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/bosdyn.api.EstopService/EstopCheckIn',
bosdyn_dot_api_dot_estop__pb2.EstopCheckInRequest.SerializeToString,
bosdyn_dot_api_dot_estop__pb2.EstopCheckInResponse.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def GetEstopConfig(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/bosdyn.api.EstopService/GetEstopConfig',
bosdyn_dot_api_dot_estop__pb2.GetEstopConfigRequest.SerializeToString,
bosdyn_dot_api_dot_estop__pb2.GetEstopConfigResponse.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def SetEstopConfig(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/bosdyn.api.EstopService/SetEstopConfig',
bosdyn_dot_api_dot_estop__pb2.SetEstopConfigRequest.SerializeToString,
bosdyn_dot_api_dot_estop__pb2.SetEstopConfigResponse.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
@staticmethod
def GetEstopSystemStatus(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/bosdyn.api.EstopService/GetEstopSystemStatus',
bosdyn_dot_api_dot_estop__pb2.GetEstopSystemStatusRequest.SerializeToString,
bosdyn_dot_api_dot_estop__pb2.GetEstopSystemStatusResponse.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)