Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add enums for player fields #73

Merged
merged 2 commits into from
Jan 10, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 11 additions & 4 deletions pyheos/command/__init__.py
Original file line number Diff line number Diff line change
@@ -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 = (
Expand All @@ -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"
Expand Down Expand Up @@ -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:
Expand Down
48 changes: 25 additions & 23 deletions pyheos/player.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand All @@ -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 []
Expand Down Expand Up @@ -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)
Expand All @@ -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],
Expand All @@ -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,
)

Expand All @@ -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.
Expand Down
4 changes: 2 additions & 2 deletions pyheos/system.py
Original file line number Diff line number Diff line change
Expand Up @@ -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":
Expand All @@ -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),
)


Expand Down
18 changes: 18 additions & 0 deletions pyheos/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""

Expand Down
6 changes: 3 additions & 3 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
)

Expand All @@ -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,
)

Expand Down
3 changes: 2 additions & 1 deletion tests/fixtures/player.get_players.json
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
5 changes: 4 additions & 1 deletion tests/test_heos.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,12 +43,14 @@
from pyheos.types import (
AddCriteriaType,
ConnectionState,
LineOutLevelType,
MediaType,
NetworkType,
PlayState,
RepeatType,
SignalHeosEvent,
SignalType,
VolumeControlType,
)
from tests.common import MediaItems

Expand Down Expand Up @@ -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
Expand Down
12 changes: 9 additions & 3 deletions tests/test_player.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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"


Expand All @@ -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"


Expand Down
Loading