Module pyatv.conf

Configuration used when connecting to a device.

A configuration describes a device, e.g. it's name, IP address and credentials. It is possible to manually create a configuration, but generally scanning for devices will provide configurations for you.

For a configuration to be usable ("ready") it must have either a DMAP or MRP configuration (or both), as connecting to plain AirPlay devices it not supported.

Expand source code 
"""Configuration used when connecting to a device.

A configuration describes a device, e.g. it's name, IP address and credentials. It is
possible to manually create a configuration, but generally scanning for devices will
provide configurations for you.

For a configuration to be usable ("ready") it must have either a `DMAP` or `MRP`
configuration (or both), as connecting to plain `AirPlay` devices it not supported.
"""
from ipaddress import IPv4Address
from typing import Dict, List, Mapping, Optional, Tuple, cast

from pyatv import exceptions
from pyatv.const import DeviceModel, OperatingSystem, Protocol
from pyatv.interface import BaseService, DeviceInfo
from pyatv.support.device_info import lookup_model, lookup_version


class AppleTV:
    """Representation of an Apple TV configuration.

    An instance of this class represents a single device. A device can have
    several services depending on the protocols it supports, e.g. DMAP or
    AirPlay.
    """

    def __init__(
        self,
        address: IPv4Address,
        name: str,
        deep_sleep: bool = False,
        model: DeviceModel = DeviceModel.Unknown,
        properties: Optional[Mapping[str, str]] = None,
    ) -> None:
        """Initialize a new AppleTV."""
        self._address = address
        self._name = name
        self._deep_sleep = deep_sleep
        self._model = model
        self._services: Dict[Protocol, BaseService] = {}
        self._properties: Mapping[str, str] = properties or {}

    @property
    def address(self) -> IPv4Address:
        """IP address of device."""
        return self._address

    @property
    def name(self) -> str:
        """Name of device."""
        return self._name

    @property
    def deep_sleep(self) -> bool:
        """If device is in deep sleep."""
        return self._deep_sleep

    @property
    def ready(self) -> bool:
        """Return if configuration is ready, i.e. has a main service."""
        ready_protocols = set(list(Protocol))

        # Companion has no unique identifier so it's the only protocol that can't be
        # used independently for now
        ready_protocols.remove(Protocol.Companion)

        intersection = ready_protocols.intersection(self._services.keys())
        return len(intersection) > 0

    @property
    def identifier(self) -> Optional[str]:
        """Return the main identifier associated with this device."""
        for prot in [Protocol.MRP, Protocol.DMAP, Protocol.AirPlay, Protocol.RAOP]:
            service = self._services.get(prot)
            if service:
                return service.identifier
        return None

    @property
    def all_identifiers(self) -> List[str]:
        """Return all unique identifiers for this device."""
        return [x.identifier for x in self.services if x.identifier is not None]

    def add_service(self, service: BaseService) -> None:
        """Add a new service.

        If the service already exists, it will be merged.
        """
        existing = self._services.get(service.protocol)
        if existing is not None:
            existing.merge(service)
        else:
            self._services[service.protocol] = service

    def get_service(self, protocol: Protocol) -> Optional[BaseService]:
        """Look up a service based on protocol.

        If a service with the specified protocol is not available, None is
        returned.
        """
        return self._services.get(protocol)

    @property
    def services(self) -> List[BaseService]:
        """Return all supported services."""
        return list(self._services.values())

    def main_service(self, protocol: Optional[Protocol] = None) -> BaseService:
        """Return suggested service used to establish connection."""
        protocols = (
            [protocol]
            if protocol is not None
            else [Protocol.MRP, Protocol.DMAP, Protocol.AirPlay, Protocol.RAOP]
        )

        for prot in protocols:
            service = self._services.get(prot)
            if service is not None:
                return service

        raise exceptions.NoServiceError("no service to connect to")

    def set_credentials(self, protocol: Protocol, credentials: str) -> bool:
        """Set credentials for a protocol if it exists."""
        service = self.get_service(protocol)
        if service:
            service.credentials = credentials
            return True
        return False

    # TODO: The extraction should be generialized and moved somewhere else. It is
    # very hard to test rght now.
    @property
    def device_info(self) -> DeviceInfo:
        """Return general device information."""
        properties = self._all_properties()

        build: Optional[str] = properties.get("systembuildversion")
        version = properties.get("ov")
        if not version:
            version = properties.get("osvers", lookup_version(build))

        model_name: Optional[str] = properties.get("model", properties.get("am"))
        if model_name:
            model = lookup_model(model_name)
        else:
            model = self._model

        # MRP devices run tvOS (as far as we know now) as well as HomePods for
        # some reason
        if Protocol.MRP in self._services or model in [
            DeviceModel.HomePod,
            DeviceModel.HomePodMini,
        ]:
            os_type = OperatingSystem.TvOS
        elif Protocol.DMAP in self._services:
            os_type = OperatingSystem.Legacy
        elif model in [DeviceModel.AirPortExpress, DeviceModel.AirPortExpressGen2]:
            os_type = OperatingSystem.AirPortOS
        else:
            os_type = OperatingSystem.Unknown

        mac = properties.get("macaddress", properties.get("deviceid"))
        if mac:
            mac = mac.upper()

        # The waMA property comes from the _airport._tcp.local service, announced by
        # AirPort Expresses (used by the admin tool). It contains various information,
        # for instance MAC address and software version.
        wama = properties.get("wama")
        if wama:
            props: Mapping[str, str] = dict(
                cast(Tuple[str, str], prop.split("=", maxsplit=1))
                for prop in ("macaddress=" + wama).split(",")
            )
            if not mac:
                mac = props["macaddress"].replace("-", ":").upper()
            version = props.get("syVs")

        return DeviceInfo(os_type, version, build, model, mac)

    def _all_properties(self) -> Mapping[str, str]:
        properties: Dict[str, str] = {}
        properties.update(self._properties)
        for service in self.services:
            properties.update(service.properties)
        return properties

    def __eq__(self, other) -> bool:
        """Compare instance with another instance."""
        if isinstance(other, self.__class__):
            return self.identifier == other.identifier
        return False

    def __str__(self) -> str:
        """Return a string representation of this object."""
        device_info = self.device_info
        services = "\n".join([" - {0}".format(s) for s in self._services.values()])
        identifiers = "\n".join([" - {0}".format(x) for x in self.all_identifiers])
        return (
            f"       Name: {self.name}\n"
            f"   Model/SW: {device_info}\n"
            f"    Address: {self.address}\n"
            f"        MAC: {self.device_info.mac}\n"
            f" Deep Sleep: {self.deep_sleep}\n"
            f"Identifiers:\n"
            f"{identifiers}\n"
            f"Services:\n"
            f"{services}"
        )


# pylint: disable=too-few-public-methods
class DmapService(BaseService):
    """Representation of a DMAP service."""

    def __init__(
        self,
        identifier: Optional[str],
        credentials: Optional[str],
        port: int = 3689,
        properties: Optional[Mapping[str, str]] = None,
    ) -> None:
        """Initialize a new DmapService."""
        super().__init__(
            identifier,
            Protocol.DMAP,
            port,
            properties,
        )
        self.credentials = credentials


# pylint: disable=too-few-public-methods
class MrpService(BaseService):
    """Representation of a MediaRemote Protocol (MRP) service."""

    def __init__(
        self,
        identifier: Optional[str],
        port: int,
        credentials: Optional[str] = None,
        properties: Optional[Mapping[str, str]] = None,
    ) -> None:
        """Initialize a new MrpService."""
        super().__init__(identifier, Protocol.MRP, port, properties)
        self.credentials = credentials


# pylint: disable=too-few-public-methods
class AirPlayService(BaseService):
    """Representation of an AirPlay service."""

    def __init__(
        self,
        identifier: Optional[str],
        port: int = 7000,
        credentials: Optional[str] = None,
        properties: Optional[Mapping[str, str]] = None,
    ) -> None:
        """Initialize a new AirPlayService."""
        super().__init__(identifier, Protocol.AirPlay, port, properties)
        self.credentials = credentials


# pylint: disable=too-few-public-methods
class CompanionService(BaseService):
    """Representation of a Companion link service."""

    def __init__(
        self,
        port: int,
        credentials: Optional[str] = None,
        properties: Optional[Mapping[str, str]] = None,
    ) -> None:
        """Initialize a new CompaniomService."""
        super().__init__(None, Protocol.Companion, port, properties)
        self.credentials = credentials


# pylint: disable=too-few-public-methods
class RaopService(BaseService):
    """Representation of an RAOP service."""

    def __init__(
        self,
        identifier: Optional[str],
        port: int = 7000,
        credentials: Optional[str] = None,
        properties: Optional[Mapping[str, str]] = None,
    ) -> None:
        """Initialize a new RaopService."""
        super().__init__(identifier, Protocol.RAOP, port, properties)
        self.credentials = credentials

Classes

class AirPlayService (identifier: Optional[str], port: int = 7000, credentials: Optional[str] = None, properties: Optional[Mapping[str, str]] = None)

Representation of an AirPlay service.

Initialize a new AirPlayService.

Expand source code 
class AirPlayService(BaseService):
    """Representation of an AirPlay service."""

    def __init__(
        self,
        identifier: Optional[str],
        port: int = 7000,
        credentials: Optional[str] = None,
        properties: Optional[Mapping[str, str]] = None,
    ) -> None:
        """Initialize a new AirPlayService."""
        super().__init__(identifier, Protocol.AirPlay, port, properties)
        self.credentials = credentials

Ancestors

Inherited members

class AppleTV (address: ipaddress.IPv4Address, name: str, deep_sleep: bool = False, model: DeviceModel = DeviceModel.Unknown, properties: Optional[Mapping[str, str]] = None)

Representation of an Apple TV configuration.

An instance of this class represents a single device. A device can have several services depending on the protocols it supports, e.g. DMAP or AirPlay.

Initialize a new AppleTV.

Expand source code 
class AppleTV:
    """Representation of an Apple TV configuration.

    An instance of this class represents a single device. A device can have
    several services depending on the protocols it supports, e.g. DMAP or
    AirPlay.
    """

    def __init__(
        self,
        address: IPv4Address,
        name: str,
        deep_sleep: bool = False,
        model: DeviceModel = DeviceModel.Unknown,
        properties: Optional[Mapping[str, str]] = None,
    ) -> None:
        """Initialize a new AppleTV."""
        self._address = address
        self._name = name
        self._deep_sleep = deep_sleep
        self._model = model
        self._services: Dict[Protocol, BaseService] = {}
        self._properties: Mapping[str, str] = properties or {}

    @property
    def address(self) -> IPv4Address:
        """IP address of device."""
        return self._address

    @property
    def name(self) -> str:
        """Name of device."""
        return self._name

    @property
    def deep_sleep(self) -> bool:
        """If device is in deep sleep."""
        return self._deep_sleep

    @property
    def ready(self) -> bool:
        """Return if configuration is ready, i.e. has a main service."""
        ready_protocols = set(list(Protocol))

        # Companion has no unique identifier so it's the only protocol that can't be
        # used independently for now
        ready_protocols.remove(Protocol.Companion)

        intersection = ready_protocols.intersection(self._services.keys())
        return len(intersection) > 0

    @property
    def identifier(self) -> Optional[str]:
        """Return the main identifier associated with this device."""
        for prot in [Protocol.MRP, Protocol.DMAP, Protocol.AirPlay, Protocol.RAOP]:
            service = self._services.get(prot)
            if service:
                return service.identifier
        return None

    @property
    def all_identifiers(self) -> List[str]:
        """Return all unique identifiers for this device."""
        return [x.identifier for x in self.services if x.identifier is not None]

    def add_service(self, service: BaseService) -> None:
        """Add a new service.

        If the service already exists, it will be merged.
        """
        existing = self._services.get(service.protocol)
        if existing is not None:
            existing.merge(service)
        else:
            self._services[service.protocol] = service

    def get_service(self, protocol: Protocol) -> Optional[BaseService]:
        """Look up a service based on protocol.

        If a service with the specified protocol is not available, None is
        returned.
        """
        return self._services.get(protocol)

    @property
    def services(self) -> List[BaseService]:
        """Return all supported services."""
        return list(self._services.values())

    def main_service(self, protocol: Optional[Protocol] = None) -> BaseService:
        """Return suggested service used to establish connection."""
        protocols = (
            [protocol]
            if protocol is not None
            else [Protocol.MRP, Protocol.DMAP, Protocol.AirPlay, Protocol.RAOP]
        )

        for prot in protocols:
            service = self._services.get(prot)
            if service is not None:
                return service

        raise exceptions.NoServiceError("no service to connect to")

    def set_credentials(self, protocol: Protocol, credentials: str) -> bool:
        """Set credentials for a protocol if it exists."""
        service = self.get_service(protocol)
        if service:
            service.credentials = credentials
            return True
        return False

    # TODO: The extraction should be generialized and moved somewhere else. It is
    # very hard to test rght now.
    @property
    def device_info(self) -> DeviceInfo:
        """Return general device information."""
        properties = self._all_properties()

        build: Optional[str] = properties.get("systembuildversion")
        version = properties.get("ov")
        if not version:
            version = properties.get("osvers", lookup_version(build))

        model_name: Optional[str] = properties.get("model", properties.get("am"))
        if model_name:
            model = lookup_model(model_name)
        else:
            model = self._model

        # MRP devices run tvOS (as far as we know now) as well as HomePods for
        # some reason
        if Protocol.MRP in self._services or model in [
            DeviceModel.HomePod,
            DeviceModel.HomePodMini,
        ]:
            os_type = OperatingSystem.TvOS
        elif Protocol.DMAP in self._services:
            os_type = OperatingSystem.Legacy
        elif model in [DeviceModel.AirPortExpress, DeviceModel.AirPortExpressGen2]:
            os_type = OperatingSystem.AirPortOS
        else:
            os_type = OperatingSystem.Unknown

        mac = properties.get("macaddress", properties.get("deviceid"))
        if mac:
            mac = mac.upper()

        # The waMA property comes from the _airport._tcp.local service, announced by
        # AirPort Expresses (used by the admin tool). It contains various information,
        # for instance MAC address and software version.
        wama = properties.get("wama")
        if wama:
            props: Mapping[str, str] = dict(
                cast(Tuple[str, str], prop.split("=", maxsplit=1))
                for prop in ("macaddress=" + wama).split(",")
            )
            if not mac:
                mac = props["macaddress"].replace("-", ":").upper()
            version = props.get("syVs")

        return DeviceInfo(os_type, version, build, model, mac)

    def _all_properties(self) -> Mapping[str, str]:
        properties: Dict[str, str] = {}
        properties.update(self._properties)
        for service in self.services:
            properties.update(service.properties)
        return properties

    def __eq__(self, other) -> bool:
        """Compare instance with another instance."""
        if isinstance(other, self.__class__):
            return self.identifier == other.identifier
        return False

    def __str__(self) -> str:
        """Return a string representation of this object."""
        device_info = self.device_info
        services = "\n".join([" - {0}".format(s) for s in self._services.values()])
        identifiers = "\n".join([" - {0}".format(x) for x in self.all_identifiers])
        return (
            f"       Name: {self.name}\n"
            f"   Model/SW: {device_info}\n"
            f"    Address: {self.address}\n"
            f"        MAC: {self.device_info.mac}\n"
            f" Deep Sleep: {self.deep_sleep}\n"
            f"Identifiers:\n"
            f"{identifiers}\n"
            f"Services:\n"
            f"{services}"
        )

Instance variables

var address -> ipaddress.IPv4Address

IP address of device.

Expand source code 
@property
def address(self) -> IPv4Address:
    """IP address of device."""
    return self._address
var all_identifiers -> List[str]

Return all unique identifiers for this device.

Expand source code 
@property
def all_identifiers(self) -> List[str]:
    """Return all unique identifiers for this device."""
    return [x.identifier for x in self.services if x.identifier is not None]
var deep_sleep -> bool

If device is in deep sleep.

Expand source code 
@property
def deep_sleep(self) -> bool:
    """If device is in deep sleep."""
    return self._deep_sleep
var device_info -> DeviceInfo

Return general device information.

Expand source code 
@property
def device_info(self) -> DeviceInfo:
    """Return general device information."""
    properties = self._all_properties()

    build: Optional[str] = properties.get("systembuildversion")
    version = properties.get("ov")
    if not version:
        version = properties.get("osvers", lookup_version(build))

    model_name: Optional[str] = properties.get("model", properties.get("am"))
    if model_name:
        model = lookup_model(model_name)
    else:
        model = self._model

    # MRP devices run tvOS (as far as we know now) as well as HomePods for
    # some reason
    if Protocol.MRP in self._services or model in [
        DeviceModel.HomePod,
        DeviceModel.HomePodMini,
    ]:
        os_type = OperatingSystem.TvOS
    elif Protocol.DMAP in self._services:
        os_type = OperatingSystem.Legacy
    elif model in [DeviceModel.AirPortExpress, DeviceModel.AirPortExpressGen2]:
        os_type = OperatingSystem.AirPortOS
    else:
        os_type = OperatingSystem.Unknown

    mac = properties.get("macaddress", properties.get("deviceid"))
    if mac:
        mac = mac.upper()

    # The waMA property comes from the _airport._tcp.local service, announced by
    # AirPort Expresses (used by the admin tool). It contains various information,
    # for instance MAC address and software version.
    wama = properties.get("wama")
    if wama:
        props: Mapping[str, str] = dict(
            cast(Tuple[str, str], prop.split("=", maxsplit=1))
            for prop in ("macaddress=" + wama).split(",")
        )
        if not mac:
            mac = props["macaddress"].replace("-", ":").upper()
        version = props.get("syVs")

    return DeviceInfo(os_type, version, build, model, mac)
var identifier -> Optional[str]

Return the main identifier associated with this device.

Expand source code 
@property
def identifier(self) -> Optional[str]:
    """Return the main identifier associated with this device."""
    for prot in [Protocol.MRP, Protocol.DMAP, Protocol.AirPlay, Protocol.RAOP]:
        service = self._services.get(prot)
        if service:
            return service.identifier
    return None
var name -> str

Name of device.

Expand source code 
@property
def name(self) -> str:
    """Name of device."""
    return self._name
var ready -> bool

Return if configuration is ready, i.e. has a main service.

Expand source code 
@property
def ready(self) -> bool:
    """Return if configuration is ready, i.e. has a main service."""
    ready_protocols = set(list(Protocol))

    # Companion has no unique identifier so it's the only protocol that can't be
    # used independently for now
    ready_protocols.remove(Protocol.Companion)

    intersection = ready_protocols.intersection(self._services.keys())
    return len(intersection) > 0
var services -> List[BaseService]

Return all supported services.

Expand source code 
@property
def services(self) -> List[BaseService]:
    """Return all supported services."""
    return list(self._services.values())

Methods

def add_service(self, service: BaseService) -> NoneType

Add a new service.

If the service already exists, it will be merged.

Expand source code 
def add_service(self, service: BaseService) -> None:
    """Add a new service.

    If the service already exists, it will be merged.
    """
    existing = self._services.get(service.protocol)
    if existing is not None:
        existing.merge(service)
    else:
        self._services[service.protocol] = service
def get_service(self, protocol: Protocol) -> Optional[BaseService]

Look up a service based on protocol.

If a service with the specified protocol is not available, None is returned.

Expand source code 
def get_service(self, protocol: Protocol) -> Optional[BaseService]:
    """Look up a service based on protocol.

    If a service with the specified protocol is not available, None is
    returned.
    """
    return self._services.get(protocol)
def main_service(self, protocol: Optional[Protocol] = None) -> BaseService

Return suggested service used to establish connection.

Expand source code 
def main_service(self, protocol: Optional[Protocol] = None) -> BaseService:
    """Return suggested service used to establish connection."""
    protocols = (
        [protocol]
        if protocol is not None
        else [Protocol.MRP, Protocol.DMAP, Protocol.AirPlay, Protocol.RAOP]
    )

    for prot in protocols:
        service = self._services.get(prot)
        if service is not None:
            return service

    raise exceptions.NoServiceError("no service to connect to")
def set_credentials(self, protocol: Protocol, credentials: str) -> bool

Set credentials for a protocol if it exists.

Expand source code 
def set_credentials(self, protocol: Protocol, credentials: str) -> bool:
    """Set credentials for a protocol if it exists."""
    service = self.get_service(protocol)
    if service:
        service.credentials = credentials
        return True
    return False
class CompanionService (port: int, credentials: Optional[str] = None, properties: Optional[Mapping[str, str]] = None)

Representation of a Companion link service.

Initialize a new CompaniomService.

Expand source code 
class CompanionService(BaseService):
    """Representation of a Companion link service."""

    def __init__(
        self,
        port: int,
        credentials: Optional[str] = None,
        properties: Optional[Mapping[str, str]] = None,
    ) -> None:
        """Initialize a new CompaniomService."""
        super().__init__(None, Protocol.Companion, port, properties)
        self.credentials = credentials

Ancestors

Inherited members

class DmapService (identifier: Optional[str], credentials: Optional[str], port: int = 3689, properties: Optional[Mapping[str, str]] = None)

Representation of a DMAP service.

Initialize a new DmapService.

Expand source code 
class DmapService(BaseService):
    """Representation of a DMAP service."""

    def __init__(
        self,
        identifier: Optional[str],
        credentials: Optional[str],
        port: int = 3689,
        properties: Optional[Mapping[str, str]] = None,
    ) -> None:
        """Initialize a new DmapService."""
        super().__init__(
            identifier,
            Protocol.DMAP,
            port,
            properties,
        )
        self.credentials = credentials

Ancestors

Inherited members

class MrpService (identifier: Optional[str], port: int, credentials: Optional[str] = None, properties: Optional[Mapping[str, str]] = None)

Representation of a MediaRemote Protocol (MRP) service.

Initialize a new MrpService.

Expand source code 
class MrpService(BaseService):
    """Representation of a MediaRemote Protocol (MRP) service."""

    def __init__(
        self,
        identifier: Optional[str],
        port: int,
        credentials: Optional[str] = None,
        properties: Optional[Mapping[str, str]] = None,
    ) -> None:
        """Initialize a new MrpService."""
        super().__init__(identifier, Protocol.MRP, port, properties)
        self.credentials = credentials

Ancestors

Inherited members

class RaopService (identifier: Optional[str], port: int = 7000, credentials: Optional[str] = None, properties: Optional[Mapping[str, str]] = None)

Representation of an RAOP service.

Initialize a new RaopService.

Expand source code 
class RaopService(BaseService):
    """Representation of an RAOP service."""

    def __init__(
        self,
        identifier: Optional[str],
        port: int = 7000,
        credentials: Optional[str] = None,
        properties: Optional[Mapping[str, str]] = None,
    ) -> None:
        """Initialize a new RaopService."""
        super().__init__(identifier, Protocol.RAOP, port, properties)
        self.credentials = credentials

Ancestors

Inherited members