diff options
| -rw-r--r-- | floss/pandora/server/bluetooth.py | 6 | ||||
| -rw-r--r-- | floss/pandora/server/hfp.py | 73 |
2 files changed, 79 insertions, 0 deletions
diff --git a/floss/pandora/server/bluetooth.py b/floss/pandora/server/bluetooth.py index 1ef1496506..41254a9ada 100644 --- a/floss/pandora/server/bluetooth.py +++ b/floss/pandora/server/bluetooth.py @@ -392,6 +392,12 @@ class Bluetooth(object): def audio_disconnect(self, address): return self.telephony_client.audio_disconnect(address) + def hangup_call(self): + return self.telephony_client.hangup_call() + + def set_battery_level(self, battery_level): + return self.telephony_client.set_battery_level(battery_level) + def gatt_connect(self, address, is_direct, transport): return self.gatt_client.connect_client(address, is_direct, transport) diff --git a/floss/pandora/server/hfp.py b/floss/pandora/server/hfp.py index d31439c6d3..59f2e81318 100644 --- a/floss/pandora/server/hfp.py +++ b/floss/pandora/server/hfp.py @@ -13,7 +13,11 @@ # limitations under the License. """HFP grpc interface.""" +import math + +from floss.pandora.floss import utils from floss.pandora.server import bluetooth as bluetooth_module +from google.protobuf import empty_pb2 import grpc from pandora_experimental import hfp_grpc_aio from pandora_experimental import hfp_pb2 @@ -35,6 +39,75 @@ class HFPService(hfp_grpc_aio.HFPServicer): self.bluetooth.set_phone_ops_enabled(True) self.bluetooth.set_mps_qualification_enabled(True) + async def EnableSlc(self, request: hfp_pb2.EnableSlcRequest, context: grpc.ServicerContext) -> empty_pb2.Empty: + self.enable_phone_for_testing() + address = utils.connection_from(request.connection).address + self.bluetooth.connect_device(address) + return empty_pb2.Empty() + + async def EnableSlcAsHandsfree(self, request: hfp_pb2.EnableSlcAsHandsfreeRequest, + context: grpc.ServicerContext) -> empty_pb2.Empty: + context.set_code(grpc.StatusCode.UNIMPLEMENTED) # type: ignore + context.set_details("Method not implemented!") # type: ignore + raise NotImplementedError("Method not implemented!") + + async def DisableSlc(self, request: hfp_pb2.DisableSlcRequest, context: grpc.ServicerContext) -> empty_pb2.Empty: + address = utils.connection_from(request.connection).address + self.bluetooth.disconnect_device(address) + return empty_pb2.Empty() + + async def DisableSlcAsHandsfree(self, request: hfp_pb2.DisableSlcAsHandsfreeRequest, + context: grpc.ServicerContext) -> empty_pb2.Empty: + context.set_code(grpc.StatusCode.UNIMPLEMENTED) # type: ignore + context.set_details("Method not implemented!") # type: ignore + raise NotImplementedError("Method not implemented!") + + async def DeclineCall(self, request: hfp_pb2.DeclineCallRequest, + context: grpc.ServicerContext) -> hfp_pb2.DeclineCallResponse: + self.enable_phone_for_testing() + self.bluetooth.hangup_call() + return hfp_pb2.DeclineCallResponse() + + async def DeclineCallAsHandsfree(self, request: hfp_pb2.DeclineCallAsHandsfreeRequest, + context: grpc.ServicerContext) -> hfp_pb2.DeclineCallAsHandsfreeResponse: + context.set_code(grpc.StatusCode.UNIMPLEMENTED) # type: ignore + context.set_details("Method not implemented!") # type: ignore + raise NotImplementedError("Method not implemented!") + + async def SetBatteryLevel(self, request: hfp_pb2.SetBatteryLevelRequest, + context: grpc.ServicerContext) -> (empty_pb2.Empty): + self.enable_phone_for_testing() + if request.battery_percentage > 100 or request.battery_percentage < 0: + await context.abort(grpc.StatusCode.INVALID_ARGUMENT, 'Wrong battery percentage.') + self.bluetooth.set_battery_level(math.floor((request.battery_percentage / 100) * 5)) + return empty_pb2.Empty() + + async def ConnectToAudioAsHandsfree(self, request: hfp_pb2.ConnectToAudioAsHandsfreeRequest, + context: grpc.ServicerContext) -> hfp_pb2.ConnectToAudioAsHandsfreeResponse: + context.set_code(grpc.StatusCode.UNIMPLEMENTED) # type: ignore + context.set_details("Method not implemented!") # type: ignore + raise NotImplementedError("Method not implemented!") + + async def DisconnectFromAudioAsHandsfree( + self, request: hfp_pb2.DisconnectFromAudioAsHandsfreeRequest, + context: grpc.ServicerContext) -> hfp_pb2.DisconnectFromAudioAsHandsfreeResponse: + context.set_code(grpc.StatusCode.UNIMPLEMENTED) # type: ignore + context.set_details("Method not implemented!") # type: ignore + raise NotImplementedError("Method not implemented!") + + async def SetVoiceRecognition(self, request: hfp_pb2.SetVoiceRecognitionRequest, + context: grpc.ServicerContext) -> hfp_pb2.SetVoiceRecognitionResponse: + context.set_code(grpc.StatusCode.UNIMPLEMENTED) # type: ignore + context.set_details("Method not implemented!") # type: ignore + raise NotImplementedError("Method not implemented!") + + async def SetVoiceRecognitionAsHandsfree( + self, request: hfp_pb2.SetVoiceRecognitionAsHandsfreeRequest, + context: grpc.ServicerContext) -> hfp_pb2.SetVoiceRecognitionAsHandsfreeResponse: + context.set_code(grpc.StatusCode.UNIMPLEMENTED) # type: ignore + context.set_details("Method not implemented!") # type: ignore + raise NotImplementedError("Method not implemented!") + async def MakeCall(self, request: hfp_pb2.MakeCallRequest, context: grpc.ServicerContext) -> hfp_pb2.MakeCallResponse: self.enable_phone_for_testing() |