<commit_before>#!/usr/bin/env python3
"""Set up a fake device.

Currently only supports MRP.
"""
import sys
import os
import socket
import asyncio
import logging
import argparse

from aiozeroconf import Zeroconf, ServiceInfo

sys.path.insert(0, os.path.dirname(os.path.realpath(__file__)) + "/..")  # noqa

from tests.mrp.fake_mrp_atv import FakeAppleTV, FakeDeviceState

_LOGGER = logging.getLogger(__name__)

DEVICE_NAME = "FakeMRP"
AIRPLAY_IDENTIFIER = "4D797FD3-3538-427E-A47B-A32FC6CF3A6A"

SERVER_IDENTIFIER = "6D797FD3-3538-427E-A47B-A32FC6CF3A69"


async def publish_zeroconf(zconf, ip_address, port):
    """Publish zeroconf service for ATV proxy instance."""
    props = {
        b"ModelName": "Apple TV",
        b"AllowPairing": b"YES",
        b"macAddress": b"40:cb:c0:12:34:56",
        b"BluetoothAddress": False,
        b"Name": DEVICE_NAME.encode(),
        b"UniqueIdentifier": SERVER_IDENTIFIER.encode(),
        b"SystemBuildVersion": b"17K499",
        b"LocalAirPlayReceiverPairingIdentity": AIRPLAY_IDENTIFIER.encode(),
    }

    service = ServiceInfo(
        "_mediaremotetv._tcp.local.",
        DEVICE_NAME + "._mediaremotetv._tcp.local.",
        address=socket.inet_aton(ip_address),
        port=port,
        weight=0,
        priority=0,
        properties=props,
    )
    await zconf.register_service(service)
    _LOGGER.debug("Published zeroconf service: %s", service)

    return service


async def appstart(loop):
    """Script starts here."""
    parser = argparse.ArgumentParser()
    parser.add_argument("--local-ip", default="127.0.0.1", help="local IP address")
    parser.add_argument(
        "-d", "--debug", default=False, action="store_true", help="enable debug logs"
    )
    args = parser.parse_args()

    level = logging.DEBUG if args.debug else logging.WARNING
    logging.basicConfig(level=level, stream=sys.stdout)

    state = FakeDeviceState()
    zconf = Zeroconf(loop)
    server = await loop.create_server(lambda: FakeAppleTV(loop, state=state), "0.0.0.0")
    port = server.sockets[0].getsockname()[1]
    _LOGGER.info("Started fake MRP device at port %d", port)

    service = await publish_zeroconf(zconf, args.local_ip, port)

    print("Press ENTER to quit")
    await loop.run_in_executor(None, sys.stdin.readline)

    await zconf.unregister_service(service)
<commit_msg>fake_device: Add demo mode in fake_device<commit_after>#!/usr/bin/env python3
"""Set up a fake device.

Currently only supports MRP.
"""
import os
import sys
import socket
import asyncio
import logging
import argparse

from aiozeroconf import Zeroconf, ServiceInfo

from pyatv.const import Protocol

sys.path.insert(0, os.path.dirname(os.path.realpath(__file__)) + "/..")  # noqa

from tests.fake_device import FakeAppleTV

_LOGGER = logging.getLogger(__name__)

DEVICE_NAME = "FakeMRP"
AIRPLAY_IDENTIFIER = "4D797FD3-3538-427E-A47B-A32FC6CF3A6A"

SERVER_IDENTIFIER = "6D797FD3-3538-427E-A47B-A32FC6CF3A69"


async def _alter_playing(usecase):
    while True:
        try:
            logging.debug("Starting new output lap")
            usecase.example_video()
            await asyncio.sleep(3)
            usecase.example_music()
            await asyncio.sleep(3)
            usecase.nothing_playing()
            await asyncio.sleep(3)
        except asyncio.CancelledError:
            break
        except Exception:
            logging.exception("Exception in output loop")


async def publish_zeroconf(zconf, ip_address, port):
    """Publish zeroconf service for ATV proxy instance."""
    props = {
        b"ModelName": "Apple TV",
        b"AllowPairing": b"YES",
        b"macAddress": b"40:cb:c0:12:34:56",
        b"BluetoothAddress": False,
        b"Name": DEVICE_NAME.encode(),
        b"UniqueIdentifier": SERVER_IDENTIFIER.encode(),
        b"SystemBuildVersion": b"17K499",
        b"LocalAirPlayReceiverPairingIdentity": AIRPLAY_IDENTIFIER.encode(),
    }

    service = ServiceInfo(
        "_mediaremotetv._tcp.local.",
        DEVICE_NAME + "._mediaremotetv._tcp.local.",
        address=socket.inet_aton(ip_address),
        port=port,
        weight=0,
        priority=0,
        properties=props,
    )
    await zconf.register_service(service)
    _LOGGER.debug("Published zeroconf service: %s", service)

    return service


async def appstart(loop):
    """Script starts here."""
    parser = argparse.ArgumentParser()
    parser.add_argument("--local-ip", default="127.0.0.1", help="local IP address")
    parser.add_argument(
        "--demo", default=False, action="store_true", help="enable demo mode"
    )
    parser.add_argument(
        "-d", "--debug", default=False, action="store_true", help="enable debug logs"
    )
    args = parser.parse_args()

    level = logging.DEBUG if args.debug else logging.WARNING
    logging.basicConfig(
        level=level,
        stream=sys.stdout,
        datefmt="%Y-%m-%d %H:%M:%S",
        format="%(asctime)s %(levelname)s: %(message)s",
    )

    tasks = []
    zconf = Zeroconf(loop)
    fake_atv = FakeAppleTV(loop)
    state, usecase = fake_atv.add_service(Protocol.MRP)
    await fake_atv.start()

    service = await publish_zeroconf(
        zconf, args.local_ip, fake_atv.get_port(Protocol.MRP)
    )

    if args.demo:
        tasks.append(asyncio.ensure_future(_alter_playing(usecase)))

    print("Press ENTER to quit")
    await loop.run_in_executor(None, sys.stdin.readline)

    for task in tasks:
        task.cancel()

    await zconf.unregister_service(service)

    print("Exiting")

{'fraction_non_alphanumeric': 0.12240101247514013, 'fraction_numerical': 0.024407882842162356, 'mean_word_length': 3.2553846153846155, 'pattern_counts': {'":': 16, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 3, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1840378', 'n_tokens_mistral': 1925, 'n_tokens_neox': 1848, 'n_words': 365}