diff --git a/pyheos/command/__init__.py b/pyheos/command/__init__.py index b5f162b..0d8e3ce 100644 --- a/pyheos/command/__init__.py +++ b/pyheos/command/__init__.py @@ -1,7 +1,7 @@ """Define the HEOS command module.""" import logging -from enum import StrEnum +from enum import ReprEnum from typing import Any, Final, TypeVar REPORT_ISSUE_TEXT: Final = ( @@ -16,6 +16,7 @@ ATTR_COMMAND: Final = "command" ATTR_CONTAINER: Final = "container" ATTR_CONTAINER_ID: Final = "cid" +ATTR_CONTROL: Final = "control" ATTR_COUNT: Final = "count" ATTR_CURRENT_POSITION: Final = "cur_pos" ATTR_DESTINATION_QUEUE_ID: Final = "dqid" @@ -156,12 +157,18 @@ _LOGGER: Final = logging.getLogger(__name__) -TStrEnum = TypeVar("TStrEnum", bound=StrEnum) +TEnum = TypeVar("TEnum", bound=ReprEnum) + + +def optional_int(value: str | None) -> int | None: + if value is not None: + return int(value) + return None def parse_enum( - key: str, data: dict[str, Any], enum_type: type[TStrEnum], default: TStrEnum -) -> TStrEnum: + key: str, data: dict[str, Any], enum_type: type[TEnum], default: TEnum +) -> TEnum: """Parse an enum value from the provided data. This is a safe operation that will return the default value if the key is missing or the value is not recognized.""" value = data.get(key) if value is None: diff --git a/pyheos/player.py b/pyheos/player.py index 0236d69..b5f8778 100644 --- a/pyheos/player.py +++ b/pyheos/player.py @@ -6,18 +6,20 @@ from datetime import datetime from typing import TYPE_CHECKING, Any, Final, Optional, cast -from pyheos.command import parse_enum +from pyheos.command import optional_int, parse_enum from pyheos.dispatch import DisconnectType, EventCallbackType, callback_wrapper from pyheos.media import MediaItem, QueueItem, ServiceOption from pyheos.message import HeosMessage from pyheos.types import ( AddCriteriaType, ControlType, + LineOutLevelType, MediaType, NetworkType, PlayState, RepeatType, SignalType, + VolumeControlType, ) from . import command as c @@ -96,7 +98,7 @@ class HeosNowPlayingMedia: current_position: int | None = None current_position_updated: datetime | None = None duration: int | None = None - supported_controls: Sequence[str] = field( + supported_controls: Sequence[ControlType] = field( default_factory=lambda: CONTROLS_ALL, init=False ) options: Sequence[ServiceOption] = field( @@ -118,19 +120,12 @@ def _update_from_message(self, message: HeosMessage) -> None: self.image_url = data.get(c.ATTR_IMAGE_URL) self.album_id = data.get(c.ATTR_ALBUM_ID) self.media_id = data.get(c.ATTR_MEDIA_ID) - self.queue_id = self.__get_optional_int(data.get(c.ATTR_QUEUE_ID)) - self.source_id = self.__get_optional_int(data.get(c.ATTR_SOURCE_ID)) + self.queue_id = optional_int(data.get(c.ATTR_QUEUE_ID)) + self.source_id = optional_int(data.get(c.ATTR_SOURCE_ID)) self.options = ServiceOption._from_options(message.options) self._update_supported_controls() self.clear_progress() - @staticmethod - def __get_optional_int(value: Any) -> int | None: - try: - return int(str(value)) - except (TypeError, ValueError): - return None - def _update_supported_controls(self) -> None: """Updates the supported controls based on the source and type.""" new_supported_controls = CONTROLS_ALL if self.source_id is not None else [] @@ -183,8 +178,11 @@ class HeosPlayer: serial: str | None = field(repr=False, hash=False, compare=False) version: str = field(repr=True, hash=False, compare=False) ip_address: str = field(repr=True, hash=False, compare=False) - network: str = field(repr=False, hash=False, compare=False) - line_out: int = field(repr=False, hash=False, compare=False) + network: NetworkType = field(repr=False, hash=False, compare=False) + line_out: LineOutLevelType = field(repr=False, hash=False, compare=False) + control: VolumeControlType = field( + repr=False, hash=False, compare=False, default=VolumeControlType.UNKNOWN + ) state: PlayState | None = field(repr=True, hash=False, compare=False, default=None) volume: int = field(repr=False, hash=False, compare=False, default=0) is_muted: bool = field(repr=False, hash=False, compare=False, default=False) @@ -200,12 +198,6 @@ class HeosPlayer: group_id: int | None = field(repr=False, hash=False, compare=False, default=None) heos: Optional["Heos"] = field(repr=False, hash=False, compare=False, default=None) - @staticmethod - def __get_optional_int(value: str | None) -> int | None: - if value is not None: - return int(value) - return None - @staticmethod def _from_data( data: dict[str, Any], @@ -221,8 +213,13 @@ def _from_data( version=data[c.ATTR_VERSION], ip_address=data[c.ATTR_IP_ADDRESS], network=parse_enum(c.ATTR_NETWORK, data, NetworkType, NetworkType.UNKNOWN), - line_out=int(data[c.ATTR_LINE_OUT]), - group_id=HeosPlayer.__get_optional_int(data.get(c.ATTR_GROUP_ID)), + line_out=parse_enum( + c.ATTR_LINE_OUT, data, LineOutLevelType, LineOutLevelType.UNKNOWN + ), + control=parse_enum( + c.ATTR_CONTROL, data, VolumeControlType, VolumeControlType.UNKNOWN + ), + group_id=optional_int(data.get(c.ATTR_GROUP_ID)), heos=heos, ) @@ -237,8 +234,13 @@ def _update_from_data(self, data: dict[str, Any]) -> None: self.network = parse_enum( c.ATTR_NETWORK, data, NetworkType, NetworkType.UNKNOWN ) - self.line_out = int(data[c.ATTR_LINE_OUT]) - self.group_id = HeosPlayer.__get_optional_int(data.get(c.ATTR_GROUP_ID)) + self.line_out = parse_enum( + c.ATTR_LINE_OUT, data, LineOutLevelType, LineOutLevelType.UNKNOWN + ) + self.control = parse_enum( + c.ATTR_CONTROL, data, VolumeControlType, VolumeControlType.UNKNOWN + ) + self.group_id = optional_int(data.get(c.ATTR_GROUP_ID)) async def _on_event(self, event: HeosMessage, all_progress_events: bool) -> bool: """Updates the player based on the received HEOS event. diff --git a/pyheos/system.py b/pyheos/system.py index 65b8092..1ebc8bf 100644 --- a/pyheos/system.py +++ b/pyheos/system.py @@ -19,7 +19,7 @@ class HeosHost: serial: str | None version: str ip_address: str - network: str + network: NetworkType @classmethod def from_data(cls, data: dict[str, str]) -> "HeosHost": @@ -37,7 +37,7 @@ def from_data(cls, data: dict[str, str]) -> "HeosHost": data.get(c.ATTR_SERIAL), data[c.ATTR_VERSION], data[c.ATTR_IP_ADDRESS], - data[c.ATTR_NETWORK], + c.parse_enum(c.ATTR_NETWORK, data, NetworkType, NetworkType.UNKNOWN), ) diff --git a/pyheos/types.py b/pyheos/types.py index 433ce21..787a7f7 100644 --- a/pyheos/types.py +++ b/pyheos/types.py @@ -20,6 +20,24 @@ class ConnectionState(StrEnum): RECONNECTING = "reconnecting" +class LineOutLevelType(IntEnum): + """Define the line out level types.""" + + UNKNOWN = 0 + VARIABLE = 1 + FIXED = 2 + + +class VolumeControlType(IntEnum): + "Define control types." + + UNKNOWN = 0 + NONE = 1 + IR = 2 + TRIGGER = 3 + NETWORK = 4 + + class NetworkType(StrEnum): """Define the network type.""" diff --git a/tests/conftest.py b/tests/conftest.py index ec995b3..921f2e1 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -10,7 +10,7 @@ from pyheos.heos import Heos, HeosOptions from pyheos.media import MediaItem, MediaMusicSource from pyheos.player import HeosPlayer -from pyheos.types import NetworkType +from pyheos.types import LineOutLevelType, NetworkType from tests.common import MediaItems, MediaMusicSources from . import MockHeos, MockHeosDevice @@ -142,7 +142,7 @@ async def player_fixture(heos: MockHeos) -> HeosPlayer: version="1.493.180", ip_address="127.0.0.1", network=NetworkType.WIRED, - line_out=1, + line_out=LineOutLevelType.FIXED, heos=heos, ) @@ -158,7 +158,7 @@ async def player_front_porch_fixture(heos: MockHeos) -> HeosPlayer: version="1.493.180", ip_address="127.0.0.2", network=NetworkType.WIFI, - line_out=1, + line_out=LineOutLevelType.FIXED, heos=heos, ) diff --git a/tests/fixtures/player.get_players.json b/tests/fixtures/player.get_players.json index ccb6cef..9d85716 100644 --- a/tests/fixtures/player.get_players.json +++ b/tests/fixtures/player.get_players.json @@ -11,7 +11,8 @@ "version": "1.493.180", "ip": "127.0.0.1", "network": "wired", - "lineout": 1, + "lineout": 2, + "control": 2, "serial": "B1A2C3K" }, { "name": "Front Porch", diff --git a/tests/test_heos.py b/tests/test_heos.py index 8e65363..357a20f 100644 --- a/tests/test_heos.py +++ b/tests/test_heos.py @@ -43,12 +43,14 @@ from pyheos.types import ( AddCriteriaType, ConnectionState, + LineOutLevelType, MediaType, NetworkType, PlayState, RepeatType, SignalHeosEvent, SignalType, + VolumeControlType, ) from tests.common import MediaItems @@ -493,7 +495,8 @@ async def test_get_players(heos: Heos) -> None: assert player.player_id == 1 assert player.name == "Back Patio" assert player.ip_address == "127.0.0.1" - assert player.line_out == 1 + assert player.line_out == LineOutLevelType.FIXED + assert player.control == VolumeControlType.IR assert player.model == "HEOS Drive" assert player.network == NetworkType.WIRED assert player.state == PlayState.STOP diff --git a/tests/test_player.py b/tests/test_player.py index 619b266..cf5a420 100644 --- a/tests/test_player.py +++ b/tests/test_player.py @@ -8,7 +8,13 @@ from pyheos.const import INPUT_AUX_IN_1, MUSIC_SOURCE_DEEZER, MUSIC_SOURCE_PLAYLISTS from pyheos.media import MediaItem from pyheos.player import HeosPlayer -from pyheos.types import AddCriteriaType, NetworkType, PlayState, RepeatType +from pyheos.types import ( + AddCriteriaType, + LineOutLevelType, + NetworkType, + PlayState, + RepeatType, +) from tests import CallCommand, calls_command, calls_commands, value from tests.common import MediaItems @@ -41,7 +47,7 @@ def test_from_data(network: str | None, expected_network: NetworkType) -> None: assert player.version == "1.493.180" assert player.ip_address == "192.168.0.1" assert player.network == expected_network - assert player.line_out == 1 + assert player.line_out == LineOutLevelType.VARIABLE assert player.serial == "1234567890" @@ -65,7 +71,7 @@ async def test_update_from_data(player: HeosPlayer) -> None: assert player.version == "2.0.0" assert player.ip_address == "192.168.0.2" assert player.network == NetworkType.WIFI - assert player.line_out == 0 + assert player.line_out == LineOutLevelType.UNKNOWN assert player.serial == "0987654321"