Automation recipes

These recipes turn a Rebind script into a control surface that any program can drive — a web app, a Python process, a home-automation hub, or a local model. The script runs Luau on the Rust core, hooks dispatched up to 8,000 Hz, and emits standard USB HID from a Teensy 4.x (or OS-level input in software mode).

The single-endpoint Net.Listen pattern — HTTP request in, output out — appears in the Quickstart section of the SDK reference; these recipes build on it. They lean on the localhost transport numbers documented in Remote control.

A WebSocket remote

Net.WSListen is the same idea over a persistent connection. Use it when you need push events (server to client) or a low-latency channel for high-frequency commands. The server handles multiple clients, broadcasts to all of them, and reacts to each message within a single tick.

--[[
  rebind: min_sdk=3.0.0
  rebind: name=WebSocket Remote
  rebind: permission=net
--]]

local cfg = UI.Schema({
  port = UI.Slider(9000, { min = 1024, max = 65535, label = "WS port" }),
})

local server = nil

function OnStart()
  server = Net.WSListen(cfg.port, {
    OnConnect = function(client)
      Log.Info(
        string.format(
          "client %d connected (%d total)",
          client.id,
          server:ClientCount()
        )
      )
      client:Send("welcome " .. client.id)
    end,

    OnMessage = function(client, payload, is_binary)
      local req = JSON.Parse(payload)
      if req.t == "type" then
        Run(function()
          HID.Type(req.text or "")
        end)
        client:Send(JSON.Stringify({ id = req.id, ok = true }))
      elseif req.t == "move" then
        HID.Move(req.dx or 0, req.dy or 0)
      elseif req.t == "broadcast" then
        server:Broadcast(payload)
      end
    end,

    OnClose = function(client)
      Log.Info(string.format("client %d disconnected", client.id))
    end,
  })
  Log.Info("WebSocket remote on ws://0.0.0.0:" .. cfg.port)
end

function OnStop()
  if server then
    server:Stop()
  end
end

Connect from Python with websocket-client:

import json
from websocket import create_connection

ws = create_connection("ws://127.0.0.1:9000")
print(ws.recv())  # "welcome 1"
ws.send(json.dumps({"t": "type", "text": "hello from python"}))
ws.send(json.dumps({"t": "move", "dx": 50, "dy": 0}))

Net.WSListen can expose arbitrary JSON-RPC surfaces — reads (screen pixels, window state, clipboard, input state), writes (HID commands), and push subscriptions (mouse position, window changes, input events) compose with the same handler. For the official typed, auto-reconnecting clients, see Remote control.

A screen-condition trigger

Screen.GetPixelColor samples a pixel; Screen.SearchForColor scans a region. Polling them on a tick gives you a desktop-automation trigger: act when a known visual state appears. This recipe watches a fixed point on screen and fires a keystroke when a build indicator turns red — the kind of thing you’d wire into a CI dashboard, a long-running render, or an accessibility cue.

Screen and Window are limited or absent on Linux; see Platforms & limits.

--[[
  rebind: min_sdk=3.0.0
  rebind: name=Build Status Watcher
--]]

local cfg = UI.Schema({
  enabled = UI.Toggle(true, { label = "Enable" }),
  x = UI.Slider(40, { min = 0, max = 4000, label = "Sample X" }),
  y = UI.Slider(40, { min = 0, max = 4000, label = "Sample Y" }),
  target = UI.Text("E53935", { maxLength = 6, label = "Alert Color (hex)" }),
  tolerance = UI.Slider(24, { min = 0, max = 100, label = "Tolerance" }),
})

local fired = false

local function hexToRgb(hex)
  return tonumber(hex:sub(1, 2), 16),
    tonumber(hex:sub(3, 4), 16),
    tonumber(hex:sub(5, 6), 16)
end

local function distance(a, b)
  local r1, g1, b1 = hexToRgb(a)
  local r2, g2, b2 = hexToRgb(b)
  return math.sqrt((r2 - r1) ^ 2 + (g2 - g1) ^ 2 + (b2 - b1) ^ 2)
end

function OnTick()
  if not cfg.enabled then
    return
  end

  local pixel = Screen.GetPixelColor(cfg.x, cfg.y)
  local matched = distance(pixel, cfg.target) <= cfg.tolerance

  -- edge-triggered: act once when the state appears, reset when it clears
  if matched and not fired then
    fired = true
    UI.Notify("Build status: alert color is showing", "warning")
    Run(function()
      HID.Press("F5")
    end) -- refresh the dashboard
  elseif not matched then
    fired = false
  end
end

Use Screen.SearchForColor instead of a fixed point when the indicator can move:

function OnTick()
  -- scan the top-left quadrant for the alert color
  -- region is {x1, y1, x2, y2}; tolerance is 0.0-1.0 (per-channel)
  local hit = Screen.SearchForColor({ 0, 0, 600, 400 }, cfg.target, 0.05)
  if hit then
    Log.Info(string.format("found at %d,%d", hit.x, hit.y))
  end
end

A timer-driven task

Timer.Every runs a callback on a fixed interval without blocking the input loop. This recipe nudges the cursor a few pixels every few minutes to keep a workstation from idling during a long unattended job, with a hotkey to toggle it.

--[[
  rebind: min_sdk=3.0.0
  rebind: name=Idle Nudge
--]]
local cfg = UI.Schema({
  interval = UI.Slider(
    180,
    { min = 30, max = 900, suffix = "s", label = "Interval" }
  ),
  toggle = UI.Keybind("F7", { label = "Toggle" }),
})

local active = false
local timer = nil

function OnDown(key)
  if key ~= cfg.toggle then
    return true
  end

  active = not active
  if active then
    timer = Timer.Every(cfg.interval * 1000, function()
      Run(function()
        HID.Move(1, 0)
        Sleep(50)
        HID.Move(-1, 0)
      end)
    end)
    UI.Notify("Idle nudge ON", "info")
  else
    if timer then
      timer:Cancel()
      timer = nil
    end
    UI.Notify("Idle nudge OFF", "info")
  end
  return false
end

function OnStop()
  if timer then
    timer:Cancel()
  end
end

A sidecar over shared-memory IPC

Keep the perception and decision loop in your own process, with your model and your stack. When the sidecar decides what to do, it hands the decision to Rebind over Pipe (shared-memory IPC, Windows-only) and the script emits the movement as standard USB HID. The sidecar never touches the hardware; the script does, deterministically, at up to 8,000 Hz.

Pipe is the low-latency path for a co-located process. For anything on the network or off-Windows, use the HTTP or WebSocket recipes above.

--[[
  rebind: min_sdk=3.0.0
  rebind: name=Sidecar Bridge
--]]

local pipe = nil

function OnStart()
  pipe = Pipe.Open("sidecar", { size = 65536 })
  Log.Info("Pipe opened: " .. pipe.name)

  Timer.Every(5, function() -- poll the pipe every 5ms
    local msg = pipe:Read()
    if not msg then
      return
    end

    local cmd = JSON.Parse(msg)
    if cmd.action == "move" then
      HID.Move(cmd.dx or 0, cmd.dy or 0)
    elseif cmd.action == "press" and cmd.key then
      Run(function()
        HID.Press(cmd.key)
      end)
    elseif cmd.action == "type" and cmd.text then
      Run(function()
        HID.Type(cmd.text, 30)
      end)
    end
  end)
end

function OnStop()
  if pipe then
    pipe:Close()
  end
end

The external process maps the same shared-memory region. The region is split into two channels: the script writes to channel A (offset 0) and reads from channel B (offset size / 2), so the peer does the inverse — it writes into channel B and reads from channel A. Each channel is framed as [u64 seq][u32 len][payload] (little-endian); the writer lays down the payload and length first and bumps the sequence number last, so the reader only acts on a complete frame. In Python:

import mmap
import json
import struct

SIZE = 65536
HALF = SIZE // 2  # channel B (the script's inbox) begins here

shm = mmap.mmap(-1, SIZE, tagname="Local\\Rebind_sidecar")
seq = 0


def send(payload: dict) -> None:
    # Frame the message the way pipe:Read() expects and write it into channel B.
    # Payload + length first, sequence number last (the read barrier).
    global seq
    data = json.dumps(payload).encode()
    seq += 1
    shm[HALF + 12 : HALF + 12 + len(data)] = data
    shm[HALF + 8 : HALF + 12] = struct.pack("<I", len(data))
    shm[HALF : HALF + 8] = struct.pack("<Q", seq)


# Each channel is a single slot — a new write overwrites an unread one — so pace
# your writes to the script's poll interval rather than bursting.
send({"action": "move", "dx": 12, "dy": -4})
send({"action": "press", "key": "Mouse1"})