Automation Recipes

These recipes expose a Rebind script to other programs: a WebSocket remote, a screen-condition watcher, a timer-driven task, and a shared-memory sidecar bridge. Their transport and queue behavior is documented in Remote control.

A WebSocket remote

Net.WSListen exposes a script 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 and broadcasts to all of them. It dispatches up to 32 queued events per tick; excess messages wait for later ticks.

--[[
  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 client libraries, 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. This is 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. A region up to 200k physical pixels, about 220x220 on a 2x display, scans inside one tick; scan the area the indicator can actually reach, not the quadrant:

function OnTick()
  -- region is {x1, y1, x2, y2}; tolerance is 0.0-1.0 (per-channel)
  local hit = Screen.SearchForColor({ 40, 40, 240, 240 }, cfg.target, 0.05)
  if hit then
    Log.Info(string.format("found at %d,%d", hit.x, hit.y))
  end
end

A larger region yields on a worker thread, so call it from Run():

Run(function()
  while true do
    local hit = Screen.SearchForColor({ 0, 0, 1200, 800 }, cfg.target, 0.05)
    if hit then Log.Info(`found at {hit.x},{hit.y}`) end
    Sleep(100)
  end
end)

When the indicator is an icon rather than a flat color, search for a PNG of it with Screen.FindImage. It captures the display live, so poll it from a coroutine, where it yields instead of blocking the tick:

function OnStart()
  Run(function()
    while true do
      -- alert-icon.png sits next to the script; transparent pixels match anything
      local hit = Screen.FindImage("alert-icon.png", { tolerance = 0.05 })
      if hit then
        Log.Info(string.format("icon at %d,%d", hit.x, hit.y))
      end
      Sleep(500)
    end
  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: through the OS input APIs in software mode, or as USB HID reports with a Rebind Link. The sidecar never emits input; the script does. Pipe is the low-latency path for a co-located process. For anything on the network or off-Windows, use the WebSocket recipe 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.Typewriter(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 channel layout and framing are specified in the Pipe wire protocol. In Python:

import mmap
import json
import struct
import time

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})
time.sleep(0.005)  # the script polls every 5 ms
send({"action": "press", "key": "Mouse1"})

Getting Started

Concepts

Cookbook

Marketplace

Reference

Remote Control

Help

Automation Recipes

A WebSocket remote

A screen-condition trigger

A timer-driven task

A sidecar over shared-memory IPC

​A WebSocket remote

​A screen-condition trigger

​A timer-driven task

​A sidecar over shared-memory IPC

A WebSocket remote

A screen-condition trigger

A timer-driven task

A sidecar over shared-memory IPC