Push Butt

This commit is contained in:
firestar5683
2026-08-29 22:13:29 -05:00
parent d426054dec
commit aa9eeae40e
14 changed files with 716 additions and 84 deletions
Binary file not shown.
+1
View File
@@ -369,6 +369,7 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> keys = {
{"StarPilotCarParamsPersistent", {PERSISTENT, BYTES, "", ""}},
{"StarPilotDongleId", {PERSISTENT | DONT_LOG, STRING, "", "", 0}},
{"StarPilotFavoriteSlots", {PERSISTENT, JSON, "[]", "[]", 1}},
{"ControllerActionSlots", {PERSISTENT, JSON, "[]", "[]", 1}},
{"WheelControlLearnSlot", {CLEAR_ON_MANAGER_START | DONT_LOG, INT}},
{"WheelControlMappings", {PERSISTENT, JSON, "[]", "[]", 1}},
{"WheelControlStatus", {CLEAR_ON_MANAGER_START | DONT_LOG, JSON, "{}", "{}"}},
Binary file not shown.
+18 -10
View File
@@ -480,20 +480,12 @@ def trigger_favorite_action(key: str | None, params_memory: Params | None = None
return True
def toggle_favorite_slot(slot_index: int, params: Params | None = None, params_memory: Params | None = None, *,
def execute_favorite_key(key: str | None, params: Params | None = None, params_memory: Params | None = None, *,
eligible_keys: Iterable[str] | None = None) -> bool:
"""Universal polymorphic favorite execution dispatcher."""
if slot_index < 0 or slot_index >= FAVORITE_SLOT_COUNT:
return False
params = params or Params(return_defaults=True)
eligible_keys = set(eligible_keys) if eligible_keys is not None else None
slots = load_favorite_slots(params, eligible_keys=eligible_keys)
slot = slots[slot_index]
key = slot.get("key")
if not slot.get("enabled") or not key:
if not favorite_key_is_valid(params, key, eligible_keys=eligible_keys):
return False
if not is_param_action_safe_onroad(key, params):
return False
@@ -516,6 +508,22 @@ def toggle_favorite_slot(slot_index: int, params: Params | None = None, params_m
return False
def toggle_favorite_slot(slot_index: int, params: Params | None = None, params_memory: Params | None = None, *,
eligible_keys: Iterable[str] | None = None) -> bool:
"""Universal polymorphic favorite execution dispatcher."""
if slot_index < 0 or slot_index >= FAVORITE_SLOT_COUNT:
return False
params = params or Params(return_defaults=True)
eligible_keys = set(eligible_keys) if eligible_keys is not None else None
slots = load_favorite_slots(params, eligible_keys=eligible_keys)
slot = slots[slot_index]
key = slot.get("key")
if not slot.get("enabled") or not key:
return False
return execute_favorite_key(key, params, params_memory, eligible_keys=eligible_keys)
def unassign_favorite_slot(slot_index: int, params: Params | None = None, params_memory: Params | None = None, *,
eligible_keys: Iterable[str] | None = None) -> list[dict[str, Any]] | None:
"""Reset a slot to the disabled/unassigned state and notify listeners."""
@@ -12,6 +12,7 @@ from openpilot.starpilot.common.favorite_slots import (
SETTINGS_CATALOG_PATH,
build_favorite_slot_options,
default_favorite_slots,
execute_favorite_key,
filter_favorite_slot_options,
load_settings_catalog,
load_favorite_slots,
@@ -156,6 +157,17 @@ def test_toggle_favorite_slot_flips_bool_and_requests_refresh():
assert memory.get_bool("StarPilotTogglesUpdated") is True
def test_execute_favorite_key_uses_same_dispatch_without_a_visible_slot():
params = FakeParams()
memory = FakeParams()
params.put("RedneckCruise", False)
assert execute_favorite_key("RedneckCruise", params, memory, eligible_keys={"RedneckCruise"}) is True
assert params.get_bool("RedneckCruise") is True
assert memory.get_bool("StarPilotTogglesUpdated") is True
assert execute_favorite_key("NotBool", params, memory, eligible_keys={"RedneckCruise"}) is False
def test_toggle_favorite_slot_blocks_alpha_longitudinal_onroad():
params = FakeParams()
params.put("IsOnroad", True)
@@ -18,6 +18,8 @@
.wheelHeader h2,
.wheelHeader p,
.wheelSectionHeading h3,
.wheelSectionHeading p,
.wheelCard h3,
.wheelCard p {
margin: 0;
@@ -43,11 +45,26 @@
}
.wheelSlotGrid,
.wheelControllerGrid,
.wheelMappings {
display: grid;
gap: 12px;
}
.wheelSection {
display: grid;
gap: 12px;
}
.wheelSectionHeading p {
margin-top: 5px;
opacity: 0.68;
}
.wheelControllerGrid {
grid-template-columns: repeat(auto-fit, minmax(340px, 1fr));
}
.wheelSlotLabel,
.wheelMapping span,
.wheelMuted,
@@ -68,6 +85,31 @@
color: #cbb8ff;
}
.wheelActionPicker {
display: grid;
gap: 6px;
margin-top: 14px;
}
.wheelActionPicker span {
opacity: 0.68;
}
.wheelActionPicker select,
.wheelActionPicker input {
width: 100%;
min-height: 42px;
border: 1px solid var(--sidebar-border-color);
border-radius: var(--border-radius-md);
padding: 8px 10px;
color: inherit;
background: var(--main-bg);
}
.wheelSpeedPicker {
max-width: 220px;
}
.wheelMapping {
margin-top: 12px;
padding-top: 12px;
@@ -9,6 +9,11 @@ const state = reactive({
joystickDevice: "",
mappings: [],
slots: [],
controllerSlots: [],
controllerOptions: [],
speedUnit: "mph",
speedMinimum: 5,
speedMaximum: 90,
learning: false,
learningSlot: null,
remainingSeconds: 0,
@@ -30,6 +35,11 @@ async function refresh() {
state.joystickDevice = typeof payload.joystick_device === "string" ? payload.joystick_device : ""
state.mappings = Array.isArray(payload.mappings) ? payload.mappings : []
state.slots = Array.isArray(payload.slots) ? payload.slots : []
state.controllerSlots = Array.isArray(payload.controller_slots) ? payload.controller_slots : []
state.controllerOptions = Array.isArray(payload.controller_options) ? payload.controller_options : []
state.speedUnit = typeof payload.speed_unit === "string" ? payload.speed_unit : "mph"
state.speedMinimum = Number(payload.speed_minimum || 5)
state.speedMaximum = Number(payload.speed_maximum || 90)
state.learning = !!payload.learning
state.learningSlot = Number.isInteger(payload.learning_slot) ? payload.learning_slot : null
state.remainingSeconds = Number(payload.remaining_seconds || 0)
@@ -117,6 +127,71 @@ function slotCard(slot, index) {
`
}
function controllerSlotCard(slot, index) {
const targetIndex = 3 + index
const selectedKey = slot?.key || ""
const selectedOption = state.controllerOptions.find(option => option.key === selectedKey)
const mappings = () => state.mappings.filter(mapping => mapping.slot === targetIndex)
const learning = () => state.learning && state.learningSlot === targetIndex
const selectAction = event => {
const key = event.currentTarget.value
const option = state.controllerOptions.find(candidate => candidate.key === key)
const value = option?.value_type === "speed" ? Number(slot?.value || option.default_value || 30) : null
request("action", { slot: index, key, value })
}
return html`
<section class="wheelCard wheelControllerCard">
<div class="wheelCardHeader">
<div>
<span class="wheelSlotLabel">Controller Action #${index + 1}</span>
<h3>${slot?.label || "Not configured"}</h3>
</div>
<button class="${() => learning() ? "learning" : ""}"
disabled="${() => !slot?.enabled || !state.offroad || state.testing || !!state.busy}"
@click="${() => request(learning() ? "cancel" : "learn", { slot: targetIndex })}">
${() => learning() ? `Listening (${Math.ceil(state.remainingSeconds)}s)` : "Learn Button"}
</button>
</div>
<label class="wheelActionPicker">
<span>Action</span>
<select disabled="${() => !state.offroad || !!state.busy}"
@change="${selectAction}">
<option value="" selected="${() => selectedKey === ""}">Not configured</option>
${state.controllerOptions.map(option => html`
<option value="${option.key}" selected="${() => selectedKey === option.key}">${option.label}</option>
`)}
</select>
</label>
${selectedOption?.value_type === "speed" ? html`
<label class="wheelActionPicker wheelSpeedPicker">
<span>Set speed (${() => state.speedUnit})</span>
<input type="number" inputmode="decimal"
min="${() => state.speedMinimum}" max="${() => state.speedMaximum}" step="1"
value="${slot?.value || selectedOption.default_value || 30}"
disabled="${() => !state.offroad || !!state.busy}"
@change="${event => request("action", { slot: index, key: selectedKey, value: Number(event.currentTarget.value) })}" />
</label>
` : ""}
<p class="wheelHint">${selectedOption?.value_type === "speed"
? "Uses the current mph/km/h setting and applies while openpilot is engaged on software-controlled cruise."
: selectedKey === "__starpilot_controller_action__:selfie"
? "Captures in the background and saves the driver-camera image in Galaxy → Sentry Mode."
: "Controller-only action. It does not create an on-screen Favorite button."}</p>
${() => learning() ? html`
<div class="wheelLearnPrompt"><span></span>Press one button on your controller, macropad, or keyboard.</div>
` : ""}
<div class="wheelMappings">
${() => mappings().length ? mappings().map(mappingRow) : html`<span class="wheelEmpty">No buttons mapped.</span>`}
</div>
</section>
`
}
function mappingTargetName(slot) {
const index = Number(slot)
return index < 3 ? `Favorite #${index + 1}` : `Controller Action #${index - 2}`
}
function testResultClass() {
if (!state.lastTested) return "waiting"
return state.lastTested.mapped ? "success" : "failure"
@@ -137,7 +212,7 @@ function testPanel() {
const device = state.lastTested.device_name || "External input"
const button = state.lastTested.event_name || `Button ${state.lastTested.event_code}`
return state.lastTested.mapped
? `${button} on ${device} is mapped to Favorite #${Number(state.lastTested.slot) + 1}.`
? `${button} on ${device} is mapped to ${mappingTargetName(state.lastTested.slot)}.`
: `${button} on ${device} does not have a mapping.`
}}</p>
</section>
@@ -170,7 +245,7 @@ export function WheelControls() {
<header class="wheelHeader">
<div>
<h2>Controllers</h2>
<p>Map buttons to favorites, or explicitly select one gamepad for Joystick Mode.</p>
<p>Map buttons to favorites or controller-only actions, or explicitly select one gamepad for Joystick Mode.</p>
</div>
<div class="wheelHeaderActions">
<button class="${() => state.testing ? "testing" : ""}"
@@ -193,16 +268,32 @@ export function WheelControls() {
<div class="wheelDeviceSummary">
<div class="wheelDeviceHeading">
<strong>Connected input devices</strong>
<span>Favorite buttons are the default. Only the selected gamepad controls Joystick Mode.</span>
<span>Favorite buttons are the default, with controller-only actions available below. Only the selected gamepad controls Joystick Mode.</span>
</div>
${() => state.devices.length
? html`<div class="wheelDeviceList">${state.devices.map(deviceRow)}</div>`
: html`<span class="wheelMuted">Connect or pair a controller, macropad, or keyboard.</span>`}
</div>
<div class="wheelSlotGrid">
${() => state.loading ? html`<div class="wheelCard">Loading...</div>` : state.slots.map(slotCard)}
</div>
<section class="wheelSection">
<div class="wheelSectionHeading">
<h3>On-screen Favorites</h3>
<p>These remain linked to the three Favorite Slots shown elsewhere in StarPilot.</p>
</div>
<div class="wheelSlotGrid">
${() => state.loading ? html`<div class="wheelCard">Loading...</div>` : state.slots.map(slotCard)}
</div>
</section>
<section class="wheelSection">
<div class="wheelSectionHeading">
<h3>Controller-only Actions</h3>
<p>Ten additional actions for physical buttons. These never appear as on-screen Favorites.</p>
</div>
<div class="wheelControllerGrid">
${() => state.controllerSlots.map(controllerSlotCard)}
</div>
</section>
</div>
`
}
@@ -320,10 +320,22 @@ def _install_server_import_stubs():
sys.modules["openpilot.starpilot.system.the_galaxy.utilities"] = utilities
sys.modules["openpilot.starpilot.system.wheel_controls"] = _simple_module(
"openpilot.starpilot.system.wheel_controls",
CONTROLLER_ACTION_OPTIONS=(
{"key": "__starpilot_controller_action__:set_speed", "label": "Set Speed To", "section": "Controller Actions", "value_type": "speed"},
{"key": "__starpilot_controller_action__:selfie", "label": "Take Comma Selfie", "section": "Controller Actions"},
),
CONTROLLER_ACTION_SET_SPEED="__starpilot_controller_action__:set_speed",
CONTROLLER_ACTION_SLOT_COUNT=10,
FAVORITE_SLOT_COUNT=3,
cancel_learning=lambda *args, **kwargs: None,
clear_mappings=lambda *args, **kwargs: None,
controller_speed_bounds=lambda is_metric: (8, 145) if is_metric else (5, 90),
delete_mapping=lambda *args, **kwargs: True,
load_controller_action_slots=lambda *args, **kwargs: [
{"enabled": False, "key": None, "label": "", "value": None} for _ in range(10)
],
public_status=lambda *args, **kwargs: {"mappings": [], "devices": [], "available": True},
set_controller_action_slot=lambda *args, **kwargs: None,
set_joystick_device=lambda *args, **kwargs: None,
start_learning=lambda *args, **kwargs: None,
start_testing=lambda *args, **kwargs: None,
@@ -55,6 +55,18 @@ def test_controller_joystick_mode_requires_explicit_device_selection():
assert 'request("joystick", { device_id: device.device_id, enabled: !selected() })' in source
def test_controller_page_has_ten_controller_only_action_slots():
source = CONTROLLERS_PATH.read_text(encoding="utf-8")
assert "Controller-only Actions" in source
assert "These never appear as on-screen Favorites" in source
assert 'request("action", { slot: index, key, value })' in source
assert "const targetIndex = 3 + index" in source
assert "state.controllerSlots.map(controllerSlotCard)" in source
assert "Set speed (${() => state.speedUnit})" in source
assert "Galaxy → Sentry Mode" in source
def test_bluetooth_and_controllers_sidebar_order():
source = SIDEBAR_PATH.read_text(encoding="utf-8")
@@ -215,6 +215,90 @@ def test_wheel_controls_status_includes_favorite_slots(monkeypatch):
assert response.status_code == 200
assert response.get_json()["slots"][0]["label"] == "Force Offroad"
assert len(response.get_json()["slots"]) == 3
assert len(response.get_json()["controller_slots"]) == 10
option_keys = {option["key"] for option in response.get_json()["controller_options"]}
assert option_keys == {
"ForceOffroad",
"__starpilot_controller_action__:set_speed",
"__starpilot_controller_action__:selfie",
}
assert response.get_json()["speed_unit"] == "mph"
def test_wheel_controls_configures_a_controller_only_action(monkeypatch):
client, _ = _params_client(monkeypatch, {"IsOffroad": True}, "mici")
calls = []
monkeypatch.setattr(the_galaxy, "_get_available_favorite_slot_options", lambda: [{"key": "ForceOffroad", "label": "Force Offroad"}])
monkeypatch.setattr(the_galaxy, "set_controller_action_slot", lambda *args, **kwargs: calls.append((args, kwargs)))
response = client.post("/api/wheel-controls/action", json={"slot": 9, "key": "ForceOffroad"})
assert response.status_code == 200
expected_keys = {
"ForceOffroad",
"__starpilot_controller_action__:set_speed",
"__starpilot_controller_action__:selfie",
}
assert calls == [((9, "ForceOffroad", "Force Offroad", the_galaxy.params), {"value": None, "eligible_keys": expected_keys})]
def test_wheel_controls_configures_set_speed_in_current_units(monkeypatch):
client, _ = _params_client(monkeypatch, {"IsOffroad": True, "IsMetric": False}, "mici")
calls = []
monkeypatch.setattr(the_galaxy, "_get_available_favorite_slot_options", list)
monkeypatch.setattr(the_galaxy, "set_controller_action_slot", lambda *args, **kwargs: calls.append((args, kwargs)))
response = client.post("/api/wheel-controls/action", json={
"slot": 0,
"key": "__starpilot_controller_action__:set_speed",
"value": 60,
})
assert response.status_code == 200
assert calls[0][0][:4] == (0, "__starpilot_controller_action__:set_speed", "Set Speed To", the_galaxy.params)
assert calls[0][1]["value"] == 60
response = client.post("/api/wheel-controls/action", json={
"slot": 0,
"key": "__starpilot_controller_action__:set_speed",
"value": 100,
})
assert response.status_code == 400
def test_controller_selfie_is_saved_to_sentry_history(monkeypatch, tmp_path):
client, fake_params = _params_client(monkeypatch, {"IsOffroad": False}, "mici")
recorded = []
monkeypatch.setattr(the_galaxy, "_get_live_driver_jpeg", lambda: b"jpeg-data")
monkeypatch.setattr(the_galaxy, "_sentry_event_roots", lambda: (tmp_path,))
monkeypatch.setattr(the_galaxy, "_record_sentry_event", lambda event: recorded.append(event))
response = client.post("/api/sentry/selfie")
assert response.status_code == 201
event = recorded[0]
assert event["kind"] == "selfie"
assert event["message"] == "Comma Selfie"
assert (tmp_path / event["eventId"] / "driver.jpg").read_bytes() == b"jpeg-data"
assert the_galaxy._normalize_sentry_event(event)["kind"] == "selfie"
assert fake_params.get("SentryModeLastEvent") is not None
def test_wheel_controls_learning_targets_controller_only_action(monkeypatch):
client, _ = _params_client(monkeypatch, {"IsOffroad": True}, "mici")
calls = []
monkeypatch.setattr(the_galaxy, "start_wheel_control_learning", lambda *args: calls.append(args))
monkeypatch.setattr(the_galaxy, "_get_available_favorite_slot_options", lambda: [{"key": "ForceOffroad", "label": "Force Offroad"}])
monkeypatch.setattr(the_galaxy, "load_controller_action_slots", lambda *_args, **_kwargs: [
{"enabled": True, "key": "ForceOffroad", "label": "Force Offroad"},
*[{"enabled": False, "key": None, "label": ""} for _ in range(9)],
])
response = client.post("/api/wheel-controls/learn", json={"slot": 3})
assert response.status_code == 200
assert calls == [(3, the_galaxy.params_memory, the_galaxy.params)]
def test_wheel_controls_learning_requires_offroad(monkeypatch):
+161 -62
View File
@@ -104,10 +104,17 @@ from openpilot.starpilot.system.the_galaxy import flm_workspace, utilities
from openpilot.starpilot.system.the_galaxy.update_recovery import inspect_interrupted_update, public_recovery_status, recover_interrupted_update
from openpilot.starpilot.system.bluetooth import BluetoothClient
from openpilot.starpilot.system.wheel_controls import (
CONTROLLER_ACTION_OPTIONS,
CONTROLLER_ACTION_SET_SPEED,
CONTROLLER_ACTION_SLOT_COUNT,
FAVORITE_SLOT_COUNT,
cancel_learning as cancel_wheel_control_learning,
clear_mappings as clear_wheel_control_mappings,
controller_speed_bounds,
delete_mapping as delete_wheel_control_mapping,
load_controller_action_slots,
public_status as wheel_control_status,
set_controller_action_slot,
set_joystick_device,
start_learning as start_wheel_control_learning,
start_testing as start_wheel_control_testing,
@@ -692,7 +699,7 @@ def _normalize_sentry_event(payload) -> dict | None:
event_id = str(payload.get("eventId") or "").strip()
kind = str(payload.get("kind") or "").strip().lower()
if not event_id or kind not in {"warning", "alarm", "power_off"}:
if not event_id or kind not in {"warning", "alarm", "power_off", "selfie"}:
return None
event = {
@@ -799,6 +806,57 @@ def _capture_sentry_live_images() -> list[str]:
return paths
def _get_live_driver_jpeg():
from openpilot.system.manager.process_config import managed_processes
started = False
try:
try:
subprocess.check_call(["pgrep", "camerad"])
except subprocess.CalledProcessError:
managed_processes['camerad'].start()
started = True
client = VisionIpcClient("camerad", VisionStreamType.VISION_STREAM_DRIVER, True)
if not client.connect(True):
return None
if started:
settle_deadline = time.monotonic() + 4.0
while time.monotonic() < settle_deadline:
client.recv(timeout_ms=100)
buf = client.recv(timeout_ms=5000)
if buf is None:
return None
y = np.array(buf.data[:buf.uv_offset], dtype=np.uint8).reshape((-1, buf.stride))[:buf.height, :buf.width]
u = np.array(buf.data[buf.uv_offset::2], dtype=np.uint8).reshape((-1, buf.stride // 2))[:buf.height // 2, :buf.width // 2]
v = np.array(buf.data[buf.uv_offset + 1::2], dtype=np.uint8).reshape((-1, buf.stride // 2))[:buf.height // 2, :buf.width // 2]
ul = np.repeat(np.repeat(u, 2).reshape(u.shape[0], y.shape[1]), 2, axis=0).reshape(y.shape)
vl = np.repeat(np.repeat(v, 2).reshape(v.shape[0], y.shape[1]), 2, axis=0).reshape(y.shape)
yuv = np.dstack((y, ul, vl)).astype(np.int16)
yuv[:, :, 1:] -= 128
m = np.array([
[1.00000, 1.00000, 1.00000],
[0.00000, -0.39465, 2.03211],
[1.13983, -0.58060, 0.00000],
])
rgb = np.dot(yuv, m).clip(0, 255).astype(np.uint8)
img = Image.fromarray(rgb)
buf_io = BytesIO()
img.save(buf_io, format="JPEG", quality=85)
return buf_io.getvalue()
except Exception:
return None
finally:
if started:
managed_processes['camerad'].stop()
_SENTRY_PUSH_LOCK = threading.Lock()
_SENTRY_NOTIFICATION_RATE_LIMIT_LOCK = threading.Lock()
_SENTRY_NOTIFICATION_LAST_AT: float | None = None
@@ -3238,6 +3296,15 @@ def _get_available_favorite_slot_options():
{"HasRivianAngleHarness": _get_has_rivian_angle_harness()},
)
def _get_available_controller_action_options():
options = [*_get_available_favorite_slot_options(), *(dict(option) for option in CONTROLLER_ACTION_OPTIONS)]
return sorted(options, key=lambda option: (
str(option.get("section") or "").casefold(),
str(option.get("label") or option.get("key") or "").casefold(),
))
def _favorite_slot_values(options):
return get_favorite_values(options, params)
@@ -4954,29 +5021,72 @@ def setup(app):
@app.route("/api/wheel-controls/status", methods=["GET"])
def wheel_controls_status():
status = wheel_control_status(params, params_memory)
options = _get_available_favorite_slot_options()
option_by_key = {option["key"]: option for option in options}
favorite_options = _get_available_favorite_slot_options()
favorite_option_by_key = {option["key"]: option for option in favorite_options}
controller_options = _get_available_controller_action_options()
controller_option_by_key = {option["key"]: option for option in controller_options}
slots = normalize_favorite_slots(
params.get(FAVORITE_SLOTS_PARAM),
params=params,
eligible_keys=set(option_by_key),
eligible_keys=set(favorite_option_by_key),
)
for slot in slots:
key = slot.get("key")
if key in option_by_key:
slot["label"] = option_by_key[key]["label"]
if key in favorite_option_by_key:
slot["label"] = favorite_option_by_key[key]["label"]
controller_slots = load_controller_action_slots(params, set(controller_option_by_key))
for slot in controller_slots:
key = slot.get("key")
if key in controller_option_by_key:
slot["label"] = controller_option_by_key[key]["label"]
status["slots"] = slots
status["controller_slots"] = controller_slots
status["controller_options"] = controller_options
is_metric = params.get_bool("IsMetric")
speed_minimum, speed_maximum = controller_speed_bounds(is_metric)
status["speed_unit"] = "km/h" if is_metric else "mph"
status["speed_minimum"] = speed_minimum
status["speed_maximum"] = speed_maximum
return jsonify(status), 200
@app.route("/api/wheel-controls/<operation>", methods=["POST"])
def wheel_controls_operation(operation):
if operation not in {"learn", "cancel", "delete", "clear", "test", "test-stop", "joystick"}:
if operation not in {"action", "learn", "cancel", "delete", "clear", "test", "test-stop", "joystick"}:
return jsonify({"error": "Unknown wheel control operation."}), 404
if not params.get_bool("IsOffroad"):
return jsonify({"error": "Wheel controls can only be configured offroad."}), 409
data = request.get_json(silent=True) or {}
try:
if operation == "action":
slot_index = int(data.get("slot", -1))
key = str(data.get("key") or "").strip()
options = _get_available_controller_action_options()
option_by_key = {option["key"]: option for option in options}
if not 0 <= slot_index < CONTROLLER_ACTION_SLOT_COUNT:
return jsonify({"error": f"Controller action must be between 1 and {CONTROLLER_ACTION_SLOT_COUNT}."}), 400
if key and key not in option_by_key:
return jsonify({"error": "That controller action is not available."}), 400
value = None
if key == CONTROLLER_ACTION_SET_SPEED:
try:
value = float(data.get("value"))
except (TypeError, ValueError):
return jsonify({"error": "Enter a valid set speed."}), 400
speed_minimum, speed_maximum = controller_speed_bounds(params.get_bool("IsMetric"))
if not math.isfinite(value) or not speed_minimum <= value <= speed_maximum:
unit = "km/h" if params.get_bool("IsMetric") else "mph"
return jsonify({"error": f"Set speed must be between {speed_minimum} and {speed_maximum} {unit}."}), 400
cancel_wheel_control_learning(params_memory, params)
set_controller_action_slot(
slot_index,
key or None,
str(option_by_key.get(key, {}).get("label") or ""),
params,
value=value,
eligible_keys=set(option_by_key),
)
return jsonify({"message": f"Controller Action #{slot_index + 1} updated."}), 200
if operation == "joystick":
device_id = str(data.get("device_id") or "").strip()
enabled = bool(data.get("enabled", False))
@@ -4992,16 +5102,28 @@ def setup(app):
if operation == "learn":
stop_wheel_control_testing(params_memory)
slot_index = int(data.get("slot", -1))
options = _get_available_favorite_slot_options()
favorite_options = _get_available_favorite_slot_options()
favorite_eligible_keys = {option["key"] for option in favorite_options}
controller_options = _get_available_controller_action_options()
controller_eligible_keys = {option["key"] for option in controller_options}
slots = normalize_favorite_slots(
params.get(FAVORITE_SLOTS_PARAM),
params=params,
eligible_keys={option["key"] for option in options},
eligible_keys=favorite_eligible_keys,
)
if not 0 <= slot_index < len(slots) or not slots[slot_index].get("enabled") or not slots[slot_index].get("key"):
return jsonify({"error": "Configure and enable that Favorite before learning a button."}), 400
if 0 <= slot_index < FAVORITE_SLOT_COUNT:
target = slots[slot_index]
target_name = f"Favorite #{slot_index + 1}"
elif FAVORITE_SLOT_COUNT <= slot_index < FAVORITE_SLOT_COUNT + CONTROLLER_ACTION_SLOT_COUNT:
controller_index = slot_index - FAVORITE_SLOT_COUNT
target = load_controller_action_slots(params, controller_eligible_keys)[controller_index]
target_name = f"Controller Action #{controller_index + 1}"
else:
return jsonify({"error": "Unknown controller mapping target."}), 400
if not target.get("enabled") or not target.get("key"):
return jsonify({"error": f"Configure {target_name} before learning a button."}), 400
start_wheel_control_learning(slot_index, params_memory, params)
return jsonify({"message": f"Press a button for Favorite #{slot_index + 1}."}), 200
return jsonify({"message": f"Press a button for {target_name}."}), 200
if operation == "cancel":
cancel_wheel_control_learning(params_memory, params)
return jsonify({"message": "Button learning cancelled."}), 200
@@ -7876,6 +7998,33 @@ def setup(app):
})
return jsonify({"capturedAt": captured_at, "imageUrls": event["imageUrls"]})
@app.route("/api/sentry/selfie", methods=["POST"])
def sentry_selfie():
if request.remote_addr not in {None, "127.0.0.1", "::1"}:
return jsonify({"error": "Comma Selfies must originate on the device."}), 403
with _SENTRY_LIVE_CAPTURE_LOCK:
jpeg = _get_live_driver_jpeg()
if jpeg is None:
return jsonify({"error": "Unable to capture the driver camera."}), 503
captured_at = datetime.now(timezone.utc).isoformat()
event_id = f"selfie-{int(time.time())}-{secrets.token_hex(4)}"
directory = _sentry_event_roots()[0] / event_id
directory.mkdir(parents=True, exist_ok=True)
image_path = directory / "driver.jpg"
image_path.write_bytes(jpeg)
event = {
"eventId": event_id,
"kind": "selfie",
"detectedAt": captured_at,
"imagePaths": [str(image_path)],
"message": "Comma Selfie",
}
_record_sentry_event(event)
params.put("SentryModeLastEvent", json.dumps(event, separators=(",", ":")))
return jsonify({"accepted": True, "capturedAt": captured_at, "eventId": event_id}), 201
@app.route("/api/sentry/test", methods=["POST"])
def sentry_test():
if request.remote_addr not in {None, "127.0.0.1", "::1"}:
@@ -9032,56 +9181,6 @@ def setup(app):
return jsonify({"error": "Unable to capture live frame from driver camera."}), 503
def _get_live_driver_jpeg():
from openpilot.system.manager.process_config import managed_processes
started = False
try:
try:
subprocess.check_call(["pgrep", "camerad"])
except subprocess.CalledProcessError:
managed_processes['camerad'].start()
started = True
client = VisionIpcClient("camerad", VisionStreamType.VISION_STREAM_DRIVER, True)
if not client.connect(True):
return None
if started:
settle_deadline = time.monotonic() + 4.0
while time.monotonic() < settle_deadline:
client.recv(timeout_ms=100)
buf = client.recv(timeout_ms=5000)
if buf is None:
return None
y = np.array(buf.data[:buf.uv_offset], dtype=np.uint8).reshape((-1, buf.stride))[:buf.height, :buf.width]
u = np.array(buf.data[buf.uv_offset::2], dtype=np.uint8).reshape((-1, buf.stride // 2))[:buf.height // 2, :buf.width // 2]
v = np.array(buf.data[buf.uv_offset + 1::2], dtype=np.uint8).reshape((-1, buf.stride // 2))[:buf.height // 2, :buf.width // 2]
ul = np.repeat(np.repeat(u, 2).reshape(u.shape[0], y.shape[1]), 2, axis=0).reshape(y.shape)
vl = np.repeat(np.repeat(v, 2).reshape(v.shape[0], y.shape[1]), 2, axis=0).reshape(y.shape)
yuv = np.dstack((y, ul, vl)).astype(np.int16)
yuv[:, :, 1:] -= 128
m = np.array([
[1.00000, 1.00000, 1.00000],
[0.00000, -0.39465, 2.03211],
[1.13983, -0.58060, 0.00000],
])
rgb = np.dot(yuv, m).clip(0, 255).astype(np.uint8)
img = Image.fromarray(rgb)
buf_io = BytesIO()
img.save(buf_io, format="JPEG", quality=85)
return buf_io.getvalue()
except Exception:
return None
finally:
if started:
managed_processes['camerad'].stop()
@app.route("/api/v_asm/config", methods=["GET"])
def v_asm_get_config():
return jsonify(_decode_json_object(params.get("VASMAnnotationConfig")))
@@ -1,28 +1,42 @@
from .wheel_controlsd import (
CONTROLLER_ACTION_SLOT_COUNT,
CONTROLLER_ACTION_OPTIONS,
CONTROLLER_ACTION_SET_SPEED,
FAVORITE_SLOT_COUNT,
LEARN_TIMEOUT_SECONDS,
cancel_learning,
clear_mappings,
connected_input_sources,
controller_speed_bounds,
delete_mapping,
load_controller_action_slots,
load_mappings,
public_status,
selected_joystick_device,
set_joystick_device,
set_controller_action_slot,
start_learning,
start_testing,
stop_testing,
)
__all__ = [
"CONTROLLER_ACTION_SLOT_COUNT",
"CONTROLLER_ACTION_OPTIONS",
"CONTROLLER_ACTION_SET_SPEED",
"FAVORITE_SLOT_COUNT",
"LEARN_TIMEOUT_SECONDS",
"cancel_learning",
"clear_mappings",
"connected_input_sources",
"controller_speed_bounds",
"delete_mapping",
"load_controller_action_slots",
"load_mappings",
"public_status",
"selected_joystick_device",
"set_joystick_device",
"set_controller_action_slot",
"start_learning",
"start_testing",
"stop_testing",
@@ -26,6 +26,12 @@ class FakeParams:
def put_bool(self, key, value):
self.values[key] = bool(value)
def get_float(self, key, default=0.0):
return float(self.values.get(key, default))
def put_float(self, key, value):
self.values[key] = float(value)
def remove(self, key):
self.values.pop(key, None)
@@ -48,6 +54,29 @@ def test_mapping_round_trip_and_reassignment():
assert not params.get_bool(wheel_controlsd.ENABLED_PARAM)
def test_controller_only_mapping_slots_extend_beyond_three_favorites():
params = FakeParams()
mapping = wheel_controlsd.upsert_mapping(source(), 30, 12, params)
assert mapping["slot"] == 12
assert wheel_controlsd.load_mappings(params) == [mapping]
assert wheel_controlsd.normalize_mappings([{**mapping, "slot": 13}]) == []
def test_controller_action_slots_are_separate_and_fixed_at_ten():
params = FakeParams()
slots = wheel_controlsd.set_controller_action_slot(
9, "RedneckCruise", "Redneck Cruise", params, eligible_keys={"RedneckCruise"},
)
assert len(slots) == 10
assert slots[9] == {"enabled": True, "key": "RedneckCruise", "label": "Redneck Cruise", "value": None}
assert wheel_controlsd.load_controller_action_slots(params, {"RedneckCruise"}) == slots
assert wheel_controlsd.CONTROLLER_ACTIONS_PARAM != "StarPilotFavoriteSlots"
def test_joystick_selection_is_explicit_and_exclusive():
params = FakeParams()
@@ -113,6 +142,81 @@ def test_mapped_key_triggers_once(monkeypatch):
daemon.close()
def test_mapped_controller_action_dispatches_without_using_a_favorite_slot(monkeypatch):
params = FakeParams({"IsOffroad": False})
memory = FakeParams()
wheel_controlsd.upsert_mapping(source(), 30, 3, params)
daemon = wheel_controlsd.WheelControlsDaemon(params, memory)
favorite_triggered = []
controller_triggered = []
monkeypatch.setattr(wheel_controlsd, "execute_favorite_slot", lambda slot, *_args: favorite_triggered.append(slot) or True)
monkeypatch.setattr(wheel_controlsd, "execute_controller_action", lambda slot, *_args: controller_triggered.append(slot) or True)
daemon._handle_key(source(), 30)
assert favorite_triggered == []
assert controller_triggered == [0]
daemon.close()
def test_controller_set_speed_uses_current_display_units_and_requires_engagement():
memory = FakeParams()
params = FakeParams({"IsOnroad": True, "IsEngaged": True, "IsMetric": False})
assert wheel_controlsd.set_controller_cruise_speed(60, params, memory)
assert memory.get_float("SLCForceCruiseSpeed") == 60 * wheel_controlsd.CV.MPH_TO_MS
params.values["IsMetric"] = True
assert wheel_controlsd.set_controller_cruise_speed(30, params, memory)
assert memory.get_float("SLCForceCruiseSpeed") == 30 * wheel_controlsd.CV.KPH_TO_MS
params.values["IsEngaged"] = False
memory.remove("SLCForceCruiseSpeed")
assert not wheel_controlsd.set_controller_cruise_speed(60, params, memory)
assert "SLCForceCruiseSpeed" not in memory.values
def test_controller_custom_actions_dispatch_their_own_payload(monkeypatch):
params = FakeParams({
wheel_controlsd.CONTROLLER_ACTIONS_PARAM: [
{
"enabled": True,
"key": wheel_controlsd.CONTROLLER_ACTION_SET_SPEED,
"label": "Set Speed To",
"value": 60,
},
{
"enabled": True,
"key": wheel_controlsd.CONTROLLER_ACTION_SELFIE,
"label": "Take Comma Selfie",
},
],
})
memory = FakeParams()
speeds = []
selfies = []
monkeypatch.setattr(wheel_controlsd, "set_controller_cruise_speed", lambda value, *_args: speeds.append(value) or True)
monkeypatch.setattr(wheel_controlsd, "request_comma_selfie", lambda: selfies.append(True) or True)
assert wheel_controlsd.execute_controller_action(0, params, memory)
assert wheel_controlsd.execute_controller_action(1, params, memory)
assert speeds == [60]
assert selfies == [True]
def test_learning_accepts_the_tenth_controller_action():
params = FakeParams({"IsOffroad": True})
memory = FakeParams()
daemon = wheel_controlsd.WheelControlsDaemon(params, memory)
wheel_controlsd.start_learning(12, memory, params)
daemon._update_learning(10.0)
assert daemon.learning_slot == 12
assert memory.get_int(wheel_controlsd.LEARN_SLOT_PARAM) == 13
daemon.close()
def test_selected_joystick_controller_does_not_trigger_favorites(monkeypatch):
params = FakeParams({"IsOffroad": False})
memory = FakeParams()
@@ -3,11 +3,14 @@ from __future__ import annotations
import hashlib
import json
import math
import os
import re
import selectors
import struct
import threading
import time
import urllib.request
from dataclasses import dataclass
from pathlib import Path
@@ -15,14 +18,38 @@ from typing import Any
from openpilot.common.params import Params
from openpilot.common.swaglog import cloudlog
from openpilot.common.constants import CV
from openpilot.starpilot.common.favorite_slots import FAVORITE_SLOT_COUNT
MAPPINGS_PARAM = "WheelControlMappings"
CONTROLLER_ACTIONS_PARAM = "ControllerActionSlots"
LEARN_SLOT_PARAM = "WheelControlLearnSlot"
STATUS_PARAM = "WheelControlStatus"
TEST_ACTIVE_PARAM = "WheelControlTestActive"
ENABLED_PARAM = "WheelControlsEnabled"
JOYSTICK_DEVICE_PARAM = "JoystickControlDevice"
CONTROLLER_ACTION_SLOT_COUNT = 10
MAPPING_SLOT_COUNT = FAVORITE_SLOT_COUNT + CONTROLLER_ACTION_SLOT_COUNT
CONTROLLER_ACTION_SET_SPEED = "__starpilot_controller_action__:set_speed"
CONTROLLER_ACTION_SELFIE = "__starpilot_controller_action__:selfie"
CONTROLLER_ACTION_OPTIONS = (
{
"key": CONTROLLER_ACTION_SET_SPEED,
"label": "Set Speed To",
"description": "Immediately changes the software-controlled cruise set speed while engaged.",
"section": "Controller Actions",
"value_type": "speed",
"default_value": 30,
},
{
"key": CONTROLLER_ACTION_SELFIE,
"label": "Take Comma Selfie",
"description": "Captures the driver camera and saves it in Sentry history.",
"section": "Controller Actions",
},
)
CONTROLLER_ACTION_KEYS = {option["key"] for option in CONTROLLER_ACTION_OPTIONS}
LEARN_TIMEOUT_SECONDS = 20.0
DEVICE_SCAN_INTERVAL_SECONDS = 1.0
STATUS_INTERVAL_SECONDS = 0.5
@@ -35,6 +62,7 @@ HAT_EVENT_BASE = 0x10000
EXTERNAL_INPUT_BUSES = {0x0003, 0x0005}
INPUT_EVENT = struct.Struct("@llHHi")
MODALIAS_RE = re.compile(r"input:b([0-9a-f]{4})v([0-9a-f]{4})p([0-9a-f]{4})e([0-9a-f]{4})", re.IGNORECASE)
_SELFIE_REQUEST_LOCK = threading.Lock()
try:
from inputs import KEYS_AND_BUTTONS
@@ -139,6 +167,110 @@ def mapping_id(device_id: str, code: int) -> str:
return hashlib.sha256(f"{device_id}:{code}".encode()).hexdigest()[:16]
def default_controller_action_slots() -> list[dict[str, Any]]:
return [{"enabled": False, "key": None, "label": "", "value": None} for _ in range(CONTROLLER_ACTION_SLOT_COUNT)]
def normalize_controller_action_slots(value: Any, eligible_keys: set[str] | None = None) -> list[dict[str, Any]]:
if isinstance(value, bytes):
value = value.decode("utf-8", errors="replace")
if isinstance(value, str):
try:
value = json.loads(value)
except json.JSONDecodeError:
value = []
if not isinstance(value, list):
value = []
slots = default_controller_action_slots()
for index, raw in enumerate(value[:CONTROLLER_ACTION_SLOT_COUNT]):
if not isinstance(raw, dict):
continue
key = str(raw.get("key") or "").strip() or None
if key is not None and eligible_keys is not None and key not in eligible_keys:
key = None
label = str(raw.get("label") or "").strip()[:64] if key else ""
value = None
if key == CONTROLLER_ACTION_SET_SPEED:
try:
candidate = float(raw.get("value"))
value = candidate if math.isfinite(candidate) and candidate > 0 else None
except (TypeError, ValueError):
pass
enabled = key is not None and (key != CONTROLLER_ACTION_SET_SPEED or value is not None)
slots[index] = {"enabled": enabled, "key": key, "label": label, "value": value}
return slots
def load_controller_action_slots(params: Params | None = None,
eligible_keys: set[str] | None = None) -> list[dict[str, Any]]:
params = params or Params(return_defaults=True)
try:
raw = params.get(CONTROLLER_ACTIONS_PARAM)
except Exception:
raw = None
return normalize_controller_action_slots(raw, eligible_keys)
def save_controller_action_slots(slots: list[dict[str, Any]], params: Params | None = None, *,
eligible_keys: set[str] | None = None) -> list[dict[str, Any]]:
params = params or Params(return_defaults=True)
normalized = normalize_controller_action_slots(slots, eligible_keys)
params.put(CONTROLLER_ACTIONS_PARAM, normalized)
return normalized
def set_controller_action_slot(index: int, key: str | None, label: str, params: Params | None = None, *, value: float | None = None,
eligible_keys: set[str] | None = None) -> list[dict[str, Any]]:
if not 0 <= index < CONTROLLER_ACTION_SLOT_COUNT:
raise ValueError(f"Controller action must be between 1 and {CONTROLLER_ACTION_SLOT_COUNT}")
key = str(key or "").strip() or None
if key is not None and eligible_keys is not None and key not in eligible_keys:
raise ValueError("That controller action is not available")
params = params or Params(return_defaults=True)
slots = load_controller_action_slots(params, eligible_keys)
slots[index] = {"enabled": key is not None, "key": key, "label": label if key else "", "value": value}
return save_controller_action_slots(slots, params, eligible_keys=eligible_keys)
def controller_speed_bounds(is_metric: bool) -> tuple[int, int]:
return (8, 145) if is_metric else (5, 90)
def set_controller_cruise_speed(value: Any, params: Params, params_memory: Params) -> bool:
if not params.get_bool("IsOnroad") or not params.get_bool("IsEngaged"):
return False
try:
native_speed = float(value)
except (TypeError, ValueError):
return False
minimum, maximum = controller_speed_bounds(params.get_bool("IsMetric"))
if not math.isfinite(native_speed) or not minimum <= native_speed <= maximum:
return False
conversion = CV.KPH_TO_MS if params.get_bool("IsMetric") else CV.MPH_TO_MS
params_memory.put_float("SLCForceCruiseSpeed", native_speed * conversion)
return True
def _request_comma_selfie() -> None:
if not _SELFIE_REQUEST_LOCK.acquire(blocking=False):
return
try:
port = os.environ.get("SP_GALAXY_PORT", "8082")
request = urllib.request.Request(f"http://127.0.0.1:{port}/api/sentry/selfie", method="POST")
with urllib.request.urlopen(request, timeout=10.0):
pass
except Exception:
cloudlog.exception("wheel controls: Comma Selfie capture failed")
finally:
_SELFIE_REQUEST_LOCK.release()
def request_comma_selfie() -> bool:
threading.Thread(target=_request_comma_selfie, name="comma-selfie", daemon=True).start()
return True
def normalize_mappings(value: Any) -> list[dict[str, Any]]:
if isinstance(value, bytes):
value = value.decode("utf-8", errors="replace")
@@ -162,7 +294,7 @@ def normalize_mappings(value: Any) -> list[dict[str, Any]]:
slot = int(raw.get("slot"))
except (TypeError, ValueError):
continue
if not device_id or code < 0 or not 0 <= slot < 3:
if not device_id or code < 0 or not 0 <= slot < MAPPING_SLOT_COUNT:
continue
signature = (device_id, code)
if signature in seen:
@@ -229,8 +361,8 @@ def clear_mappings(params: Params | None = None) -> None:
def start_learning(slot: int, params_memory: Params | None = None, params: Params | None = None) -> None:
if not 0 <= slot < 3:
raise ValueError("Favorite slot must be between 1 and 3")
if not 0 <= slot < MAPPING_SLOT_COUNT:
raise ValueError(f"Mapping target must be between 1 and {MAPPING_SLOT_COUNT}")
params_memory = params_memory or Params(memory=True)
(params or Params()).put_bool(ENABLED_PARAM, True)
params_memory.put_int(LEARN_SLOT_PARAM, slot + 1)
@@ -299,6 +431,27 @@ def execute_favorite_slot(slot: int, params: Params, params_memory: Params) -> b
return toggle_favorite_slot(slot, params, params_memory)
def execute_controller_action(index: int, params: Params, params_memory: Params) -> bool:
from openpilot.starpilot.common.favorite_slots import execute_favorite_key
slots = load_controller_action_slots(params)
if not 0 <= index < len(slots):
return False
slot = slots[index]
if not slot.get("enabled"):
return False
if slot.get("key") == CONTROLLER_ACTION_SET_SPEED:
return set_controller_cruise_speed(slot.get("value"), params, params_memory)
if slot.get("key") == CONTROLLER_ACTION_SELFIE:
return request_comma_selfie()
return execute_favorite_key(slot.get("key"), params, params_memory)
def execute_mapping_slot(slot: int, params: Params, params_memory: Params) -> bool:
if slot < FAVORITE_SLOT_COUNT:
return execute_favorite_slot(slot, params, params_memory)
return execute_controller_action(slot - FAVORITE_SLOT_COUNT, params, params_memory)
class WheelControlsDaemon:
def __init__(self, params: Params | None = None, params_memory: Params | None = None):
self.params = params or Params(return_defaults=True)
@@ -364,7 +517,7 @@ class WheelControlsDaemon:
return
requested = self.params_memory.get_int(LEARN_SLOT_PARAM)
if 1 <= requested <= 3:
if 1 <= requested <= MAPPING_SLOT_COUNT:
slot = requested - 1
if slot != self.learning_slot:
self.learning_slot = slot
@@ -413,9 +566,9 @@ class WheelControlsDaemon:
for mapping in mappings:
if mapping["device_id"] == source.device_id and mapping["event_code"] == code:
try:
execute_favorite_slot(mapping["slot"], self.params, self.params_memory)
execute_mapping_slot(mapping["slot"], self.params, self.params_memory)
except Exception:
cloudlog.exception("wheel control favorite action failed")
cloudlog.exception("wheel control action failed")
return
def _read_events(self, fd: int) -> None: