diff --git a/apps/auracast.py b/apps/auracast.py index 37e19b6d..f51dfa91 100644 --- a/apps/auracast.py +++ b/apps/auracast.py @@ -1195,7 +1195,7 @@ def transmit( input, input_format, ): - """Transmit a broadcast source.""" + """Transmit a broadcast source""" if manufacturer_data: vendor_id_str, data_hex = manufacturer_data.split(':') vendor_id = int(vendor_id_str) diff --git a/bumble/gatt.py b/bumble/gatt.py index f237b805..b3397ae9 100644 --- a/bumble/gatt.py +++ b/bumble/gatt.py @@ -42,7 +42,7 @@ from typing import ( ) from bumble.colors import color -from bumble.core import BaseBumbleError, UUID +from bumble.core import BaseBumbleError, InvalidOperationError, UUID from bumble.att import Attribute, AttributeValue from bumble.utils import ByteSerializable @@ -679,10 +679,14 @@ class DelegatedCharacteristicAdapter(CharacteristicAdapter): self.decode = decode def encode_value(self, value): - return self.encode(value) if self.encode else value + if self.encode is None: + raise InvalidOperationError('delegated adapter does not have an encoder') + return self.encode(value) def decode_value(self, value): - return self.decode(value) if self.decode else value + if self.decode is None: + raise InvalidOperationError('delegate adapter does not have a decoder') + return self.decode(value) # ----------------------------------------------------------------------------- diff --git a/bumble/profiles/vocs.py b/bumble/profiles/vocs.py index 70764fba..af5447fe 100644 --- a/bumble/profiles/vocs.py +++ b/bumble/profiles/vocs.py @@ -82,9 +82,7 @@ class VolumeOffsetState: async def notify_subscribers_via_connection(self, connection: Connection) -> None: assert self.attribute_value is not None - await connection.device.notify_subscribers( - attribute=self.attribute_value, value=bytes(self) - ) + await connection.device.notify_subscribers(attribute=self.attribute_value) def on_read(self, _connection: Optional[Connection]) -> bytes: return bytes(self) @@ -111,9 +109,7 @@ class VocsAudioLocation: assert self.attribute_value self.audio_location = AudioLocation(int.from_bytes(value, 'little')) - await connection.device.notify_subscribers( - attribute=self.attribute_value, value=value - ) + await connection.device.notify_subscribers(attribute=self.attribute_value) @dataclass @@ -169,9 +165,7 @@ class AudioOutputDescription: assert self.attribute_value self.audio_output_description = value.decode('utf-8') - await connection.device.notify_subscribers( - attribute=self.attribute_value, value=value - ) + await connection.device.notify_subscribers(attribute=self.attribute_value) # ----------------------------------------------------------------------------- @@ -203,37 +197,30 @@ class VolumeOffsetControlService(TemplateService): VolumeOffsetControlPoint(self.volume_offset_state) ) - self.volume_offset_state_characteristic = DelegatedCharacteristicAdapter( - Characteristic( - uuid=GATT_VOLUME_OFFSET_STATE_CHARACTERISTIC, - properties=( - Characteristic.Properties.READ | Characteristic.Properties.NOTIFY - ), - permissions=Characteristic.Permissions.READ_REQUIRES_ENCRYPTION, - value=CharacteristicValue(read=self.volume_offset_state.on_read), + self.volume_offset_state_characteristic = Characteristic( + uuid=GATT_VOLUME_OFFSET_STATE_CHARACTERISTIC, + properties=( + Characteristic.Properties.READ | Characteristic.Properties.NOTIFY ), - encode=lambda value: bytes(value), + permissions=Characteristic.Permissions.READ_REQUIRES_ENCRYPTION, + value=CharacteristicValue(read=self.volume_offset_state.on_read), ) - self.audio_location_characteristic = DelegatedCharacteristicAdapter( - Characteristic( - uuid=GATT_AUDIO_LOCATION_CHARACTERISTIC, - properties=( - Characteristic.Properties.READ - | Characteristic.Properties.NOTIFY - | Characteristic.Properties.WRITE_WITHOUT_RESPONSE - ), - permissions=( - Characteristic.Permissions.READ_REQUIRES_ENCRYPTION - | Characteristic.Permissions.WRITE_REQUIRES_ENCRYPTION - ), - value=CharacteristicValue( - read=self.audio_location.on_read, - write=self.audio_location.on_write, - ), + self.audio_location_characteristic = Characteristic( + uuid=GATT_AUDIO_LOCATION_CHARACTERISTIC, + properties=( + Characteristic.Properties.READ + | Characteristic.Properties.NOTIFY + | Characteristic.Properties.WRITE_WITHOUT_RESPONSE + ), + permissions=( + Characteristic.Permissions.READ_REQUIRES_ENCRYPTION + | Characteristic.Permissions.WRITE_REQUIRES_ENCRYPTION + ), + value=CharacteristicValue( + read=self.audio_location.on_read, + write=self.audio_location.on_write, ), - encode=lambda value: bytes(value), - decode=VocsAudioLocation.from_bytes, ) self.audio_location.attribute_value = self.audio_location_characteristic.value @@ -244,25 +231,22 @@ class VolumeOffsetControlService(TemplateService): value=CharacteristicValue(write=self.volume_offset_control_point.on_write), ) - self.audio_output_description_characteristic = DelegatedCharacteristicAdapter( - Characteristic( - uuid=GATT_AUDIO_OUTPUT_DESCRIPTION_CHARACTERISTIC, - properties=( - Characteristic.Properties.READ - | Characteristic.Properties.NOTIFY - | Characteristic.Properties.WRITE_WITHOUT_RESPONSE - ), - permissions=( - Characteristic.Permissions.READ_REQUIRES_ENCRYPTION - | Characteristic.Permissions.WRITE_REQUIRES_ENCRYPTION - ), - value=CharacteristicValue( - read=self.audio_output_description.on_read, - write=self.audio_output_description.on_write, - ), - ) + self.audio_output_description_characteristic = Characteristic( + uuid=GATT_AUDIO_OUTPUT_DESCRIPTION_CHARACTERISTIC, + properties=( + Characteristic.Properties.READ + | Characteristic.Properties.NOTIFY + | Characteristic.Properties.WRITE_WITHOUT_RESPONSE + ), + permissions=( + Characteristic.Permissions.READ_REQUIRES_ENCRYPTION + | Characteristic.Permissions.WRITE_REQUIRES_ENCRYPTION + ), + value=CharacteristicValue( + read=self.audio_output_description.on_read, + write=self.audio_output_description.on_write, + ), ) - self.audio_output_description.attribute_value = ( self.audio_output_description_characteristic.value ) @@ -287,18 +271,19 @@ class VolumeOffsetControlServiceProxy(ProfileServiceProxy): def __init__(self, service_proxy: ServiceProxy) -> None: self.service_proxy = service_proxy - self.volume_offset_state = DelegatedCharacteristicAdapter( + self.volume_offset_state = SerializableCharacteristicAdapter( service_proxy.get_required_characteristic_by_uuid( GATT_VOLUME_OFFSET_STATE_CHARACTERISTIC ), - decode=VolumeOffsetState.from_bytes, + VolumeOffsetState, ) - self.audio_location = SerializableCharacteristicAdapter( + self.audio_location = DelegatedCharacteristicAdapter( service_proxy.get_required_characteristic_by_uuid( GATT_AUDIO_LOCATION_CHARACTERISTIC ), - VocsAudioLocation, + encode=lambda value: bytes([int(value)]), + decode=lambda data: AudioLocation(data[0]), ) self.volume_offset_control_point = ( diff --git a/tests/vocs_test.py b/tests/vocs_test.py index 683654a9..599fc6d1 100644 --- a/tests/vocs_test.py +++ b/tests/vocs_test.py @@ -32,7 +32,6 @@ from bumble.profiles.vocs import ( SetVolumeOffsetOpCode, VolumeOffsetControlServiceProxy, VolumeOffsetState, - VocsAudioLocation, ) from bumble.profiles.vcs import VolumeControlService, VolumeControlServiceProxy from bumble.profiles.bap import AudioLocation @@ -81,9 +80,7 @@ async def test_init_service(vocs_client: VolumeOffsetControlServiceProxy): volume_offset=0, change_counter=0, ) - assert await vocs_client.audio_location.read_value() == VocsAudioLocation( - audio_location=AudioLocation.NOT_ALLOWED - ) + assert await vocs_client.audio_location.read_value() == AudioLocation.NOT_ALLOWED description = await vocs_client.audio_output_description.read_value() assert description == '' @@ -162,11 +159,9 @@ async def test_set_volume_offset(vocs_client: VolumeOffsetControlServiceProxy): @pytest.mark.asyncio async def test_set_audio_channel_location(vocs_client: VolumeOffsetControlServiceProxy): - new_audio_location = VocsAudioLocation(audio_location=AudioLocation.FRONT_LEFT) + new_audio_location = AudioLocation.FRONT_LEFT - await vocs_client.audio_location.write_value( - struct.pack('