apple-tv.py

"""Apple TV node for Node-BLUE.

Connects to a configured Apple TV via pyatv, reports power, play state and
content as node outputs and accepts power/remote commands on its inputs.

Device selection and pairing (including PIN entry) happen in the node's edit
dialog in the frontend. The frontend talks to this process through node data:
it writes a command to "uiCommand" and reads the answer from "uiState".
"""

from homegear import Homegear

import asyncio
import faulthandler
import json
import signal
import sys
import threading
import time
import traceback


# Node data keys. "uiCommand" is written by the frontend, "uiState" by us.
STORAGE_KEY = "pyatvStorage"
ADDRESS_KEY = "lastAddress"
CONFIG_KEY = "config"

# Last resort for the configuration: Node-BLUE keeps the deployed flows here.
MAIN_CONF = "/etc/homegear/main.conf"
DEFAULT_FLOWS_PATH = "/var/lib/homegear/node-blue/data/"
COMMAND_KEY = "uiCommand"
STATE_KEY = "uiState"

# Log levels of Homegear's nodeLog.
LOG_ERROR = 2
LOG_WARN = 3
LOG_INFO = 4

RECONNECT_MIN = 5
RECONNECT_MAX = 60

# An mDNS scan always waits its full window -- it does not stop early once the
# device answered -- so these are collection windows, not timeouts. Probing a
# known address returns immediately and ignores them.
# The dialog search is the one a user waits for and must not miss a device;
# the runtime fallback only runs when a known address stopped answering, and a
# missed device there simply means another retry.
SCAN_TIMEOUT_UI = 15
SCAN_TIMEOUT_RUNTIME = 10
COMMAND_POLL_INTERVAL = 1.0

# Watchdog. A session can go silent while its TCP connections stay up and keep
# carrying traffic -- observed after ~2.8 days, with a real power change
# producing no callback at all. Neither pyatv nor the socket notices, so the
# node has to check for itself.
WATCHDOG_INTERVAL = 60
# Silence up to this point is normal: an idle Apple TV sends nothing.
WATCHDOG_PROBE_AFTER = 300
WATCHDOG_PROBE_TIMEOUT = 10
# Silence beyond this is not trusted, even if the device answers probes -- the
# push channel can be dead while request/response still works.
WATCHDOG_MAX_SILENCE = 900

# Power state is polled in addition to the push updates. A switch-off was
# observed to arrive only as a play status change, never as a power event, and
# a rebuilt session reported Unknown -- so the node kept reporting "on" while
# the device slept. Companion can be asked for the attention state directly,
# which is a real request rather than a cached push value.
POWER_POLL_INTERVAL = 30
POWER_POLL_TIMEOUT = 5
PIN_TIMEOUT = 180

# How long to wait for Node-BLUE's startUpComplete before falling back to the
# configuration on disk. Short, because the fallback carries the same values.
STARTUP_TIMEOUT = 10

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

nodeId = sys.argv[2]
nodeInfo = None
hg = None
node = None

startUpComplete = threading.Condition()

# Diagnostics for the startup handshake: if startUpComplete never arrives we
# want to know whether any event reached this process at all.
receivedEvents = 0
nodeBlueEvents = []


def log(level, message):
	"""Write to Homegear's log and to stderr, which Node-BLUE also captures."""
	print(message, file=sys.stderr, flush=True)
	try:
		hg.nodeLog(level, message)
	except Exception:
		pass


# This callback method is called on Homegear variable changes.
def eventHandler(eventSource, peerId, channel, variableName, value):
	# Note that the event handler is called by a different thread than the main thread. I. e. thread synchronization is
	# needed when you access non local variables.
	# See https://effbot.org/zone/thread-synchronization.htm

	global receivedEvents
	receivedEvents += 1
	if eventSource == "nodeBlue":
		nodeBlueEvents.append(str(variableName))

	# When the flows are fully started, "startUpComplete" is set to "true". Wait for this event.
	if eventSource == "nodeBlue" and peerId == 0 and variableName == "startUpComplete":
		startUpComplete.acquire()
		global nodeInfo
		nodeInfo = value
		startUpComplete.notify()
		startUpComplete.release()


# This callback method is called when a message arrives on one of the node's inputs.
def nodeInput(info, inputIndex, message):
	# A message carries the node info too, so a node that missed startUpComplete
	# can still pick up its configuration from here.
	global nodeInfo
	if nodeInfo is None and info:
		nodeInfo = info

	# Note that the event handler is called by a different thread than the main thread. I. e. thread synchronization is
	# needed when you access non local variables.
	if node is None:
		log(LOG_WARN, "Input ignored: node is still starting")
		return

	future = asyncio.run_coroutine_threadsafe(
		nodeInputThreadSafe(info, inputIndex, message), loop
	)
	try:
		# Bounded so a hanging device command cannot block Homegear's caller
		# thread indefinitely.
		future.result(30)
	except Exception as ex:
		log(LOG_ERROR, "Input handler failed: " + str(ex))


# Connect to Homegear before anything else is imported.
#
# The wrapper pings this process until it answers and pushes
# startUpComplete as soon as it does. Everything imported before that
# point widens the window in which the wrapper considers the process
# started while it is not yet reachable -- and an event sent into that
# window is dropped without a trace on either side.
# hg waits until the connection is established (but for a maximum of 2 seconds).
# The socket path is passed in sys.argv[1], the node's ID in sys.argv[2]
# Lets a stuck process be inspected with "kill -USR1 <pid>" -- the stacks of
# all threads then land in the node's log.
faulthandler.register(signal.SIGUSR1, file=sys.stderr, all_threads=True)

hg = Homegear(sys.argv[1], eventHandler, sys.argv[2], nodeInput)

import pyatv
from pyatv.const import (
	FeatureName,
	FeatureState,
	PowerState,
	Protocol,
	RepeatState,
	ShuffleState,
)
from pyatv.exceptions import NotSupportedError
from pyatv.interface import DeviceListener, Playing, PowerListener, PushListener
from pyatv.storage import AbstractStorage, StorageModel

# Protocols that need to be paired for power and play state to work. On
# tvOS 15+ MRP -- which carries the power state -- is tunneled through AirPlay,
# so AirPlay has to be paired even though MRP is the protocol we rely on.
PAIRING_PROTOCOLS = [Protocol.AirPlay, Protocol.Companion]


class NodeDataStorage(AbstractStorage):
	"""pyatv storage that keeps credentials in Homegear's node data.

	Credentials survive a redeploy this way, so pairing is a one-time action
	instead of something the user has to repeat after every flow change.
	"""

	def __init__(self, hg, loop):
		super().__init__()
		self._hg = hg
		self._loop = loop

	async def save(self):
		dumped = dict(self)
		if not self.has_changed(dumped):
			return
		# Stored as a JSON string so the nested model survives Homegear's
		# struct conversion unchanged.
		await self._loop.run_in_executor(
			None, self._hg.setNodeData, STORAGE_KEY, json.dumps(dumped)
		)
		self.update_hash(dumped)

	async def load(self):
		raw = await self._loop.run_in_executor(None, self._hg.getNodeData, STORAGE_KEY)
		if not raw:
			return
		data = json.loads(raw)
		self.storage_model = StorageModel.model_validate(data)
		self.update_hash(data)

	def __str__(self):
		return "NodeDataStorage:" + nodeId


class Listeners(PowerListener, DeviceListener, PushListener):
	"""All pyatv callbacks in one object.

	pyatv keeps listeners as weak references only, so this instance has to be
	held for as long as the connection lives -- see AppleTvNode.session().
	"""

	def __init__(self, node):
		self.node = node
		self.disconnected = asyncio.Event()

	# -- PowerListener ---------------------------------------------------
	def powerstate_update(self, old_state, new_state):
		self.node.note_event()
		self.node.report_power(new_state)

	# -- DeviceListener --------------------------------------------------
	def connection_lost(self, exception):
		self.node.note_event()
		log(LOG_WARN, "Apple TV connection lost: " + str(exception))
		self.disconnected.set()

	def connection_closed(self):
		self.node.note_event()
		if not self.node.maintenance:
			log(LOG_INFO, "Apple TV connection closed")
		self.disconnected.set()

	# -- PushListener ----------------------------------------------------
	def playstatus_update(self, updater, playstatus: Playing):
		self.node.note_event()
		self.node.report_playing(playstatus)

	def playstatus_error(self, updater, exception: Exception):
		self.node.note_event()
		log(LOG_WARN, "Play status error: " + str(exception))


class AppleTvNode:
	def __init__(self, hg, info):
		self.hg = hg
		self.atv = None
		self.storage = None
		self.listeners = None
		# Configuration of the device we are currently connected to.
		self.current_conf = None

		self.config_source = "startUpComplete"
		self.apply_config(info)

		# Last values sent on each output, so we only emit real changes.
		self.last_connected = None
		self.last_power = None
		self.last_play_state = None
		self.last_status = None

		# Last address the device actually answered on.
		self.last_address = ""

		# Monotonic timestamp of the last callback pyatv delivered.
		self.last_event = 0.0
		# True while the watchdog rebuilds the session on purpose. Such a
		# reconnect must not look like an outage to the flow.
		self.maintenance = False

		# Pairing state.
		self.pairing_active = False
		self.pin_event = asyncio.Event()
		self.pin_value = None
		self.last_command_seq = None
		# Long running UI command (scan/pair) running in the background.
		self.command_task = None

	def apply_config(self, info):
		info = info or {}
		self.identifier = str(info.get("deviceId", "") or "").strip()
		self.host = str(info.get("host", "") or "").strip()
		# Address the edit dialog saw last. Used only as a fallback probe when
		# the identifier cannot be resolved via mDNS.
		self.device_address = str(info.get("deviceAddress", "") or "").strip()

	@staticmethod
	def flows_file():
		"""Path of the deployed flows, as configured in Homegear's main.conf."""
		try:
			with open(MAIN_CONF, "r") as handle:
				for line in handle:
					line = line.strip()
					if line.startswith("#") or "=" not in line:
						continue
					key, value = line.split("=", 1)
					if key.strip().lower() == "nodebluedatapath":
						return value.strip().rstrip("/") + "/flows.json"
		except Exception:
			pass
		return DEFAULT_FLOWS_PATH + "flows.json"

	def read_config_from_flows(self):
		"""Pick this node's configuration out of the deployed flows.

		Only reached when neither startUpComplete nor the dialog's copy in node
		data provided anything -- but then it is the difference between a node
		that works after a restart and one that needs a manual save first.
		"""
		path = self.flows_file()
		with open(path, "r") as handle:
			flows = json.load(handle)
		for entry in flows:
			if isinstance(entry, dict) and entry.get("id") == nodeId:
				return entry
		return None

	async def load_fallback_config(self):
		"""Take the configuration from node data if startUpComplete never came.

		A Python node normally learns its configuration only from that event.
		When it does not arrive the node would sit there unconfigured, so the
		edit dialog mirrors the same values into node data when it is saved.
		"""
		if self.identifier or self.host:
			return
		try:
			raw = await loop.run_in_executor(None, self.hg.getNodeData, CONFIG_KEY)
		except Exception as ex:
			log(LOG_WARN, "Could not read fallback configuration: " + str(ex))
			return
		if raw:
			self.apply_config(json.loads(raw))
			if self.identifier or self.host:
				self.config_source = "node data"
				log(LOG_INFO, "Configuration taken from node data: "
					+ (self.identifier or self.host))
				return

		try:
			entry = await loop.run_in_executor(None, self.read_config_from_flows)
		except Exception as ex:
			log(LOG_WARN, "Could not read flows file: " + str(ex))
			return
		if entry:
			self.apply_config(entry)
			if self.identifier or self.host:
				self.config_source = "flows file"
				log(LOG_INFO, "Configuration taken from the flows file: "
					+ (self.identifier or self.host))

	# -- outputs ---------------------------------------------------------

	def output(self, index, payload):
		try:
			self.hg.nodeOutput(index, {"payload": payload})
		except Exception as ex:
			log(LOG_ERROR, "nodeOutput failed: " + str(ex))

	def status(self, text, fill="grey", shape="dot"):
		# Only push on change -- the connection loop calls this every second
		# and Node-BLUE would otherwise get a status event per tick.
		if (text, fill, shape) == self.last_status:
			return
		self.last_status = (text, fill, shape)
		try:
			self.hg.nodeEvent(
				"statusBottom/" + nodeId,
				{"text": text, "fill": fill, "shape": shape},
				True,
			)
		except Exception as ex:
			log(LOG_ERROR, "nodeEvent failed: " + str(ex))

	async def fetch_power_state(self, atv):
		"""Ask the device for its power state instead of trusting the cache.

		Uses Companion's attention state when available. That goes through
		pyatv internals, so every step falls back: first to the per-protocol
		cached states, then to the facade value.
		"""
		interfaces = {}
		try:
			interfaces = dict(atv.power._interfaces)
		except Exception:
			pass

		companion = interfaces.get(Protocol.Companion)
		if companion is not None:
			try:
				status = await asyncio.wait_for(
					companion.api.fetch_attention_state(), timeout=POWER_POLL_TIMEOUT
				)
				state = companion._system_status_to_power_state(status)
				if state in (PowerState.On, PowerState.Off):
					return state
			except asyncio.CancelledError:
				raise
			except Exception:
				pass

		# The facade prefers Companion even when Companion only knows Unknown,
		# while MRP may well know the answer -- so look at each protocol.
		for candidate in [atv.power] + list(interfaces.values()):
			try:
				state = candidate.power_state
			except Exception:
				continue
			if state in (PowerState.On, PowerState.Off):
				return state
		return PowerState.Unknown

	async def power_poll(self, atv, listeners):
		"""Keep the reported power state in line with the device."""
		while not listeners.disconnected.is_set():
			try:
				state = await self.fetch_power_state(atv)
			except asyncio.CancelledError:
				raise
			except Exception as ex:
				log(LOG_WARN, "Power poll failed: " + str(ex))
				state = PowerState.Unknown

			expected = {PowerState.On: True, PowerState.Off: False}.get(state)
			if expected is not None and expected != self.last_power:
				if self.last_power is not None:
					log(LOG_WARN, "Power state " + state.name
						+ " was not delivered as an event. Reporting it from the poll.")
				self.report_power(state)

			await asyncio.sleep(POWER_POLL_INTERVAL)

	def note_event(self):
		self.last_event = time.monotonic()

	async def watchdog(self, atv, listeners):
		"""Detect a session that stopped delivering events.

		A connection can stay established and keep exchanging bytes while pyatv
		delivers nothing any more -- power changes then go unnoticed for days.
		Nothing below us reports that, so check here and force a reconnect,
		which re-reads the real state and emits it.
		"""
		while not listeners.disconnected.is_set():
			await asyncio.sleep(WATCHDOG_INTERVAL)
			if listeners.disconnected.is_set():
				return

			silence = int(time.monotonic() - self.last_event)
			if silence < WATCHDOG_PROBE_AFTER:
				continue

			try:
				await asyncio.wait_for(
					atv.metadata.playing(), timeout=WATCHDOG_PROBE_TIMEOUT
				)
			except asyncio.CancelledError:
				raise
			except Exception as ex:
				log(LOG_WARN, "Watchdog: no answer after " + str(silence)
					+ "s without events (" + str(ex) + "). Reconnecting.")
				self.maintenance = True
				listeners.disconnected.set()
				return

			if silence >= WATCHDOG_MAX_SILENCE:
				log(LOG_INFO, "Watchdog: " + str(silence)
					+ "s without any event although the device answers. Rebuilding the session.")
				self.maintenance = True
				listeners.disconnected.set()
				return

	def report_connected(self, connected):
		if not connected and self.maintenance:
			# The watchdog is rebuilding the session deliberately. Announcing a
			# disconnect would flap INIT, and forgetting the last values would
			# re-emit them unchanged -- the fresh session reports only what
			# really differs.
			return
		if connected == self.last_connected:
			return
		self.last_connected = connected
		self.output(0, connected)
		if not connected:
			# The power state is unknown while disconnected. Do not guess --
			# forget it so the true value is emitted again on reconnect.
			self.last_power = None
			self.last_play_state = None

	def report_power(self, state):
		# PowerState.Unknown carries no information; emitting False for it
		# would fake an "off" event. Only On and Off are real transitions.
		if state == PowerState.On:
			value = True
		elif state == PowerState.Off:
			value = False
		else:
			# Unknown carries no information. The power poll resolves it.
			return

		if value == self.last_power:
			return
		self.last_power = value
		self.output(1, value)
		self.status("on" if value else "off", "green" if value else "grey")

	def report_playing(self, playstatus: Playing):
		state = playstatus.device_state.name.lower()
		if state != self.last_play_state:
			self.last_play_state = state
			self.output(2, state)

		app = None
		try:
			app = self.atv.metadata.app if self.atv else None
		except Exception:
			app = None

		# Everything here has to be a plain type -- Homegear cannot serialize
		# pyatv objects such as App, which is why the app is split into two
		# strings instead of being passed through.
		self.output(3, {
			"state": state,
			"album": playstatus.album,
			"app": app.name if app else None,
			"app_id": app.identifier if app else None,
			"artist": playstatus.artist,
			"genre": playstatus.genre,
			"hash": playstatus.hash,
			"media_type": playstatus.media_type.name.lower(),
			"position": playstatus.position,
			"repeat": playstatus.repeat.name.lower(),
			"series_name": playstatus.series_name,
			"season_number": playstatus.season_number,
			"episode_number": playstatus.episode_number,
			"shuffle": playstatus.shuffle.name.lower(),
			"title": playstatus.title,
			"total_time": playstatus.total_time,
		})

	# -- device lookup ---------------------------------------------------

	async def scan(self, identifier=None, host=None, timeout=SCAN_TIMEOUT_RUNTIME):
		hosts = [host] if host else None
		return await pyatv.scan(
			loop,
			timeout=timeout,
			identifier=identifier,
			hosts=hosts,
			storage=self.storage,
		)

	async def resolve_device(self, identifier, host):
		"""Find one device by IP address or by identifier -- either works alone.

		The IP address wins when both are given. Addressing by identifier
		depends on mDNS, and a sleeping Apple TV answers so slowly that it is
		regularly missing from a scan; addressing by IP queries the device
		directly and does not have that problem.
		"""
		identifier = (identifier or "").strip()
		host = (host or "").strip()

		if host:
			confs = await self.scan(host=host)
			if not confs:
				return None
			# The host was asked directly, so whatever answered is the device.
			# An identifier configured alongside it is not used for filtering,
			# otherwise a renewed identifier would break a working IP.
			return confs[0]

		if not identifier:
			return None

		# Known addresses first: probing one is a directed query that answers in
		# about a second, while an mDNS scan costs the full scan timeout and
		# regularly misses a sleeping Apple TV altogether. This is also what
		# makes reconnecting fast enough to catch a standby transition.
		conf = None
		for address in self.candidate_addresses():
			conf = self.match_identifier(await self.scan(host=address), identifier)
			if conf is not None:
				break

		if conf is None:
			conf = self.match_identifier(await self.scan(identifier=identifier), identifier)

		if conf is None:
			# A device has several identifiers (MAC, UUID, MAC without colons)
			# and the filtered scan does not always match the configured form.
			conf = self.match_identifier(await self.scan(), identifier)

		if conf is not None:
			await self.remember_address(identifier, str(conf.address))
		return conf

	def candidate_addresses(self):
		"""Known addresses to probe, freshest first."""
		addresses = []
		for address in (self.last_address, self.device_address):
			if address and address not in addresses:
				addresses.append(address)
		return addresses

	async def remember_address(self, identifier, address):
		if not address or address == self.last_address:
			return
		self.last_address = address
		try:
			await loop.run_in_executor(
				None, self.hg.setNodeData, ADDRESS_KEY,
				json.dumps({"identifier": identifier, "address": address}),
			)
		except Exception as ex:
			log(LOG_WARN, "Could not store last address: " + str(ex))

	async def load_last_address(self):
		try:
			raw = await loop.run_in_executor(None, self.hg.getNodeData, ADDRESS_KEY)
			if raw:
				data = json.loads(raw)
				if data.get("identifier", "").lower() == self.identifier.lower():
					self.last_address = data.get("address") or ""
		except Exception as ex:
			log(LOG_WARN, "Could not read last address: " + str(ex))

	@staticmethod
	def match_identifier(confs, identifier):
		wanted = identifier.lower()
		for conf in confs or []:
			for candidate in conf.all_identifiers:
				if candidate and candidate.lower() == wanted:
					return conf
		return None

	async def find_configured_device(self):
		"""Look up exactly the configured device, never just the first hit.

		A plain scan also returns HomePods and other AirPlay receivers, and
		picking the first one connects the node to an arbitrary device.
		"""
		return await self.resolve_device(self.identifier, self.host)

	# -- connection ------------------------------------------------------

	async def session(self):
		"""Hold one connection. Returns True if it was established."""
		conf = await self.find_configured_device()
		if conf is None:
			self.status("device not found", "red", "ring")
			return False

		atv = await pyatv.connect(conf, loop, storage=self.storage)
		self.atv = atv
		self.current_conf = conf
		try:
			if atv.features.get_feature(FeatureName.PowerState).state != FeatureState.Available:
				raise NotSupportedError(
					conf.name + " does not report a power state. Pair the device "
					"in the node's edit dialog (AirPlay and Companion)."
				)

			listeners = Listeners(self)
			self.listeners = listeners
			atv.listener = listeners
			atv.power.listener = listeners
			atv.push_updater.listener = listeners
			atv.push_updater.start()

			if not self.maintenance:
				log(LOG_INFO, "Connected to " + conf.name + " (" + str(conf.address) + ")")
			self.maintenance = False
			self.note_event()
			self.report_connected(True)

			# The poll runs its first check right away, so the state reported at
			# connect is a fetched one -- a cached Unknown used to be dropped and
			# never looked at again.
			helpers = [
				asyncio.ensure_future(self.power_poll(atv, listeners)),
				asyncio.ensure_future(self.watchdog(atv, listeners)),
			]
			try:
				await listeners.disconnected.wait()
			finally:
				for helper in helpers:
					helper.cancel()
				for helper in helpers:
					try:
						await helper
					except BaseException:
						pass
		finally:
			self.report_connected(False)
			self.listeners = None
			self.atv = None
			self.current_conf = None
			# close() returns cleanup tasks. A hanging peer must not block the
			# reconnect, but the tasks still have to be harvested: cancelling a
			# gather() leaves them running and Python later reports their
			# exceptions as "Task exception was never retrieved".
			pending = atv.close()
			if pending:
				done, running = await asyncio.wait(pending, timeout=5)
				for task in done:
					if not task.cancelled():
						task.exception()
				for task in running:
					task.add_done_callback(lambda t: t.cancelled() or t.exception())
		return True

	async def connection_loop(self):
		delay = RECONNECT_MIN
		while self.hg.connected():
			if not self.identifier and not self.host:
				self.status("not configured", "grey", "ring")
				await asyncio.sleep(COMMAND_POLL_INTERVAL)
				continue
			if self.pairing_active:
				await asyncio.sleep(COMMAND_POLL_INTERVAL)
				continue
			try:
				connected = await self.session()
				if not connected:
					self.maintenance = False
					self.report_connected(False)
				delay = RECONNECT_MIN if connected else min(delay * 2, RECONNECT_MAX)
			except asyncio.CancelledError:
				raise
			except NotSupportedError as ex:
				# Missing pairing: retrying does not help until the user pairs,
				# but the process must stay alive to serve the pairing UI.
				log(LOG_ERROR, str(ex))
				self.status("not paired", "red", "ring")
				delay = RECONNECT_MAX
			except Exception as ex:
				log(LOG_ERROR, "Connection failed: " + str(ex))
				# The rebuild did not work out, so this is a real outage after all.
				self.maintenance = False
				self.report_connected(False)
				self.status("connection failed", "red", "ring")
				delay = min(delay * 2, RECONNECT_MAX)
			await asyncio.sleep(delay)

	# -- commands from the node's inputs ---------------------------------

	async def power(self, on):
		if self.atv is None:
			log(LOG_WARN, "Power command ignored: not connected")
			return
		if on:
			await self.atv.power.turn_on()
		else:
			await self.atv.power.turn_off()

	async def remote(self, message):
		if self.atv is None:
			log(LOG_WARN, "Remote command ignored: not connected")
			return

		command = message["payload"]
		rc = self.atv.remote_control

		if command == "set_position":
			if "position" in message:
				await rc.set_position(message["position"])
			return
		if command == "set_repeat":
			states = {
				"all": RepeatState.All,
				"off": RepeatState.Off,
				"track": RepeatState.Track,
			}
			if message.get("state") in states:
				await rc.set_repeat(states[message["state"]])
			return
		if command == "set_shuffle":
			states = {
				"albums": ShuffleState.Albums,
				"off": ShuffleState.Off,
				"songs": ShuffleState.Songs,
			}
			if message.get("state") in states:
				await rc.set_shuffle(states[message["state"]])
			return
		if command in ("volume_up", "volume_down"):
			# Volume lives on the audio interface; remote_control still has it
			# but only as a deprecated alias.
			await getattr(self.atv.audio, command)()
			return

		method = getattr(rc, command, None)
		if method is None or command.startswith("_"):
			log(LOG_WARN, "Unknown remote command: " + str(command))
			return
		await method()

	# -- frontend channel ------------------------------------------------

	def write_state(self, **fields):
		try:
			self.hg.setNodeData(STATE_KEY, json.dumps(fields))
		except Exception as ex:
			log(LOG_ERROR, "Writing UI state failed: " + str(ex))

	async def load_last_command_seq(self):
		"""Remember which command was already answered before this restart.

		Without this the last command left in node data would be executed again
		on every redeploy -- which would restart a pairing nobody asked for.
		"""
		try:
			raw = await loop.run_in_executor(None, self.hg.getNodeData, STATE_KEY)
			if raw:
				self.last_command_seq = json.loads(raw).get("seq")
		except Exception as ex:
			log(LOG_WARN, "Could not read last UI state: " + str(ex))

	async def command_loop(self):
		"""Poll for commands the edit dialog leaves in node data.

		Node data is the only channel the frontend and a Python node share:
		Node-BLUE forwards neither setNodeVariable nor custom HTTP endpoints to
		Python nodes, so the dialog writes here and we pick it up.
		"""
		while self.hg.connected():
			try:
				raw = await loop.run_in_executor(None, self.hg.getNodeData, COMMAND_KEY)
				if raw:
					command = json.loads(raw)
					if command.get("seq") != self.last_command_seq:
						self.last_command_seq = command.get("seq")
						await self.dispatch(command)
			except asyncio.CancelledError:
				raise
			except Exception as ex:
				log(LOG_ERROR, "UI command failed: " + traceback.format_exc())
				self.write_state(seq=self.last_command_seq, busy=False, error=str(ex))
			await asyncio.sleep(COMMAND_POLL_INTERVAL)

	async def dispatch(self, command):
		"""Accept a command without blocking the poll loop.

		Running a command inline stalls this loop for as long as it takes -- a
		scan for its whole window, a pairing until someone types the PIN. The
		next command was then not even read, so the dialog saw no answer at all
		and the PIN, which has to arrive during a running pairing, could never
		get through.
		"""
		action = command.get("action")
		seq = command.get("seq")

		if action == "pin":
			# Instant, and it must be handled while pair() is still running.
			self.pin_value = str(command.get("pin", "")).strip()
			self.pin_event.set()
			return

		# Acknowledge before doing the work so the dialog knows we are alive.
		# No message: the dialog already shows what it asked for, and a raw
		# string here would replace that translated text.
		self.write_state(seq=seq, busy=True)

		if self.command_task is not None and not self.command_task.done():
			self.command_task.cancel()
			try:
				await asyncio.wait_for(self.command_task, timeout=5)
			except (asyncio.CancelledError, asyncio.TimeoutError):
				pass
			except Exception:
				pass

		self.command_task = asyncio.ensure_future(self.run_command(command))

	async def run_command(self, command):
		try:
			await self.handle_command(command)
		except asyncio.CancelledError:
			raise
		except Exception as ex:
			log(LOG_ERROR, "UI command failed: " + traceback.format_exc())
			self.write_state(seq=command.get("seq"), busy=False, error=str(ex))

	@staticmethod
	def describe(conf):
		return {
			"identifier": conf.identifier,
			"name": conf.name,
			"address": str(conf.address),
			"model": conf.device_info.model_str,
			"paired": all(
				any(
					service.protocol == protocol and service.credentials
					for service in conf.services
				)
				for protocol in PAIRING_PROTOCOLS
			),
		}

	async def handle_command(self, command):
		action = command.get("action")
		seq = command.get("seq")

		if action == "scan":
			self.write_state(seq=seq, busy=True, message="scanning")
			confs = await self.scan(
				host=command.get("host") or None, timeout=SCAN_TIMEOUT_UI
			)

			# A device we hold an open connection to does not show up in the
			# scan, so the node's own Apple TV would be missing from the list it
			# is configured in. Put it back in from the live connection.
			if self.current_conf is not None and self.match_identifier(
				confs, self.current_conf.identifier
			) is None:
				confs = [self.current_conf] + list(confs)

			devices = [self.describe(conf) for conf in confs]
			self.write_state(seq=seq, busy=False, devices=devices)
			return

		if action == "pair":
			await self.pair(seq, command)
			return

		log(LOG_WARN, "Unknown UI action: " + str(action))

	async def pair(self, seq, command):
		"""Pair AirPlay and Companion, asking the frontend for each PIN."""
		identifier = str(command.get("identifier", "") or "").strip()
		host = str(command.get("host", "") or "").strip()

		self.pairing_active = True
		self.status("pairing", "blue", "ring")
		try:
			# Drop an existing connection: the device will not pair while pyatv
			# holds a session to it. Wait for the teardown to actually finish
			# instead of guessing a delay.
			if self.listeners is not None:
				self.listeners.disconnected.set()
			for _ in range(100):
				if self.atv is None:
					break
				await asyncio.sleep(0.1)

			conf = await self.resolve_device(identifier, host)
			if conf is None:
				self.write_state(seq=seq, busy=False, error="device not found")
				return

			for protocol in PAIRING_PROTOCOLS:
				if not any(service.protocol == protocol for service in conf.services):
					continue
				await self.pair_protocol(seq, conf, protocol)

			await self.storage.save()
			self.write_state(
				seq=seq,
				busy=False,
				paired=True,
				identifier=conf.identifier,
				name=conf.name,
				address=str(conf.address),
				message="pairing complete",
			)
			log(LOG_INFO, "Pairing complete for " + conf.name)
		finally:
			self.pairing_active = False

	async def pair_protocol(self, seq, conf, protocol):
		handler = await pyatv.pair(conf, protocol, loop, storage=self.storage)
		try:
			await handler.begin()

			if handler.device_provides_pin:
				self.pin_value = None
				self.pin_event.clear()
				self.write_state(
					seq=seq, busy=True, needPin=True, protocol=protocol.name,
					message="enter the PIN shown on the Apple TV",
				)
				try:
					await asyncio.wait_for(self.pin_event.wait(), timeout=PIN_TIMEOUT)
				except asyncio.TimeoutError:
					raise TimeoutError("no PIN entered for " + protocol.name)
				handler.pin(self.pin_value)
			else:
				handler.pin(1234)
				self.write_state(
					seq=seq, busy=True, needPin=False, protocol=protocol.name,
					message="enter PIN 1234 on the Apple TV",
				)

			await handler.finish()
		finally:
			await handler.close()

		if not handler.has_paired:
			raise RuntimeError("pairing failed for " + protocol.name)

	# -- lifecycle -------------------------------------------------------

	async def run(self):
		self.storage = NodeDataStorage(self.hg, loop)
		await self.storage.load()
		await self.load_fallback_config()
		await self.load_last_address()
		await self.load_last_command_seq()

		tasks = [
			asyncio.ensure_future(self.connection_loop()),
			asyncio.ensure_future(self.command_loop()),
		]
		try:
			while self.hg.connected():
				await asyncio.sleep(1)
		finally:
			if self.command_task is not None:
				tasks.append(self.command_task)
			for task in tasks:
				task.cancel()
			await asyncio.gather(*tasks, return_exceptions=True)


async def nodeInputThreadSafe(nodeInfo, inputIndex, message):
	if node is None:
		return
	if inputIndex == 0:
		await node.power(bool(message["payload"]))
	elif inputIndex == 1:
		await node.remote(message)



# Wait for the flows to start.
#
# Waiting for the notification alone loses the race: the wrapper pings the
# process until it answers, which happens inside the Homegear() constructor
# above, and it then sends startUpComplete right away -- possibly before this
# thread is waiting. A notify() with no waiter is simply dropped and the node
# would wait forever even though nodeInfo had already arrived. So wait on the
# state, not on the notification, and give up after a while instead of hanging.
startUpComplete.acquire()
waited = 0
while hg.connected() and nodeInfo is None and waited < STARTUP_TIMEOUT:
	startUpComplete.wait(1)
	waited += 1
startedWith = "startUpComplete" if nodeInfo is not None else "timeout"
startUpComplete.release()

if nodeInfo is None:
	log(LOG_WARN,
		"No startUpComplete received within " + str(STARTUP_TIMEOUT) + "s ("
		+ str(receivedEvents) + " events received, nodeBlue: "
		+ (",".join(nodeBlueEvents) if nodeBlueEvents else "none")
		+ "). Falling back to the configuration stored in node data.")
else:
	log(LOG_INFO, "Node started after " + str(waited) + "s via " + startedWith)

# The node is now fully started. Start event loop.
node = AppleTvNode(hg, (nodeInfo or {}).get("info"))
try:
	loop.run_until_complete(node.run())
except KeyboardInterrupt:
	pass