"""One cast: capture → local HTTP stream → receiver, and the watching of it.""" from __future__ import annotations import asyncio import shutil import time from .backends.base import BackendError from .capture import Capture, PortalError, pick_encoder from .devices import Device from .stream import Fanout from .util import firewall_hint, local_ip_for, log, runtime_dir, url_host IDLE, STARTING, STREAMING, STOPPING, ERROR = "idle", "starting", "streaming", "stopping", "error" def choose_container(backend, cfg) -> str: """The best container this receiver and this machine can agree on.""" preference = str(cfg["encoder"]) for container in backend.containers: if container == "hls": return "hls" if preference in ("vp8", "vp9") and container != "webm": continue try: pick_encoder(preference, container) return container except PortalError: continue # Nothing matched the preference; fall back to whatever the machine has. for container in backend.containers: try: pick_encoder("auto", container) return container except PortalError: continue raise PortalError("no usable encoder for this receiver") class Session: def __init__(self, daemon) -> None: self.daemon = daemon self.state = IDLE self.phase = "" # what "starting" is waiting on, for the panel self.device: Device | None = None self.backend = None self.container = "" self.encoder = "" self.url = "" self.error = "" self.since = 0.0 self.player_state = "" self.clients = 0 self.last_client = 0.0 self.capture = Capture(on_stopped=self._capture_stopped) self.fanout: Fanout | None = None self.hls_dir = None self._watchdog: asyncio.Task | None = None self._reconnects = 0 # ---- reporting ---- @property def active(self) -> bool: return self.state in (STARTING, STREAMING) def to_json(self) -> dict: width, height = self.capture.size return { "state": self.state, "phase": self.phase, "deviceId": self.device.id if self.device else "", "deviceName": self.device.name if self.device else "", "deviceKind": self.device.kind if self.device else "", "since": self.since, "url": self.url, "container": self.container, "encoder": self.encoder, "source": self.capture.source_name, "preview": str(self.capture.preview_path) if self.capture.preview_at else "", "previewAt": self.capture.preview_at, "width": width, "height": height, "clients": self.clients, "playerState": self.player_state, "bytes": self.fanout.bytes_out if self.fanout else 0, # How far behind the receiver is letting itself fall, in bytes. "backlog": self.fanout.backlog if self.fanout else 0, "dropped": self.fanout.dropped if self.fanout else 0, "error": self.error, } # ---- start / stop ---- async def start(self, device: Device, backend) -> None: await self.stop() cfg = self.daemon.cfg self.device, self.backend = device, backend self.state, self.error, self.since = STARTING, "", time.time() self.phase = "screen" # the portal is asking which screen to share self.player_state, self.clients, self._reconnects = "", 0, 0 self.daemon.push_state(now=True) try: self.container = choose_container(backend, cfg) hls_dir = None if self.container == "hls": hls_dir = runtime_dir() / "hls" shutil.rmtree(hls_dir, ignore_errors=True) hls_dir.mkdir(parents=True, exist_ok=True) self.hls_dir = hls_dir self.daemon.stream.hls_dir = hls_dir self.fanout = None else: self.fanout = Fanout(self.container, int(cfg["bitrate"])) self.daemon.stream.fanout = self.fanout await self.daemon.stream.start(int(cfg["port"])) await self.capture.start( container="mpegts" if self.container == "hls" else self.container, cfg=cfg, fanout=self.fanout, hls_dir=str(hls_dir) if hls_dir else None, on_phase=self._set_phase, ) self.encoder = self.capture.encoder ip = local_ip_for(device.host) # A fresh token per session is both the access control and the reason # a receiver cannot carry its old buffer - and the delay in it - into # the new stream: to it, this is a URL it has never seen. self.daemon.stream.new_token() path = (self.daemon.stream.hls_path() if self.container == "hls" else self.daemon.stream.url_path(self.container)) self.url = f"http://{url_host(ip)}:{self.daemon.stream.port}{path}" if self.container == "hls": await self._await_playlist() self._set_phase("connect") await backend.play(device, self.url, container=self.container, title="Screencast") except asyncio.CancelledError: await self._teardown() self.state, self.device, self.backend = IDLE, None, None self.daemon.push_state(now=True) raise except (PortalError, BackendError) as exc: await self.fail(str(exc)) return except Exception as exc: # noqa: BLE001 - never leave a half-started cast behind log.exception("cast failed") await self.fail(str(exc)) return self.state = STREAMING # Handed over, but nothing is on screen until the receiver fetches it. self.phase = "live" if self.clients else "waiting" self.daemon.push_state(now=True) self._watchdog = asyncio.create_task(self._watch()) def _set_phase(self, phase: str) -> None: self.phase = phase self.daemon.push_state(now=True) async def _await_playlist(self) -> None: """AirPlay fetches the playlist immediately; give hlssink2 time to write one.""" assert self.hls_dir is not None for _ in range(100): if (self.hls_dir / "index.m3u8").exists(): await asyncio.sleep(1.0) # one full segment, so the receiver has something to play return await asyncio.sleep(0.1) raise PortalError("the encoder produced no HLS playlist") async def fail(self, message: str) -> None: log.warning("cast failed: %s", message) device, backend = self.device, self.backend await self._teardown() self.device, self.backend = device, backend self.state, self.error, self.phase = ERROR, message, "" self.daemon.push_state(now=True) async def stop(self, *, quiet: bool = False) -> None: if self.state == IDLE and self.device is None: return self.state = STOPPING if not quiet: self.daemon.push_state(now=True) device, backend = self.device, self.backend if backend is not None and device is not None: try: await backend.stop(device) except Exception as exc: # noqa: BLE001 log.debug("backend stop: %s", exc) await self._teardown() self.state, self.error, self.phase = IDLE, "", "" self.device, self.backend = None, None if not quiet: self.daemon.push_state(now=True) async def _teardown(self) -> None: # The watchdog is usually the one calling this (through fail()), and a # task that cancels itself never gets to finish the job: the session # would stay "streaming" with a dead encoder behind it, which is exactly # what a receiver stuck on a spinner looks like from here. watchdog, self._watchdog = self._watchdog, None if watchdog is not None and watchdog is not asyncio.current_task(): watchdog.cancel() await self.capture.stop() if self.fanout is not None: self.fanout.close() self.fanout = None self.daemon.stream.fanout = None self.daemon.stream.hls_dir = None if self.hls_dir is not None: shutil.rmtree(self.hls_dir, ignore_errors=True) self.hls_dir = None self.url = "" self.clients = 0 # ---- live ---- def note_client(self, peer: str, connected: bool) -> None: if self.fanout is None and self.hls_dir is None: return self.clients = self.fanout.clients if self.fanout else self.clients + (1 if connected else -1) self.clients = max(0, self.clients) if connected: self.last_client = time.time() self._reconnects = 0 if self.state == STREAMING: self.phase = "live" if self.clients else "waiting" self.daemon.push_state(now=True) def _capture_stopped(self, code: int, error: str) -> None: if not self.active: return asyncio.create_task(self.fail(error or f"screen capture stopped ({code})")) async def _watch(self) -> None: """Poll the receiver, and notice when the picture stops arriving.""" last_bytes, last_progress = -1, time.time() ticks = 0 while self.active: await asyncio.sleep(2) ticks += 1 if self.device is None or self.backend is None: return # A client that connected and then went quiet (a half-open socket, a # receiver that gave up without closing) used to keep the session # looking healthy forever, so check the encoder itself rather than # trusting the client count. if not self.capture.running: await self.fail(self.capture.error or "the screen capture stopped") return if self.fanout is not None: if self.fanout.bytes_out != last_bytes: last_bytes, last_progress = self.fanout.bytes_out, time.time() elif time.time() - last_progress > 15: await self.fail("the encoder stopped producing video") return try: status = await self.backend.poll(self.device) except Exception as exc: # noqa: BLE001 log.debug("poll: %s", exc) status = {} if status: self.player_state = status.get("playerState", self.player_state) if "volume" in status and status["volume"] is not None: self.device.volume = float(status["volume"]) if "muted" in status: self.device.muted = bool(status["muted"]) self.device.app = status.get("app", self.device.app) if status.get("connected") is False: await self.fail(f"{self.device.name} dropped the connection") return # A receiver that never fetched the stream, or that let go of it, is # not casting — retry the play once before giving up on it. stale = self.fanout is not None and self.clients == 0 and time.time() - self.since > 12 if stale and time.time() - self.last_client > 12: if bool(self.daemon.cfg["autoReconnect"]) and self._reconnects < 2: self._reconnects += 1 log.info("no receiver on the stream; retrying play (%d)", self._reconnects) try: await self.backend.play(self.device, self.url, container=self.container, title="Screencast") self.last_client = time.time() except BackendError as exc: await self.fail(str(exc)) return else: hint = firewall_hint(self.daemon.stream.port, self.device.host) reason = f"{self.device.name} never picked up the stream" if hint: # The panel turns this into an "Open port" button. self.daemon.firewall_blocked = True await self.fail(f"{reason} — {hint}" if hint else reason) return # How much of the delay is ours: bytes the receiver has not taken # yet, and bytes we skipped past because it never would have. if self.fanout is not None and ticks % 8 == 0 and self.clients: rate = self.fanout.bytes_out / max(1e-3, time.time() - self.fanout.started) log.info("stream: %.1f KiB queued (%.1fs behind), %.0f KiB skipped, %.0f KiB/s", self.fanout.backlog / 1024, self.fanout.backlog / max(1.0, rate), self.fanout.dropped / 1024, rate / 1024) self.daemon.push_state(now=True) # ---- source ---- async def repick_source(self) -> None: """Restart the cast so the portal asks which screen to share again.""" if self.active and self.device is not None and self.backend is not None: await self.start(self.device, self.backend)