diff --git a/examples/teleop/python/keyboard_printer_example.py b/examples/teleop/python/keyboard_printer_example.py new file mode 100644 index 0000000000..7c8dba603d --- /dev/null +++ b/examples/teleop/python/keyboard_printer_example.py @@ -0,0 +1,185 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Keyboard Printer Example. + +Prints every currently-held key (not just the SE3-relevant subset) each frame, +via KeyboardSource's "keyboard_all_keys" bitmap output. The keyboard plugin +self-discovers its device and is auto-launched by TeleopSession -- no external +process to start manually. +""" + +import sys +import time +from pathlib import Path + +from isaaccapture.cloudxr import CloudXRLauncher +from isaaccapture.retargeting_engine.deviceio_source_nodes import KeyboardSource +from isaaccapture.teleop_session_manager import ( + TeleopSession, + TeleopSessionConfig, + PluginConfig, +) + + +PLUGIN_ROOT_DIR = Path(__file__).resolve().parent.parent.parent.parent / "plugins" +PLUGIN_NAME = "keyboard" +PLUGIN_ROOT_ID = "keyboard" + +# Evdev key codes (linux/input-event-codes.h) -> display name, for the keys on a +# standard PC keyboard. Codes not listed here still show up (as "code") -- +# this table is for readability, not a completeness gate. +KEY_NAMES = { + 1: "ESC", + 14: "BACKSPACE", + 15: "TAB", + 28: "ENTER", + 29: "LCTRL", + 42: "LSHIFT", + 54: "RSHIFT", + 56: "LALT", + 57: "SPACE", + 58: "CAPSLOCK", + 69: "NUMLOCK", + 70: "SCROLLLOCK", + 97: "RCTRL", + 100: "RALT", + 102: "HOME", + 103: "UP", + 104: "PAGEUP", + 105: "LEFT", + 106: "RIGHT", + 107: "END", + 108: "DOWN", + 109: "PAGEDOWN", + 110: "INSERT", + 111: "DELETE", + **{2 + i: str((i + 1) % 10) for i in range(10)}, # KEY_1..KEY_0 -> "1".."9","0" + **{59 + i: f"F{i + 1}" for i in range(10)}, # KEY_F1..KEY_F10 + 87: "F11", + 88: "F12", + 16: "Q", + 17: "W", + 18: "E", + 19: "R", + 20: "T", + 21: "Y", + 22: "U", + 23: "I", + 24: "O", + 25: "P", + 30: "A", + 31: "S", + 32: "D", + 33: "F", + 34: "G", + 35: "H", + 36: "J", + 37: "K", + 38: "L", + 44: "Z", + 45: "X", + 46: "C", + 47: "V", + 48: "B", + 49: "N", + 50: "M", +} + +ALL_KEYS_BITMAP_SIZE = 256 + + +def main(): + import argparse + + parser = argparse.ArgumentParser(description=__doc__) + CloudXRLauncher.add_launcher_arguments(parser) + args = parser.parse_args() + + print("\n" + "=" * 80) + print(" Keyboard Printer Example") + print("=" * 80) + print("Press any key on the machine running this example.") + print("=" * 80 + "\n") + + # ================================================================== + # Setup: Create keyboard source + # ================================================================== + keyboard_source = KeyboardSource(name="keyboard") + + # ================================================================== + # Configure Plugins + # ================================================================== + + plugins = [] + if PLUGIN_ROOT_DIR.exists(): + plugins.append( + PluginConfig( + plugin_name=PLUGIN_NAME, + plugin_root_id=PLUGIN_ROOT_ID, + search_paths=[PLUGIN_ROOT_DIR], + ) + ) + + # ================================================================== + # Create and run TeleopSession + # ================================================================== + + session_config = TeleopSessionConfig( + app_name="KeyboardPrinterExample", + trackers=[], + pipeline=keyboard_source, + plugins=plugins, + ) + + with CloudXRLauncher.launch_context(args): + with TeleopSession(session_config) as session: + start_time = time.time() + prev_pressed: set[int] = set() + + while time.time() - start_time < 30.0: + result = session.step() + bitmap_group = result["keyboard_all_keys"] + + elapsed = session.get_elapsed_time() + if bitmap_group.is_none: + print( + f"[{elapsed:5.1f}s] (no keyboard data yet)", + end="\r", + flush=True, + ) + time.sleep(0.01) + continue + + bitmap = bitmap_group[0] + pressed = {code for code in range(ALL_KEYS_BITMAP_SIZE) if bitmap[code]} + names = [KEY_NAMES.get(code, f"code{code}") for code in sorted(pressed)] + + # Live status line (overwritten each frame). + print( + f"[{elapsed:5.1f}s] Held: {' '.join(names) or '-'}" + " " * 20, + end="\r", + flush=True, + ) + + # Permanent, scrollable log of every press/release transition -- a + # quick tap can flash by on the status line above before you notice + # it, but every transition is logged here. + for code in sorted(pressed - prev_pressed): + print( + f"[{elapsed:5.1f}s] {KEY_NAMES.get(code, f'code{code}')} down" + ) + for code in sorted(prev_pressed - pressed): + print(f"[{elapsed:5.1f}s] {KEY_NAMES.get(code, f'code{code}')} up") + prev_pressed = pressed + + time.sleep(0.01) # ~100 FPS + + print("\nTime limit reached.") + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/core/deviceio_trackers/trackers.toml b/src/core/deviceio_trackers/trackers.toml index e3afd3878b..82e4a210a2 100644 --- a/src/core/deviceio_trackers/trackers.toml +++ b/src/core/deviceio_trackers/trackers.toml @@ -43,6 +43,12 @@ traits = "PedalRecordingTraits" max_flatbuffer_size = 256 python_accessor = "get_pedal_data" +[[tracker]] +name = "keyboard" +table = "KeyboardOutput" +max_flatbuffer_size = 2048 +python_accessor = "get_keyboard_data" + [[tracker]] name = "haptic_command" direction = "push" diff --git a/src/core/schema/fbs/keyboard.fbs b/src/core/schema/fbs/keyboard.fbs new file mode 100644 index 0000000000..aa39af68ee --- /dev/null +++ b/src/core/schema/fbs/keyboard.fbs @@ -0,0 +1,28 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +include "timestamp.fbs"; + +namespace core; + +// Raw press state of a keyboard: the set of currently-held Linux evdev key codes +// (see linux/input-event-codes.h, e.g. KEY_W = 17). Carries no semantic mapping -- +// which keys mean what (an axis, a toggle, a mode switch) is entirely up to the +// consuming retargeter. +// +// All fields are always present whenever this table itself is present. +table KeyboardOutput { + // Evdev key codes currently held down, in no particular order. + pressed_keys: [ushort] (id: 0); + + // Whether the tracker has emitted at least one sample. + is_valid: bool (id: 1); +} + +// MCAP recording wrapper for KeyboardOutput. +table KeyboardOutputRecord { + data: KeyboardOutput (id: 0); + timestamp: DeviceDataTimestamp (id: 1); +} + +root_type KeyboardOutputRecord; diff --git a/src/core/schema/golden/keyboard.bfbs b/src/core/schema/golden/keyboard.bfbs new file mode 100644 index 0000000000..b336573505 Binary files /dev/null and b/src/core/schema/golden/keyboard.bfbs differ diff --git a/src/core/schema/python/CMakeLists.txt b/src/core/schema/python/CMakeLists.txt index 05e9feeefb..3a9fff4f32 100644 --- a/src/core/schema/python/CMakeLists.txt +++ b/src/core/schema/python/CMakeLists.txt @@ -10,6 +10,7 @@ pybind11_add_module(schema_py head_bindings.h joint_se3_pose_bindings.h joint_state_bindings.h + keyboard_bindings.h message_channel_bindings.h oglo_tactile_bindings.h pedals_bindings.h diff --git a/src/core/schema/python/keyboard_bindings.h b/src/core/schema/python/keyboard_bindings.h new file mode 100644 index 0000000000..a3a6589f24 --- /dev/null +++ b/src/core/schema/python/keyboard_bindings.h @@ -0,0 +1,58 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +// Python bindings for the Keyboard FlatBuffer schema. +// Types: KeyboardOutput (table), exposed as an encoded view. + +#pragma once + +#include "schema_serialized.h" + +#include +#include + +#include +#include +#include + +namespace py = pybind11; + +namespace core +{ + +inline void bind_keyboard(py::module& m) +{ + serialized_class(m, "KeyboardOutput", "Encoded raw keyboard press state.") + .def(py::init( + [](std::vector pressed_keys, bool is_valid) + { + KeyboardOutputT native; + native.pressed_keys = std::move(pressed_keys); + native.is_valid = is_valid; + return pack(native); + }), + py::arg("pressed_keys"), py::arg("is_valid"), "Encode a keyboard press-state snapshot.") + .def_property_readonly("pressed_keys", vector_field(&KeyboardOutput::pressed_keys)) + .def_property_readonly("is_valid", field(&KeyboardOutput::is_valid)) + .def("__repr__", + [](const Serialized& self) + { + std::string result = "KeyboardOutput(pressed_keys=["; + const auto* keys = self->pressed_keys(); + if (keys != nullptr) + { + for (size_t i = 0; i < keys->size(); ++i) + { + if (i > 0) + result += ", "; + result += std::to_string((*keys)[i]); + } + } + result += "], is_valid=" + std::to_string(self->is_valid()) + ")"; + return result; + }); + + bind_record(m, "KeyboardOutputRecord", "KeyboardOutput"); +} + +} // namespace core diff --git a/src/core/schema/python/schema_module.cpp b/src/core/schema/python/schema_module.cpp index 14f0a27073..3751d0be35 100644 --- a/src/core/schema/python/schema_module.cpp +++ b/src/core/schema/python/schema_module.cpp @@ -13,6 +13,7 @@ #include "head_bindings.h" #include "joint_se3_pose_bindings.h" #include "joint_state_bindings.h" +#include "keyboard_bindings.h" #include "message_channel_bindings.h" #include "oak_bindings.h" #include "oglo_tactile_bindings.h" @@ -46,6 +47,9 @@ PYBIND11_MODULE(_schema, m) // Bind pedals types (Generic3AxisPedalOutput table). core::bind_pedals(m); + // Bind keyboard types (KeyboardOutput table) for raw keyboard press state. + core::bind_keyboard(m); + // Bind OGLO tactile glove types (OgloGloveSample table). core::bind_oglo_tactile(m); diff --git a/src/plugins/CMakeLists.txt b/src/plugins/CMakeLists.txt index 127717f682..94292238f9 100644 --- a/src/plugins/CMakeLists.txt +++ b/src/plugins/CMakeLists.txt @@ -17,6 +17,7 @@ add_subdirectory(controller_se3_tracker) add_subdirectory(vive_se3_tracker) add_subdirectory(controller_synthetic_hands) add_subdirectory(generic_3axis_pedal) +add_subdirectory(keyboard) add_subdirectory(xsens_full_body) add_subdirectory(so101_leader) add_subdirectory(rebot_devarm_leader) diff --git a/src/plugins/keyboard/CMakeLists.txt b/src/plugins/keyboard/CMakeLists.txt new file mode 100644 index 0000000000..96c490715d --- /dev/null +++ b/src/plugins/keyboard/CMakeLists.txt @@ -0,0 +1,23 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +if(NOT CMAKE_SYSTEM_NAME STREQUAL "Linux") + message(STATUS "Skipping keyboard plugin (Linux only)") + add_custom_target(keyboard_plugin + COMMAND ${CMAKE_COMMAND} -E echo "Skipping keyboard: Linux only") + return() +endif() + +add_executable(keyboard_plugin + main.cpp + keyboard_plugin.cpp +) + +target_link_libraries(keyboard_plugin PRIVATE + pusherio::pusherio + oxr::oxr_core + isaacteleop_schema +) + +install(TARGETS keyboard_plugin RUNTIME DESTINATION plugins/keyboard) +install(FILES plugin.yaml README.md DESTINATION plugins/keyboard) diff --git a/src/plugins/keyboard/README.md b/src/plugins/keyboard/README.md new file mode 100644 index 0000000000..022d2f42cc --- /dev/null +++ b/src/plugins/keyboard/README.md @@ -0,0 +1,42 @@ + + +# Keyboard Plugin + +Reads a keyboard from `/dev/input/event*` and pushes `KeyboardOutput` via OpenXR. Use with `KeyboardTracker` with the same `collection_id`. + +Reports raw press state only, with no semantic mapping to axes or commands -- that mapping belongs +in a retargeter (e.g. `KeyboardToSe3RelRetargeter`) consuming this tracker's output. + +Self-discovers its device (the first `*-event-kbd` entry under `/dev/input/by-path/`), so it needs +no arguments to run and can be auto-launched by `PluginManager` via `PluginConfig` -- no manual +process to start. + +## Usage + +Auto-launched (recommended -- matches how `PluginManager` invokes plugins): + +```bash +./keyboard_plugin --plugin-root-id=keyboard +``` + +Manual / standalone, with an explicit device: + +```bash +./keyboard_plugin [device_path] [--plugin-root-id=] +``` + +- **device_path**: Optional. Defaults to the first `*-event-kbd` entry under `/dev/input/by-path/`. + Identify a specific keyboard event device with `cat /proc/bus/input/devices` (look for a + `Handlers=... eventN` line under a keyboard entry) or `evtest`. Reading `/dev/input/event*` + typically requires membership in the `input` group. +- **collection_id**: Default `keyboard`. Match this when creating `KeyboardTracker`. + +## Key mapping + +Reports the current press state of: `W`, `A`, `S`, `D`, `Q`, `E`, `Z`, `X`, `T`, `G`, `C`, `V`, `K`. +Autorepeat events are ignored (press state is already true). + +Linux only. diff --git a/src/plugins/keyboard/keyboard_plugin.cpp b/src/plugins/keyboard/keyboard_plugin.cpp new file mode 100644 index 0000000000..d254b656b5 --- /dev/null +++ b/src/plugins/keyboard/keyboard_plugin.cpp @@ -0,0 +1,194 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include "keyboard_plugin.hpp" + +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +namespace plugins +{ +namespace keyboard +{ + +namespace +{ + +constexpr size_t kInputEventSize = sizeof(input_event); +constexpr size_t kMaxFlatbufferSize = 2048; + +} // namespace + +KeyboardPlugin::KeyboardPlugin(const std::string& device_path, const std::string& collection_id) + : device_path_(device_path), + session_(std::make_shared("KeyboardPlugin", core::SchemaPusher::get_required_extensions())), + pusher_(session_->get_handles(), + core::SchemaPusherConfig{ .collection_id = collection_id, + .max_flatbuffer_size = kMaxFlatbufferSize, + .tensor_identifier = "keyboard", + .localized_name = "Keyboard", + .app_name = "KeyboardPlugin" }) +{ + if (!open_device()) + throw std::runtime_error("KeyboardPlugin: Failed to open " + device_path + " (" + strerror(errno) + ")"); +} + +KeyboardPlugin::~KeyboardPlugin() +{ + if (device_fd_ >= 0) + close_device(); +} + +void KeyboardPlugin::update() +{ + if (device_fd_ < 0) + { + open_device(); + if (device_fd_ < 0) + { + push_current_state(); + return; + } + } + + fd_set read_fds; + struct timeval timeout = { 0, 0 }; + + while (true) + { + FD_ZERO(&read_fds); + FD_SET(device_fd_, &read_fds); + timeout = { 0, 0 }; + + int ret = select(device_fd_ + 1, &read_fds, nullptr, nullptr, &timeout); + if (ret < 0) + { + if (errno == EINTR) + return; + close_device(); + push_current_state(); + return; + } + if (ret == 0 || !FD_ISSET(device_fd_, &read_fds)) + { + // If there is no data to read (ret == 0) or the device file descriptor is not set in + // the read set, break out of the loop; this means there's no new event available. + break; + } + + input_event event; + ssize_t n = read(device_fd_, &event, kInputEventSize); + if (n != static_cast(kInputEventSize)) + { + if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) + break; + close_device(); + push_current_state(); + return; + } + + if (event.type == EV_SYN) + { + if (event.code == SYN_DROPPED) + { + // The kernel-side event queue overran; everything until the next + // SYN_REPORT is incomplete and must be discarded rather than applied. + awaiting_syn_report_ = true; + } + else if (event.code == SYN_REPORT && awaiting_syn_report_) + { + // Only leave recovery mode on a successful resync -- if EVIOCGKEY + // failed, pressed_keys_ is still whatever it was before the drop and + // must not be trusted yet; keep discarding events until the next + // SYN_REPORT gives us another chance to resynchronize. + if (resync_pressed_keys()) + awaiting_syn_report_ = false; + } + continue; + } + + if (awaiting_syn_report_) + continue; + + // value: 0 = release, 1 = press, 2 = autorepeat (ignored -- key is already tracked as held). + if (event.type == EV_KEY && event.value != 2) + { + if (event.value != 0) + pressed_keys_.insert(event.code); + else + pressed_keys_.erase(event.code); + } + } + + push_current_state(); +} + +bool KeyboardPlugin::open_device() +{ + assert(device_fd_ < 0); + + int fd = open(device_path_.c_str(), O_RDONLY | O_NONBLOCK); + if (fd < 0) + return false; + + device_fd_ = fd; + awaiting_syn_report_ = false; + resync_pressed_keys(); + std::cout << "KeyboardPlugin: Opened " << device_path_ << std::endl; + return true; +} + +void KeyboardPlugin::close_device() +{ + assert(device_fd_ >= 0); + + close(device_fd_); + device_fd_ = -1; + awaiting_syn_report_ = false; + // A closed device can no longer report releases -- forget everything it + // last reported as held so a stale key doesn't stick "pressed" forever. + pressed_keys_.clear(); +} + +bool KeyboardPlugin::resync_pressed_keys() +{ + unsigned char bitmap[(KEY_CNT + 7) / 8] = {}; + if (ioctl(device_fd_, EVIOCGKEY(sizeof(bitmap)), bitmap) < 0) + return false; + + pressed_keys_.clear(); + for (uint16_t code = 0; code < KEY_CNT; ++code) + { + if (bitmap[code / 8] & (1u << (code % 8))) + pressed_keys_.insert(code); + } + return true; +} + +void KeyboardPlugin::push_current_state() +{ + core::KeyboardOutputT out; + out.pressed_keys.assign(pressed_keys_.begin(), pressed_keys_.end()); + out.is_valid = true; + + auto sample_time_ns = core::os_monotonic_now_ns(); + + flatbuffers::FlatBufferBuilder builder(kMaxFlatbufferSize); + auto offset = core::KeyboardOutput::Pack(builder, &out); + builder.Finish(offset); + pusher_.push_buffer(builder.GetBufferPointer(), builder.GetSize(), sample_time_ns, sample_time_ns); +} + +} // namespace keyboard +} // namespace plugins diff --git a/src/plugins/keyboard/keyboard_plugin.hpp b/src/plugins/keyboard/keyboard_plugin.hpp new file mode 100644 index 0000000000..7caf1bd892 --- /dev/null +++ b/src/plugins/keyboard/keyboard_plugin.hpp @@ -0,0 +1,62 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include + +#include +#include +#include +#include + +namespace core +{ +class OpenXRSession; +} + +namespace plugins +{ +namespace keyboard +{ + +/*! + * @brief Reads a Linux evdev keyboard device (e.g. /dev/input/event3), tracks + * the set of currently-held key codes, and pushes KeyboardOutput via + * OpenXR SchemaPusher. Carries no semantic mapping -- keys are reported + * as-is (evdev key codes, see linux/input-event-codes.h). + */ +class KeyboardPlugin +{ +public: + KeyboardPlugin(const std::string& device_path, const std::string& collection_id); + ~KeyboardPlugin(); + + void update(); + +private: + bool open_device(); + void close_device(); + void push_current_state(); + // Rebuilds pressed_keys_ from the kernel's authoritative EVIOCGKEY bitmap -- + // used on open and after a SYN_DROPPED (evdev client-side buffer overrun) so a + // dropped release event can't leave a key stuck "pressed" forever. Returns + // false on ioctl failure, in which case pressed_keys_ is left untouched -- + // callers recovering from a drop must not treat that as resynchronized. + bool resync_pressed_keys(); + + std::string device_path_; + int device_fd_ = -1; + // Set between a SYN_DROPPED and the next SYN_REPORT: per the evdev protocol, + // events in that window are incomplete and must be discarded rather than + // applied individually. + bool awaiting_syn_report_ = false; + + std::set pressed_keys_; + + std::shared_ptr session_; + core::SchemaPusher pusher_; +}; + +} // namespace keyboard +} // namespace plugins diff --git a/src/plugins/keyboard/main.cpp b/src/plugins/keyboard/main.cpp new file mode 100644 index 0000000000..af64ea3037 --- /dev/null +++ b/src/plugins/keyboard/main.cpp @@ -0,0 +1,123 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include "keyboard_plugin.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace plugins::keyboard; + +namespace +{ + +// Returns the first udev-classified keyboard event device, or nullopt if none found. +// Lets the plugin run with zero required arguments when launched via PluginManager +// (which invokes plugins as ` --plugin-root-id=`, not positional args). +std::optional discover_keyboard_device_path() +{ + const std::filesystem::path by_path_dir = "/dev/input/by-path"; + std::vector candidates; + std::error_code ec; + if (!std::filesystem::exists(by_path_dir, ec)) + return std::nullopt; + + for (const auto& entry : std::filesystem::directory_iterator(by_path_dir, ec)) + { + const std::string name = entry.path().filename().string(); + if (name.ends_with("-event-kbd")) + candidates.push_back(entry.path().string()); + } + if (candidates.empty()) + return std::nullopt; + + std::sort(candidates.begin(), candidates.end()); + return candidates.front(); +} + +// PluginManager invokes plugins as ` --plugin-root-id= [plugin_args...]`. +// A bare positional token (no leading `--`) is treated as an explicit device path +// override, matching manual/standalone invocation. +struct ParsedArgs +{ + std::optional device_path; + std::string collection_id = "keyboard"; +}; + +ParsedArgs parse_args(int argc, char** argv) +{ + ParsedArgs parsed; + constexpr std::string_view kRootIdPrefix = "--plugin-root-id="; + for (int i = 1; i < argc; ++i) + { + const std::string_view arg = argv[i]; + if (arg.starts_with(kRootIdPrefix)) + { + parsed.collection_id = std::string(arg.substr(kRootIdPrefix.size())); + } + else if (!arg.starts_with("--")) + { + parsed.device_path = std::string(arg); + } + } + return parsed; +} + +} // namespace + +int main(int argc, char** argv) +try +{ + if (argc == 0) + { + std::cerr << "Usage: keyboard_plugin [device_path] [--plugin-root-id=]" << std::endl; + return 1; + } + + const ParsedArgs args = parse_args(argc, argv); + std::optional device_path = args.device_path; + if (!device_path) + device_path = discover_keyboard_device_path(); + if (!device_path) + { + std::cerr << argv[0] << ": No keyboard event device found under /dev/input/by-path/ and none given explicitly." + << std::endl; + return 1; + } + + std::cout << "Keyboard (device: " << *device_path << ", collection: " << args.collection_id << ")" << std::endl; + + KeyboardPlugin plugin(*device_path, args.collection_id); + + // Push data at 90 Hz. + const auto frame_duration = std::chrono::nanoseconds(1000000000 / 90); + const auto program_start = std::chrono::steady_clock::now(); + std::size_t frame_count = 0; + + while (true) + { + plugin.update(); + frame_count++; + std::this_thread::sleep_until(program_start + frame_duration * frame_count); + } + + return 0; +} +catch (const std::exception& e) +{ + std::cerr << argv[0] << ": " << e.what() << std::endl; + return 1; +} +catch (...) +{ + std::cerr << argv[0] << ": Unknown error" << std::endl; + return 1; +} diff --git a/src/plugins/keyboard/plugin.yaml b/src/plugins/keyboard/plugin.yaml new file mode 100644 index 0000000000..106bef6cba --- /dev/null +++ b/src/plugins/keyboard/plugin.yaml @@ -0,0 +1,11 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +name: keyboard +description: "Raw keyboard press state (W/A/S/D/Q/E/Z/X/T/G/C/V/K) via Linux evdev device" +command: "./keyboard_plugin" +version: "1.0.0" +devices: + - path: "/keyboard" + type: "keyboard" + description: "Keyboard press state from /dev/input/event*" diff --git a/src/python/isaaccapture/deviceio/__init__.py b/src/python/isaaccapture/deviceio/__init__.py index 728675434b..b144b93e8b 100644 --- a/src/python/isaaccapture/deviceio/__init__.py +++ b/src/python/isaaccapture/deviceio/__init__.py @@ -19,6 +19,7 @@ MessageChannelTracker, FrameMetadataTrackerOak, Generic3AxisPedalTracker, + KeyboardTracker, OgloTactileTracker, TensorPushTracker, JointStateTracker, @@ -49,6 +50,7 @@ StreamType, FrameMetadataOak, Generic3AxisPedalOutput, + KeyboardOutput, OgloGloveSample, ) @@ -60,6 +62,7 @@ "StreamType", "FrameMetadataOak", "Generic3AxisPedalOutput", + "KeyboardOutput", "OgloGloveSample", "ITracker", "HandTracker", @@ -69,6 +72,7 @@ "MessageChannelTracker", "FrameMetadataTrackerOak", "Generic3AxisPedalTracker", + "KeyboardTracker", "OgloTactileTracker", "TensorPushTracker", "JointStateTracker", diff --git a/src/python/isaaccapture/retargeters/__init__.py b/src/python/isaaccapture/retargeters/__init__.py index 341d77f6ad..f474016270 100644 --- a/src/python/isaaccapture/retargeters/__init__.py +++ b/src/python/isaaccapture/retargeters/__init__.py @@ -16,6 +16,9 @@ - LocomotionRootCmdRetargeter: Locomotion from controller inputs - FootPedalRootCmdRetargeter: Root command from 3-axis foot pedal (horizontal/vertical + rudder) - GripperRetargeter: Pinch-based gripper control + - KeyboardToSe3RelRetargeter: Keyboard press state -> relative EE delta control + - KeyboardGripperRetargeter: Keyboard K-key toggle -> gripper open/closed + - KeyboardToSe2Retargeter: Keyboard press state -> base velocity command (v_x, v_y, omega_z) - SO101ClutchRetargeter: Clutch-rebased absolute EE pose for the SO-101 5-DOF arm -- re-latches BOTH home position and orientation on every engage, base-frame left-composed, no fixed offset @@ -107,6 +110,33 @@ # .gripper_retargeter "GripperRetargeter": (".gripper_retargeter", "GripperRetargeter", None), "GripperRetargeterConfig": (".gripper_retargeter", "GripperRetargeterConfig", None), + # .keyboard_se3_retargeter (requires retargeters-lite extra: scipy) + "KeyboardToSe3RelRetargeter": ( + ".keyboard_se3_retargeter", + "KeyboardToSe3RelRetargeter", + "retargeters-lite", + ), + "KeyboardToSe3RelRetargeterConfig": ( + ".keyboard_se3_retargeter", + "KeyboardToSe3RelRetargeterConfig", + "retargeters-lite", + ), + "KeyboardGripperRetargeter": ( + ".keyboard_se3_retargeter", + "KeyboardGripperRetargeter", + "retargeters-lite", + ), + # .keyboard_se2_retargeter + "KeyboardToSe2Retargeter": ( + ".keyboard_se2_retargeter", + "KeyboardToSe2Retargeter", + None, + ), + "KeyboardToSe2RetargeterConfig": ( + ".keyboard_se2_retargeter", + "KeyboardToSe2RetargeterConfig", + None, + ), # .SO101 (SO-101 5-DOF arm: clutch EE-pose, analog gripper) "SO101ClutchRetargeter": ( ".SO101.clutch_retargeter", @@ -239,6 +269,11 @@ def __getattr__(name: str): # Manipulator retargeters "GripperRetargeter", "GripperRetargeterConfig", + "KeyboardToSe3RelRetargeter", + "KeyboardToSe3RelRetargeterConfig", + "KeyboardGripperRetargeter", + "KeyboardToSe2Retargeter", + "KeyboardToSe2RetargeterConfig", # SO-101 5-DOF arm retargeters "SO101ClutchRetargeter", "SO101GripperRetargeter", diff --git a/src/python/isaaccapture/retargeters/keyboard_se2_retargeter.py b/src/python/isaaccapture/retargeters/keyboard_se2_retargeter.py new file mode 100644 index 0000000000..7fc8567876 --- /dev/null +++ b/src/python/isaaccapture/retargeters/keyboard_se2_retargeter.py @@ -0,0 +1,118 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Keyboard SE2 Retargeter Module. + +Maps raw keyboard press state to a base velocity command (v_x, v_y, omega_z). +""" + +import numpy as np +from dataclasses import dataclass + +from isaaccapture.retargeting_engine.deviceio_source_nodes import ( + EvdevKeyCode, + KeyboardAllKeysType, +) +from isaaccapture.retargeting_engine.interface import ( + BaseRetargeter, + RetargeterIOType, +) +from isaaccapture.retargeting_engine.interface.retargeter_core_types import RetargeterIO +from isaaccapture.retargeting_engine.interface.tensor_group_type import ( + TensorGroupType, + OptionalType, +) +from isaaccapture.retargeting_engine.tensor_types import NDArrayType, DLDataType + + +@dataclass +class KeyboardToSe2RetargeterConfig: + """Configuration for the keyboard-to-SE2 base-velocity retargeter.""" + + v_x_sensitivity: float = 0.8 + v_y_sensitivity: float = 0.4 + omega_z_sensitivity: float = 1.0 + + +class KeyboardToSe2Retargeter(BaseRetargeter): + """ + Maps keyboard press state to a 3D base velocity command (v_x, v_y, omega_z). + + Key bindings (matching Isaac Lab's legacy Se2Keyboard): + Numpad 8 / Arrow Up: +v_x Numpad 2 / Arrow Down: -v_x + Numpad 4 / Arrow Left: +v_y Numpad 6 / Arrow Right: -v_y + Numpad 7 / Z: +omega_z Numpad 9 / X: -omega_z + + Consumes the "keyboard_all_keys" bitmap (rather than the fixed 13-key SE3 + subset) since numpad and arrow keys fall outside it. + + Output is the instantaneous command implied by the currently held keys (scaled + by sensitivity), not an integrated velocity -- matching a continuous-axis input + device. + """ + + def __init__(self, config: KeyboardToSe2RetargeterConfig, name: str) -> None: + self._config = config + super().__init__(name=name) + + def input_spec(self) -> RetargeterIOType: + return {"keyboard_all_keys": OptionalType(KeyboardAllKeysType())} + + def output_spec(self) -> RetargeterIOType: + return { + "base_command": TensorGroupType( + "base_command", + [ + NDArrayType( + "velocity", shape=(3,), dtype=DLDataType.FLOAT, dtype_bits=32 + ) + ], + ) + } + + def _compute_fn(self, inputs: RetargeterIO, outputs: RetargeterIO, context) -> None: + base_command = outputs["base_command"] + all_keys = inputs["keyboard_all_keys"] + if all_keys.is_none: + base_command[0] = np.zeros(3, dtype=np.float32) + return + + bitmap = np.asarray(all_keys[0]) + v_x_sens = self._config.v_x_sensitivity + v_y_sens = self._config.v_y_sensitivity + omega_z_sens = self._config.omega_z_sensitivity + + velocity = np.zeros(3) + velocity[0] += ( + v_x_sens + if (bitmap[EvdevKeyCode.KEY_KP8] or bitmap[EvdevKeyCode.KEY_UP]) + else 0.0 + ) + velocity[0] -= ( + v_x_sens + if (bitmap[EvdevKeyCode.KEY_KP2] or bitmap[EvdevKeyCode.KEY_DOWN]) + else 0.0 + ) + velocity[1] += ( + v_y_sens + if (bitmap[EvdevKeyCode.KEY_KP4] or bitmap[EvdevKeyCode.KEY_LEFT]) + else 0.0 + ) + velocity[1] -= ( + v_y_sens + if (bitmap[EvdevKeyCode.KEY_KP6] or bitmap[EvdevKeyCode.KEY_RIGHT]) + else 0.0 + ) + velocity[2] += ( + omega_z_sens + if (bitmap[EvdevKeyCode.KEY_KP7] or bitmap[EvdevKeyCode.KEY_Z]) + else 0.0 + ) + velocity[2] -= ( + omega_z_sens + if (bitmap[EvdevKeyCode.KEY_KP9] or bitmap[EvdevKeyCode.KEY_X]) + else 0.0 + ) + + base_command[0] = velocity.astype(np.float32) diff --git a/src/python/isaaccapture/retargeters/keyboard_se3_retargeter.py b/src/python/isaaccapture/retargeters/keyboard_se3_retargeter.py new file mode 100644 index 0000000000..e245de955a --- /dev/null +++ b/src/python/isaaccapture/retargeters/keyboard_se3_retargeter.py @@ -0,0 +1,157 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Keyboard SE3 Retargeter Module. + +Maps raw keyboard press state to end-effector delta commands and a gripper toggle. +""" + +import numpy as np +from dataclasses import dataclass + +from isaaccapture.retargeting_engine.deviceio_source_nodes import ( + EvdevKeyCode, + KeyboardAllKeysType, +) +from isaaccapture.retargeting_engine.interface import ( + BaseRetargeter, + RetargeterIOType, +) +from isaaccapture.retargeting_engine.interface.retargeter_core_types import RetargeterIO +from isaaccapture.retargeting_engine.interface.tensor_group_type import ( + TensorGroupType, + OptionalType, +) +from isaaccapture.retargeting_engine.tensor_types import ( + NDArrayType, + DLDataType, + FloatType, +) + +from scipy.spatial.transform import Rotation + + +@dataclass +class KeyboardToSe3RelRetargeterConfig: + """Configuration for the keyboard-to-SE3-relative retargeter.""" + + pos_sensitivity: float = 0.4 + rot_sensitivity: float = 0.8 + + +class KeyboardToSe3RelRetargeter(BaseRetargeter): + """ + Maps keyboard press state to a 6D end-effector delta command. + + Key bindings: + W/S: +/-X, A/D: +/-Y, Q/E: +/-Z (position) + Z/X: +/-roll, T/G: +/-pitch, C/V: +/-yaw (rotation) + + Output is the instantaneous command implied by the currently held keys (scaled by + sensitivity), not an integrated delta -- matching a continuous-axis input device. + """ + + def __init__(self, config: KeyboardToSe3RelRetargeterConfig, name: str) -> None: + self._config = config + super().__init__(name=name) + + def input_spec(self) -> RetargeterIOType: + return {"keyboard_all_keys": OptionalType(KeyboardAllKeysType())} + + def output_spec(self) -> RetargeterIOType: + return { + "ee_delta": TensorGroupType( + "ee_delta", + [ + NDArrayType( + "delta", shape=(6,), dtype=DLDataType.FLOAT, dtype_bits=32 + ) + ], + ) + } + + def _compute_fn(self, inputs: RetargeterIO, outputs: RetargeterIO, context) -> None: + ee_delta = outputs["ee_delta"] + all_keys = inputs["keyboard_all_keys"] + if all_keys.is_none: + ee_delta[0] = np.zeros(6, dtype=np.float32) + return + + bitmap = np.asarray(all_keys[0]) + pos_sens = self._config.pos_sensitivity + rot_sens = self._config.rot_sensitivity + + delta_pos = np.zeros(3) + delta_pos[0] += pos_sens if bitmap[EvdevKeyCode.KEY_W] else 0.0 + delta_pos[0] -= pos_sens if bitmap[EvdevKeyCode.KEY_S] else 0.0 + delta_pos[1] += pos_sens if bitmap[EvdevKeyCode.KEY_A] else 0.0 + delta_pos[1] -= pos_sens if bitmap[EvdevKeyCode.KEY_D] else 0.0 + delta_pos[2] += pos_sens if bitmap[EvdevKeyCode.KEY_Q] else 0.0 + delta_pos[2] -= pos_sens if bitmap[EvdevKeyCode.KEY_E] else 0.0 + + delta_euler = np.zeros(3) + delta_euler[0] += rot_sens if bitmap[EvdevKeyCode.KEY_Z] else 0.0 + delta_euler[0] -= rot_sens if bitmap[EvdevKeyCode.KEY_X] else 0.0 + delta_euler[1] += rot_sens if bitmap[EvdevKeyCode.KEY_T] else 0.0 + delta_euler[1] -= rot_sens if bitmap[EvdevKeyCode.KEY_G] else 0.0 + delta_euler[2] += rot_sens if bitmap[EvdevKeyCode.KEY_C] else 0.0 + delta_euler[2] -= rot_sens if bitmap[EvdevKeyCode.KEY_V] else 0.0 + + delta_rot = Rotation.from_euler("XYZ", delta_euler).as_rotvec() + + ee_delta[0] = np.concatenate([delta_pos, delta_rot]).astype(np.float32) + + +class KeyboardGripperRetargeter(BaseRetargeter): + """ + Toggles a gripper open/closed state on each rising edge of the K key. + + Output matches GripperRetargeter's convention: -1.0 when closed, 1.0 when open. + """ + + def __init__(self, name: str) -> None: + super().__init__(name=name) + self._closed = False + self._prev_k_pressed = False + + def input_spec(self) -> RetargeterIOType: + return {"keyboard_all_keys": OptionalType(KeyboardAllKeysType())} + + def output_spec(self) -> RetargeterIOType: + return { + "gripper_command": TensorGroupType( + "gripper_command", [FloatType("command")] + ) + } + + def _compute_fn(self, inputs: RetargeterIO, outputs: RetargeterIO, context) -> None: + gripper_out = outputs["gripper_command"] + all_keys = inputs["keyboard_all_keys"] + k_pressed = ( + False + if all_keys.is_none + else bool(np.asarray(all_keys[0])[EvdevKeyCode.KEY_K]) + ) + + if context.execution_events.reset: + self._closed = False + # Sync to the current key state without toggling -- K may already be + # held on a reset frame, and that isn't a rising edge. Leave + # _prev_k_pressed alone when the device is inactive this frame; + # overwriting it to False would misread a still-held key as a fresh + # rising edge once data resumes. + if not all_keys.is_none: + self._prev_k_pressed = k_pressed + gripper_out[0] = -1.0 if self._closed else 1.0 + return + + if all_keys.is_none: + gripper_out[0] = -1.0 if self._closed else 1.0 + return + + if k_pressed and not self._prev_k_pressed: + self._closed = not self._closed + self._prev_k_pressed = k_pressed + + gripper_out[0] = -1.0 if self._closed else 1.0 diff --git a/src/python/isaaccapture/retargeting_engine/deviceio_source_nodes/__init__.py b/src/python/isaaccapture/retargeting_engine/deviceio_source_nodes/__init__.py index b1b583e572..e7653e14f5 100644 --- a/src/python/isaaccapture/retargeting_engine/deviceio_source_nodes/__init__.py +++ b/src/python/isaaccapture/retargeting_engine/deviceio_source_nodes/__init__.py @@ -11,6 +11,7 @@ from .hands_source import HandsSource from .controllers_source import ControllersSource from .pedals_source import Generic3AxisPedalSource +from .keyboard_source import EvdevKeyCode, KeyboardAllKeysType, KeyboardSource from .joint_state_source import JointStateSource from .full_body_source import FullBodySource from .message_channel_source import MessageChannelSource @@ -26,12 +27,14 @@ HandPoseTrackedType, ControllerSnapshotTrackedType, Generic3AxisPedalOutputTrackedType, + KeyboardOutputTrackedType, JointStateOutputTrackedType, FullBodyPoseTrackedType, DeviceIOHeadPoseTracked, DeviceIOHandPoseTracked, DeviceIOControllerSnapshotTracked, DeviceIOGeneric3AxisPedalOutputTracked, + DeviceIOKeyboardOutputTracked, DeviceIOJointStateOutputTracked, DeviceIOFullBodyPoseTracked, MessageChannelMessagesTrackedType, @@ -49,6 +52,9 @@ "HandsSource", "ControllersSource", "Generic3AxisPedalSource", + "EvdevKeyCode", + "KeyboardAllKeysType", + "KeyboardSource", "JointStateSource", "FullBodySource", "MessageChannelSource", @@ -61,6 +67,7 @@ "HandPoseTrackedType", "ControllerSnapshotTrackedType", "Generic3AxisPedalOutputTrackedType", + "KeyboardOutputTrackedType", "JointStateOutputTrackedType", "FullBodyPoseTrackedType", "MessageChannelMessagesTrackedType", @@ -70,6 +77,7 @@ "DeviceIOHandPoseTracked", "DeviceIOControllerSnapshotTracked", "DeviceIOGeneric3AxisPedalOutputTracked", + "DeviceIOKeyboardOutputTracked", "DeviceIOJointStateOutputTracked", "DeviceIOFullBodyPoseTracked", "DeviceIOMessageChannelMessagesTracked", diff --git a/src/python/isaaccapture/retargeting_engine/deviceio_source_nodes/deviceio_tensor_types.py b/src/python/isaaccapture/retargeting_engine/deviceio_source_nodes/deviceio_tensor_types.py index 5bfad97384..5cebd9a090 100644 --- a/src/python/isaaccapture/retargeting_engine/deviceio_source_nodes/deviceio_tensor_types.py +++ b/src/python/isaaccapture/retargeting_engine/deviceio_source_nodes/deviceio_tensor_types.py @@ -19,6 +19,7 @@ HandPose, ControllerSnapshot, Generic3AxisPedalOutput, + KeyboardOutput, JointStateOutput, FullBodyPose, MessageChannelMessagesTracked, @@ -92,6 +93,12 @@ class Generic3AxisPedalOutputTrackedType(_PayloadTensorType): _payload_cls = Generic3AxisPedalOutput +class KeyboardOutputTrackedType(_PayloadTensorType): + """KeyboardOutput payload from DeviceIO KeyboardTracker.""" + + _payload_cls = KeyboardOutput + + class JointStateOutputTrackedType(_PayloadTensorType): """JointStateOutput payload from DeviceIO JointStateTracker.""" @@ -172,6 +179,18 @@ def DeviceIOGeneric3AxisPedalOutputTracked() -> TensorGroupType: ) +def DeviceIOKeyboardOutputTracked() -> TensorGroupType: + """Tracked keyboard data from DeviceIO KeyboardTracker. + + Contains: + keyboard_tracked: KeyboardOutput handle, or None when inactive + """ + return TensorGroupType( + "deviceio_keyboard_output", + [KeyboardOutputTrackedType("keyboard_tracked")], + ) + + def DeviceIOJointStateOutputTracked() -> TensorGroupType: """Tracked joint-state data from DeviceIO JointStateTracker. diff --git a/src/python/isaaccapture/retargeting_engine/deviceio_source_nodes/keyboard_source.py b/src/python/isaaccapture/retargeting_engine/deviceio_source_nodes/keyboard_source.py new file mode 100644 index 0000000000..0901edef2b --- /dev/null +++ b/src/python/isaaccapture/retargeting_engine/deviceio_source_nodes/keyboard_source.py @@ -0,0 +1,268 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +""" +Keyboard Source Node - DeviceIO to Retargeting Engine converter. + +Converts raw KeyboardOutput flatbuffer data (a set of held evdev key codes, see +linux/input-event-codes.h) to a single standard output: a 256-entry bitmap covering +every standard key. Carries no semantic mapping -- which keys mean what (an axis, a +toggle, a mode switch) is entirely up to the consuming retargeter, which indexes the +bitmap via :class:`EvdevKeyCode`. +""" + +from __future__ import annotations + +from enum import IntEnum +from typing import Any, TYPE_CHECKING +from .interface import IDeviceIOSource +from ..interface.retargeter_core_types import RetargeterIO, RetargeterIOType +from ..interface.tensor_group import TensorGroup +from ..tensor_types import NDArrayType, DLDataType +from ..interface.tensor_group_type import OptionalType, TensorGroupType +from .deviceio_tensor_types import DeviceIOKeyboardOutputTracked + +if TYPE_CHECKING: + from isaaccapture.deviceio import ITracker + from isaaccapture.schema import KeyboardOutput + +# Default collection_id matching the keyboard plugin and KeyboardTracker. +DEFAULT_KEYBOARD_COLLECTION_ID = "keyboard" + + +class EvdevKeyCode(IntEnum): + """Evdev key codes (linux/input-event-codes.h) for every standard PC keyboard key. + + Values double as indices into the "keyboard_all_keys" bitmap. Covers the full + standard keyboard (letters, digits, function keys, navigation/editing cluster, + numpad, punctuation, and modifiers) so any retargeter can bind any of them without + extending this table; multimedia/vendor-specific keys above this range are not + covered. + """ + + KEY_ESC = 1 + KEY_1 = 2 + KEY_2 = 3 + KEY_3 = 4 + KEY_4 = 5 + KEY_5 = 6 + KEY_6 = 7 + KEY_7 = 8 + KEY_8 = 9 + KEY_9 = 10 + KEY_0 = 11 + KEY_MINUS = 12 + KEY_EQUAL = 13 + KEY_BACKSPACE = 14 + KEY_TAB = 15 + KEY_Q = 16 + KEY_W = 17 + KEY_E = 18 + KEY_R = 19 + KEY_T = 20 + KEY_Y = 21 + KEY_U = 22 + KEY_I = 23 + KEY_O = 24 + KEY_P = 25 + KEY_LEFTBRACE = 26 + KEY_RIGHTBRACE = 27 + KEY_ENTER = 28 + KEY_LEFTCTRL = 29 + KEY_A = 30 + KEY_S = 31 + KEY_D = 32 + KEY_F = 33 + KEY_G = 34 + KEY_H = 35 + KEY_J = 36 + KEY_K = 37 + KEY_L = 38 + KEY_SEMICOLON = 39 + KEY_APOSTROPHE = 40 + KEY_GRAVE = 41 + KEY_LEFTSHIFT = 42 + KEY_BACKSLASH = 43 + KEY_Z = 44 + KEY_X = 45 + KEY_C = 46 + KEY_V = 47 + KEY_B = 48 + KEY_N = 49 + KEY_M = 50 + KEY_COMMA = 51 + KEY_DOT = 52 + KEY_SLASH = 53 + KEY_RIGHTSHIFT = 54 + KEY_KPASTERISK = 55 + KEY_LEFTALT = 56 + KEY_SPACE = 57 + KEY_CAPSLOCK = 58 + KEY_F1 = 59 + KEY_F2 = 60 + KEY_F3 = 61 + KEY_F4 = 62 + KEY_F5 = 63 + KEY_F6 = 64 + KEY_F7 = 65 + KEY_F8 = 66 + KEY_F9 = 67 + KEY_F10 = 68 + KEY_NUMLOCK = 69 + KEY_SCROLLLOCK = 70 + KEY_KP7 = 71 + KEY_KP8 = 72 + KEY_KP9 = 73 + KEY_KPMINUS = 74 + KEY_KP4 = 75 + KEY_KP5 = 76 + KEY_KP6 = 77 + KEY_KPPLUS = 78 + KEY_KP1 = 79 + KEY_KP2 = 80 + KEY_KP3 = 81 + KEY_KP0 = 82 + KEY_KPDOT = 83 + KEY_102ND = 86 + KEY_F11 = 87 + KEY_F12 = 88 + KEY_KPENTER = 96 + KEY_RIGHTCTRL = 97 + KEY_KPSLASH = 98 + KEY_SYSRQ = 99 + KEY_RIGHTALT = 100 + KEY_HOME = 102 + KEY_UP = 103 + KEY_PAGEUP = 104 + KEY_LEFT = 105 + KEY_RIGHT = 106 + KEY_END = 107 + KEY_DOWN = 108 + KEY_PAGEDOWN = 109 + KEY_INSERT = 110 + KEY_DELETE = 111 + KEY_KPEQUAL = 117 + KEY_PAUSE = 119 + KEY_LEFTMETA = 125 + KEY_RIGHTMETA = 126 + KEY_COMPOSE = 127 + + +# Every standard PC keyboard key (letters, digits, function keys, navigation, +# modifiers, numpad, punctuation) fits under evdev code 255; anything above that +# is an exotic/vendor key not tracked here. +ALL_KEYS_BITMAP_SIZE = 256 + + +def KeyboardAllKeysType() -> TensorGroupType: + """Type for the "keyboard_all_keys" output: a 256-entry uint8 bitmap indexed by evdev key code.""" + return TensorGroupType( + "keyboard_all_keys", + [ + NDArrayType( + "bitmap", + shape=(ALL_KEYS_BITMAP_SIZE,), + dtype=DLDataType.UINT, + dtype_bits=8, + ) + ], + ) + + +class KeyboardSource(IDeviceIOSource): + """ + Stateless converter: DeviceIO KeyboardOutput → keyboard_all_keys bitmap tensor. + + Inputs: + - "deviceio_keyboard": Raw KeyboardOutput flatbuffer from KeyboardTracker + + Outputs (Optional — absent when the keyboard plugin has not yet streamed): + - "keyboard_all_keys": OptionalTensorGroup, a 256-entry uint8 bitmap + indexed by evdev key code (1 = held, 0 = released; see + :class:`EvdevKeyCode`) covering every standard key. + + Usage: + # In TeleopSession, the keyboard tracker is discovered from the pipeline; + # data is polled via poll_tracker. Or manually: + tracked = keyboard_tracker.get_keyboard_data(session) + result = keyboard_source_node({ + "deviceio_keyboard": TensorGroup(DeviceIOKeyboardOutputTracked(), [tracked]) + }) + """ + + def __init__( + self, name: str, collection_id: str = DEFAULT_KEYBOARD_COLLECTION_ID + ) -> None: + """Initialize stateless keyboard source node. + + Creates a KeyboardTracker instance for TeleopSession to discover and use. + + Args: + name: Unique name for this source node + collection_id: Tensor collection ID for keyboard data (must match the keyboard plugin). + """ + import isaaccapture.deviceio as deviceio + + self._keyboard_tracker = deviceio.KeyboardTracker(collection_id) + self._collection_id = collection_id + super().__init__(name) + + def get_tracker(self) -> "ITracker": + """Get the KeyboardTracker instance. + + Returns: + The KeyboardTracker instance for TeleopSession to initialize + """ + return self._keyboard_tracker + + def poll_tracker(self, deviceio_session: Any) -> RetargeterIO: + """Poll the keyboard tracker and return input data. + + Args: + deviceio_session: The active DeviceIO session. + + Returns: + Dict with "deviceio_keyboard" TensorGroup containing KeyboardOutput | None. + """ + keys = self._keyboard_tracker.get_keyboard_data(deviceio_session) + tg = TensorGroup(DeviceIOKeyboardOutputTracked()) + tg[0] = keys + return {"deviceio_keyboard": tg} + + def input_spec(self) -> RetargeterIOType: + """Declare DeviceIO keyboard input.""" + return { + "deviceio_keyboard": DeviceIOKeyboardOutputTracked(), + } + + def output_spec(self) -> RetargeterIOType: + """Declare standard keyboard input output (Optional — may be absent).""" + return { + "keyboard_all_keys": OptionalType(KeyboardAllKeysType()), + } + + def _compute_fn(self, inputs: RetargeterIO, outputs: RetargeterIO, context) -> None: + """ + Convert DeviceIO KeyboardOutput to the standard keyboard_all_keys bitmap. + + Calls ``set_none()`` on the output when the keyboard plugin has not yet streamed. + + Args: + inputs: Dict with "deviceio_keyboard" containing KeyboardOutput | None + outputs: Dict with "keyboard_all_keys" OptionalTensorGroup + context: Shared ComputeContext for the current step (carries GraphTime). + """ + import numpy as np + + keys: KeyboardOutput | None = inputs["deviceio_keyboard"][0] + + all_keys_out = outputs["keyboard_all_keys"] + if keys is None: + all_keys_out.set_none() + return + + bitmap = np.zeros(ALL_KEYS_BITMAP_SIZE, dtype=np.uint8) + for code in keys.pressed_keys: + if code < ALL_KEYS_BITMAP_SIZE: + bitmap[code] = 1 + all_keys_out[0] = bitmap diff --git a/src/python/isaaccapture/schema/__init__.py b/src/python/isaaccapture/schema/__init__.py index b994e674a5..4438cebf6f 100644 --- a/src/python/isaaccapture/schema/__init__.py +++ b/src/python/isaaccapture/schema/__init__.py @@ -38,6 +38,9 @@ # Pedals-related types. Generic3AxisPedalOutput, Generic3AxisPedalOutputRecord, + # Keyboard types (raw keyboard press state). + KeyboardOutput, + KeyboardOutputRecord, # OGLO tactile glove types. OgloGloveSample, OgloGloveSampleRecord, @@ -135,6 +138,9 @@ def __getattr__(name: str): # Pedals types. "Generic3AxisPedalOutput", "Generic3AxisPedalOutputRecord", + # Keyboard types (raw keyboard press state). + "KeyboardOutput", + "KeyboardOutputRecord", # OGLO tactile glove types. "OgloGloveSample", "OgloGloveSampleRecord", diff --git a/tests/python/core/retargeting_engine/test_keyboard_retargeter.py b/tests/python/core/retargeting_engine/test_keyboard_retargeter.py new file mode 100644 index 0000000000..71c1a68015 --- /dev/null +++ b/tests/python/core/retargeting_engine/test_keyboard_retargeter.py @@ -0,0 +1,239 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Sim-free unit tests for KeyboardToSe3RelRetargeter, KeyboardGripperRetargeter, and +KeyboardToSe2Retargeter, exercised through KeyboardSource so a regression anywhere in the +schema -> source -> retargeter chain (a field rename, an index drift, a sign flip) fails +here, with no OpenXR device involved. +""" + +import numpy as np +import pytest + +from isaaccapture.retargeting_engine.deviceio_source_nodes import KeyboardSource +from isaaccapture.retargeting_engine.interface.base_retargeter import _make_output_group +from isaaccapture.retargeting_engine.interface.execution_events import ExecutionEvents +from isaaccapture.retargeting_engine.interface.retargeter_core_types import ( + ComputeContext, +) +from isaaccapture.retargeting_engine.interface.tensor_group import TensorGroup +from isaaccapture.retargeters import ( + KeyboardGripperRetargeter, + KeyboardToSe2Retargeter, + KeyboardToSe2RetargeterConfig, + KeyboardToSe3RelRetargeter, + KeyboardToSe3RelRetargeterConfig, +) +from isaaccapture.schema import KeyboardOutput + +# Evdev key codes (linux/input-event-codes.h), matching keyboard_plugin.cpp / KeyboardSource. +KEY_W, KEY_A, KEY_S, KEY_D, KEY_Q, KEY_E = 17, 30, 31, 32, 16, 18 +KEY_K = 37 # gripper toggle +KEY_UP, KEY_DOWN, KEY_LEFT, KEY_RIGHT = 103, 108, 105, 106 +KEY_KP8, KEY_KP9 = 72, 73 + + +def _keyboard_source(): + return KeyboardSource(name="keyboard") + + +def _run_source(src, pressed_keys: list[int] | None): + """Feed raw pressed_keys (None = inactive device) through KeyboardSource.compute().""" + keys = None if pressed_keys is None else KeyboardOutput(pressed_keys, True) + + input_spec = src.input_spec() + tg = TensorGroup(input_spec["deviceio_keyboard"]) + tg[0] = keys + + outputs = {name: _make_output_group(gt) for name, gt in src.output_spec().items()} + src.compute({"deviceio_keyboard": tg}, outputs) + return outputs + + +class TestKeyboardToSe3RelRetargeter: + def test_movement_key_produces_ee_delta(self): + """W held -> KeyboardSource -> KeyboardToSe3RelRetargeter -> +X delta.""" + src = _keyboard_source() + src_outputs = _run_source(src, [KEY_W]) + + retargeter = KeyboardToSe3RelRetargeter( + KeyboardToSe3RelRetargeterConfig(), name="se3" + ) + out = {"ee_delta": _make_output_group(retargeter.output_spec()["ee_delta"])} + retargeter.compute({"keyboard_all_keys": src_outputs["keyboard_all_keys"]}, out) + + delta = np.asarray(out["ee_delta"][0]) + assert delta[0] == pytest.approx(0.4) # default pos_sensitivity + assert np.allclose(delta[1:], 0.0) + + def test_opposing_keys_combine(self): + """W (+X) and Q (+Z) held together combine on independent axes.""" + src = _keyboard_source() + src_outputs = _run_source(src, [KEY_W, KEY_Q]) + + retargeter = KeyboardToSe3RelRetargeter( + KeyboardToSe3RelRetargeterConfig(), name="se3" + ) + out = {"ee_delta": _make_output_group(retargeter.output_spec()["ee_delta"])} + retargeter.compute({"keyboard_all_keys": src_outputs["keyboard_all_keys"]}, out) + + delta = np.asarray(out["ee_delta"][0]) + assert delta[0] == pytest.approx(0.4) # W: +X + assert delta[2] == pytest.approx(0.4) # Q: +Z + assert delta[1] == pytest.approx(0.0) + assert np.allclose(delta[3:], 0.0) # no rotation keys held + + def test_inactive_device_yields_zero_delta(self): + src = _keyboard_source() + src_outputs = _run_source(src, None) + + se3 = KeyboardToSe3RelRetargeter(KeyboardToSe3RelRetargeterConfig(), name="se3") + se3_out = {"ee_delta": _make_output_group(se3.output_spec()["ee_delta"])} + se3.compute({"keyboard_all_keys": src_outputs["keyboard_all_keys"]}, se3_out) + assert np.allclose(np.asarray(se3_out["ee_delta"][0]), 0.0) + + +class TestKeyboardGripperRetargeter: + def test_gripper_toggles_on_rising_edge_only(self): + """K press/release/press across three frames toggles exactly on each rising edge.""" + src = _keyboard_source() + retargeter = KeyboardGripperRetargeter(name="gripper") + + def step(pressed_keys): + src_outputs = _run_source(src, pressed_keys) + out = { + "gripper_command": _make_output_group( + retargeter.output_spec()["gripper_command"] + ) + } + retargeter.compute( + {"keyboard_all_keys": src_outputs["keyboard_all_keys"]}, out + ) + return float(out["gripper_command"][0]) + + assert step([]) == pytest.approx(1.0) # open (default) + assert step([KEY_K]) == pytest.approx(-1.0) # rising edge -> close + assert step([KEY_K]) == pytest.approx( + -1.0 + ) # held -> stays closed, no re-toggle + assert step([]) == pytest.approx(-1.0) # release -> stays closed + assert step([KEY_K]) == pytest.approx(1.0) # rising edge again -> open + + def test_inactive_device_yields_default_open(self): + src = _keyboard_source() + src_outputs = _run_source(src, None) + + gripper = KeyboardGripperRetargeter(name="gripper") + gripper_out = { + "gripper_command": _make_output_group( + gripper.output_spec()["gripper_command"] + ) + } + gripper.compute( + {"keyboard_all_keys": src_outputs["keyboard_all_keys"]}, gripper_out + ) + assert float(gripper_out["gripper_command"][0]) == pytest.approx( + 1.0 + ) # default open + + def test_reset_does_not_toggle_gripper_while_k_is_held(self): + """K held across a reset frame is not a rising edge and must not toggle the gripper.""" + src = _keyboard_source() + retargeter = KeyboardGripperRetargeter(name="gripper") + + def step(pressed_keys, reset=False): + src_outputs = _run_source(src, pressed_keys) + out = { + "gripper_command": _make_output_group( + retargeter.output_spec()["gripper_command"] + ) + } + context = ComputeContext(execution_events=ExecutionEvents(reset=reset)) + retargeter.compute( + {"keyboard_all_keys": src_outputs["keyboard_all_keys"]}, out, context + ) + return float(out["gripper_command"][0]) + + assert step([KEY_K]) == pytest.approx(-1.0) # rising edge -> close + # Reset resets the gripper to open, but K is still held -- not a new rising + # edge, so this must not immediately re-close it. + assert step([KEY_K], reset=True) == pytest.approx(1.0) + assert step([KEY_K]) == pytest.approx(1.0) # still held -> stays open + assert step([]) == pytest.approx(1.0) # release + assert step([KEY_K]) == pytest.approx(-1.0) # genuine rising edge -> close + + def test_reset_with_inactive_device_preserves_prior_edge_state(self): + """A reset frame with no keyboard data must not clobber _prev_k_pressed.""" + src = _keyboard_source() + retargeter = KeyboardGripperRetargeter(name="gripper") + + def step(pressed_keys, reset=False): + src_outputs = _run_source(src, pressed_keys) + out = { + "gripper_command": _make_output_group( + retargeter.output_spec()["gripper_command"] + ) + } + context = ComputeContext(execution_events=ExecutionEvents(reset=reset)) + retargeter.compute( + {"keyboard_all_keys": src_outputs["keyboard_all_keys"]}, out, context + ) + return float(out["gripper_command"][0]) + + assert step([KEY_K]) == pytest.approx(-1.0) # rising edge -> close + # Reset while the device is inactive (keyboard.is_none) -- must not force + # _prev_k_pressed to False, or the next frame (K still held) would be + # misread as a fresh rising edge. + assert step(None, reset=True) == pytest.approx(1.0) # gripper still resets + assert step([KEY_K]) == pytest.approx(1.0) # still held -> no spurious toggle + + +class TestKeyboardToSe2Retargeter: + def test_arrow_and_numpad_keys_combine(self): + """Arrow-Up (+v_x) and Numpad-9 (-omega_z) held together -> combined base_command.""" + src = _keyboard_source() + src_outputs = _run_source(src, [KEY_UP, KEY_KP9]) + + retargeter = KeyboardToSe2Retargeter( + KeyboardToSe2RetargeterConfig(), name="se2" + ) + out = { + "base_command": _make_output_group(retargeter.output_spec()["base_command"]) + } + retargeter.compute({"keyboard_all_keys": src_outputs["keyboard_all_keys"]}, out) + + velocity = np.asarray(out["base_command"][0]) + assert velocity[0] == pytest.approx(0.8) # default v_x_sensitivity + assert velocity[1] == pytest.approx(0.0) + assert velocity[2] == pytest.approx(-1.0) # default omega_z_sensitivity + + def test_aliased_keys_do_not_double_count(self): + """Numpad-8 and Arrow-Up both map to +v_x; holding both is not double the sensitivity.""" + src = _keyboard_source() + src_outputs = _run_source(src, [KEY_UP, KEY_KP8]) + + retargeter = KeyboardToSe2Retargeter( + KeyboardToSe2RetargeterConfig(), name="se2" + ) + out = { + "base_command": _make_output_group(retargeter.output_spec()["base_command"]) + } + retargeter.compute({"keyboard_all_keys": src_outputs["keyboard_all_keys"]}, out) + + velocity = np.asarray(out["base_command"][0]) + assert velocity[0] == pytest.approx(0.8) + assert np.allclose(velocity[1:], 0.0) + + def test_inactive_device_yields_zero_velocity(self): + src = _keyboard_source() + src_outputs = _run_source(src, None) + + retargeter = KeyboardToSe2Retargeter( + KeyboardToSe2RetargeterConfig(), name="se2" + ) + out = { + "base_command": _make_output_group(retargeter.output_spec()["base_command"]) + } + retargeter.compute({"keyboard_all_keys": src_outputs["keyboard_all_keys"]}, out) + + assert np.allclose(np.asarray(out["base_command"][0]), 0.0) diff --git a/tests/python/core/retargeting_engine/test_keyboard_source.py b/tests/python/core/retargeting_engine/test_keyboard_source.py new file mode 100644 index 0000000000..1412d1e0b7 --- /dev/null +++ b/tests/python/core/retargeting_engine/test_keyboard_source.py @@ -0,0 +1,68 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the KeyboardSource DeviceIO converter. + +Exercises the stateless converter from a raw ``KeyboardOutput`` FlatBuffer (constructed via +the real schema Python bindings) into the 256-entry keyboard_all_keys bitmap, with no OpenXR +device involved. +""" + +import numpy as np + +from isaaccapture.retargeting_engine.deviceio_source_nodes import KeyboardSource +from isaaccapture.retargeting_engine.interface.base_retargeter import _make_output_group +from isaaccapture.retargeting_engine.interface.tensor_group import TensorGroup +from isaaccapture.schema import KeyboardOutput + +# Evdev key codes (linux/input-event-codes.h), matching keyboard_plugin.cpp / KeyboardSource. +KEY_W = 17 +KEY_F1 = 59 # exercises "every key" coverage, not just the SE2/SE3 bindings + + +def _keyboard_source(): + return KeyboardSource(name="keyboard") + + +def _run_source(src, pressed_keys: list[int] | None): + """Feed raw pressed_keys (None = inactive device) through KeyboardSource.compute().""" + keys = None if pressed_keys is None else KeyboardOutput(pressed_keys, True) + + input_spec = src.input_spec() + tg = TensorGroup(input_spec["deviceio_keyboard"]) + tg[0] = keys + + outputs = {name: _make_output_group(gt) for name, gt in src.output_spec().items()} + src.compute({"deviceio_keyboard": tg}, outputs) + return outputs + + +class TestKeyboardSource: + def test_source_creates_real_tracker(self): + src = _keyboard_source() + tracker = src.get_tracker() + assert tracker is not None + assert tracker.get_name() == "KeyboardTracker" + + def test_movement_key_marks_bitmap(self): + """W held shows up in the all-keys bitmap.""" + src = _keyboard_source() + outputs = _run_source(src, [KEY_W]) + + assert not outputs["keyboard_all_keys"].is_none + bitmap = np.asarray(outputs["keyboard_all_keys"][0]) + assert bitmap[KEY_W] == 1 + + def test_inactive_device_yields_none(self): + src = _keyboard_source() + outputs = _run_source(src, None) + assert outputs["keyboard_all_keys"].is_none + + def test_all_keys_bitmap_covers_keys_outside_se3_subset(self): + """F1 (not bound by any SE2/SE3 retargeter) still shows up in keyboard_all_keys.""" + src = _keyboard_source() + outputs = _run_source(src, [KEY_F1]) + + bitmap = np.asarray(outputs["keyboard_all_keys"][0]) + assert bitmap[KEY_F1] == 1 + assert bitmap.sum() == 1 # only F1 is set diff --git a/tests/python/core/schema/test_keyboard.py b/tests/python/core/schema/test_keyboard.py new file mode 100644 index 0000000000..ea1cc2e126 --- /dev/null +++ b/tests/python/core/schema/test_keyboard.py @@ -0,0 +1,103 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for KeyboardOutput type in isaaccapture.schema. + +Tests the following FlatBuffers types: +- KeyboardOutput: Table with pressed_keys (evdev key codes) and is_valid +- KeyboardOutputRecord: Record wrapper carrying DeviceDataTimestamp + +Timestamps are carried by KeyboardOutputRecord, not KeyboardOutput. +""" + +from isaaccapture.schema import ( + DeviceDataTimestamp, + KeyboardOutput, + KeyboardOutputRecord, +) + +# Evdev key codes (linux/input-event-codes.h), matching keyboard_plugin.cpp. +KEY_W, KEY_A = 17, 30 + + +class TestKeyboardOutputConstruction: + """Tests for KeyboardOutput table construction.""" + + def test_construction(self): + """Test construction with explicit fields.""" + output = KeyboardOutput(pressed_keys=[KEY_W], is_valid=True) + + assert list(output.pressed_keys) == [KEY_W] + assert output.is_valid is True + + def test_repr(self): + """Test __repr__ returns meaningful string.""" + output = KeyboardOutput(pressed_keys=[], is_valid=False) + repr_str = repr(output) + + assert "KeyboardOutput" in repr_str + + +class TestKeyboardOutputPressedKeys: + """Tests that pressed_keys round-trips through the encoding.""" + + def test_empty_pressed_keys(self): + """Test encoding with no keys held.""" + output = KeyboardOutput(pressed_keys=[], is_valid=True) + + assert list(output.pressed_keys) == [] + + def test_multiple_pressed_keys(self): + """Test encoding multiple simultaneously-held keys.""" + output = KeyboardOutput(pressed_keys=[KEY_W, KEY_A], is_valid=True) + + assert list(output.pressed_keys) == [KEY_W, KEY_A] + + def test_encodings_are_independent(self): + """Test each encoding carries its own values, not a shared buffer's.""" + first = KeyboardOutput(pressed_keys=[KEY_W], is_valid=True) + second = KeyboardOutput(pressed_keys=[KEY_A], is_valid=True) + + assert list(first.pressed_keys) == [KEY_W] + assert list(second.pressed_keys) == [KEY_A] + + +class TestKeyboardOutputEncoding: + """Tests that an encoded payload reads back. + + A tracker with no keyboard data returns None rather than an empty payload, so + absence needs no case here; the source-node tests cover feeding None through. + """ + + def test_encoded_payload_reads_back(self): + """An encoded payload gates as True and its fields read directly.""" + output = KeyboardOutput(pressed_keys=[KEY_W], is_valid=True) + + assert output + assert list(output.pressed_keys) == [KEY_W] + assert output.is_valid is True + + def test_repr_present(self): + """Repr of a present payload names the type.""" + assert "KeyboardOutput" in repr(KeyboardOutput(pressed_keys=[], is_valid=True)) + + +class TestKeyboardOutputRecordTimestamp: + """Tests for KeyboardOutputRecord with DeviceDataTimestamp.""" + + def test_construction_with_timestamp(self): + """Test KeyboardOutputRecord carries DeviceDataTimestamp.""" + data = KeyboardOutput(pressed_keys=[KEY_W], is_valid=True) + ts = DeviceDataTimestamp(1000000000, 2000000000, 3000000000) + record = KeyboardOutputRecord(data, ts) + + assert record.timestamp.available_time_local_common_clock == 1000000000 + assert record.timestamp.sample_time_local_common_clock == 2000000000 + assert record.timestamp.sample_time_raw_device_clock == 3000000000 + assert list(record.data.pressed_keys) == [KEY_W] + + def test_payload_less_record(self): + """A record may carry a timestamp and no payload: MCAP's frame sentinel.""" + record = KeyboardOutputRecord(None, DeviceDataTimestamp(1, 2, 3)) + assert record.data is None + assert record.timestamp.available_time_local_common_clock == 1