RRebind

05-03-26

Remote Input Monitoring with Lua and Python

This guide shows how to monitor input states on a remote computer using a Lua script on Rebind hardware and a Python client. This allows for real-time input monitoring and conditional automation.

rebindlua scriptpythonremote inputinput monitoringhardware inputscript

Question

Control a device with Rebind using Python from another computer.

Rebind AI

Remote Control Device with Python

This example shows how to control a Rebind device from Python on another computer using WebSocket. The Rebind script runs a WebSocket server, and Python connects to it and sends commands.

Rebind Script

luau
-- rebind: min_sdk=0.1.0
-- rebind: name=Python Remote Control
-- rebind: permission=net

local cfg = UI.Schema({
    port = UI.Slider(9000, { min = 1024, max = 65535, label = "WebSocket Port" }),
    password = UI.Text("rebind123", { label = "Password", maxLength = 50 }),
})

local server = nil
local authenticated = {}

function OnStart()
    server = Net.WSListen(cfg.port, {
        OnConnect = function(client)
            authenticated[client.id] = false
            Log.Info("Client " .. client.id .. " connected (unauthenticated)")
            client:Send(JSON.Stringify({ type = "auth_required" }))
        end,

        OnMessage = function(client, payload, is_binary)
            local msg = JSON.Parse(payload)

            -- authenticate first
            if not authenticated[client.id] then
                if msg.type == "auth" and msg.password == cfg.password then
                    authenticated[client.id] = true
                    client:Send(JSON.Stringify({ type = "auth_ok" }))
                    Log.Info("Client " .. client.id .. " authenticated")
                    return
                else
                    client:Send(JSON.Stringify({ type = "auth_failed" }))
                    return
                end
            end

            -- process commands
            if msg.type == "press" then
                Run(function()
                    HID.Press(msg.key or "Space")
                end)
                client:Send(JSON.Stringify({ type = "ok", action = "press", key = msg.key }))

            elseif msg.type == "type" then
                Run(function()
                    HID.Type(msg.text or "", msg.delay or 30)
                end)
                client:Send(JSON.Stringify({ type = "ok", action = "type", text = msg.text }))

            elseif msg.type == "move" then
                HID.Move(msg.dx or 0, msg.dy or 0)
                client:Send(JSON.Stringify({ type = "ok", action = "move", dx = msg.dx, dy = msg.dy }))

            elseif msg.type == "scroll" then
                HID.Scroll(msg.delta or 1)
                client:Send(JSON.Stringify({ type = "ok", action = "scroll", delta = msg.delta }))

            elseif msg.type == "down" then
                HID.Down(msg.key or "Space")
                client:Send(JSON.Stringify({ type = "ok", action = "down", key = msg.key }))

            elseif msg.type == "up" then
                HID.Up(msg.key or "Space")
                client:Send(JSON.Stringify({ type = "ok", action = "up", key = msg.key }))

            elseif msg.type == "mouse_pos" then
                local mx, my = System.Mouse()
                client:Send(JSON.Stringify({ type = "mouse_pos", x = mx, y = my }))

            elseif msg.type == "screenshot_color" then
                local color = Screen.GetPixelColor(msg.x, msg.y)
                client:Send(JSON.Stringify({ type = "color", hex = color, x = msg.x, y = msg.y }))

            else
                client:Send(JSON.Stringify({ type = "error", message = "Unknown command" }))
            end
        end,

        OnClose = function(client)
            authenticated[client.id] = nil
            Log.Info("Client " .. client.id .. " disconnected")
        end,
    })
    Log.Info("WebSocket remote listening on ws://0.0.0.0:" .. cfg.port)
    UI.Notify("Remote control active on port " .. cfg.port, "success")
end

function OnStop()
    if server then
        server:Stop()
        Log.Info("Remote control server stopped")
    end
end

Python Client

Install dependencies:

bash
pip install websocket-client

Save as rebind_remote.py:

python
import json
import time
import websocket
import sys

class RebindRemote:
    def __init__(self, host="127.0.0.1", port=9000, password="rebind123"):
        self.url = f"ws://{host}:{port}"
        self.password = password
        self.ws = None
        self.authenticated = False

    def connect(self):
        """Connect to the Rebind WebSocket server"""
        print(f"Connecting to {self.url}...")
        self.ws = websocket.WebSocketApp(
            self.url,
            on_open=self._on_open,
            on_message=self._on_message,
            on_error=self._on_error,
            on_close=self._on_close,
        )
        self.ws.run_forever()

    def _on_open(self, ws):
        print("Connected to Rebind device")

    def _on_message(self, ws, message):
        msg = json.loads(message)
        msg_type = msg.get("type")

        if msg_type == "auth_required":
            print("Authentication required. Sending password...")
            self._send({"type": "auth", "password": self.password})

        elif msg_type == "auth_ok":
            print("✓ Authenticated successfully")
            self.authenticated = True

        elif msg_type == "auth_failed":
            print("✗ Authentication failed")
            self.authenticated = False

        elif msg_type == "ok":
            action = msg.get("action", "unknown")
            print(f"✓ {action} completed")

        elif msg_type == "error":
            print(f"✗ Error: {msg.get('message')}")

        elif msg_type == "mouse_pos":
            print(f"Mouse position: ({msg['x']}, {msg['y']})")

        elif msg_type == "color":
            print(f"Color at ({msg['x']}, {msg['y']}): #{msg['hex']}")

        else:
            print(f"Server: {message}")

    def _on_error(self, ws, error):
        print(f"Error: {error}")

    def _on_close(self, ws, close_status_code, close_msg):
        print("Connection closed")

    def _send(self, data):
        """Send a command to the device"""
        if self.ws:
            self.ws.send(json.dumps(data))

    # High-level commands

    def press(self, key):
        """Press and release a key"""
        self._send({"type": "press", "key": key})
        time.sleep(0.1)

    def type(self, text, delay=30):
        """Type a string"""
        self._send({"type": "type", "text": text, "delay": delay})
        time.sleep(0.1 + len(text) * delay / 1000)

    def move(self, dx, dy):
        """Move mouse by (dx, dy)"""
        self._send({"type": "move", "dx": dx, "dy": dy})
        time.sleep(0.05)

    def scroll(self, delta):
        """Scroll the mouse wheel"""
        self._send({"type": "scroll", "delta": delta})
        time.sleep(0.05)

    def hold(self, key, duration=1.0):
        """Hold a key for a duration (seconds)"""
        self._send({"type": "down", "key": key})
        time.sleep(duration)
        self._send({"type": "up", "key": key})

    def combo(self, keys):
        """Press a key combination (e.g., "LCtrl+V")"""
        self.press(keys)

    def get_mouse_pos(self):
        """Get current mouse position"""
        self._send({"type": "mouse_pos"})
        time.sleep(0.1)

    def get_pixel_color(self, x, y):
        """Get pixel color at (x, y)"""
        self._send({"type": "screenshot_color", "x": x, "y": y})
        time.sleep(0.1)


if __name__ == "__main__":
    # Example usage
    remote = RebindRemote(host="127.0.0.1", port=9000, password="rebind123")

    # Connect in a separate thread to allow commands
    import threading
    thread = threading.Thread(target=remote.connect, daemon=True)
    thread.start()

    # Wait for authentication
    time.sleep(2)

    if remote.authenticated:
        print("\n=== Testing Commands ===")

        # Type some text
        print("\nTyping text...")
        remote.type("Hello from Python!")
        time.sleep(1)

        # Press a key
        print("\nPressing Enter...")
        remote.press("Enter")
        time.sleep(0.5)

        # Mouse movement
        print("\nMoving mouse...")
        remote.move(100, 0)
        time.sleep(0.5)

        # Scroll
        print("\nScrolling...")
        remote.scroll(3)
        time.sleep(0.5)

        # Key combo
        print("\nCopying (Ctrl+C)...")
        remote.combo("LCtrl+C")
        time.sleep(0.5)

        # Hold a key
        print("\nHolding W for 1 second...")
        remote.hold("W", 1.0)

        # Get mouse position
        print("\nGetting mouse position...")
        remote.get_mouse_pos()

    print("\nDone!")

Usage

On the Rebind device computer:

  1. Create a new script in the Rebind UI and paste the Lua code above
  2. Note the WebSocket port (default 9000)

On the remote computer (Python):

bash
python rebind_remote.py

Customization

Change the host/port/password:

python
remote = RebindRemote(
    host="192.168.1.100",  # IP of Rebind device
    port=9001,
    password="mypassword"
)

Add more commands as methods:

python
def click(self):
    """Left click"""
    self.press("Mouse1")

def right_click(self):
    """Right click"""
    self.press("Mouse2")

def ctrl_v(self):
    """Paste"""
    self.combo("LCtrl+V")

Security Notes

  • Change the default password in both the Lua script and Python client
  • Only expose the WebSocket port to trusted networks
  • Use wss:// (WebSocket Secure) if accessing over the internet — requires self-signed certificate setup
  • The script validates the password on every unauthenticated message

Learn more

  • Net.WSListen — Rebind WebSocket server
  • HID — Keyboard and mouse output
  • System — Read mouse position and screen
  • Screen — Sample pixel colors
  • JSON — Parse and stringify JSON

Question

Can we check if inputs are being held down on the remote PC

Rebind AI

Remote Control with Input State Polling

Yes! Use the Input namespace to query the physical state of keys on the Rebind device. This example polls held keys and sends them to the remote Python client.

Rebind Script (Enhanced)

luau
-- rebind: min_sdk=0.1.0
-- rebind: name=Python Remote Control with Input Polling
-- rebind: permission=net

local cfg = UI.Schema({
    port = UI.Slider(9000, { min = 1024, max = 65535, label = "WebSocket Port" }),
    password = UI.Text("rebind123", { label = "Password", maxLength = 50 }),
    poll_interval = UI.Slider(50, { min = 10, max = 500, suffix = "ms", label = "Poll Interval" }),
})

local server = nil
local authenticated = {}
local pollTimers = {}

function OnStart()
    server = Net.WSListen(cfg.port, {
        OnConnect = function(client)
            authenticated[client.id] = false
            Log.Info("Client " .. client.id .. " connected (unauthenticated)")
            client:Send(JSON.Stringify({ type = "auth_required" }))
        end,

        OnMessage = function(client, payload, is_binary)
            local msg = JSON.Parse(payload)

            -- authenticate first
            if not authenticated[client.id] then
                if msg.type == "auth" and msg.password == cfg.password then
                    authenticated[client.id] = true
                    client:Send(JSON.Stringify({ type = "auth_ok" }))
                    Log.Info("Client " .. client.id .. " authenticated")
                    return
                else
                    client:Send(JSON.Stringify({ type = "auth_failed" }))
                    return
                end
            end

            -- process commands
            if msg.type == "press" then
                Run(function()
                    HID.Press(msg.key or "Space")
                end)
                client:Send(JSON.Stringify({ type = "ok", action = "press", key = msg.key }))

            elseif msg.type == "type" then
                Run(function()
                    HID.Type(msg.text or "", msg.delay or 30)
                end)
                client:Send(JSON.Stringify({ type = "ok", action = "type", text = msg.text }))

            elseif msg.type == "move" then
                HID.Move(msg.dx or 0, msg.dy or 0)
                client:Send(JSON.Stringify({ type = "ok", action = "move", dx = msg.dx, dy = msg.dy }))

            elseif msg.type == "scroll" then
                HID.Scroll(msg.delta or 1)
                client:Send(JSON.Stringify({ type = "ok", action = "scroll", delta = msg.delta }))

            elseif msg.type == "down" then
                HID.Down(msg.key or "Space")
                client:Send(JSON.Stringify({ type = "ok", action = "down", key = msg.key }))

            elseif msg.type == "up" then
                HID.Up(msg.key or "Space")
                client:Send(JSON.Stringify({ type = "ok", action = "up", key = msg.key }))

            elseif msg.type == "mouse_pos" then
                local mx, my = System.Mouse()
                client:Send(JSON.Stringify({ type = "mouse_pos", x = mx, y = my }))

            elseif msg.type == "screenshot_color" then
                local color = Screen.GetPixelColor(msg.x, msg.y)
                client:Send(JSON.Stringify({ type = "color", hex = color, x = msg.x, y = msg.y }))

            elseif msg.type == "is_down" then
                -- check if a specific key is held
                local key = msg.key
                local down = Input.IsDown(key)
                local duration = Input.GetDuration(key)
                client:Send(JSON.Stringify({
                    type = "key_state",
                    key = key,
                    down = down,
                    duration = duration,
                }))

            elseif msg.type == "get_active_keys" then
                -- get all currently held keys
                local keys = Input.GetActiveKeys()
                client:Send(JSON.Stringify({
                    type = "active_keys",
                    keys = keys,
                }))

            elseif msg.type == "get_modifiers" then
                -- get modifier state
                local mods = Input.GetModifiers()
                client:Send(JSON.Stringify({
                    type = "modifiers",
                    ctrl = mods.ctrl,
                    shift = mods.shift,
                    alt = mods.alt,
                    win = mods.win,
                }))

            elseif msg.type == "start_poll" then
                -- start polling a key and send updates every interval
                local key = msg.key
                if pollTimers[client.id .. ":" .. key] then
                    pollTimers[client.id .. ":" .. key]:Cancel()
                end
                pollTimers[client.id .. ":" .. key] = Timer.Every(cfg.poll_interval, function()
                    if not authenticated[client.id] then
                        pollTimers[client.id .. ":" .. key]:Cancel()
                        return
                    end
                    local down = Input.IsDown(key)
                    local duration = Input.GetDuration(key)
                    client:Send(JSON.Stringify({
                        type = "key_poll",
                        key = key,
                        down = down,
                        duration = duration,
                    }))
                end)
                Log.Info("Started polling " .. key .. " for client " .. client.id)

            elseif msg.type == "stop_poll" then
                -- stop polling a key
                local key = msg.key
                local pollKey = client.id .. ":" .. key
                if pollTimers[pollKey] then
                    pollTimers[pollKey]:Cancel()
                    pollTimers[pollKey] = nil
                end
                Log.Info("Stopped polling " .. key .. " for client " .. client.id)

            elseif msg.type == "start_poll_all" then
                -- start polling all keys
                if pollTimers[client.id .. ":all"] then
                    pollTimers[client.id .. ":all"]:Cancel()
                end
                pollTimers[client.id .. ":all"] = Timer.Every(cfg.poll_interval, function()
                    if not authenticated[client.id] then
                        pollTimers[client.id .. ":all"]:Cancel()
                        return
                    end
                    local keys = Input.GetActiveKeys()
                    local mods = Input.GetModifiers()
                    client:Send(JSON.Stringify({
                        type = "all_input_state",
                        keys = keys,
                        ctrl = mods.ctrl,
                        shift = mods.shift,
                        alt = mods.alt,
                        win = mods.win,
                    }))
                end)
                Log.Info("Started polling all keys for client " .. client.id)

            elseif msg.type == "stop_poll_all" then
                -- stop polling all keys
                if pollTimers[client.id .. ":all"] then
                    pollTimers[client.id .. ":all"]:Cancel()
                    pollTimers[client.id .. ":all"] = nil
                end
                Log.Info("Stopped polling all keys for client " .. client.id)

            else
                client:Send(JSON.Stringify({ type = "error", message = "Unknown command" }))
            end
        end,

        OnClose = function(client)
            authenticated[client.id] = nil
            -- clean up all timers for this client
            for pollKey, timer in pairs(pollTimers) do
                if pollKey:find("^" .. client.id) then
                    timer:Cancel()
                    pollTimers[pollKey] = nil
                end
            end
            Log.Info("Client " .. client.id .. " disconnected")
        end,
    })
    Log.Info("WebSocket remote listening on ws://0.0.0.0:" .. cfg.port)
    UI.Notify("Remote control active on port " .. cfg.port, "success")
end

function OnStop()
    if server then
        server:Stop()
        Log.Info("Remote control server stopped")
    end
    -- cancel all timers
    for _, timer in pairs(pollTimers) do
        timer:Cancel()
    end
end

Python Client (Enhanced)

python
import json
import time
import websocket
import sys
import threading

class RebindRemote:
    def __init__(self, host="127.0.0.1", port=9000, password="rebind123"):
        self.url = f"ws://{host}:{port}"
        self.password = password
        self.ws = None
        self.authenticated = False
        self.input_callbacks = {}
        self.lock = threading.Lock()

    def connect(self):
        """Connect to the Rebind WebSocket server"""
        print(f"Connecting to {self.url}...")
        self.ws = websocket.WebSocketApp(
            self.url,
            on_open=self._on_open,
            on_message=self._on_message,
            on_error=self._on_error,
            on_close=self._on_close,
        )
        self.ws.run_forever()

    def _on_open(self, ws):
        print("Connected to Rebind device")

    def _on_message(self, ws, message):
        msg = json.loads(message)
        msg_type = msg.get("type")

        if msg_type == "auth_required":
            print("Authentication required. Sending password...")
            self._send({"type": "auth", "password": self.password})

        elif msg_type == "auth_ok":
            print("✓ Authenticated successfully")
            self.authenticated = True

        elif msg_type == "auth_failed":
            print("✗ Authentication failed")
            self.authenticated = False

        elif msg_type == "ok":
            action = msg.get("action", "unknown")
            print(f"✓ {action} completed")

        elif msg_type == "error":
            print(f"✗ Error: {msg.get('message')}")

        elif msg_type == "mouse_pos":
            print(f"Mouse position: ({msg['x']}, {msg['y']})")

        elif msg_type == "color":
            print(f"Color at ({msg['x']}, {msg['y']}): #{msg['hex']}")

        # Input state messages
        elif msg_type == "key_state":
            key = msg.get("key")
            down = msg.get("down")
            duration = msg.get("duration")
            state = "DOWN" if down else "UP"
            print(f"[{key}] {state} (held {duration}ms)")

        elif msg_type == "active_keys":
            keys = msg.get("keys", [])
            if keys:
                print(f"Active keys: {', '.join(keys)}")
            else:
                print("No keys pressed")

        elif msg_type == "modifiers":
            mods = [k for k, v in msg.items() if k in ["ctrl", "shift", "alt", "win"] and v]
            if mods:
                print(f"Modifiers held: {', '.join(mods)}")
            else:
                print("No modifiers held")

        elif msg_type == "key_poll":
            key = msg.get("key")
            down = msg.get("down")
            duration = msg.get("duration")
            state = "DOWN" if down else "UP"
            # Call registered callback if any
            with self.lock:
                if key in self.input_callbacks:
                    self.input_callbacks[key](down, duration)
            # Uncomment for verbose output:
            # print(f"[POLL {key}] {state} ({duration}ms)")

        elif msg_type == "all_input_state":
            keys = msg.get("keys", [])
            ctrl = msg.get("ctrl")
            shift = msg.get("shift")
            alt = msg.get("alt")
            win = msg.get("win")
            mods = [m for m in ["Ctrl", "Shift", "Alt", "Win"] if [ctrl, shift, alt, win][["Ctrl", "Shift", "Alt", "Win"].index(m)]]
            print(f"Held keys: {', '.join(keys) if keys else 'None'} | Mods: {', '.join(mods) if mods else 'None'}")

        else:
            print(f"Server: {message}")

    def _on_error(self, ws, error):
        print(f"Error: {error}")

    def _on_close(self, ws, close_status_code, close_msg):
        print("Connection closed")

    def _send(self, data):
        """Send a command to the device"""
        if self.ws:
            self.ws.send(json.dumps(data))

    # Output commands

    def press(self, key):
        """Press and release a key"""
        self._send({"type": "press", "key": key})
        time.sleep(0.1)

    def type(self, text, delay=30):
        """Type a string"""
        self._send({"type": "type", "text": text, "delay": delay})
        time.sleep(0.1 + len(text) * delay / 1000)

    def move(self, dx, dy):
        """Move mouse by (dx, dy)"""
        self._send({"type": "move", "dx": dx, "dy": dy})
        time.sleep(0.05)

    def scroll(self, delta):
        """Scroll the mouse wheel"""
        self._send({"type": "scroll", "delta": delta})
        time.sleep(0.05)

    def hold(self, key, duration=1.0):
        """Hold a key for a duration (seconds)"""
        self._send({"type": "down", "key": key})
        time.sleep(duration)
        self._send({"type": "up", "key": key})

    def combo(self, keys):
        """Press a key combination (e.g., "LCtrl+V")"""
        self.press(keys)

    def get_mouse_pos(self):
        """Get current mouse position"""
        self._send({"type": "mouse_pos"})
        time.sleep(0.1)

    def get_pixel_color(self, x, y):
        """Get pixel color at (x, y)"""
        self._send({"type": "screenshot_color", "x": x, "y": y})
        time.sleep(0.1)

    # Input state queries

    def is_down(self, key):
        """Check if a key is currently held"""
        self._send({"type": "is_down", "key": key})
        time.sleep(0.05)

    def get_active_keys(self):
        """Get all currently held keys"""
        self._send({"type": "get_active_keys"})
        time.sleep(0.05)

    def get_modifiers(self):
        """Get current modifier state (Ctrl, Shift, Alt, Win)"""
        self._send({"type": "get_modifiers"})
        time.sleep(0.05)

    def poll_key(self, key, callback=None):
        """
        Start polling a specific key.
        Callback receives (is_down, duration_ms)
        """
        if callback:
            with self.lock:
                self.input_callbacks[key] = callback
        self._send({"type": "start_poll", "key": key})

    def stop_poll_key(self, key):
        """Stop polling a specific key"""
        self._send({"type": "stop_poll", "key": key})
        with self.lock:
            if key in self.input_callbacks:
                del self.input_callbacks[key]

    def poll_all_keys(self):
        """Start polling all keys and modifiers"""
        self._send({"type": "start_poll_all"})

    def stop_poll_all_keys(self):
        """Stop polling all keys"""
        self._send({"type": "stop_poll_all"})


if __name__ == "__main__":
    # Example usage
    remote = RebindRemote(host="127.0.0.1", port=9000, password="rebind123")

    # Connect in a separate thread
    thread = threading.Thread(target=remote.connect, daemon=True)
    thread.start()

    # Wait for authentication
    time.sleep(2)

    if remote.authenticated:
        print("\n=== Testing Input Queries ===\n")

        # Query individual key state
        print("Checking if 'W' is pressed...")
        remote.is_down("W")
        time.sleep(0.2)

        # Get all active keys
        print("\nGetting all held keys...")
        remote.get_active_keys()
        time.sleep(0.2)

        # Get modifier state
        print("\nGetting modifier state...")
        remote.get_modifiers()
        time.sleep(0.2)

        # Poll a specific key with callback
        print("\nPolling 'Mouse1' for 5 seconds (press/release it on the remote PC)...")

        def on_mouse1_change(is_down, duration):
            state = "PRESSED" if is_down else "RELEASED"
            print(f"  Mouse1 {state} (was held {duration}ms)")

        remote.poll_key("Mouse1", on_mouse1_change)
        time.sleep(5)
        remote.stop_poll_key("Mouse1")

        # Poll all keys
        print("\nPolling all keys for 5 seconds...")
        remote.poll_all_keys()
        time.sleep(5)
        remote.stop_poll_all_keys()

        print("\nDone!")

    # Keep the connection alive
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("Exiting...")

Usage Examples

Check if a key is currently held:

python
remote.is_down("W")  # prints: [W] DOWN (held 1234ms) or [W] UP

Get all currently pressed keys:

python
remote.get_active_keys()  # prints: Active keys: W, LCtrl, A

Get modifier state:

python
remote.get_modifiers()  # prints: Modifiers held: Ctrl, Shift

Poll a key and react to changes:

python
def on_w_pressed(is_down, duration):
    if is_down:
        print("W pressed! Starting movement...")
    else:
        print(f"W released after {duration}ms")

remote.poll_key("W", on_w_pressed)
time.sleep(10)  # listen for 10 seconds
remote.stop_poll_key("W")

Poll all keys continuously:

python
remote.poll_all_keys()
time.sleep(30)  # monitor for 30 seconds
remote.stop_poll_all_keys()

Use Cases

  • Input mirroring: Record key presses on the remote PC and replay them locally
  • Diagnostics: Check if keys are stuck or not registering
  • Conditional automation: Trigger actions based on what the user is currently holding
  • Training/coaching: Monitor player inputs in real-time
  • Accessibility: Verify input state for users with limited mobility

Learn more

  • Input — Query physical key state
  • Net.WSListen — WebSocket server
  • Timer — Polling timers
  • HID — Send keyboard/mouse output