Module pyatv

Main routines for interacting with an Apple TV.

Expand source code

"""Main routines for interacting with an Apple TV."""

import asyncio
import datetime  # noqa
from ipaddress import IPv4Address
from typing import List

import aiohttp

from pyatv import conf, exceptions, interface
from pyatv.airplay import AirPlayStreamAPI
from pyatv.const import Protocol
from pyatv.dmap import DmapAppleTV
from pyatv.dmap.pairing import DmapPairingHandler
from pyatv.mrp import MrpAppleTV
from pyatv.mrp.pairing import MrpPairingHandler
from pyatv.airplay.pairing import AirPlayPairingHandler
from pyatv.support import net
from pyatv.support.scan import BaseScanner, UnicastMdnsScanner, MulticastMdnsScanner


async def scan(
    loop: asyncio.AbstractEventLoop,
    timeout: int = 5,
    identifier: str = None,
    protocol: Protocol = None,
    hosts: List[str] = None,
) -> List[conf.AppleTV]:
    """Scan for Apple TVs on network and return their configurations."""

    def _should_include(atv):
        if not atv.ready:
            return False

        if identifier and identifier not in atv.all_identifiers:
            return False

        if protocol and atv.get_service(protocol) is None:
            return False

        return True

    scanner: BaseScanner
    if hosts:
        scanner = UnicastMdnsScanner([IPv4Address(host) for host in hosts], loop)
    else:
        scanner = MulticastMdnsScanner(loop)

    devices = (await scanner.discover(timeout)).values()
    return [device for device in devices if _should_include(device)]


async def connect(
    config: conf.AppleTV,
    loop: asyncio.AbstractEventLoop,
    protocol: Protocol = None,
    session: aiohttp.ClientSession = None,
) -> interface.AppleTV:
    """Connect to a device based on a configuration."""
    if config.identifier is None:
        raise exceptions.DeviceIdMissingError("no device identifier")

    service = config.main_service(protocol=protocol)

    implementation = {Protocol.DMAP: DmapAppleTV, Protocol.MRP: MrpAppleTV}.get(
        service.protocol
    )

    if not implementation:
        raise exceptions.UnsupportedProtocolError(str(service.protocol))

    # AirPlay stream API is the same for both DMAP and MRP
    airplay = AirPlayStreamAPI(config, loop)

    atv = implementation(loop, await net.create_session(session), config, airplay)
    await atv.connect()
    return atv


async def pair(
    config: conf.AppleTV,
    protocol: Protocol,
    loop: asyncio.AbstractEventLoop,
    session: aiohttp.ClientSession = None,
    **kwargs
):
    """Pair a protocol for an Apple TV."""
    service = config.get_service(protocol)
    if not service:
        raise exceptions.NoServiceError(
            "no service available for protocol " + str(protocol)
        )

    handler = {
        Protocol.DMAP: DmapPairingHandler,
        Protocol.MRP: MrpPairingHandler,
        Protocol.AirPlay: AirPlayPairingHandler,
    }.get(protocol)

    if handler is None:
        raise exceptions.UnsupportedProtocolError(str(protocol))

    return handler(config, await net.create_session(session), loop, **kwargs)

Sub-modules

pyatv.conf

Configuration used when connecting to a device …

pyatv.const

Constants used in the public API.

pyatv.convert

Various types of extraction and conversion functions.

pyatv.exceptions

Local exceptions used by library.

pyatv.helpers

Various helper methods.

pyatv.interface

Public interface exposed by library …

Functions

async def connect(config: AppleTV, loop: asyncio.events.AbstractEventLoop, protocol: Protocol = None, session: aiohttp.client.ClientSession = None) -> AppleTV

Connect to a device based on a configuration.

Expand source code

async def connect(
    config: conf.AppleTV,
    loop: asyncio.AbstractEventLoop,
    protocol: Protocol = None,
    session: aiohttp.ClientSession = None,
) -> interface.AppleTV:
    """Connect to a device based on a configuration."""
    if config.identifier is None:
        raise exceptions.DeviceIdMissingError("no device identifier")

    service = config.main_service(protocol=protocol)

    implementation = {Protocol.DMAP: DmapAppleTV, Protocol.MRP: MrpAppleTV}.get(
        service.protocol
    )

    if not implementation:
        raise exceptions.UnsupportedProtocolError(str(service.protocol))

    # AirPlay stream API is the same for both DMAP and MRP
    airplay = AirPlayStreamAPI(config, loop)

    atv = implementation(loop, await net.create_session(session), config, airplay)
    await atv.connect()
    return atv

async def pair(config: AppleTV, protocol: Protocol, loop: asyncio.events.AbstractEventLoop, session: aiohttp.client.ClientSession = None, **kwargs)

Pair a protocol for an Apple TV.

Expand source code

async def pair(
    config: conf.AppleTV,
    protocol: Protocol,
    loop: asyncio.AbstractEventLoop,
    session: aiohttp.ClientSession = None,
    **kwargs
):
    """Pair a protocol for an Apple TV."""
    service = config.get_service(protocol)
    if not service:
        raise exceptions.NoServiceError(
            "no service available for protocol " + str(protocol)
        )

    handler = {
        Protocol.DMAP: DmapPairingHandler,
        Protocol.MRP: MrpPairingHandler,
        Protocol.AirPlay: AirPlayPairingHandler,
    }.get(protocol)

    if handler is None:
        raise exceptions.UnsupportedProtocolError(str(protocol))

    return handler(config, await net.create_session(session), loop, **kwargs)

async def scan(loop: asyncio.events.AbstractEventLoop, timeout: int = 5, identifier: str = None, protocol: Protocol = None, hosts: List[str] = None) -> List[AppleTV]

Scan for Apple TVs on network and return their configurations.

Expand source code

async def scan(
    loop: asyncio.AbstractEventLoop,
    timeout: int = 5,
    identifier: str = None,
    protocol: Protocol = None,
    hosts: List[str] = None,
) -> List[conf.AppleTV]:
    """Scan for Apple TVs on network and return their configurations."""

    def _should_include(atv):
        if not atv.ready:
            return False

        if identifier and identifier not in atv.all_identifiers:
            return False

        if protocol and atv.get_service(protocol) is None:
            return False

        return True

    scanner: BaseScanner
    if hosts:
        scanner = UnicastMdnsScanner([IPv4Address(host) for host in hosts], loop)
    else:
        scanner = MulticastMdnsScanner(loop)

    devices = (await scanner.discover(timeout)).values()
    return [device for device in devices if _should_include(device)]