diff --git a/pyatv/protocols/mrp/__init__.py b/pyatv/protocols/mrp/__init__.py index c83355cbd..ed512598a 100644 --- a/pyatv/protocols/mrp/__init__.py +++ b/pyatv/protocols/mrp/__init__.py @@ -685,12 +685,20 @@ async def _update_power_state(self, message): @staticmethod def _get_power_state(device_info_message) -> PowerState: - logical_device_count = protobuf.extract_inner( - device_info_message - ).logicalDeviceCount + info = protobuf.extract_inner(device_info_message) + logical_device_count = info.logicalDeviceCount if logical_device_count >= 1: return PowerState.On if logical_device_count == 0: + # The leader of a HomePod stereo pair reports zero logical devices + # even while it is playing, so it is identified as the leader of + # its cluster instead + if ( + info.isClusterAware + and info.HasField("clusterLeaderID") + and info.clusterLeaderID == info.deviceUID + ): + return PowerState.On return PowerState.Off return PowerState.Unknown diff --git a/tests/protocols/mrp/test_mrp.py b/tests/protocols/mrp/test_mrp.py index c28894a34..782b27b86 100644 --- a/tests/protocols/mrp/test_mrp.py +++ b/tests/protocols/mrp/test_mrp.py @@ -5,10 +5,17 @@ from deepdiff import DeepDiff import pytest -from pyatv.const import OperatingSystem, PairingRequirement, Protocol +from pyatv.const import OperatingSystem, PairingRequirement, PowerState, Protocol from pyatv.core import MutableService, mdns from pyatv.interface import DeviceInfo -from pyatv.protocols.mrp import device_info, scan, service_info +from pyatv.protocols.mrp import ( + MrpPower, + device_info, + messages, + protobuf, + scan, + service_info, +) from pyatv.support.device_info import lookup_version MRP_SERVICE = "_mediaremotetv._tcp.local" @@ -113,3 +120,45 @@ async def test_disabled_service_no_pairing(): {Protocol.MRP: mrp_service, Protocol.AirPlay: airplay_service}, ) assert mrp_service.pairing == PairingRequirement.NotNeeded + + +def _device_info(**fields): + message = messages.create(protobuf.ProtocolMessage.DEVICE_INFO_MESSAGE) + inner = protobuf.extract_inner(message) + for name, value in fields.items(): + setattr(inner, name, value) + return message + + +@pytest.mark.parametrize( + "fields,expected", + [ + ({"logicalDeviceCount": 1}, PowerState.On), + ({"logicalDeviceCount": 0}, PowerState.Off), + # The leader of a HomePod stereo pair reports zero logical devices + # while it is playing, so the cluster fields have to decide it + ( + { + "logicalDeviceCount": 0, + "deviceUID": "leader", + "isClusterAware": True, + "clusterLeaderID": "leader", + }, + PowerState.On, + ), + # The other member of the pair is not the leader and stays off + ( + { + "logicalDeviceCount": 0, + "deviceUID": "follower", + "isClusterAware": True, + "clusterLeaderID": "leader", + }, + PowerState.Off, + ), + # An Apple TV that is asleep has no cluster fields at all + ({"logicalDeviceCount": 0, "deviceUID": "atv"}, PowerState.Off), + ], +) +def test_power_state_from_device_info(fields, expected): + assert MrpPower._get_power_state(_device_info(**fields)) == expected