feat: add Set B profiles and low-latency native haptics
Add schema-6 profiles, atomic catalog migration, shortcuts, Shift, configurable Turbo/Burst, macro recording and playback. Promote qualified 300 MHz native DualSense transport to AIO/XInput defaults with bounded bus transactions, credit batching and generation-safe persistent rumble.
This commit is contained in:
parent
44a474e103
commit
3c2d9fa723
75 changed files with 9599 additions and 1819 deletions
|
|
@ -6,6 +6,7 @@ import urllib.error
|
|||
import urllib.request
|
||||
from collections.abc import Iterator
|
||||
from contextlib import contextmanager
|
||||
from dataclasses import replace
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
|
@ -49,9 +50,7 @@ def request_json(
|
|||
headers["Content-Type"] = "application/json"
|
||||
if token is not None:
|
||||
headers["X-Switch-Pico-Token"] = token
|
||||
request = urllib.request.Request(
|
||||
url, data=data, headers=headers, method=method
|
||||
)
|
||||
request = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=2) as response:
|
||||
return response.status, json.loads(response.read())
|
||||
|
|
@ -75,7 +74,6 @@ def test_editor_serves_assets_and_complete_schema(
|
|||
artwork_size = len(response.read())
|
||||
assert response.headers["Content-Type"] == "image/svg+xml"
|
||||
|
||||
|
||||
status, schema = request_json(f"{base_url}/api/schema")
|
||||
|
||||
assert status == 200
|
||||
|
|
@ -136,9 +134,7 @@ def test_editor_identifies_connected_controller_artwork(
|
|||
status, listing = request_json(f"{base_url}/api/profiles")
|
||||
|
||||
assert status == 200
|
||||
assert [
|
||||
identity["controller"] for identity in listing["identities"]
|
||||
] == [
|
||||
assert [identity["controller"] for identity in listing["identities"]] == [
|
||||
{"model": "Generic controller", "style": "generic"},
|
||||
{"model": "Nintendo Switch Pro Controller", "style": "switch"},
|
||||
{"model": "Sony DualSense", "style": "playstation"},
|
||||
|
|
@ -171,9 +167,7 @@ def test_editor_reads_writes_and_activates_profiles_atomically(
|
|||
"style": "xbox",
|
||||
}
|
||||
|
||||
status, playtest = request_json(
|
||||
f"{base_url}/api/profiles/1/8/playtest"
|
||||
)
|
||||
status, playtest = request_json(f"{base_url}/api/profiles/1/8/playtest")
|
||||
assert status == 200
|
||||
assert playtest["connected"] is True
|
||||
assert playtest["label"] == "Xbox · 50:60"
|
||||
|
|
@ -190,7 +184,10 @@ def test_editor_reads_writes_and_activates_profiles_atomically(
|
|||
assert playtest["triggers"] == {"left": 123, "right": 65000}
|
||||
assert playtest["battery"] == 100
|
||||
assert playtest["capabilities"] == [
|
||||
"rumble", "lightbar", "player_leds", "motion"
|
||||
"rumble",
|
||||
"lightbar",
|
||||
"player_leds",
|
||||
"motion",
|
||||
]
|
||||
|
||||
status, selected = request_json(f"{base_url}/api/profiles/1/8")
|
||||
|
|
@ -212,7 +209,6 @@ def test_editor_reads_writes_and_activates_profiles_atomically(
|
|||
)
|
||||
== custom_profile()
|
||||
)
|
||||
assert device.profile_chunk_sizes == [40, 40, 40, 40, 40, 40, 16]
|
||||
status, renamed = request_json(
|
||||
f"{base_url}/api/profiles/1/8/name",
|
||||
method="PUT",
|
||||
|
|
@ -221,9 +217,7 @@ def test_editor_reads_writes_and_activates_profiles_atomically(
|
|||
)
|
||||
assert status == 200
|
||||
assert renamed["stored_generation"] == 9
|
||||
assert device.profile_names[
|
||||
(device.stable_identity.to_bytes(), 7)
|
||||
] == "Desktop"
|
||||
assert device.profile_names[(device.stable_identity.to_bytes(), 7)] == "Desktop"
|
||||
|
||||
status, aliased = request_json(
|
||||
f"{base_url}/api/identities/1/alias",
|
||||
|
|
@ -241,25 +235,48 @@ def test_editor_reads_writes_and_activates_profiles_atomically(
|
|||
)
|
||||
assert status == 200
|
||||
assert identified == {"identified": True}
|
||||
assert device.identified_identities == [
|
||||
device.stable_identity.to_bytes()
|
||||
]
|
||||
assert device.identified_identities == [device.stable_identity.to_bytes()]
|
||||
|
||||
draft = config_manager.ControllerProfile.default().to_json_object()
|
||||
draft["shortcuts"] = {
|
||||
"modifier": "left_shoulder",
|
||||
"profiles": ["south", None, None, None, None, None, None, "dpad_up"],
|
||||
}
|
||||
draft["shift"]["mode"] = "hold"
|
||||
draft["shift"]["modifier"] = "right_shoulder"
|
||||
draft["shift"]["button_map"]["south"] = "north"
|
||||
draft["turbo"]["south"] = "burst"
|
||||
draft["turbo_settings"]["overrides"]["south"] = {
|
||||
"rate_hz": 7,
|
||||
"duty_percent": 25,
|
||||
"burst_count": 9,
|
||||
}
|
||||
draft["macros"][0]["playback"] = "repeat"
|
||||
draft["macros"][0]["repeat_count"] = 4
|
||||
status, copied = request_json(
|
||||
f"{base_url}/api/profiles/1/8/copy",
|
||||
method="POST",
|
||||
value={"identity_index": 1, "profile_number": 4},
|
||||
value={
|
||||
"identity_index": 1,
|
||||
"profile_number": 4,
|
||||
"profile": draft,
|
||||
"name": "Draft copy",
|
||||
},
|
||||
token=token,
|
||||
)
|
||||
assert status == 200
|
||||
assert copied["stored_generation"] == 12
|
||||
assert device.profiles[
|
||||
(device.stable_identity.to_bytes(), 3)
|
||||
] == custom_profile().to_bytes()
|
||||
assert device.profile_names[
|
||||
(device.stable_identity.to_bytes(), 3)
|
||||
] == "Desktop"
|
||||
|
||||
assert (
|
||||
device.profiles[(device.stable_identity.to_bytes(), 3)]
|
||||
== config_manager.ControllerProfile.from_json_object(draft).to_bytes()
|
||||
)
|
||||
assert (
|
||||
device.profile_names[(device.stable_identity.to_bytes(), 3)] == "Draft copy"
|
||||
)
|
||||
assert (
|
||||
device.profiles[(device.stable_identity.to_bytes(), 7)]
|
||||
== custom_profile().to_bytes()
|
||||
)
|
||||
|
||||
status, activated = request_json(
|
||||
f"{base_url}/api/profiles/1/8/activate",
|
||||
|
|
@ -298,3 +315,82 @@ def test_editor_rejects_invalid_or_unauthorized_mutations(
|
|||
assert status == 400
|
||||
assert "outer_saturation" in invalid["error"]
|
||||
assert device.profiles[(device.global_identity.to_bytes(), 0)] == original
|
||||
|
||||
|
||||
def test_recorder_accepts_first_connection_generation_zero(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
device = FakeDevice()
|
||||
device.playtest_connection_generation = 0
|
||||
page = config_manager.MacroCapturePage(
|
||||
0,
|
||||
0,
|
||||
0,
|
||||
device.playtest_slot,
|
||||
0,
|
||||
1,
|
||||
0,
|
||||
0,
|
||||
512,
|
||||
1024,
|
||||
10000,
|
||||
8,
|
||||
(),
|
||||
)
|
||||
|
||||
def start(*args: Any, **kwargs: Any) -> config_manager.MacroCapturePage:
|
||||
nonlocal page
|
||||
page = replace(
|
||||
page,
|
||||
run_id=1,
|
||||
state=1,
|
||||
total_events=1,
|
||||
events=(config_manager.MacroCaptureEvent(0, 0, 0, 0, 0, 0, 0, 0),),
|
||||
)
|
||||
return page
|
||||
|
||||
def stop(*args: Any) -> config_manager.MacroCapturePage:
|
||||
nonlocal page
|
||||
page = replace(page, state=2, elapsed_us=100000)
|
||||
return page
|
||||
|
||||
monkeypatch.setattr(config_manager, "read_macro_capture", lambda *args: page)
|
||||
monkeypatch.setattr(config_manager, "start_macro_capture", start)
|
||||
monkeypatch.setattr(config_manager, "stop_macro_capture", stop)
|
||||
monkeypatch.setattr(config_manager, "collect_macro_capture", lambda *args: page)
|
||||
with running_server(monkeypatch, device) as (base_url, token):
|
||||
status, _ = request_json(f"{base_url}/api/profiles/1/1/playtest")
|
||||
assert status == 200
|
||||
identity = device.stable_identity.to_bytes().hex()
|
||||
status, started = request_json(
|
||||
f"{base_url}/api/profiles/1/1/capture/start",
|
||||
method="POST",
|
||||
token=token,
|
||||
value={
|
||||
"capture_id": "first-connection",
|
||||
"owner_key": identity,
|
||||
"slot": device.playtest_slot,
|
||||
"connection_generation": 0,
|
||||
"macro_index": 0,
|
||||
"profile": config_manager.ControllerProfile.default().to_json_object(),
|
||||
"channels": 1,
|
||||
"max_events": 8,
|
||||
"axis_quantum": 512,
|
||||
"trigger_quantum": 1024,
|
||||
"max_duration_ms": 10000,
|
||||
},
|
||||
)
|
||||
assert status == 200 and started["state_name"] == "recording"
|
||||
status, stopped = request_json(
|
||||
f"{base_url}/api/profiles/1/1/capture/stop",
|
||||
method="POST",
|
||||
token=token,
|
||||
value={
|
||||
"capture_id": "first-connection",
|
||||
"owner_key": identity,
|
||||
"connection_generation": 0,
|
||||
"run_id": started["run_id"],
|
||||
},
|
||||
)
|
||||
assert status == 200 and stopped["state_name"] == "stopped"
|
||||
assert stopped["steps"][0]["duration_ms"] == 100
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue