hex-flow-node-teleop 0.0.1a1

Hex Flow Node for Teleoperation Input (Keyboard & Joystick)

HEX FLOW NODE TELEOP

---

Overview

Hex Flow nodes for reading teleoperation inputs from keyboard and joystick/gamepad using evdev. Each node maintains an internal state table updated by background threads and publishes the full state at the configured rate (default 100 Hz) over Zenoh via hex_flow_core, using FlatBuffer messages from hex_util_msg.

Nodes

Node	Description	Publishes	Subscribes
hex-flow-teleop-keyboard	Keyboard letter-key state reader	teleop_keyboard	-
hex-flow-teleop-joystick	Joystick/Gamepad button & axis state reader	teleop_joy	-
hex-flow-teleop-keyboard-test	Terminal viewer for keyboard state	-	teleop_keyboard
hex-flow-teleop-joystick-test	Terminal viewer for joystick state	-	teleop_joy

Prerequisites

Accessing /dev/input/* requires the current user to be in the input group:

sudo usermod -aG input $USER
# Log out and back in for the change to take effect

Installation

Requirements

• Python >= 3.10
• OS: Ubuntu (or other Linux)
• Core dependencies:
  • evdev >= 1.7.0
  • hex_flow_core >= 0.0.0, < 0.1.0
  • hex_util_msg >= 0.0.0, < 0.1.0
  • hex_util_runtime >= 0.0.0, < 0.1.0

Install hex-flow-cli

For Ubuntu or any Debian-based system, install Zenoh and hex-flow CLI:

curl -L https://download.eclipse.org/zenoh/debian-repo/zenoh-public-key | sudo gpg --dearmor --yes --output /etc/apt/keyrings/zenoh-public-key.gpg
echo "deb [signed-by=/etc/apt/keyrings/zenoh-public-key.gpg] https://download.eclipse.org/zenoh/debian-repo/ /" | sudo tee -a /etc/apt/sources.list > /dev/null
sudo apt update
sudo apt install zenoh curl
curl -fsSL https://raw.githubusercontent.com/hexfellow/hex-flow/main/install.sh | sh

Install from PyPI

pip install hex_flow_node_teleop

Python Config API

The package provides helper functions that return NodeConfig objects for easy integration into your LaunchConfig.

from hex_flow_core import LaunchConfig
from hex_flow_node_teleop import (
    default_teleop_keyboard_node,
    default_teleop_keyboard_test_node,
    default_teleop_joystick_node,
    default_teleop_joystick_test_node,
)

config = LaunchConfig(local_only=True, save_path="/tmp/teleop_launch.yml")

nodes = {
    "teleop_keyboard": default_teleop_keyboard_node(),
    "test_keyboard": default_teleop_keyboard_test_node(),
    "teleop_joystick": default_teleop_joystick_node(),
    "test_joystick": default_teleop_joystick_test_node(),
}

config.set_nodes(nodes)
print(config.export())

default_teleop_keyboard_node

Parameter	Type	Default	Description
name	str	"teleop_keyboard"	Node name and remap prefix
device_path	str	""	/dev/input/eventN or empty for auto-detect
rate_hz	float	100.0	Publishing rate in Hz
required	bool	True	Required for launch
remap_dict	dict	None	Custom remap; defaults to {name}/teleop_keyboard

default_teleop_keyboard_test_node

Parameter	Type	Default	Description
name	str	"test_keyboard"	Node name
rate_hz	float	10.0	Display refresh rate in Hz
required	bool	False	Required for launch
remap_dict	dict	None	Custom remap; defaults to {teleop_name}/teleop_keyboard
teleop_name	str	"teleop_keyboard"	Keyboard teleop node to subscribe to

default_teleop_joystick_node

Parameter	Type	Default	Description
name	str	"teleop_joystick"	Node name and remap prefix
device_path	str	""	/dev/input/eventN or empty for auto-detect
rate_hz	float	100.0	Publishing rate in Hz
required	bool	True	Required for launch
remap_dict	dict	None	Custom remap; defaults to {name}/teleop_joy

default_teleop_joystick_test_node

Parameter	Type	Default	Description
name	str	"test_joystick"	Node name
rate_hz	float	10.0	Display refresh rate in Hz
required	bool	False	Required for launch
remap_dict	dict	None	Custom remap; defaults to {teleop_name}/teleop_joy
teleop_name	str	"teleop_joystick"	Joystick teleop node to subscribe to

Examples

Ready-to-run config scripts are provided in the example/ directory. Each script prints a launch YAML to stdout, intended for use with hexflow run:

Keyboard

# Basic — auto-detect device, 100 Hz publish, 10 Hz test display
hexflow run $(python example/keyboard_basic_config.py)

# Custom — specify device path, 50 Hz publish, 5 Hz test display
hexflow run $(python example/keyboard_custom_config.py)

Joystick

# Basic — auto-detect device, 100 Hz publish, 10 Hz test display
hexflow run $(python example/joystick_basic_config.py)

# Custom — specify device path, 50 Hz publish, 5 Hz test display
hexflow run $(python example/joystick_custom_config.py)

YAML Examples

Keyboard (100 Hz)

nodes:
  - name: teleop_keyboard
    build: pip install hex_flow_node_teleop
    run: hex-flow-teleop-keyboard
    required: true
    remap:
      teleop_keyboard: teleop_keyboard/teleop_keyboard
    env:
      DEVICE_PATH: ""
      RATE_HZ: "100"

  - name: test_keyboard
    build: pip install hex_flow_node_teleop
    run: hex-flow-teleop-keyboard-test
    required: false
    remap:
      teleop_keyboard: teleop_keyboard/teleop_keyboard
    env:
      RATE_HZ: "10"

Joystick / Gamepad (100 Hz)

nodes:
  - name: teleop_joystick
    build: pip install hex_flow_node_teleop
    run: hex-flow-teleop-joystick
    required: true
    remap:
      teleop_joy: teleop_joystick/teleop_joy
    env:
      DEVICE_PATH: ""
      RATE_HZ: "100"

  - name: test_joystick
    build: pip install hex_flow_node_teleop
    run: hex-flow-teleop-joystick-test
    required: false
    remap:
      teleop_joy: teleop_joystick/teleop_joy
    env:
      RATE_HZ: "10"

Message Types (FlatBuffer)

All topics use FlatBuffer messages from hex_util_msg.msg_teleop.

teleop_keyboard — HexTeleopKeyboard

Field	Type	Description
ts_ns	int64	Timestamp in nanoseconds
key_a … key_z	bool	Per-letter key state (true = pressed)

Schema: msgs/msg_teleop/teleop_keyboard.fbs

teleop_joy — HexTeleopJoy

Field	Type	Description
ts_ns	int64	Timestamp in nanoseconds
btn_x, btn_y, btn_a, btn_b	bool	Face buttons
btn_lb, btn_rb	bool	Shoulder bumpers
btn_lt, btn_rt	bool	Trigger buttons
btn_lthumb, btn_rthumb	bool	Thumbstick clicks
axis_lx, axis_ly	float64	Left stick axes (normalized to [-1, 1])
axis_rx, axis_ry	float64	Right stick axes (normalized to [-1, 1])
axis_lt, axis_rt	float64	Trigger analog values (normalized to [-1, 1])
hat_x, hat_y	float64	D-pad / hat switch values

Schema: msgs/msg_teleop/teleop_joy.fbs

Environment Variables

All Nodes

Variable	Type	Default	Description
HEX_FLOW_NODE_NAME	str	constructor arg	Overrides node name (handled by hex_flow_core)
HEX_FLOW_REMAP	str	{}	JSON dict for topic remapping (handled by hex_flow_core)
RUST_LOG	str	info	Log level for envlog

Teleop Nodes (keyboard / joystick)

Variable	Type	Default	Description
DEVICE_PATH	str	"" (auto)	Specific /dev/input/eventN path; empty = auto-detect
RATE_HZ	float	100.0	Publishing rate in Hz

Test Nodes (test-keyboard / test-joystick)

Variable	Type	Default	Description
RATE_HZ	float	10.0	Display refresh rate in Hz

Architecture

Both keyboard and joystick nodes follow the same pattern:

1. Device discovery — enumerate /dev/input/* via evdev.list_devices() and filter by capabilities (letter keys for keyboard, analog axes for joystick).
2. Background reader thread — a daemon thread continuously reads evdev events and updates an in-memory state table protected by a threading.Lock.
3. FlatBuffer serialization — on each publish cycle, the node snapshots the current state, builds a FlatBuffer message (HexTeleopKeyboard or HexTeleopJoy) with a nanosecond timestamp, and publishes the binary payload via hex_flow_core.Node.pub().
4. Rate-controlled publish loop — using HexRate from hex_util_runtime, the node publishes at the configured rate (default 100 Hz).

Test nodes subscribe via hex_flow_core.Node.create_sub() and poll with Node.get(topic, latest=True) at a lower rate (default 10 Hz), deserializing the FlatBuffer to refresh the terminal display.

This decouples the variable-rate hardware events from the fixed-rate publish output, ensuring consistent state updates regardless of actual input event frequency.

License

Apache License 2.0. See LICENSE.