Skip to content

Commit

Permalink
Merge pull request #14 from zachariahpifer/type-annotations
Browse files Browse the repository at this point in the history
Add Type Annotations
  • Loading branch information
jepler committed Apr 25, 2023
2 parents f51a46e + 824b56e commit 4998b27
Showing 1 changed file with 36 additions and 29 deletions.
65 changes: 36 additions & 29 deletions adafruit_ble_apple_media.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,13 @@
https://developer.apple.com/library/archive/documentation/CoreBluetooth/Reference/AppleMediaService_Reference/Introduction/Introduction.html#//apple_ref/doc/uid/TP40014716-CH2-SW1
"""
try:
from typing import Union, Type

AppleMediaServiceType = Union["AppleMediaService", Type["AppleMediaService"]]
except ImportError:
pass

import struct
import time

Expand All @@ -36,7 +43,7 @@ class _RemoteCommand(ComplexCharacteristic):

uuid = VendorUUID("9B3C81D8-57B1-4A8A-B8DF-0E56F7CA51C2")

def __init__(self):
def __init__(self) -> None:
super().__init__(
properties=Characteristic.WRITE_NO_RESPONSE | Characteristic.NOTIFY,
read_perm=Attribute.OPEN,
Expand All @@ -45,7 +52,7 @@ def __init__(self):
fixed_length=False,
)

def bind(self, service):
def bind(self, service: Service) -> _bleio.PacketBuffer:
"""Binds the characteristic to the given Service."""
bound_characteristic = super().bind(service)
return _bleio.PacketBuffer(bound_characteristic, buffer_size=1)
Expand All @@ -56,7 +63,7 @@ class _EntityUpdate(ComplexCharacteristic):

uuid = VendorUUID("2F7CABCE-808D-411F-9A0C-BB92BA96C102")

def __init__(self):
def __init__(self) -> None:
super().__init__(
properties=Characteristic.WRITE | Characteristic.NOTIFY,
read_perm=Attribute.OPEN,
Expand All @@ -65,7 +72,7 @@ def __init__(self):
fixed_length=False,
)

def bind(self, service):
def bind(self, service: Service) -> _bleio.PacketBuffer:
"""Binds the characteristic to the given Service."""
bound_characteristic = super().bind(service)
return _bleio.PacketBuffer(bound_characteristic, buffer_size=8)
Expand All @@ -76,7 +83,7 @@ class _EntityAttribute(Characteristic): # pylint: disable=too-few-public-method

uuid = VendorUUID("C6B2F38C-23AB-46D8-A6AB-A3A870BBD5D7")

def __init__(self):
def __init__(self) -> None:
super().__init__(
properties=Characteristic.WRITE | Characteristic.READ,
read_perm=Attribute.OPEN,
Expand All @@ -86,11 +93,11 @@ def __init__(self):


class _MediaAttribute:
def __init__(self, entity_id, attribute_id):
def __init__(self, entity_id: int, attribute_id: int) -> None:
self.key = (entity_id, attribute_id)

@staticmethod
def _update(obj):
def _update(obj: AppleMediaServiceType) -> None:
if not obj._buffer:
obj._buffer = bytearray(128)
length_read = obj._entity_update.readinto(obj._buffer)
Expand All @@ -107,7 +114,7 @@ def _update(obj):
value = str(obj._buffer[3:length_read], "utf-8")
obj._attribute_cache[(entity_id, attribute_id)] = value

def __get__(self, obj, cls):
def __get__(self, obj: AppleMediaServiceType, cls) -> str:
self._update(obj)
if self.key not in obj._attribute_cache:
siblings = [self.key[1]]
Expand All @@ -123,25 +130,25 @@ def __get__(self, obj, cls):


class _MediaAttributePlaybackState:
def __init__(self, playback_value):
def __init__(self, playback_value: int):
self._playback_value = playback_value

def __get__(self, obj, cls):
def __get__(self, obj: AppleMediaServiceType, cls) -> bool:
info = obj._playback_info
if info:
return int(info.split(",")[0]) == self._playback_value
return False


class _MediaAttributePlaybackInfo:
def __init__(self, position):
def __init__(self, position: int) -> None:
self._position = position

def __get__(self, obj, cls):
def __get__(self, obj: AppleMediaServiceType, cls) -> float:
info = obj._playback_info
if info:
return float(info.split(",")[self._position])
return 0
return 0.0


class UnsupportedCommand(Exception):
Expand Down Expand Up @@ -200,7 +207,7 @@ class AppleMediaService(Service):
duration = _MediaAttribute(2, 3)
"""Current track's duration as a string."""

def __init__(self, **kwargs):
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
self._buffer = None
self._cmd = None
Expand All @@ -209,7 +216,7 @@ def __init__(self, **kwargs):
self._supported_commands = []
self._command_buffer = None

def _send_command(self, command_id):
def _send_command(self, command_id: bytearray) -> None:
if not self._command_buffer:
self._command_buffer = bytearray(13)
i = self._remote_command.readinto( # pylint: disable=no-member
Expand All @@ -226,58 +233,58 @@ def _send_command(self, command_id):
self._cmd[0] = command_id
self._remote_command.write(self._cmd) # pylint: disable=no-member

def play(self):
def play(self) -> None:
"""Plays the current track. Does nothing if already playing."""
self._send_command(0)

def pause(self):
def pause(self) -> None:
"""Pauses the current track. Does nothing if already paused."""
self._send_command(1)

def toggle_play_pause(self):
def toggle_play_pause(self) -> None:
"""Plays the current track if it is paused. Otherwise it pauses the track."""
self._send_command(2)

def next_track(self):
def next_track(self) -> None:
"""Stops playing the current track and plays the next one."""
self._send_command(3)

def previous_track(self):
def previous_track(self) -> None:
"""Stops playing the current track and plays the previous track."""
self._send_command(4)

def volume_up(self):
def volume_up(self) -> None:
"""Increases the playback volume."""
self._send_command(5)

def volume_down(self):
def volume_down(self) -> None:
"""Decreases the playback volume."""
self._send_command(6)

def advance_repeat_mode(self):
def advance_repeat_mode(self) -> None:
"""Advances the repeat mode. Modes are: Off, One and All"""
self._send_command(7)

def advance_shuffle_mode(self):
def advance_shuffle_mode(self) -> None:
"""Advances the shuffle mode. Modes are: Off, One and All"""
self._send_command(8)

def skip_forward(self):
def skip_forward(self) -> None:
"""Skips forwards in the current track"""
self._send_command(9)

def skip_backward(self):
def skip_backward(self) -> None:
"""Skips backwards in the current track"""
self._send_command(10)

def like_track(self):
def like_track(self) -> None:
"""Likes the current track"""
self._send_command(11)

def dislike_track(self):
def dislike_track(self) -> None:
"""Dislikes the current track"""
self._send_command(12)

def bookmark_track(self):
def bookmark_track(self) -> None:
"""Bookmarks the current track"""
self._send_command(13)

0 comments on commit 4998b27

Please sign in to comment.