How to use this
Two ways, same document. Pick whichever suits you.
- AI assisted (fastest). On your Mac, open your favourite AI coding agent (Claude Code, Codex, Cursor, or similar). Paste in this entire document and say: "Build this project on my Mac and walk me through the setup." The agent creates the files, installs the dependencies, builds the native helper, and guides you through the two macOS permissions and the one router step.
- By hand. Prefer to do it yourself? Everything is here. Create the files from the Reference implementation, then follow the Setup runbook.
Either way, do not skip the Platform gotchas section. It captures the traps that cost real time to solve on modern macOS, so you (or your agent) get a smooth, low-latency result on the first try instead of the fifth.
You can name the project folder anything you like. This brief uses
remote-desktop/, with two subfolders: mac-agent/ and laptop-server/.
What you are building
The Mac at the office runs an agent that captures both screens and injects the mouse and keyboard it receives. The Windows laptop at home runs a small server plus a browser viewer. You see and control the Mac in the browser.
The clever bit: office networks usually block incoming connections, so instead of reaching in, the Mac dials out to your home laptop. Nothing on the office side is ever exposed, and you only configure your own home router.
Office (locked network) Home (your own router)
Mac agent -- dials out --------------> Laptop server + browser
captures screens, injects input relays video, serves the viewer
(hardware H.264 over TLS) http://localhost:8000
Video is hardware-encoded H.264 and decoded in the browser with WebCodecs, so it is smooth and low-bandwidth. On Apple Silicon the Mac uses a native ScreenCaptureKit capture path for true ~60fps that also shows the mouse cursor.
Requirements
- A Mac that stays on at the office (Apple Silicon recommended, macOS 12.3+).
- A Windows laptop at home.
- A home internet connection where you control the router (one port forward). A fixed public IP is ideal; with a dynamic IP, use a Dynamic-DNS hostname.
- Homebrew on the Mac, plus Xcode command line tools
(
xcode-select --install) for the native 60fps path. Without them the agent falls back to a slower pure-Python capture path automatically. - Chrome or Edge on the laptop (for WebCodecs).
There are no secrets in this document. Each machine generates its own private token and certificate on first run.
Platform gotchas (do not skip)
These are the things that are non-obvious and expensive to rediscover. Make sure your agent applies them.
- Capture with ScreenCaptureKit, not ffmpeg's avfoundation. On macOS 26,
ffmpeg's
avfoundationscreen input opens the device but never delivers frames (the old AVCaptureScreenInput path is deprecated). Capture with a native ScreenCaptureKit plus VideoToolbox helper (the Swift file below) for true ~60fps and the cursor. A pure-Pythonmsspath is included as a fallback, with ffmpeg used only as the H.264 encoder. - ScreenCaptureKit -3805 at startup. Starting one SCStream per display at the same instant sometimes drops one with error -3805, "application connection being interrupted." Stagger the starts by about a second and supervise each helper so it restarts. Both streams then run cleanly.
- Low-latency typing. ScreenCaptureKit only delivers a frame when the
screen changes, so sparse updates like typing feel laggy. Decouple capture
from encode: keep the most recent frame and encode it at a constant fps on a
timer. Also make the encoder emit each frame immediately
(
MaxFrameDelayCount0) and keep the capturequeueDepthsmall. That restores fixed-rate, low-latency streaming while keeping the cursor. - Permissions attach to the exact binary. The agent needs both Screen
Recording and Accessibility, granted to the venv's
python3. Build the venv from a real Python with--copies(prefer Homebrew) sovenv/bin/python3is a standalone binary you can actually select in the macOS permission picker. A symlinked interpreter (the Xcodepython3) cannot be selected. The ScreenCaptureKit helper, spawned bypython3, inherits its Screen Recording grant. - Headless display sleep. macOS only streams an active display. When
nobody is at the Mac the display sleeps and it reports 0 displays. Hold a
caffeinate -dassertion for the agent's life and wake the display on start. If you physically power the monitors off and they disconnect from macOS, add a cheap HDMI dummy plug so a display stays registered. - Keyboard mapping. Keystrokes go to the Mac, where copy and paste use Cmd. Map the laptop's Ctrl (and the Windows key) to Cmd so Ctrl+C, Ctrl+V and friends behave as a Windows user expects. Do not rely on the Windows key alone, Windows grabs it.
- Show one cursor, not two. Once the native helper renders the real Mac
cursor into the video, hide the browser's own cursor over the canvas
(
cursor: none), or the two pointers look misaligned. - Do not test on the office LAN. Many office and guest networks, and phone hotspots, isolate devices from each other, so the two machines cannot see each other locally. The design dials out over the internet anyway, so set up the home port-forward and test that path. Confirm reachability with a plain TCP check to the forwarded port before debugging anything else.
Setup runbook
Your agent can do most of this; the manual steps are the permission grants and the router.
1. Windows laptop (creates the token and certificate).
Install Python 3 (tick "Add python.exe to PATH"). In laptop-server, run
setup.bat. It installs two packages and starts the server, which on first run
prints an access token and creates cert.pem. Note the token and keep
cert.pem; never share key.pem. If Windows Firewall asks, allow Python.
2. Mac agent.
Install Homebrew and the Xcode command line tools. Put the laptop's cert.pem
into mac-agent/. Copy config.example.json to config.json and paste in the
token. Run bash install.sh: it installs ffmpeg, builds the native helper,
creates the Python environment, and registers the agent to start at login.
3. Grant both permissions to python3 in System Settings, Privacy and
Security: Screen Recording and Accessibility. If python3 is not
listed, add mac-agent/venv/bin/python3 with the + button. Then restart the
agent: launchctl kickstart -k gui/$(id -u)/com.remotedesktop.agent.
4. Home router. Find your home public IP, reserve the laptop's local IP, and forward TCP 8443 to it. Make sure Windows Firewall allows the server on your home network.
5. Connect. Set server_url in mac-agent/config.json to
wss://YOUR-HOME-IP:8443 and restart the agent. On the laptop, with the server
running, open http://localhost:8000 in Chrome or Edge. Both Mac screens appear
within a few seconds. The Windows key and Ctrl act as Cmd; keyboard is sent
while the mouse is over a screen.
Security
- The stream is TLS-encrypted with a pinned self-signed certificate; the
agent only talks to a server presenting exactly your
cert.pem. - Nobody can connect without the secret token.
- The viewer WebSocket only accepts the local viewer page, so no other website can reach it.
- Keep
key.pemand bothconfig.jsonfiles private, and never commit them. To revoke access, delete the server'sconfig.jsonand restart it; a new token is generated and the Mac needs the new one.
Reference implementation
Create each file at the path shown, with the exact contents. Paths are relative to your project folder.
laptop-server/server.py
#!/usr/bin/env python3
"""Remote Desktop - laptop server.
Runs on the Windows laptop. The Mac agent dials in on the agent port
(TLS + token). Your browser watches on http://localhost:8000 and the
viewer WebSocket, which are bound to localhost only.
"""
import asyncio
import hmac
import json
import secrets
import ssl
import threading
from datetime import datetime, timedelta, timezone
from functools import partial
from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler
from pathlib import Path
import websockets
BASE = Path(__file__).resolve().parent
CONFIG_FILE = BASE / "config.json"
CERT_FILE = BASE / "cert.pem"
KEY_FILE = BASE / "key.pem"
def load_config():
if CONFIG_FILE.exists():
return json.loads(CONFIG_FILE.read_text())
cfg = {
"token": secrets.token_urlsafe(32),
"agent_port": 8443,
"web_port": 8000,
"viewer_ws_port": 8001,
}
CONFIG_FILE.write_text(json.dumps(cfg, indent=2))
print("Created config.json with a fresh access token.")
return cfg
def ensure_certificate():
if CERT_FILE.exists() and KEY_FILE.exists():
return
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "Remote Desktop")])
now = datetime.now(timezone.utc)
cert = (x509.CertificateBuilder()
.subject_name(name).issuer_name(name)
.public_key(key.public_key())
.serial_number(x509.random_serial_number())
.not_valid_before(now - timedelta(days=1))
.not_valid_after(now + timedelta(days=3650))
.sign(key, hashes.SHA256()))
KEY_FILE.write_bytes(key.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption()))
CERT_FILE.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
print("Created cert.pem and key.pem.")
print(">> Copy cert.pem (only cert.pem!) to the Mac's mac-agent folder.")
CFG = load_config()
AGENT = None # the Mac's websocket, or None
HELLO = None # last hello message (display list) from the agent
VIEWERS = set() # browser websockets
async def broadcast(payload):
dead = []
for viewer in list(VIEWERS):
try:
await viewer.send(payload)
except Exception:
dead.append(viewer)
for viewer in dead:
VIEWERS.discard(viewer)
async def handle_agent(ws, path=None):
global AGENT, HELLO
try:
first = json.loads(await asyncio.wait_for(ws.recv(), timeout=10))
except Exception:
return
if first.get("type") != "auth" or not hmac.compare_digest(
str(first.get("token", "")), CFG["token"]):
print("Rejected a connection with a wrong token.")
await ws.close(4401, "bad token")
return
if AGENT is not None:
await AGENT.close()
AGENT = ws
print("Mac agent connected from", ws.remote_address)
await broadcast(json.dumps({"type": "status", "agent": True}))
try:
async for msg in ws:
if isinstance(msg, (bytes, bytearray)):
await broadcast(msg)
else:
try:
data = json.loads(msg)
except ValueError:
continue
if data.get("type") == "hello":
HELLO = msg
await broadcast(msg)
finally:
if AGENT is ws:
AGENT = None
HELLO = None
await broadcast(json.dumps({"type": "status", "agent": False}))
print("Mac agent disconnected.")
def request_origin(ws):
"""Read the Origin header across websockets library versions."""
headers = getattr(getattr(ws, "request", None), "headers", None)
if headers is None:
headers = getattr(ws, "request_headers", {})
try:
return headers.get("Origin")
except Exception:
return None
# Only the local viewer page may open the control WebSocket. Without this,
# any website you happen to have open could connect to ws://127.0.0.1 and
# drive the Mac (browsers don't apply CORS to WebSockets).
ALLOWED_ORIGINS = {
"http://localhost:%d" % CFG["web_port"],
"http://127.0.0.1:%d" % CFG["web_port"],
}
async def handle_viewer(ws, path=None):
origin = request_origin(ws)
if origin not in ALLOWED_ORIGINS:
print("Rejected a viewer from origin:", origin)
await ws.close(4403, "bad origin")
return
VIEWERS.add(ws)
try:
await ws.send(json.dumps({"type": "status", "agent": AGENT is not None}))
if HELLO:
await ws.send(HELLO)
async for msg in ws:
if AGENT is not None and isinstance(msg, str):
try:
await AGENT.send(msg)
except Exception:
pass
finally:
VIEWERS.discard(ws)
def start_web_server():
handler = partial(SimpleHTTPRequestHandler, directory=str(BASE / "ui"))
handler.log_message = lambda *a, **k: None
httpd = ThreadingHTTPServer(("127.0.0.1", CFG["web_port"]), handler)
threading.Thread(target=httpd.serve_forever, daemon=True).start()
def local_ips():
"""This machine's LAN IPv4 address(es), best guess first."""
import socket
ips = []
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80)) # no packet sent; reveals the LAN iface
ips.append(s.getsockname()[0])
s.close()
except Exception:
pass
try:
for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET):
ip = info[4][0]
if ip not in ips:
ips.append(ip)
except Exception:
pass
return [ip for ip in ips if not ip.startswith("127.")]
async def main():
ensure_certificate()
ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ssl_ctx.load_cert_chain(CERT_FILE, KEY_FILE)
start_web_server()
async with websockets.serve(handle_agent, "0.0.0.0", CFG["agent_port"],
ssl=ssl_ctx, max_size=None), \
websockets.serve(handle_viewer, "127.0.0.1", CFG["viewer_ws_port"],
max_size=None):
ips = local_ips()
print("")
print("=== Remote Desktop server is running ===")
print("Viewer (open in Chrome/Edge): http://localhost:%d" % CFG["web_port"])
print("Agent port (TLS): %d" % CFG["agent_port"])
print("")
print("--- Put these on the Mac (mac-agent/config.json) ---")
if ips:
print(" server_url : wss://%s:%d <-- same-network (office) test"
% (ips[0], CFG["agent_port"]))
for extra in ips[1:]:
print(" wss://%s:%d (other network interface)"
% (extra, CFG["agent_port"]))
print(" (for use FROM HOME, swap this for your home public IP)")
else:
print(" server_url : wss://<this-laptop-IP>:%d (run ipconfig to find it)"
% CFG["agent_port"])
print(" token : %s" % CFG["token"])
print("")
print("Also copy cert.pem (NOT key.pem) from this folder to the Mac's mac-agent folder.")
print("Waiting for the Mac to dial in...")
print("")
await asyncio.Future()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
pass
laptop-server/requirements.txt
websockets>=12
cryptography>=42
laptop-server/setup.bat
@echo off
cd /d "%~dp0"
title Remote Desktop - laptop setup
echo ==================================================
echo Remote Desktop - Windows laptop setup
echo ==================================================
echo.
REM --- 1. Check Python is installed and on PATH ---------------------------
where python >nul 2>nul
if errorlevel 1 (
echo [X] Python was not found on this laptop.
echo.
echo 1. Install Python 3 from https://www.python.org/downloads/
echo 2. During install, TICK "Add python.exe to PATH"
echo 3. Then run this setup.bat again.
echo.
pause
exit /b 1
)
for /f "delims=" %%v in ('python --version 2^>^&1') do echo Using %%v
echo.
REM --- 2. Install the two Python packages --------------------------------
echo Installing required packages (websockets, cryptography)...
python -m pip install --upgrade pip
python -m pip install -r requirements.txt
if errorlevel 1 (
echo.
echo [X] Package install failed - see the messages above.
pause
exit /b 1
)
echo.
REM --- 3. Start the server (creates token + cert.pem on first run) --------
echo ==================================================
echo Starting server. On first run this creates:
echo - config.json (your secret access token)
echo - cert.pem (copy this to the Mac)
echo - key.pem (stays private on this laptop)
echo.
echo If Windows Firewall asks, click ALLOW (private + public).
echo Leave this window OPEN while you use Remote Desktop.
echo Press Ctrl+C in here to stop the server.
echo ==================================================
echo.
python server.py
pause
laptop-server/ui/index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Remote Desktop</title>
<style>
:root {
--bg: #0d1117;
--panel: #161c26;
--edge: #232c3b;
--text: #dbe4f0;
--muted: #7d8aa0;
--accent: #4f8cff;
--ok: #2ecc8f;
--bad: #e05a5a;
}
* { margin: 0; padding: 0; box-sizing: border-box; }
body {
background: var(--bg);
color: var(--text);
font-family: -apple-system, "Segoe UI", system-ui, sans-serif;
min-height: 100vh;
display: flex;
flex-direction: column;
}
header {
display: flex;
align-items: center;
gap: 18px;
padding: 12px 20px;
background: var(--panel);
border-bottom: 1px solid var(--edge);
}
.brand {
display: flex;
align-items: center;
gap: 10px;
font-weight: 700;
letter-spacing: 2.5px;
font-size: 14px;
}
.brand svg { color: var(--accent); }
.brand span { color: var(--accent); font-weight: 400; }
.status {
display: flex;
align-items: center;
gap: 8px;
font-size: 13px;
color: var(--muted);
}
.dot {
width: 9px; height: 9px;
border-radius: 50%;
background: var(--bad);
box-shadow: 0 0 8px var(--bad);
}
.dot.on { background: var(--ok); box-shadow: 0 0 8px var(--ok); }
.spacer { flex: 1; }
.metric { font-size: 13px; color: var(--muted); font-variant-numeric: tabular-nums; }
button {
background: transparent;
color: var(--text);
border: 1px solid var(--edge);
border-radius: 6px;
padding: 6px 14px;
font-size: 13px;
cursor: pointer;
display: inline-flex;
align-items: center;
gap: 7px;
}
button:hover { border-color: var(--accent); }
button.active { border-color: var(--accent); color: var(--accent); }
main {
flex: 1;
display: flex;
flex-wrap: wrap;
gap: 14px;
padding: 14px;
align-items: flex-start;
justify-content: center;
}
.display {
background: var(--panel);
border: 1px solid var(--edge);
border-radius: 10px;
overflow: hidden;
flex: 1 1 46%;
min-width: 380px;
max-width: 960px;
}
.display .bar {
display: flex;
align-items: center;
gap: 8px;
padding: 7px 12px;
font-size: 12px;
color: var(--muted);
border-bottom: 1px solid var(--edge);
}
.display .bar svg { color: var(--accent); }
.display .bar .grow { flex: 1; }
.display .bar button { padding: 3px 8px; border: none; color: var(--muted); }
.display .bar button:hover { color: var(--accent); }
canvas {
display: block;
width: 100%;
height: auto;
/* Hide the browser's own pointer over the screens: the Mac's real cursor is
shown in the video, so a second local cursor would just misalign with it. */
cursor: none;
background: #000;
}
#offline {
position: fixed;
inset: 0;
display: flex;
flex-direction: column;
gap: 12px;
align-items: center;
justify-content: center;
background: rgba(13, 17, 23, 0.88);
font-size: 15px;
color: var(--muted);
text-align: center;
padding: 20px;
}
#offline svg { color: var(--edge); }
#offline.hidden { display: none; }
footer {
padding: 8px 20px;
font-size: 11px;
color: var(--muted);
border-top: 1px solid var(--edge);
background: var(--panel);
}
</style>
</head>
<body>
<header>
<div class="brand">
<svg width="20" height="20" viewBox="0 0 24 24" fill="none" stroke="currentColor"
stroke-width="2" stroke-linecap="round" stroke-linejoin="round">
<rect x="2" y="3" width="20" height="14" rx="2"></rect>
<line x1="8" y1="21" x2="16" y2="21"></line>
<line x1="12" y1="17" x2="12" y2="21"></line>
</svg>
REMOTE <span>DESKTOP</span>
</div>
<div class="status"><div class="dot" id="agentDot"></div><span id="agentText">Waiting for the Mac…</span></div>
<div class="spacer"></div>
<div class="metric" id="fps">0 fps</div>
<button id="controlBtn" class="active" title="When on, your mouse and keyboard control the Mac while hovering a screen">
<svg width="14" height="14" viewBox="0 0 24 24" fill="none" stroke="currentColor"
stroke-width="2" stroke-linecap="round" stroke-linejoin="round">
<path d="m3 3 7.07 16.97 2.51-7.39 7.39-2.51L3 3z"></path>
</svg>
Input: on
</button>
</header>
<main id="displays"></main>
<div id="offline">
<svg width="56" height="56" viewBox="0 0 24 24" fill="none" stroke="currentColor"
stroke-width="1.5" stroke-linecap="round" stroke-linejoin="round">
<rect x="2" y="3" width="20" height="14" rx="2"></rect>
<line x1="8" y1="21" x2="16" y2="21"></line>
<line x1="12" y1="17" x2="12" y2="21"></line>
</svg>
<div id="offlineText">Waiting for the Mac Mini to dial in…</div>
<div style="font-size:12px">The agent retries every few seconds. Check mac-agent/agent.log if this takes long.</div>
</div>
<footer>Remote Desktop - hardware H.264. Ctrl and the Windows key act as Cmd (so Ctrl+C / Ctrl+V work), Alt as Option. Keyboard is sent while hovering a screen.</footer>
<script>
"use strict";
const BUTTON_NAMES = { 0: "left", 1: "middle", 2: "right" };
const container = document.getElementById("displays");
const agentDot = document.getElementById("agentDot");
const agentText = document.getElementById("agentText");
const offline = document.getElementById("offline");
const fpsLabel = document.getElementById("fps");
const controlBtn = document.getElementById("controlBtn");
let ws = null;
let panels = []; // one entry per display: {canvas, ctx, decoder, ...}
let control = true;
let hoverDisplay = null;
let frameCount = 0;
if (typeof VideoDecoder === "undefined") {
offline.classList.remove("hidden");
document.getElementById("offlineText").textContent =
"This browser has no WebCodecs support - open Remote Desktop in Chrome or Edge.";
}
controlBtn.addEventListener("click", () => {
control = !control;
controlBtn.classList.toggle("active", control);
controlBtn.lastChild.textContent = control ? " Input: on" : " Input: off";
if (!control) flush();
});
setInterval(() => {
fpsLabel.textContent = frameCount + " fps";
frameCount = 0;
}, 1000);
function send(obj) {
if (ws && ws.readyState === 1 && control) ws.send(JSON.stringify(obj));
}
// Release every key/button held on the Mac (prevents stuck modifiers when the
// mouse leaves a screen or the window loses focus mid-keypress).
function flush() {
if (ws && ws.readyState === 1) ws.send(JSON.stringify({ type: "flush" }));
}
function setStatus(on) {
agentDot.classList.toggle("on", on);
agentText.textContent = on ? "Mac connected" : "Waiting for the Mac…";
offline.classList.toggle("hidden", on);
if (!on) { teardownPanels(); }
}
// --- H.264 decoding -------------------------------------------------------
// Build the WebCodecs codec string ("avc1.PPCCLL") from the SPS in an access
// unit, so the decoder is configured with the exact profile/level ffmpeg used.
function codecFromAccessUnit(bytes) {
for (let i = 0; i + 4 < bytes.length; i++) {
if (bytes[i] === 0 && bytes[i + 1] === 0 && bytes[i + 2] === 1) {
const nalType = bytes[i + 3] & 0x1f;
if (nalType === 7) { // SPS
const p = bytes[i + 4], c = bytes[i + 5], l = bytes[i + 6];
const hex = (n) => n.toString(16).padStart(2, "0");
return "avc1." + hex(p) + hex(c) + hex(l);
}
}
}
return null;
}
function makeDecoder(panel) {
panel.decoder = new VideoDecoder({
output: (frame) => {
const cv = panel.canvas;
if (cv.width !== frame.displayWidth || cv.height !== frame.displayHeight) {
cv.width = frame.displayWidth;
cv.height = frame.displayHeight;
}
panel.ctx.drawImage(frame, 0, 0);
frame.close();
frameCount++;
},
error: (e) => {
// Reset - the next keyframe will reconfigure a fresh decoder.
try { panel.decoder.close(); } catch (_) {}
panel.configured = false;
panel.sawKey = false;
},
});
}
function feed(idx, keyframe, data) {
const panel = panels[idx];
if (!panel) return;
if (!panel.decoder || panel.decoder.state === "closed") makeDecoder(panel);
if (!panel.configured) {
if (!keyframe) return; // must start on a keyframe
const codec = codecFromAccessUnit(data);
if (!codec) return;
try {
panel.decoder.configure({ codec: codec, optimizeForLatency: true });
panel.configured = true;
panel.sawKey = true;
} catch (_) { return; }
}
try {
panel.decoder.decode(new EncodedVideoChunk({
type: keyframe ? "key" : "delta",
timestamp: panel.ts,
data: data,
}));
panel.ts += Math.round(1e6 / 30);
} catch (_) { /* out-of-order before a keyframe - ignore */ }
}
// --- panels ---------------------------------------------------------------
function teardownPanels() {
for (const p of panels) {
if (p.decoder && p.decoder.state !== "closed") {
try { p.decoder.close(); } catch (_) {}
}
}
container.innerHTML = "";
panels = [];
}
function buildPanels(displays) {
teardownPanels();
panels = displays.map((d) => {
const panel = document.createElement("div");
panel.className = "display";
panel.innerHTML =
'<div class="bar">' +
'<svg width="13" height="13" viewBox="0 0 24 24" fill="none" stroke="currentColor" ' +
'stroke-width="2" stroke-linecap="round" stroke-linejoin="round">' +
'<rect x="2" y="3" width="20" height="14" rx="2"></rect>' +
'<line x1="8" y1="21" x2="16" y2="21"></line><line x1="12" y1="17" x2="12" y2="21"></line></svg>' +
"<span>Display " + (d.i + 1) + " - " + d.w + "×" + d.h + '</span><span class="grow"></span>' +
'<button title="Fullscreen">' +
'<svg width="13" height="13" viewBox="0 0 24 24" fill="none" stroke="currentColor" ' +
'stroke-width="2" stroke-linecap="round" stroke-linejoin="round">' +
'<path d="M8 3H5a2 2 0 0 0-2 2v3"></path><path d="M21 8V5a2 2 0 0 0-2-2h-3"></path>' +
'<path d="M3 16v3a2 2 0 0 0 2 2h3"></path><path d="M16 21h3a2 2 0 0 0 2-2v-3"></path></svg>' +
"</button></div>";
const canvas = document.createElement("canvas");
canvas.width = d.w;
canvas.height = d.h;
panel.appendChild(canvas);
container.appendChild(panel);
panel.querySelector("button").addEventListener("click", () => {
(canvas.requestFullscreen || canvas.webkitRequestFullscreen).call(canvas);
});
attachInput(canvas, d.i);
return {
canvas: canvas, ctx: canvas.getContext("2d"),
decoder: null, configured: false, sawKey: false, ts: 0,
};
});
}
function attachInput(canvas, index) {
const norm = (e) => {
const r = canvas.getBoundingClientRect();
return {
nx: Math.min(1, Math.max(0, (e.clientX - r.left) / r.width)),
ny: Math.min(1, Math.max(0, (e.clientY - r.top) / r.height)),
};
};
let lastMove = 0;
canvas.addEventListener("mouseenter", () => { hoverDisplay = index; });
canvas.addEventListener("mouseleave", () => {
if (hoverDisplay === index) { hoverDisplay = null; flush(); }
});
canvas.addEventListener("mousemove", (e) => {
const now = performance.now();
if (now - lastMove < 16) return;
lastMove = now;
const p = norm(e);
send({ type: "mouse", event: "move", display: index, nx: p.nx, ny: p.ny });
});
canvas.addEventListener("mousedown", (e) => {
e.preventDefault();
const p = norm(e);
send({ type: "mouse", event: "down", display: index,
button: BUTTON_NAMES[e.button] || "left", nx: p.nx, ny: p.ny });
});
canvas.addEventListener("mouseup", (e) => {
e.preventDefault();
const p = norm(e);
send({ type: "mouse", event: "up", display: index,
button: BUTTON_NAMES[e.button] || "left", nx: p.nx, ny: p.ny });
});
canvas.addEventListener("wheel", (e) => {
e.preventDefault();
send({ type: "mouse", event: "scroll", display: index, dx: e.deltaX, dy: e.deltaY });
}, { passive: false });
canvas.addEventListener("contextmenu", (e) => e.preventDefault());
}
// Map the Windows keyboard to the Mac: the Windows key AND Ctrl both act as
// Cmd, so Ctrl+C / Ctrl+V / Ctrl+A / Ctrl+Z etc. work the way a Windows user
// expects (macOS uses Cmd for those). Alt stays as Option.
function macKey(k) {
if (k === "OS" || k === "Control") return "Meta";
return k;
}
window.addEventListener("keydown", (e) => {
if (!control || hoverDisplay === null) return;
e.preventDefault();
send({ type: "key", event: "down", key: macKey(e.key) });
});
window.addEventListener("keyup", (e) => {
if (!control || hoverDisplay === null) return;
e.preventDefault();
send({ type: "key", event: "up", key: macKey(e.key) });
});
// Never leave a key stuck down if the window loses focus mid-press.
window.addEventListener("blur", () => { hoverDisplay = null; flush(); });
function handleFrame(buffer) {
const bytes = new Uint8Array(buffer);
const idx = bytes[0];
const keyframe = (bytes[1] & 1) === 1;
feed(idx, keyframe, bytes.subarray(2));
}
function connect() {
ws = new WebSocket("ws://127.0.0.1:8001");
ws.binaryType = "arraybuffer";
ws.onmessage = (e) => {
if (typeof e.data === "string") {
const msg = JSON.parse(e.data);
if (msg.type === "status") setStatus(msg.agent);
else if (msg.type === "hello") buildPanels(msg.displays);
} else {
handleFrame(e.data);
}
};
ws.onclose = () => {
setStatus(false);
document.getElementById("offlineText").textContent =
"Server not running - start start.bat, this page reconnects automatically.";
setTimeout(connect, 2000);
};
}
connect();
</script>
</body>
</html>
mac-agent/agent.py
#!/usr/bin/env python3
"""Remote Desktop - Mac agent (v2, hardware H.264).
Runs on the Mac Mini. Dials OUT to the laptop server (so nothing needs to be
opened on the office network), streams every display as a hardware-encoded
H.264 elementary stream, and injects the mouse/keyboard events that come back.
Video path: mss screen capture (CoreGraphics) -> raw frames piped to ffmpeg
-> h264_videotoolbox (Apple hardware encoder) -> Annex-B access
units -> WebSocket -> browser WebCodecs VideoDecoder.
(ffmpeg's own avfoundation screen input is deprecated and does not
deliver frames on modern macOS, so we capture with mss instead and
use ffmpeg only as the encoder.)
Input path: browser JSON -> Quartz/CGEvent injection.
"""
import asyncio
import json
import os
import shutil
import ssl
import subprocess
import threading
import time
from pathlib import Path
import mss
import websockets
import Quartz
from pynput.keyboard import Controller as KeyboardController, Key, KeyCode
BASE = Path(__file__).resolve().parent
CONFIG = json.loads((BASE / "config.json").read_text())
SERVER_URL = CONFIG["server_url"]
TOKEN = CONFIG["token"]
CERT_FILE = CONFIG.get("cert_file", "cert.pem")
FPS = int(CONFIG.get("fps", 30))
BITRATE = str(CONFIG.get("bitrate", "8M"))
SCALE = float(CONFIG.get("scale", 1.0))
RECONNECT_DELAY = CONFIG.get("reconnect_seconds", 5)
# Keep the display awake so screen capture keeps working while nobody is at the
# Mac. Without this the display sleeps after idle and macOS reports 0 active
# displays, so there is nothing to capture. (Physically powered-off monitors
# that fully disconnect still need an HDMI dummy plug - this only prevents the
# software display-sleep.)
KEEP_AWAKE = bool(CONFIG.get("keep_awake", True))
# GOP length in frames - a fresh viewer can start decoding within this many
# frames, so keep it around two seconds.
GOP = int(CONFIG.get("gop_seconds", 2) * FPS)
# Capture backend: "sck" = native ScreenCaptureKit + VideoToolbox helper
# (true 60fps, includes the cursor); "mss" = the pure-Python fallback.
CAPTURE_BACKEND = CONFIG.get("capture_backend", "mss")
CAPTURE_CURSOR = bool(CONFIG.get("capture_cursor", True))
FFMPEG = shutil.which("ffmpeg") or "/opt/homebrew/bin/ffmpeg"
SCREENCAP = str(BASE / "screencap")
if CAPTURE_BACKEND == "sck" and not Path(SCREENCAP).exists():
print("capture_backend 'sck' set but the screencap helper is missing "
"(run install.sh, needs Xcode command line tools) - falling back to mss")
CAPTURE_BACKEND = "mss"
keyboard = KeyboardController()
# ---------------------------------------------------------------------------
# Input injection
# ---------------------------------------------------------------------------
BUTTONS = {
"left": (Quartz.kCGMouseButtonLeft, Quartz.kCGEventLeftMouseDown,
Quartz.kCGEventLeftMouseUp, Quartz.kCGEventLeftMouseDragged),
"right": (Quartz.kCGMouseButtonRight, Quartz.kCGEventRightMouseDown,
Quartz.kCGEventRightMouseUp, Quartz.kCGEventRightMouseDragged),
"middle": (Quartz.kCGMouseButtonCenter, Quartz.kCGEventOtherMouseDown,
Quartz.kCGEventOtherMouseUp, Quartz.kCGEventOtherMouseDragged),
}
DOUBLE_CLICK_SECONDS = 0.4
DOUBLE_CLICK_RADIUS = 6
class Mouse:
def __init__(self):
self.pos = (0.0, 0.0)
self.pressed = set()
self.last_click = {} # button -> (time, x, y, count)
def _post(self, etype, button=Quartz.kCGMouseButtonLeft, click_state=0):
ev = Quartz.CGEventCreateMouseEvent(None, etype, self.pos, button)
if click_state:
Quartz.CGEventSetIntegerValueField(
ev, Quartz.kCGMouseEventClickState, click_state)
Quartz.CGEventPost(Quartz.kCGHIDEventTap, ev)
def move(self, x, y):
self.pos = (x, y)
for name in ("left", "right", "middle"):
if name in self.pressed:
btn, _, _, dragged = BUTTONS[name]
self._post(dragged, btn)
return
self._post(Quartz.kCGEventMouseMoved)
def _click_state(self, name, x, y):
now = time.monotonic()
last = self.last_click.get(name)
count = 1
if last:
t, lx, ly, c = last
if now - t < DOUBLE_CLICK_SECONDS and abs(x - lx) < DOUBLE_CLICK_RADIUS \
and abs(y - ly) < DOUBLE_CLICK_RADIUS:
count = c + 1
self.last_click[name] = (now, x, y, count)
return count
def down(self, name, x, y):
if name not in BUTTONS:
return
self.pos = (x, y)
self.pressed.add(name)
btn, down_ev, _, _ = BUTTONS[name]
self._post(down_ev, btn, self._click_state(name, x, y))
def up(self, name, x, y):
if name not in BUTTONS:
return
self.pos = (x, y)
self.pressed.discard(name)
btn, _, up_ev, _ = BUTTONS[name]
count = self.last_click.get(name, (0, 0, 0, 1))[3]
self._post(up_ev, btn, count)
def release_all(self):
for name in list(self.pressed):
self.up(name, *self.pos)
def scroll(self, dx, dy):
ev = Quartz.CGEventCreateScrollWheelEvent(
None, Quartz.kCGScrollEventUnitPixel, 2, int(-dy), int(-dx))
Quartz.CGEventPost(Quartz.kCGHIDEventTap, ev)
SPECIAL_KEYS = {
"Enter": Key.enter, "Backspace": Key.backspace, "Tab": Key.tab,
"Escape": Key.esc, "Delete": Key.delete, "CapsLock": Key.caps_lock,
"ArrowUp": Key.up, "ArrowDown": Key.down,
"ArrowLeft": Key.left, "ArrowRight": Key.right,
"Home": Key.home, "End": Key.end,
"PageUp": Key.page_up, "PageDown": Key.page_down,
"Shift": Key.shift, "Control": Key.ctrl, "Alt": Key.alt, "Meta": Key.cmd,
" ": Key.space, "Spacebar": Key.space,
}
for _i in range(1, 13):
SPECIAL_KEYS["F%d" % _i] = getattr(Key, "f%d" % _i)
def to_key(name):
if name in SPECIAL_KEYS:
return SPECIAL_KEYS[name]
if len(name) == 1:
return KeyCode.from_char(name.lower() if name.isalpha() else name)
return None
class Keys:
"""Tracks which keys are held so we can release them all on demand
(prevents a modifier sticking down if its key-up never arrives)."""
def __init__(self):
self.held = set()
def down(self, name):
key = to_key(name)
if key is None:
return
self.held.add(name)
try:
keyboard.press(key)
except Exception:
pass
def up(self, name):
key = to_key(name)
if key is None:
return
self.held.discard(name)
try:
keyboard.release(key)
except Exception:
pass
def release_all(self):
for name in list(self.held):
self.up(name)
def handle_input(data, meta, mouse, keys):
kind = data.get("type")
if kind == "mouse":
idx = data.get("display", 0)
if not 0 <= idx < len(meta):
return
d = meta[idx]
event = data.get("event")
if event == "scroll":
mouse.scroll(data.get("dx", 0), data.get("dy", 0))
return
x = d["x"] + float(data.get("nx", 0)) * d["w"]
y = d["y"] + float(data.get("ny", 0)) * d["h"]
if event == "move":
mouse.move(x, y)
elif event == "down":
mouse.down(data.get("button", "left"), x, y)
elif event == "up":
mouse.up(data.get("button", "left"), x, y)
elif kind == "key":
if data.get("event") == "down":
keys.down(data.get("key", ""))
else:
keys.up(data.get("key", ""))
elif kind == "flush":
# Browser lost focus / mouse left the screen - drop everything held so
# nothing stays stuck down on the Mac.
keys.release_all()
mouse.release_all()
# ---------------------------------------------------------------------------
# Video - capture each display with mss, hardware-encode with ffmpeg
# ---------------------------------------------------------------------------
AUD = b"\x00\x00\x01\x09" # start code + Access Unit Delimiter NAL header
def split_access_units(buffer):
"""Split a raw Annex-B buffer on AUD boundaries.
Returns (list_of_complete_access_units, remainder). Each access unit begins
with an AUD NAL and contains the frame's slice(s) plus, on keyframes, the
in-band SPS/PPS that VideoToolbox emits before each IDR.
"""
units = []
start = buffer.find(AUD)
if start == -1:
return units, buffer
while True:
nxt = buffer.find(AUD, start + len(AUD))
if nxt == -1:
return units, buffer[start:]
units.append(buffer[start:nxt])
start = nxt
def is_keyframe(au):
"""True if the access unit carries an IDR (NAL type 5) or SPS (type 7)."""
i = 0
while True:
j = au.find(b"\x00\x00\x01", i)
if j == -1:
return False
nal_type = au[j + 3] & 0x1F if j + 3 < len(au) else 0
if nal_type in (5, 7):
return True
i = j + 3
def _offer(queue, data):
"""Enqueue a frame, dropping the oldest if we're backed up (congestion)."""
if queue.full():
try:
queue.get_nowait()
except asyncio.QueueEmpty:
pass
queue.put_nowait(data)
class DisplayStream:
"""Captures one display with mss (CoreGraphics) and hardware-encodes it to
H.264 with ffmpeg (rawvideo in -> h264_videotoolbox out). Two threads: one
grabs frames and feeds ffmpeg, one reads the encoded access units back and
hands them to the asyncio queue."""
def __init__(self, idx, monitor, loop, queue):
self.idx = idx
self.monitor = monitor # mss monitor dict for this display
self.loop = loop
self.queue = queue
self.stop = threading.Event()
self.proc = None
def start(self):
# Probe one frame to learn the exact captured pixel size.
with mss.MSS() as sct:
shot = sct.grab(self.monitor)
w, h = shot.width, shot.height
args = [FFMPEG, "-hide_banner", "-loglevel", "error", "-nostdin",
"-f", "rawvideo", "-pix_fmt", "bgra",
"-s", "%dx%d" % (w, h), "-r", str(FPS), "-i", "pipe:0"]
if SCALE != 1.0:
args += ["-vf",
"scale=trunc(iw*%g/2)*2:trunc(ih*%g/2)*2" % (SCALE, SCALE)]
args += ["-c:v", "h264_videotoolbox", "-realtime", "true",
"-profile:v", "high",
"-b:v", BITRATE, "-maxrate", BITRATE, "-bufsize", BITRATE,
"-g", str(GOP), "-bf", "0", "-pix_fmt", "yuv420p",
"-bsf:v", "h264_metadata=aud=insert", "-f", "h264", "pipe:1"]
self.proc = subprocess.Popen(
args, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL, bufsize=0)
threading.Thread(target=self._feed, daemon=True).start()
threading.Thread(target=self._drain, daemon=True).start()
def _feed(self):
period = 1.0 / max(1, FPS)
try:
with mss.MSS() as sct:
while not self.stop.is_set():
t0 = time.monotonic()
shot = sct.grab(self.monitor)
try:
self.proc.stdin.write(shot.raw)
except (BrokenPipeError, ValueError, OSError):
break
# Pace to the target fps with short (<=4ms) sleeps and re-check
# the clock. A single long sleep gets descheduled under thread
# contention and overshoots the interval, which is what capped
# the real frame rate well below the target.
deadline = t0 + period
while not self.stop.is_set():
slack = deadline - time.monotonic()
if slack <= 0.0005:
break
time.sleep(min(slack, 0.004))
except Exception:
pass
finally:
try:
self.proc.stdin.close()
except Exception:
pass
def _drain(self):
buf = b""
out = self.proc.stdout
while not self.stop.is_set():
chunk = out.read(65536)
if not chunk:
break
buf += chunk
units, buf = split_access_units(buf)
for au in units:
flags = 1 if is_keyframe(au) else 0
data = bytes([self.idx, flags]) + au
self.loop.call_soon_threadsafe(_offer, self.queue, data)
def close(self):
self.stop.set()
if self.proc and self.proc.poll() is None:
try:
self.proc.terminate()
except Exception:
pass
def bitrate_kbps():
s = str(BITRATE).strip().upper()
if s.endswith("M"):
return int(float(s[:-1]) * 1000)
if s.endswith("K"):
return int(float(s[:-1]))
return max(1, int(float(s) / 1000)) # assume bits/sec
def cg_displays():
"""Active displays as {id, x, y, w, h} in points (top-left origin)."""
err, ids, n = Quartz.CGGetActiveDisplayList(16, None, None)
out = []
for did in list(ids)[:n]:
b = Quartz.CGDisplayBounds(did)
out.append({"id": int(did),
"x": float(b.origin.x), "y": float(b.origin.y),
"w": float(b.size.width), "h": float(b.size.height)})
return out
class SCKStream:
"""Native capture backend: one `screencap` (ScreenCaptureKit +
VideoToolbox) process per display, already emitting Annex-B to stdout."""
def __init__(self, idx, display_id, loop, queue):
self.idx = idx
self.display_id = display_id
self.loop = loop
self.queue = queue
self.stop = threading.Event()
self.proc = None
def start(self):
threading.Thread(target=self._run, daemon=True).start()
def _run(self):
# Supervisor: (re)spawn the helper and pump its output. ScreenCaptureKit
# occasionally drops a stream with -3805 during startup, so just restart.
while not self.stop.is_set():
args = [SCREENCAP, str(self.display_id), str(FPS), str(bitrate_kbps()),
"1" if CAPTURE_CURSOR else "0", str(SCALE)]
self.proc = subprocess.Popen(
args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=0)
threading.Thread(target=self._logerr, args=(self.proc,), daemon=True).start()
self._drain(self.proc) # blocks until the helper exits
if self.proc.poll() is None:
try:
self.proc.terminate()
except Exception:
pass
if not self.stop.is_set():
time.sleep(0.5) # brief backoff before restarting
def _drain(self, proc):
buf = b""
out = proc.stdout
while not self.stop.is_set():
chunk = out.read(65536)
if not chunk:
break
buf += chunk
units, buf = split_access_units(buf)
for au in units:
flags = 1 if is_keyframe(au) else 0
data = bytes([self.idx, flags]) + au
self.loop.call_soon_threadsafe(_offer, self.queue, data)
def _logerr(self, proc):
while not self.stop.is_set():
line = proc.stderr.readline()
if not line:
break
text = line.decode("utf-8", "replace").strip()
if text:
print("screencap[%d]: %s" % (self.idx, text))
def close(self):
self.stop.set()
if self.proc and self.proc.poll() is None:
try:
self.proc.terminate()
except Exception:
pass
# ---------------------------------------------------------------------------
# Session - connect out, authenticate, stream, obey
# ---------------------------------------------------------------------------
async def run_session():
ssl_ctx = ssl.create_default_context(cafile=str(BASE / CERT_FILE))
ssl_ctx.check_hostname = False # we pin the exact certificate instead
async with websockets.connect(
SERVER_URL, ssl=ssl_ctx, max_size=None,
ping_interval=20, ping_timeout=20) as ws:
await ws.send(json.dumps({"type": "auth", "role": "agent", "token": TOKEN}))
if CAPTURE_BACKEND == "sck":
disps = cg_displays()
meta = [{"x": d["x"], "y": d["y"], "w": d["w"], "h": d["h"]}
for d in disps]
else:
with mss.MSS() as sct:
monitors = [dict(m) for m in sct.monitors[1:]]
meta = [{"x": float(m["left"]), "y": float(m["top"]),
"w": float(m["width"]), "h": float(m["height"])}
for m in monitors]
n = len(meta)
await ws.send(json.dumps({
"type": "hello",
"codec": "h264",
"displays": [{"i": i, "w": int(m["w"]), "h": int(m["h"])}
for i, m in enumerate(meta)],
}))
print("Connected - streaming %d display(s) as H.264 (%s)"
% (n, CAPTURE_BACKEND))
queue = asyncio.Queue(maxsize=FPS) # ~1s of frames before we drop
loop = asyncio.get_running_loop()
if CAPTURE_BACKEND == "sck":
streams = [SCKStream(i, disps[i]["id"], loop, queue) for i in range(n)]
else:
streams = [DisplayStream(i, monitors[i], loop, queue) for i in range(n)]
for i, s in enumerate(streams):
s.start()
if CAPTURE_BACKEND == "sck" and i < len(streams) - 1:
await asyncio.sleep(0.8) # stagger SCK stream startup
mouse, keys = Mouse(), Keys()
async def sender():
while True:
await ws.send(await queue.get())
send_task = asyncio.create_task(sender())
try:
async for msg in ws:
if isinstance(msg, str):
try:
data = json.loads(msg)
except ValueError:
continue
handle_input(data, meta, mouse, keys)
finally:
for s in streams:
s.close()
keys.release_all()
mouse.release_all()
send_task.cancel()
def keep_display_awake():
"""Prevent display sleep for the life of this process, and wake the display
now if it is already asleep, so mss always sees an active display."""
try:
# -d prevents display sleep; -w ties caffeinate's life to ours.
subprocess.Popen(["caffeinate", "-d", "-w", str(os.getpid())])
# -u -t declares brief user activity, which wakes a slept display.
subprocess.run(["caffeinate", "-u", "-t", "2"])
except Exception:
pass
async def main():
if not Path(FFMPEG).exists() and shutil.which("ffmpeg") is None:
print("ffmpeg not found - install it with: brew install ffmpeg")
return
if KEEP_AWAKE:
keep_display_awake()
await asyncio.sleep(1.5) # let a just-woken display become active
while True:
try:
print("Connecting to", SERVER_URL)
await run_session()
print("Connection closed")
except Exception as exc:
print("Connection failed:", exc)
await asyncio.sleep(RECONNECT_DELAY)
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
pass
mac-agent/screencap.swift
// Remote Desktop - native screen capturer.
//
// Captures one display with ScreenCaptureKit and hardware-encodes it to H.264
// with VideoToolbox, writing an Annex-B elementary stream (AUD-delimited, with
// in-band SPS/PPS on keyframes) to stdout - the same format the Python agent
// already parses. One process per display.
//
// Usage: screencap <displayID> <fps> <bitrate-kbps> <showCursor 0|1> [scale]
//
// Build: swiftc -O screencap.swift -o screencap
import Foundation
import ScreenCaptureKit
import VideoToolbox
import CoreMedia
// ---- args ---------------------------------------------------------------
let a = CommandLine.arguments
guard a.count >= 5,
let displayID = UInt32(a[1]), let fps = Int(a[2]),
let kbps = Int(a[3]), let cursor = Int(a[4]) else {
FileHandle.standardError.write("usage: screencap <displayID> <fps> <kbps> <cursor> [scale]\n".data(using: .utf8)!)
exit(2)
}
let scale = a.count >= 6 ? (Double(a[5]) ?? 1.0) : 1.0
let gopSeconds = 2.0
func err(_ s: String) { FileHandle.standardError.write((s + "\n").data(using: .utf8)!) }
// ---- Annex-B writing ----------------------------------------------------
let stdoutHandle = FileHandle.standardOutput
let writeLock = NSLock()
let startCode = Data([0, 0, 0, 1])
let aud = Data([0, 0, 0, 1, 0x09, 0xF0]) // Access Unit Delimiter
func writeAnnexB(_ sbuf: CMSampleBuffer) {
guard let block = CMSampleBufferGetDataBuffer(sbuf) else { return }
var isKey = true
if let att = CMSampleBufferGetSampleAttachmentsArray(sbuf, createIfNecessary: false)
as? [[CFString: Any]], let f = att.first,
let notSync = f[kCMSampleAttachmentKey_NotSync] as? Bool {
isKey = !notSync
}
var out = Data()
out.append(aud)
if isKey, let fmt = CMSampleBufferGetFormatDescription(sbuf) {
var count = 0
CMVideoFormatDescriptionGetH264ParameterSetAtIndex(
fmt, parameterSetIndex: 0, parameterSetPointerOut: nil,
parameterSetSizeOut: nil, parameterSetCountOut: &count, nalUnitHeaderLengthOut: nil)
for i in 0..<count {
var ptr: UnsafePointer<UInt8>?
var size = 0
CMVideoFormatDescriptionGetH264ParameterSetAtIndex(
fmt, parameterSetIndex: i, parameterSetPointerOut: &ptr,
parameterSetSizeOut: &size, parameterSetCountOut: nil, nalUnitHeaderLengthOut: nil)
if let p = ptr {
out.append(startCode)
out.append(p, count: size)
}
}
}
var total = 0
var dp: UnsafeMutablePointer<Int8>?
CMBlockBufferGetDataPointer(block, atOffset: 0, lengthAtOffsetOut: nil,
totalLengthOut: &total, dataPointerOut: &dp)
if let base = dp {
let bytes = UnsafeRawPointer(base).assumingMemoryBound(to: UInt8.self)
var off = 0
while off + 4 <= total {
let n = (Int(bytes[off]) << 24) | (Int(bytes[off+1]) << 16)
| (Int(bytes[off+2]) << 8) | Int(bytes[off+3])
off += 4
if n <= 0 || off + n > total { break }
out.append(startCode)
out.append(UnsafeBufferPointer(start: bytes + off, count: n))
off += n
}
}
writeLock.lock()
stdoutHandle.write(out)
writeLock.unlock()
}
// ---- VideoToolbox encoder ----------------------------------------------
var session: VTCompressionSession?
func makeEncoder(_ w: Int, _ h: Int, fps: Int, kbps: Int) {
VTCompressionSessionCreate(
allocator: nil, width: Int32(w), height: Int32(h),
codecType: kCMVideoCodecType_H264, encoderSpecification: nil,
imageBufferAttributes: nil, compressedDataAllocator: nil,
outputCallback: nil, refcon: nil, compressionSessionOut: &session)
guard let s = session else { err("VTCompressionSessionCreate failed"); exit(1) }
VTSessionSetProperty(s, key: kVTCompressionPropertyKey_RealTime, value: kCFBooleanTrue)
VTSessionSetProperty(s, key: kVTCompressionPropertyKey_ProfileLevel, value: kVTProfileLevel_H264_High_AutoLevel)
VTSessionSetProperty(s, key: kVTCompressionPropertyKey_AllowFrameReordering, value: kCFBooleanFalse)
// Emit each frame as soon as it is encoded - no lookahead buffering (latency).
VTSessionSetProperty(s, key: kVTCompressionPropertyKey_MaxFrameDelayCount, value: 0 as CFNumber)
VTSessionSetProperty(s, key: kVTCompressionPropertyKey_MaximizePowerEfficiency, value: kCFBooleanFalse)
VTSessionSetProperty(s, key: kVTCompressionPropertyKey_AverageBitRate, value: (kbps * 1000) as CFNumber)
VTSessionSetProperty(s, key: kVTCompressionPropertyKey_MaxKeyFrameInterval, value: Int(Double(fps) * gopSeconds) as CFNumber)
VTSessionSetProperty(s, key: kVTCompressionPropertyKey_MaxKeyFrameIntervalDuration, value: gopSeconds as CFNumber)
VTCompressionSessionPrepareToEncodeFrames(s)
}
// ---- ScreenCaptureKit ---------------------------------------------------
final class Capturer: NSObject, SCStreamOutput, SCStreamDelegate {
let displayID: UInt32
let fps: Int
let kbps: Int
let cursor: Bool
let scale: Double
var stream: SCStream?
// Decouple capture from encode: ScreenCaptureKit only delivers a frame when
// the screen changes, which makes sparse updates (like typing) feel laggy.
// We keep the most recent frame and encode at a *constant* rate on a timer,
// exactly like a fixed-fps capture - smooth and low-latency, plus the cursor.
let frameLock = NSLock()
var latest: CVImageBuffer?
var counter: Int64 = 0
var timer: DispatchSourceTimer?
init(displayID: UInt32, fps: Int, kbps: Int, cursor: Bool, scale: Double) {
self.displayID = displayID; self.fps = fps; self.kbps = kbps
self.cursor = cursor; self.scale = scale
super.init()
}
func start() async throws {
let content = try await SCShareableContent.excludingDesktopWindows(false, onScreenWindowsOnly: false)
guard let display = content.displays.first(where: { $0.displayID == self.displayID })
?? content.displays.first else {
err("no matching display for id \(self.displayID)"); exit(1)
}
let w = Int(Double(display.width) * self.scale)
let h = Int(Double(display.height) * self.scale)
makeEncoder(w, h, fps: self.fps, kbps: self.kbps)
let filter = SCContentFilter(display: display, excludingWindows: [])
let cfg = SCStreamConfiguration()
cfg.width = w
cfg.height = h
cfg.minimumFrameInterval = CMTime(value: 1, timescale: Int32(self.fps))
cfg.pixelFormat = kCVPixelFormatType_32BGRA
cfg.queueDepth = 5
cfg.showsCursor = self.cursor
let s = SCStream(filter: filter, configuration: cfg, delegate: self)
try s.addStreamOutput(self, type: .screen, sampleHandlerQueue: DispatchQueue(label: "capture"))
try await s.startCapture()
stream = s
let t = DispatchSource.makeTimerSource(queue: DispatchQueue(label: "encode"))
t.schedule(deadline: .now(), repeating: 1.0 / Double(self.fps), leeway: .milliseconds(1))
t.setEventHandler { [weak self] in self?.tick() }
t.resume()
timer = t
err("capturing display \(self.displayID) at \(w)x\(h)@\(self.fps)")
}
func stream(_ stream: SCStream, didOutputSampleBuffer sampleBuffer: CMSampleBuffer,
of type: SCStreamOutputType) {
guard type == .screen, CMSampleBufferIsValid(sampleBuffer),
let px = CMSampleBufferGetImageBuffer(sampleBuffer) else { return }
frameLock.lock(); latest = px; frameLock.unlock()
}
func tick() {
frameLock.lock(); let px = latest; frameLock.unlock()
guard let px = px, let s = session else { return }
let pts = CMTime(value: counter, timescale: Int32(self.fps))
counter += 1
VTCompressionSessionEncodeFrame(
s, imageBuffer: px, presentationTimeStamp: pts, duration: .invalid,
frameProperties: nil, infoFlagsOut: nil) { status, _, sbuf in
if status == noErr, let sbuf = sbuf { writeAnnexB(sbuf) }
}
}
func stream(_ stream: SCStream, didStopWithError error: Error) {
err("stream stopped: \(error)"); exit(1)
}
}
// exit cleanly when our stdout reader (the Python agent) goes away
signal(SIGPIPE) { _ in exit(0) }
signal(SIGTERM) { _ in exit(0) }
let cap = Capturer(displayID: displayID, fps: fps, kbps: kbps, cursor: cursor != 0, scale: scale)
Task {
do { try await cap.start() }
catch { err("start failed: \(error)"); exit(1) }
}
dispatchMain()
mac-agent/config.example.json
{
"server_url": "wss://YOUR-HOME-IP:8443",
"token": "PASTE-TOKEN-FROM-LAPTOP-SERVER-CONFIG",
"cert_file": "cert.pem",
"capture_backend": "sck",
"fps": 60,
"bitrate": "8M",
"scale": 1.0,
"gop_seconds": 2,
"capture_cursor": true,
"reconnect_seconds": 5,
"keep_awake": true
}
mac-agent/requirements.txt
websockets>=12
pynput>=1.7
pyobjc-framework-Quartz>=10
mss>=9
mac-agent/install.sh
#!/bin/bash
# Remote Desktop - Mac agent installer.
# Creates a private Python environment and registers the agent to start at login.
set -e
cd "$(dirname "$0")"
DIR="$(pwd)"
echo "== Remote Desktop agent installer =="
if [ ! -f config.json ]; then
cp config.example.json config.json
echo ""
echo "Created config.json - EDIT IT FIRST:"
echo " - server_url : address of your laptop (wss://...:8443)"
echo " - token : from laptop-server/config.json"
echo "Then run this installer again."
exit 1
fi
if [ ! -f cert.pem ]; then
echo "cert.pem is missing - copy it from the laptop's laptop-server folder first."
exit 1
fi
if ! command -v ffmpeg >/dev/null 2>&1; then
echo "ffmpeg is required for hardware H.264 capture but was not found."
if command -v brew >/dev/null 2>&1; then
echo "Installing it now with Homebrew..."
brew install ffmpeg
else
echo "Install Homebrew (https://brew.sh) then run: brew install ffmpeg"
exit 1
fi
fi
# Build the native ScreenCaptureKit capture helper (true 60fps + cursor).
# Falls back to the mss backend automatically if this can't be built.
if command -v swiftc >/dev/null 2>&1; then
echo "Building native capture helper (screencap)..."
if swiftc -O screencap.swift -o screencap 2>/dev/null; then
echo " built screencap."
else
echo " could not build screencap - the agent will use the mss backend."
fi
else
echo "swiftc not found (install Xcode command line tools for the native"
echo "60fps backend) - the agent will use the mss backend."
fi
echo "Creating Python environment..."
# Use a real Python binary (prefer Homebrew) and --copies, so venv/bin/python3
# is a standalone binary you can grant Screen Recording / Accessibility to.
# A symlinked interpreter (e.g. the Xcode python3) can't be selected in the
# macOS permission picker.
PYBIN="$(ls /opt/homebrew/bin/python3.1? 2>/dev/null | sort -V | tail -1)"
[ -z "$PYBIN" ] && PYBIN="$(command -v python3)"
echo " using $PYBIN"
"$PYBIN" -m venv --copies venv
./venv/bin/pip install --quiet --upgrade pip
./venv/bin/pip install --quiet -r requirements.txt
echo "Registering start-at-login..."
PLIST="$HOME/Library/LaunchAgents/com.remotedesktop.agent.plist"
mkdir -p "$HOME/Library/LaunchAgents"
sed "s|__DIR__|$DIR|g" com.remotedesktop.agent.plist > "$PLIST"
launchctl bootout "gui/$(id -u)/com.remotedesktop.agent" 2>/dev/null || true
launchctl bootstrap "gui/$(id -u)" "$PLIST"
echo ""
echo "Done. The agent is now running and will start automatically at login."
echo ""
echo "IMPORTANT - grant BOTH permissions to python3 once, in"
echo "System Settings > Privacy & Security:"
echo " 1. Screen Recording -> enable python3 (captures the screen)"
echo " 2. Accessibility -> enable python3 (injects mouse + keyboard)"
echo " If python3 is not listed, click '+' and add this exact binary:"
echo " $DIR/venv/bin/python3"
echo "Then restart the agent with:"
echo " launchctl kickstart -k gui/\$(id -u)/com.remotedesktop.agent"
echo ""
echo "Log file: $DIR/agent.log"
mac-agent/com.remotedesktop.agent.plist
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
"http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
<key>Label</key>
<string>com.remotedesktop.agent</string>
<key>ProgramArguments</key>
<array>
<string>__DIR__/venv/bin/python3</string>
<string>-u</string>
<string>__DIR__/agent.py</string>
</array>
<key>WorkingDirectory</key>
<string>__DIR__</string>
<key>RunAtLoad</key>
<true/>
<key>KeepAlive</key>
<true/>
<key>StandardOutPath</key>
<string>__DIR__/agent.log</string>
<key>StandardErrorPath</key>
<string>__DIR__/agent.log</string>
</dict>
</plist>
Shared by Studio Barbanson Data Solutions. Build it, use it, make it yours.