Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 40 additions & 14 deletions apps/crazy_robotaxi/crazy_robotaxi/controls.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,12 @@
ControlDevice = Literal["keyboard", "gamepad", "wheel"]
ControlDirection = Literal["negative", "positive", "bidirectional"]
GamepadButtonStyle = Literal["Xbox", "PlayStation", "Nintendo Switch"]
_GAMEPAD_STEERING_DEADZONE = 0.15
"""Centered stick dead zone, rescaled so full deflection remains full lock."""

_GAMEPAD_PEDAL_DEADZONE = 0.05
"""Trigger dead zone, rescaled so the usable pedal range still reaches one."""

_GAMEPAD_BUTTON_NAMES: dict[GamepadButtonStyle, tuple[str, ...]] = {
"Xbox": (
"A",
Expand Down Expand Up @@ -520,11 +526,12 @@ def keyboard_driver_command(
steer = _keyboard_action_value(settings.steer_left, pressed)
steer -= _keyboard_action_value(settings.steer_right, pressed)
return DriverCommand(
throttle=1.0 if forward != reverse and not handbrake else 0.0,
brake=1.0 if handbrake else 0.0,
throttle=1.0 if forward and not reverse and not handbrake else 0.0,
brake=1.0 if reverse and not forward and not handbrake else 0.0,
steer=steer,
reverse=reverse and not forward,
manual_control=handbrake,
handbrake=handbrake,
steer_is_direct=True,
manual_control=True,
)


Expand Down Expand Up @@ -558,9 +565,18 @@ def gamepad_driver_command(
if event.action != "state":
return None
return DriverCommand(
throttle=_event_action_value(settings.throttle, event),
brake=_event_action_value(settings.brake, event),
steer=_steering_value(settings.steer, event),
throttle=_rescale_deadzone(
_event_action_value(settings.throttle, event),
_GAMEPAD_PEDAL_DEADZONE,
),
brake=_rescale_deadzone(
_event_action_value(settings.brake, event),
_GAMEPAD_PEDAL_DEADZONE,
),
steer=_rescale_deadzone(
_steering_value(settings.steer, event),
_GAMEPAD_STEERING_DEADZONE,
),
handbrake=_event_action_value(settings.handbrake, event) > 0.5,
steer_is_direct=True,
manual_control=True,
Expand Down Expand Up @@ -1005,9 +1021,8 @@ def _binding_value(
if binding.kind == "button":
index = int(binding.code)
if isinstance(event, GamepadUserInputEvent):
analog = event.buttons[index] if index < len(event.buttons) else 0.0
digital = index < len(event.pressed) and event.pressed[index]
return max(analog, float(digital))
values = _gamepad_button_values(event)
return values[index] if index < len(values) else 0.0
return float(event.buttons[index]) if index < len(event.buttons) else 0.0
if isinstance(event, GamepadUserInputEvent):
index = int(binding.code)
Expand All @@ -1019,6 +1034,16 @@ def _binding_value(
return max(0.0, value if binding.direction == "positive" else -value)


def _rescale_deadzone(value: float, deadzone: float) -> float:
"""Remove centered analog noise without reducing the reachable range."""
value = min(1.0, max(-1.0, value))
magnitude = abs(value)
if magnitude <= deadzone:
return 0.0
scaled = (magnitude - deadzone) / (1.0 - deadzone)
return scaled if value > 0.0 else -scaled


def _moved_numeric(
values: Sequence[float], baseline: Sequence[float]
) -> tuple[int, float] | None:
Expand Down Expand Up @@ -1046,12 +1071,13 @@ def _pressed_numeric(values: Sequence[float], baseline: Sequence[float]) -> int


def _gamepad_button_values(event: GamepadUserInputEvent) -> tuple[float, ...]:
"""Return button values including clients that only report digital state."""
"""Prefer analog button values, falling back to digital-only clients."""
size = max(len(event.buttons), len(event.pressed))
return tuple(
max(
event.buttons[index] if index < len(event.buttons) else 0.0,
float(event.pressed[index]) if index < len(event.pressed) else 0.0,
(
event.buttons[index]
if index < len(event.buttons)
else float(event.pressed[index])
)
for index in range(size)
)
Expand Down
18 changes: 17 additions & 1 deletion apps/crazy_robotaxi/crazy_robotaxi/dynamics.py
Original file line number Diff line number Diff line change
Expand Up @@ -121,7 +121,23 @@ def integrate_taxi_vehicle(
) -> VehicleState:
steer_rad = state.steer_rad
if command.steer_is_direct:
steer_rad = command.steer * vehicle.max_steer_rad
target_steer_rad = (
float(np.clip(command.steer, -1.0, 1.0)) * vehicle.max_steer_rad
)
returning = target_steer_rad == 0.0 or (
steer_rad * target_steer_rad > 0.0
and abs(target_steer_rad) < abs(steer_rad)
)
response_rate = (
vehicle.steer_return_rate_rad_per_s
if returning
else vehicle.steer_rate_rad_per_s
)
steer_rad = _move_towards(
steer_rad,
target_steer_rad,
response_rate * dt_s,
)
elif abs(command.steer) > 1e-5:
steer_rad += command.steer * vehicle.steer_rate_rad_per_s * dt_s
else:
Expand Down
82 changes: 82 additions & 0 deletions apps/crazy_robotaxi/tests/test_controls.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,22 @@ def test_keyboard_bindings_drive_and_dispatch_actions() -> None:
assert driver_input.command().throttle == 1.0


def test_keyboard_driving_uses_the_shared_arcade_command_shape() -> None:
settings = ControlsConfig().keyboard

reverse_left = keyboard_driver_command(settings, {"a", "s"})
handbrake = keyboard_driver_command(settings, {"space"})

assert reverse_left.steer == 1.0
assert reverse_left.steer_is_direct
assert reverse_left.manual_control
assert reverse_left.brake == 1.0
assert reverse_left.throttle == 0.0
assert not reverse_left.reverse
assert handbrake.handbrake
assert handbrake.brake == 0.0


def test_gamepad_return_to_menu_uses_digital_or_analog_button_state() -> None:
settings = ControlsConfig().gamepad
digital = GamepadUserInputEvent(
Expand All @@ -85,6 +101,52 @@ def test_gamepad_return_to_menu_uses_digital_or_analog_button_state() -> None:
assert gamepad_driver_command(settings, analog) is not None


def test_gamepad_axes_have_rescaled_deadzones() -> None:
settings = ControlsConfig().gamepad
noisy = GamepadUserInputEvent(
timestamp=np.uint64(1),
axes=(-0.1,),
buttons=(*((0.0,) * 6), 0.03, 0.04),
pressed=(*((False,) * 6), True, True),
)
halfway = GamepadUserInputEvent(
timestamp=np.uint64(2),
axes=(-0.575,),
buttons=(*((0.0,) * 6), 0.525, 0.525),
)

neutral = gamepad_driver_command(settings, noisy)
half = gamepad_driver_command(settings, halfway)

assert neutral is not None
assert neutral.steer == neutral.throttle == neutral.brake == 0.0
assert half is not None
assert half.steer == pytest.approx(0.5)
assert half.throttle == pytest.approx(0.5)
assert half.brake == pytest.approx(0.5)


def test_full_gamepad_and_keyboard_inputs_produce_the_same_drive_commands() -> None:
controls = ControlsConfig()
gamepad_forward_left = GamepadUserInputEvent(
timestamp=np.uint64(1),
axes=(-1.0,),
buttons=(*((0.0,) * 7), 1.0),
)
gamepad_reverse = replace(
gamepad_forward_left,
axes=(0.0,),
buttons=(*((0.0,) * 6), 1.0, 0.0),
)

assert keyboard_driver_command(
controls.keyboard, {"w", "a"}
) == gamepad_driver_command(controls.gamepad, gamepad_forward_left)
assert keyboard_driver_command(controls.keyboard, {"s"}) == gamepad_driver_command(
controls.gamepad, gamepad_reverse
)


def test_gamepad_defaults_use_standard_button_indices() -> None:
controls = ControlsConfig()
gamepad = controls.gamepad
Expand Down Expand Up @@ -167,6 +229,26 @@ def test_axis_capture_ignores_baseline_and_derives_steering_inversion() -> None:
assert capture_binding("gamepad", "steering", baseline, held) is None


def test_button_capture_prefers_analog_value_over_pressed_state() -> None:
baseline = GamepadUserInputEvent(
timestamp=np.uint64(1),
action="state",
buttons=(0.0,) * 8,
pressed=(False,) * 8,
)
light_trigger = replace(
baseline,
buttons=(*((0.0,) * 7), 0.1),
pressed=(*((False,) * 7), True),
)
deliberate_trigger = replace(light_trigger, buttons=(*((0.0,) * 7), 0.75))

assert capture_binding("gamepad", "scalar", light_trigger, baseline) is None
assert capture_binding(
"gamepad", "scalar", deliberate_trigger, baseline
) == InputBinding("button", 7)


def test_controls_document_round_trips_sparse_yaml_and_comments(tmp_path: Path) -> None:
path = tmp_path / "keyboard.yaml"
path.write_text(
Expand Down
30 changes: 30 additions & 0 deletions apps/crazy_robotaxi/tests/test_gameplay.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,36 @@ def test_taxi_brake_from_rest_enters_reverse() -> None:
assert result.speed_mps < 0.0


def test_direct_steering_preserves_keyboard_arcade_response() -> None:
vehicle = TaxiVehicleConfig()
direct = integrate_taxi_vehicle(
_state(),
DriverCommand(steer=1.0, steer_is_direct=True),
dt_s=0.1,
vehicle=vehicle,
)
legacy_keyboard = integrate_taxi_vehicle(
_state(),
DriverCommand(steer=1.0),
dt_s=0.1,
vehicle=vehicle,
)

assert direct.steer_rad == pytest.approx(legacy_keyboard.steer_rad)

half_lock = _state()
half_lock.steer_rad = vehicle.max_steer_rad * 0.5
released = integrate_taxi_vehicle(
half_lock,
DriverCommand(steer_is_direct=True),
dt_s=0.1,
vehicle=vehicle,
)
assert released.steer_rad == pytest.approx(
half_lock.steer_rad - vehicle.steer_return_rate_rad_per_s * 0.1
)


def test_fare_and_game_over_flow_reaches_v2_name_entry(tmp_path: Path) -> None:
store = HighScoreStore(tmp_path / "scores.csv")
controller = _controller(
Expand Down
33 changes: 28 additions & 5 deletions apps/omnidreams_game_engine/omnidreams_game_engine/input.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ class DriverInput:
"""Normalized keyboard driving keys currently held down."""

controller_command: DriverCommand | None = None
"""Latest wheel or gamepad command; ``None`` enables keyboard input."""
"""Latest wheel or gamepad command, retained while keyboard is active."""

keyboard_command: Callable[[set[str]], DriverCommand] = field(
default=lambda keys: _keyboard_command(keys)
Expand Down Expand Up @@ -70,6 +70,8 @@ class DriverInput:
init=False,
repr=False,
)
_use_controller: bool = field(default=False, init=False, repr=False)
"""Whether the most recent deliberate driving input came from a controller."""

def apply(self, events: UserInputEvents) -> tuple[float, ...]:
"""Retain new input and return command-transition times in seconds."""
Expand Down Expand Up @@ -154,20 +156,21 @@ def sample(

def command(self) -> DriverCommand:
"""Return the command represented by the current retained input state."""
if self.controller_command is not None:
if self._use_controller and self.controller_command is not None:
return self.controller_command
return self.keyboard_command(self.pressed_keys)

def source(self) -> str:
"""Return the currently active input source."""
if self.controller_command is not None:
if self._use_controller and self.controller_command is not None:
return "wheel/gamepad"
return "keyboard" if self.pressed_keys else "idle"

def reset(self) -> None:
"""Clear retained, sampled, and pending driving input."""
self.pressed_keys.clear()
self.controller_command = None
self._use_controller = False
self._sampled_command = DriverCommand()
self._pending_transitions.clear()

Expand All @@ -183,17 +186,37 @@ def _apply_event(self, event: object) -> bool:
return False
if event.state is KeyboardInputState.PRESSED:
self.pressed_keys.add(key)
self._use_controller = False
else:
self.pressed_keys.discard(key)
return True
if isinstance(event, GameWheelUserInputEvent):
self.controller_command = self.wheel_command(event)
self._apply_controller_command(self.wheel_command(event))
return True
if isinstance(event, GamepadUserInputEvent):
self.controller_command = self.gamepad_command(event)
self._apply_controller_command(self.gamepad_command(event))
return True
return False

def _apply_controller_command(self, command: DriverCommand | None) -> None:
"""Switch sources only when a controller supplies deliberate input."""
self.controller_command = command
if command is None:
self._use_controller = False
elif self._use_controller or _command_has_input(command):
self._use_controller = True


def _command_has_input(command: DriverCommand) -> bool:
return (
abs(command.throttle) > 0.01
or abs(command.brake) > 0.01
or abs(command.steer) > 0.01
or command.stop
or command.handbrake
or command.reverse
)


def _keyboard_command(pressed_keys: set[str]) -> DriverCommand:
"""Map retained keyboard state to a simulation command."""
Expand Down
Loading
Loading