cast.py

#!/usr/bin/env python3
"""
cast.py — stream Mac screen to Chromecast via HLS
Uses mss (CoreGraphics) for capture — works on macOS 15 Sequoia.

Usage:
    python cast.py                  # auto-pick the one Chromecast TV
    python cast.py --name "TV"      # match by device name (partial)
    python cast.py --list           # list discovered devices and exit
    python cast.py --display 1      # display index (default: 1 = main)
    python cast.py --port 8765      # HTTP server port (default: 8765)
    python cast.py --fps 15         # capture fps (default: 15)
    python cast.py --scale 1280x720 # output resolution (default: native)
"""

import argparse
import http.server
import os
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import threading
import time

import mss
import pychromecast


# ── helpers ──────────────────────────────────────────────────────────────────

def get_local_ip() -> str:
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        s.connect(("8.8.8.8", 80))
        return s.getsockname()[0]
    finally:
        s.close()


def discover_chromecasts(timeout: int = 5):
    print(f"Scanning for Chromecast devices ({timeout}s)…")
    chromecasts, browser = pychromecast.get_chromecasts(timeout=timeout)
    return chromecasts, browser


def pick_device(chromecasts, name_filter: str | None):
    if not chromecasts:
        print("No Chromecast devices found. Make sure you're on the same WiFi.")
        sys.exit(1)

    if name_filter:
        matches = [
            c for c in chromecasts
            if name_filter.lower() in c.cast_info.friendly_name.lower()
        ]
        if not matches:
            names = [c.cast_info.friendly_name for c in chromecasts]
            print(f"No device matching '{name_filter}'. Found: {names}")
            sys.exit(1)
        return matches[0]

    # Prefer video Chromecasts over speakers/groups
    video_devices = [
        c for c in chromecasts
        if c.cast_info.cast_type == pychromecast.CAST_TYPE_CHROMECAST
    ]
    candidates = video_devices if video_devices else chromecasts

    if len(candidates) == 1:
        return candidates[0]

    print("\nMultiple devices found:")
    for i, c in enumerate(candidates):
        print(f"  [{i}] {c.cast_info.friendly_name}")
    while True:
        try:
            idx = int(input("Pick a device number: "))
            return candidates[idx]
        except (ValueError, IndexError):
            print("Invalid choice, try again.")


def cleanup_stale():
    import glob
    for d in glob.glob("/var/folders/*/*/T/screencast_hls_*"):
        # kill ffmpeg pid stored by previous run
        pid_file = os.path.join(d, "ffmpeg.pid")
        if os.path.exists(pid_file):
            try:
                pid = int(open(pid_file).read().strip())
                os.kill(pid, 9)
            except Exception:
                pass
        shutil.rmtree(d, ignore_errors=True)


def make_hls_dir() -> str:
    return tempfile.mkdtemp(prefix="screencast_hls_")


def make_handler(hls_dir: str):
    class SilentHandler(http.server.SimpleHTTPRequestHandler):
        def __init__(self, *args, **kwargs):
            super().__init__(*args, directory=hls_dir, **kwargs)

        def log_message(self, *_):
            pass

        def end_headers(self):
            self.send_header("Access-Control-Allow-Origin", "*")
            super().end_headers()
    return SilentHandler


def start_http_server(hls_dir: str, port: int) -> http.server.HTTPServer:
    handler = make_handler(hls_dir)
    server = http.server.HTTPServer(("0.0.0.0", port), handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server


def wait_for_playlist(hls_dir: str, timeout: int = 20):
    playlist = os.path.join(hls_dir, "stream.m3u8")
    print("Waiting for stream to initialise…", end="", flush=True)
    deadline = time.time() + timeout
    while time.time() < deadline:
        if os.path.exists(playlist):
            print(" ready.")
            return
        time.sleep(0.5)
        print(".", end="", flush=True)
    print()
    print("Timed out. Check ffmpeg.log in the temp dir for errors.")
    sys.exit(1)


def start_capture(hls_dir: str, display_idx: int, fps: int, scale: str | None, log_path: str):
    """
    Capture screen with mss (CoreGraphics — works on macOS 15) and pipe
    raw BGRA frames into ffmpeg for HLS encoding.
    """
    with mss.mss() as sct:
        # mss.monitors[0] = all monitors combined; [1] = first/main display
        monitor = sct.monitors[display_idx]
        w, h = monitor["width"], monitor["height"]

    vf = f"scale={scale}" if scale else "null"

    playlist = os.path.join(hls_dir, "stream.m3u8")
    segment  = os.path.join(hls_dir, "seg%03d.ts")

    cmd = [
        "ffmpeg",
        "-loglevel", "error",
        # raw video from stdin
        "-f", "rawvideo",
        "-vcodec", "rawvideo",
        "-pix_fmt", "bgra",
        "-s", f"{w}x{h}",
        "-r", str(fps),
        "-i", "pipe:0",
        # silent audio (Chromecast requires an audio track)
        "-f", "lavfi",
        "-i", "anullsrc=r=44100:cl=stereo",
        # video
        "-vf", vf,
        "-c:v", "h264_videotoolbox",
        "-b:v", "4000k",
        # audio
        "-c:a", "aac",
        "-b:a", "32k",
        # HLS
        "-f", "hls",
        "-hls_time", "0.5",
        "-hls_list_size", "10",
        "-hls_flags", "delete_segments+append_list",
        "-hls_segment_filename", segment,
        playlist,
    ]

    log_file = open(log_path, "w")
    proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stderr=log_file, stdout=log_file)
    open(os.path.join(hls_dir, "ffmpeg.pid"), "w").write(str(proc.pid))

    frame_interval = 1.0 / fps

    def capture_loop():
        with mss.mss() as sct:
            monitor = sct.monitors[display_idx]
            while True:
                t0 = time.monotonic()
                frame = sct.grab(monitor)
                try:
                    proc.stdin.write(frame.raw)
                except (BrokenPipeError, ValueError):
                    break
                elapsed = time.monotonic() - t0
                sleep = frame_interval - elapsed
                if sleep > 0:
                    time.sleep(sleep)

    threading.Thread(target=capture_loop, daemon=True).start()
    return proc, log_file


# ── main ─────────────────────────────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(description="Cast Mac screen to Chromecast")
    parser.add_argument("--name",    help="Chromecast friendly name (partial match)")
    parser.add_argument("--list",    action="store_true", help="List devices and exit")
    parser.add_argument("--display", type=int, default=1,
                        help="Display index: 1=main (default), 2=second monitor…")
    parser.add_argument("--port",    type=int, default=8765)
    parser.add_argument("--fps",     type=int, default=15,
                        help="Capture frame rate (default 15 — good balance for screen sharing)")
    parser.add_argument("--scale",   default=None,
                        help="Output resolution e.g. 1280x720 (default: native)")
    args = parser.parse_args()

    cleanup_stale()
    chromecasts, browser = discover_chromecasts()

    if args.list:
        for c in chromecasts:
            print(f"  • {c.cast_info.friendly_name}  ({c.cast_info.host})")
        pychromecast.discovery.stop_discovery(browser)
        sys.exit(0)

    cast = pick_device(chromecasts, args.name)
    local_ip = get_local_ip()
    stream_url = f"http://{local_ip}:{args.port}/stream.m3u8"

    # Show display info
    with mss.mss() as sct:
        mon = sct.monitors[args.display]
        res = f"{mon['width']}x{mon['height']}"

    print(f"\nTarget  : {cast.cast_info.friendly_name}")
    print(f"Stream  : {stream_url}")
    print(f"Display : {args.display}  ({res})  {args.fps}fps")

    hls_dir  = make_hls_dir()
    log_path = os.path.join(hls_dir, "ffmpeg.log")

    server = start_http_server(hls_dir, args.port)
    ffmpeg_proc, ffmpeg_log = start_capture(
        hls_dir, args.display, args.fps, args.scale, log_path
    )

    def shutdown(sig, frame):
        print("\nStopping…")
        try:
            cast.quit_app()
        except Exception:
            pass
        try:
            ffmpeg_proc.stdin.close()
        except Exception:
            pass
        ffmpeg_proc.terminate()
        ffmpeg_log.close()
        server.shutdown()
        pychromecast.discovery.stop_discovery(browser)
        shutil.rmtree(hls_dir, ignore_errors=True)
        sys.exit(0)

    signal.signal(signal.SIGINT, shutdown)
    signal.signal(signal.SIGTERM, shutdown)

    wait_for_playlist(hls_dir)

    cast.wait(timeout=10)
    cast.quit_app()          # stop whatever is currently playing (YouTube etc.)
    time.sleep(2)            # give it a moment to return to idle
    mc = cast.media_controller
    mc.play_media(stream_url, "application/x-mpegURL")
    mc.block_until_active(timeout=15)

    print("Casting! Press Ctrl-C to stop.\n")

    while True:
        time.sleep(1)


if __name__ == "__main__":
    main()