Skip to content
Open
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
100 changes: 44 additions & 56 deletions src/linkplay/bridge.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,57 +34,57 @@
)


class LinkPlayDevice:
"""Represents a LinkPlay device."""
class LinkPlayPlayer:
"""Represents a LinkPlay player."""

bridge: LinkPlayBridge
properties: dict[DeviceAttribute, str]
device_properties: dict[DeviceAttribute, str]
properties: dict[PlayerAttribute, str]
custom_properties: dict[PlayerAttribute, str]
metainfo: dict[MetaInfo, dict[MetaInfoMetaData, str]]

previous_playing_mode: PlayingMode | None = None

controller: Callable[[], None] | None = None

def __init__(self, bridge: LinkPlayBridge):
self.bridge = bridge
self.properties = dict.fromkeys(DeviceAttribute.__members__.values(), "")

def to_dict(self):
"""Return the state of the LinkPlayDevice."""
return {"properties": self.properties}
self.device_properties = dict.fromkeys(DeviceAttribute.__members__.values(), "")
self.properties = dict.fromkeys(PlayerAttribute.__members__.values(), "")
self.custom_properties = dict.fromkeys(PlayerAttribute.__members__.values(), "")
self.metainfo = dict.fromkeys(MetaInfo.__members__.values(), {})

def set_callback(self, controller: Callable[[], None]) -> None:
"""Sets a callback function to notify events."""
self.controller = controller

async def update_status(self) -> None:
"""Update the device status."""
self.properties = await self.bridge.json_request(LinkPlayCommand.DEVICE_STATUS) # type: ignore[assignment]

async def reboot(self) -> None:
"""Reboot the device."""
await self.bridge.request(LinkPlayCommand.REBOOT)

@property
def uuid(self) -> str:
"""The UUID of the device."""
return self.properties.get(DeviceAttribute.UUID, "")
return self.device_properties.get(DeviceAttribute.UUID, "")

@property
def name(self) -> str:
"""The name of the device."""
return self.properties.get(DeviceAttribute.DEVICE_NAME, "")
return self.device_properties.get(DeviceAttribute.DEVICE_NAME, "")

@property
def manufacturer(self) -> str:
"""The manufacturer of the device."""
manufacturer, _ = get_info_from_project(
self.properties.get(DeviceAttribute.PROJECT, "")
self.device_properties.get(DeviceAttribute.PROJECT, "")
)
return manufacturer

@property
def model(self) -> str:
"""The model of the device."""
_, model = get_info_from_project(
self.properties.get(DeviceAttribute.PROJECT, "")
self.device_properties.get(DeviceAttribute.PROJECT, "")
)
return model

Expand All @@ -93,7 +93,7 @@ def playmode_support(self) -> list[PlayingMode]:
"""Returns the player playmode support."""

flags = InputMode(
int(self.properties[DeviceAttribute.PLAYMODE_SUPPORT], base=16)
int(self.device_properties[DeviceAttribute.PLAYMODE_SUPPORT], base=16)
)

playing_modes = [INPUT_MODE_MAP[flag] for flag in flags]
Expand All @@ -103,57 +103,48 @@ def playmode_support(self) -> list[PlayingMode]:
@property
def mac(self) -> str | None:
"""Returns the mac address."""
mac = self.properties.get(DeviceAttribute.ETH_MAC_ADDRESS)
mac = self.device_properties.get(DeviceAttribute.ETH_MAC_ADDRESS)
if mac == "00:00:00:00:00:00" or mac is None:
mac = self.properties.get(DeviceAttribute.STA_MAC_ADDRESS)
mac = self.device_properties.get(DeviceAttribute.STA_MAC_ADDRESS)
if mac == "00:00:00:00:00:00" or mac is None:
mac = self.properties.get(DeviceAttribute.MAC_ADDRESS)
mac = self.device_properties.get(DeviceAttribute.MAC_ADDRESS)
return mac

@property
def eth(self) -> str | None:
"""Returns the ethernet address."""
eth = self.properties.get(DeviceAttribute.ETH2)
eth = self.device_properties.get(DeviceAttribute.ETH2)
if eth == "0.0.0.0" or eth == "" or eth is None:
eth = self.properties.get(DeviceAttribute.ETH0)
eth = self.device_properties.get(DeviceAttribute.ETH0)
if eth == "0.0.0.0" or eth == "" or eth is None:
eth = self.properties.get(DeviceAttribute.APCLI0)
eth = self.device_properties.get(DeviceAttribute.APCLI0)
return eth

async def timesync(self) -> None:
"""Sync the time."""
timestamp = time.strftime("%Y%m%d%H%M%S")
await self.bridge.request(LinkPlayCommand.TIMESYNC.format(timestamp))


class LinkPlayPlayer:
"""Represents a LinkPlay player."""

bridge: LinkPlayBridge
properties: dict[PlayerAttribute, str]
custom_properties: dict[PlayerAttribute, str]
metainfo: dict[MetaInfo, dict[MetaInfoMetaData, str]]

previous_playing_mode: PlayingMode | None = None

def __init__(self, bridge: LinkPlayBridge):
self.bridge = bridge
self.properties = dict.fromkeys(PlayerAttribute.__members__.values(), "")
self.custom_properties = dict.fromkeys(PlayerAttribute.__members__.values(), "")
self.metainfo = dict.fromkeys(MetaInfo.__members__.values(), {})

def to_dict(self):
"""Return the state of the LinkPlayPlayer."""
return {"properties": self.properties}
return {
"device_properties": self.device_properties,
"properties": self.properties,
}

async def update_status(self) -> None:
"""Update the player status."""

self.device_properties = await self.bridge.json_request(
LinkPlayCommand.DEVICE_STATUS
) # type: ignore[assignment]

properties: dict[PlayerAttribute, str] = await self.bridge.json_request(
LinkPlayCommand.PLAYER_STATUS
) # type: ignore[assignment]

self.properties = fixup_player_properties(properties)
if self.bridge.device.manufacturer == MANUFACTURER_WIIM:
if self.bridge.player.manufacturer == MANUFACTURER_WIIM:
try:
self.metainfo: dict[
MetaInfo, dict[MetaInfoMetaData, str]
Expand All @@ -167,7 +158,7 @@ async def update_status(self) -> None:
self.metainfo = {}

# handle multiroom changes
if self.bridge.device.controller is not None and (
if self.bridge.player.controller is not None and (
(
self.previous_playing_mode != PlayingMode.FOLLOWER
and self.play_mode == PlayingMode.FOLLOWER
Expand All @@ -177,7 +168,7 @@ async def update_status(self) -> None:
and self.play_mode != PlayingMode.FOLLOWER
)
):
self.bridge.device.controller()
self.bridge.player.controller()
self.previous_playing_mode = self.play_mode

async def next(self) -> None:
Expand Down Expand Up @@ -233,7 +224,7 @@ async def set_volume(self, value: int) -> None:

async def set_equalizer_mode(self, mode: EqualizerMode) -> None:
"""Set the equalizer mode."""
if self.bridge.device.manufacturer == MANUFACTURER_WIIM:
if self.bridge.player.manufacturer == MANUFACTURER_WIIM:
await self._set_wiim_equalizer_mode(mode)
else:
await self._set_normal_equalizer_mode(mode)
Expand Down Expand Up @@ -278,7 +269,7 @@ async def set_play_mode(self, mode: PlayingMode) -> None:
async def play_preset(self, preset_number: int) -> None:
"""Play a preset."""
max_number_of_presets_allowed = int(
self.bridge.device.properties.get(DeviceAttribute.PRESET_KEY) or "10"
self.bridge.player.device_properties.get(DeviceAttribute.PRESET_KEY) or "10"
)
if not 0 < preset_number <= max_number_of_presets_allowed:
raise ValueError(
Expand Down Expand Up @@ -374,7 +365,7 @@ def status(self) -> PlayingStatus:
def equalizer_mode(self) -> EqualizerMode:
"""Returns the current equalizer mode."""
try:
if self.bridge.device.manufacturer == MANUFACTURER_WIIM:
if self.bridge.player.manufacturer == MANUFACTURER_WIIM:
# WiiM devices have a different equalizer mode handling
# and will never ever return what equalizer mode they are in
return EqualizerMode(
Expand All @@ -395,7 +386,7 @@ def equalizer_mode(self) -> EqualizerMode:
@property
def available_equalizer_modes(self) -> list[EqualizerMode]:
"""Returns the available equalizer modes."""
if self.bridge.device.manufacturer == MANUFACTURER_WIIM:
if self.bridge.player.manufacturer == MANUFACTURER_WIIM:
return [
EqualizerMode.NONE,
EqualizerMode.FLAT,
Expand Down Expand Up @@ -468,27 +459,24 @@ class LinkPlayBridge:
"""Represents a LinkPlay bridge to control the device and player attached to it."""

endpoint: LinkPlayEndpoint
device: LinkPlayDevice
player: LinkPlayPlayer
multiroom: LinkPlayMultiroom | None

def __init__(self, *, endpoint: LinkPlayEndpoint):
self.endpoint = endpoint
self.device = LinkPlayDevice(self)
self.player = LinkPlayPlayer(self)
self.multiroom = None

def __str__(self) -> str:
if self.device.name == "":
if self.player.name == "":
return f"{self.endpoint}"

return self.device.name
return self.player.name

def to_dict(self):
"""Return the state of the LinkPlayBridge."""
return {
"endpoint": self.endpoint.to_dict(),
"device": self.device.to_dict(),
"player": self.player.to_dict(),
"multiroom": self.multiroom.to_dict() if self.multiroom else None,
}
Expand Down Expand Up @@ -541,7 +529,7 @@ async def update_status(self, bridges: list[LinkPlayBridge]) -> None:
for follower in properties[MultiroomAttribute.FOLLOWER_LIST]
]
new_followers = [
bridge for bridge in bridges if bridge.device.uuid in follower_uuids
bridge for bridge in bridges if bridge.player.uuid in follower_uuids
]
self.followers.extend(new_followers)
except LinkPlayInvalidDataException as exc:
Expand All @@ -557,7 +545,7 @@ async def ungroup(self) -> None:
async def add_follower(self, follower: LinkPlayBridge) -> None:
"""Adds a follower to the multiroom group."""
await follower.request(
LinkPlayCommand.MULTIROOM_JOIN.format(self.leader.device.eth)
LinkPlayCommand.MULTIROOM_JOIN.format(self.leader.player.eth)
) # type: ignore[str-format]
if follower not in self.followers:
follower.multiroom = self
Expand All @@ -566,7 +554,7 @@ async def add_follower(self, follower: LinkPlayBridge) -> None:
async def remove_follower(self, follower: LinkPlayBridge) -> None:
"""Removes a follower from the multiroom group."""
await self.leader.request(
LinkPlayCommand.MULTIROOM_KICK.format(follower.device.eth)
LinkPlayCommand.MULTIROOM_KICK.format(follower.player.eth)
) # type: ignore[str-format]
if follower in self.followers:
follower.multiroom = None
Expand Down
16 changes: 8 additions & 8 deletions src/linkplay/controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,19 +33,19 @@ async def discover_bridges(self) -> None:

# Discover new bridges
discovered_bridges = await discover_linkplay_bridges(self.session)
current_bridges = [bridge.device.uuid for bridge in self.bridges]
current_bridges = [bridge.player.uuid for bridge in self.bridges]
new_bridges = [
discovered_bridge
for discovered_bridge in discovered_bridges
if discovered_bridge.device.uuid not in current_bridges
if discovered_bridge.player.uuid not in current_bridges
]
self.bridges.extend(new_bridges)

async def find_bridge(self, bridge_uuid: str) -> LinkPlayBridge | None:
"""Find a LinkPlay device by its bridge uuid."""

for bridge in self.bridges:
if bridge.device.uuid == bridge_uuid:
if bridge.player.uuid == bridge_uuid:
return bridge

return None
Expand All @@ -54,17 +54,17 @@ async def add_bridge(self, bridge_to_add: LinkPlayBridge) -> None:
"""Add given LinkPlay device if not already added."""

# Add bridge
current_bridges = [bridge.device.uuid for bridge in self.bridges]
if bridge_to_add.device.uuid not in current_bridges:
bridge_to_add.device.set_callback(self.get_bridge_callback())
current_bridges = [bridge.player.uuid for bridge in self.bridges]
if bridge_to_add.player.uuid not in current_bridges:
bridge_to_add.player.set_callback(self.get_bridge_callback())
self.bridges.append(bridge_to_add)

async def remove_bridge(self, bridge_to_remove: LinkPlayBridge) -> None:
"""Remove given LinkPlay device if not already deleted."""

# Remove bridge
current_bridges = [bridge.device.uuid for bridge in self.bridges]
if bridge_to_remove.device.uuid in current_bridges:
current_bridges = [bridge.player.uuid for bridge in self.bridges]
if bridge_to_remove.player.uuid in current_bridges:
self.bridges.remove(bridge_to_remove)

async def discover_multirooms(self) -> None:
Expand Down
6 changes: 3 additions & 3 deletions src/linkplay/discovery.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ async def linkplay_factory_bridge_endpoint(
Raises LinkPlayRequestException if the device is not an expected LinkPlay device."""

bridge: LinkPlayBridge = LinkPlayBridge(endpoint=endpoint)
await bridge.device.update_status()
await bridge.player.update_status()
await bridge.player.update_status()
return bridge

Expand Down Expand Up @@ -80,7 +80,7 @@ async def add_linkplay_device_to_list(upnp_device: CaseInsensitiveDict):

try:
bridge = await linkplay_factory_httpapi_bridge(ip_address, session)
bridges[bridge.device.uuid] = bridge
bridges[bridge.player.uuid] = bridge
except LinkPlayRequestException:
pass

Expand All @@ -93,7 +93,7 @@ async def add_linkplay_device_to_list(upnp_device: CaseInsensitiveDict):
multiroom_discovered_bridges: dict[str, LinkPlayBridge] = {}
for bridge in bridges.values():
for new_bridge in await discover_bridges_through_multiroom(bridge, session):
multiroom_discovered_bridges[new_bridge.device.uuid] = new_bridge
multiroom_discovered_bridges[new_bridge.player.uuid] = new_bridge

bridges = bridges | multiroom_discovered_bridges

Expand Down
Loading