vee/Cognitive_Agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: integrate cabin CAN and verified dynamic seat modes

Browse Source
This commit is contained in:
vee
2026-08-04 17:51:38 +08:00
parent 155fbb59ba
commit dc7c91800e
99 changed files with 51897 additions and 3618 deletions
Download Patch File Download Diff File Expand all files Collapse all files
domains/seat/can/seat-motion-hardware-matrix.yaml
+12
View File
@@ -89,6 +89,18 @@ messages:
- {name: Current_Raw, start_bit: 16, bit_length: 16}
- {name: Current_Baseline_Raw, start_bit: 32, bit_length: 16}
- {name: Stall_Current_Delta_Threshold_Raw, start_bit: 48, bit_length: 16}
- name: T2G_Seat_Dynamic_Mode_Status_306
frame_id: 0x306
role: feedback
minimum_payload_bytes: 8
signals:
- {name: Dynamic_Mode_ID, start_bit: 0, bit_length: 8}
- {name: Dynamic_State, start_bit: 8, bit_length: 8}
- {name: Dynamic_Phase, start_bit: 16, bit_length: 8}
- {name: Dynamic_Active_Mask, start_bit: 24, bit_length: 4}
- {name: Dynamic_Fault_Mask, start_bit: 28, bit_length: 4}
- {name: Dynamic_Cycle_Index, start_bit: 32, bit_length: 16}
- {name: Dynamic_Phase_Elapsed_Ms, start_bit: 48, bit_length: 16}
- name: T2G_Seat_Thermal_Current_Diagnostic_305
frame_id: 0x305
role: feedback
domains/seat/can/signal-mapping.hardware.yaml
+21 -1
View File
@@ -3,7 +3,7 @@ kind: SignalMapping
metadata:
id: seat-t2g-bench-hardware-mapping
name: Verified T2G seat bench mapping
version: 3.0.0
version: 3.1.0
description: Front-passenger-only mapping implemented by the unified Seat_Controller_demo IAR firmware.
spec:
domain: seat
@@ -80,3 +80,23 @@ spec:
feedback_signal: Value_X100
calibration_ref: ../../../platforms/t2g_seat_bench/calibration/seat.yaml
unit: percent
- capability: seat.start_dynamic_mode
parameter: mode_id
target_zone: front_right
physical_target_alias: t2g_primary_seat
compiler_handler: compile_t2g_dynamic_mode
bus: seat_can
message: T2G_Seat_Move_Command_300
signal: Param_A
feedback_message: T2G_Seat_Dynamic_Mode_Status_306
feedback_signal: Dynamic_Mode_ID
unit: enum
- capability: seat.stop_dynamic_mode
target_zone: front_right
physical_target_alias: t2g_primary_seat
compiler_handler: compile_t2g_dynamic_mode
bus: seat_can
message: T2G_Seat_Move_Command_300
feedback_message: T2G_Seat_Dynamic_Mode_Status_306
feedback_signal: Dynamic_State
unit: enum
domains/seat/config/capabilities.yaml
+39 -3
View File
@@ -3,7 +3,7 @@ kind: CapabilityCatalog
metadata:
id: seat-capabilities
name: Seat capabilities
version: 1.3.0
version: 1.4.0
spec:
domain: seat
capabilities:
@@ -27,11 +27,13 @@ spec:
- name: level
type: integer
required: false
description: Desired 1-3 comfort level; simulated until a level command signal is supplied.
description: Desired 1-5 comfort level; levels 4-5 require the vehicle native-CAN backend.
allowed_values:
- 1
- 2
- 3
- 4
- 5
risk_level: medium
confirmation_policy: on_ambiguity
- name: seat.set_heating
@@ -53,11 +55,13 @@ spec:
- name: level
type: integer
required: false
description: Desired 1-3 comfort level; simulated until a level command signal is supplied.
description: Desired 1-5 comfort level; levels 4-5 require the vehicle native-CAN backend.
allowed_values:
- 1
- 2
- 3
- 4
- 5
risk_level: medium
confirmation_policy: on_ambiguity
- name: seat.set_longitudinal_position
@@ -198,6 +202,38 @@ spec:
description: Absolute normalized leg-rest extension target.
risk_level: high
confirmation_policy: on_ambiguity
- name: seat.start_dynamic_mode
description: >-
Start a firmware-owned cyclic four-axis trajectory on the front-passenger seat.
Mode 1 is the short-travel rocking-horse pitch cycle; mode 2 is the long-travel
dynamic cradle. Both use four-axis target rendezvous, repeat until explicitly
stopped, and stop automatically when the firmware-enforced ten-minute runtime
limit expires. A target phase faults safely if it cannot finish within its limit.
target_devices:
- seat_motion_controller
target_scopes:
- passenger
parameters:
- name: mode_id
type: integer
required: true
description: Native dynamic mode identifier implemented by Seat_Controller_demo.
allowed_values:
- 1
- 2
risk_level: high
confirmation_policy: on_ambiguity
- name: seat.stop_dynamic_mode
description: >-
Stop whichever firmware-owned dynamic trajectory is currently running on the
front-passenger seat. Use for natural-language stop, close, cancel, or end requests.
target_devices:
- seat_motion_controller
target_scopes:
- passenger
parameters: []
risk_level: low
confirmation_policy: never
- name: seat.save_posture_preset
description: >-
Save the current five-axis posture of the front-passenger seat as a conversation-owned named preset.
domains/seat/schemas/function-parameters.schema.json
+27 -2
View File
@@ -22,7 +22,7 @@
"required": ["state"],
"properties": {
"state": {"type": "string", "enum": ["on", "off"]},
"level": {"type": "integer", "minimum": 1, "maximum": 3}
"level": {"type": "integer", "minimum": 1, "maximum": 5}
}
}
}
@@ -37,7 +37,7 @@
"required": ["state"],
"properties": {
"state": {"type": "string", "enum": ["on", "off"]},
"level": {"type": "integer", "minimum": 1, "maximum": 3}
"level": {"type": "integer", "minimum": 1, "maximum": 5}
}
}
}
@@ -65,6 +65,31 @@
}
}
}
},
{
"type": "object",
"properties": {
"capability": {"const": "seat.start_dynamic_mode"},
"parameters": {
"type": "object",
"additionalProperties": false,
"required": ["mode_id"],
"properties": {
"mode_id": {"type": "integer", "enum": [1, 2]}
}
}
}
},
{
"type": "object",
"properties": {
"capability": {"const": "seat.stop_dynamic_mode"},
"parameters": {
"type": "object",
"additionalProperties": false,
"maxProperties": 0
}
}
}
]
}
domains/seat/src/seat_domain/compiler.py
+44 -1
View File
@@ -24,7 +24,7 @@ MOTION_STATUS_MESSAGES = {
class SeatCompiler:
version = "1.1.0"
version = "1.2.0"
def __init__(self, mapping_path: Path):
self.mapping = yaml.safe_load(mapping_path.read_text(encoding="utf-8"))
@@ -42,6 +42,49 @@ class SeatCompiler:
motion_index = 0
for operation in function_config["spec"]["operations"]:
if operation["capability"] == "seat.start_dynamic_mode":
mode_id = int(operation["parameters"]["mode_id"])
if mode_id not in {1, 2}:
raise ValueError("dynamic seat mode_id must be 1 or 2")
motion_commands.append(
{
"id": f"dynamic_{operation['id']}",
"operation_ref": operation["id"],
"bus": "seat_can",
"message": "Seat_Dynamic_Mode_Virtual",
"signals": {"Dynamic_Mode_ID": mode_id},
"send_offset_ms": motion_index * 50,
"completion": {
"message": "Seat_Dynamic_Mode_Status_Virtual",
"signal": "Dynamic_Mode_ID",
"operator": "eq",
"value": mode_id,
"timeout_ms": int(operation.get("timeout_ms", 5000)),
},
}
)
motion_index += 1
continue
if operation["capability"] == "seat.stop_dynamic_mode":
motion_commands.append(
{
"id": f"dynamic_stop_{operation['id']}",
"operation_ref": operation["id"],
"bus": "seat_can",
"message": "Seat_Dynamic_Mode_Stop_Virtual",
"signals": {"Dynamic_State": 0},
"send_offset_ms": motion_index * 50,
"completion": {
"message": "Seat_Dynamic_Mode_Status_Virtual",
"signal": "Dynamic_State",
"operator": "eq",
"value": 0,
"timeout_ms": int(operation.get("timeout_ms", 5000)),
},
}
)
motion_index += 1
continue
mapping = self._mapping_for(operation["capability"], operation["target"]["zone"])
handler = mapping["compiler_handler"]
if handler == "compile_binary_control":
domains/seat/src/seat_domain/executor.py
+65
View File
@@ -58,6 +58,15 @@ class VirtualSeatExecutor:
operations = [operation_by_id[item] for item in operation_ids]
if command["message"] == "Seat_Thermal_Control_42A":
results.extend(self._execute_thermal(command, operations))
elif command["message"] in {
"Seat_Dynamic_Mode_Virtual",
"Seat_Dynamic_Mode_Stop_Virtual",
}:
if len(operations) != 1:
raise ValueError(
"A dynamic-mode command must reference exactly one operation"
)
results.append(self._execute_dynamic_mode(command, operations[0]))
else:
if len(operations) != 1:
raise ValueError("A motion command must reference exactly one operation")
@@ -134,6 +143,62 @@ class VirtualSeatExecutor:
)
return results
@staticmethod
def _execute_dynamic_mode(
command: dict[str, Any], operation: dict[str, Any]
) -> dict[str, Any]:
if operation["capability"] == "seat.stop_dynamic_mode":
return {
"ref": operation["id"],
"domain": "seat",
"status": "succeeded",
"message": "虚拟控制器已停止动态座椅模式",
"telemetry": {
"dynamic_mode_id": 0,
"dynamic_mode_name": "",
"dynamic_state": "idle",
"repeat": "none",
"max_runtime_ms": 600_000,
"command_frame": {
"frame_id": "virtual",
"data_hex": "dynamic-stop",
},
"feedback_frame": {
"frame_id": "virtual",
"data_hex": "dynamic-idle",
},
"simulated": True,
},
}
mode_id = int(operation["parameters"]["mode_id"])
cycle_ms = None
mode_name = "木马模式" if mode_id == 1 else "动态摇篮模式"
return {
"ref": operation["id"],
"domain": "seat",
"status": "succeeded",
"message": (
f"虚拟控制器已启动{mode_name}，持续循环至停止指令或十分钟上限"
),
"telemetry": {
"dynamic_mode_id": mode_id,
"dynamic_mode_name": mode_name,
"dynamic_state": "running",
"repeat": "until_stopped_or_timeout",
"max_runtime_ms": 600_000,
"cycle_duration_ms": cycle_ms,
"command_frame": {
"frame_id": "virtual",
"data_hex": f"dynamic-mode-{mode_id}",
},
"feedback_frame": {
"frame_id": "virtual",
"data_hex": f"dynamic-running-{mode_id}",
},
"simulated": True,
},
}
def _execute_motion(
self,
command: dict[str, Any],
domains/seat/src/seat_domain/generator.py
+69
View File
@@ -20,6 +20,9 @@ MOTION_CAPABILITIES = {
"legrest_height": "seat.set_legrest_height",
"legrest_extension": "seat.set_legrest_extension",
}
DYNAMIC_MODE_CAPABILITY = "seat.start_dynamic_mode"
DYNAMIC_MODE_STOP_CAPABILITY = "seat.stop_dynamic_mode"
DYNAMIC_MODE_MAX_RUNTIME_MS = 600_000
def _now() -> str:
@@ -31,6 +34,36 @@ def build_intent_ir(
) -> dict[str, Any]:
requests = []
for action in parsed["actions"]:
if action.get("action_type") == "dynamic_mode_stop":
requests.append(
{
"domain": "seat",
"goal": f"Stop the active dynamic seat mode on {action['target']}",
"target": TARGETS[action["target"]],
"parameters": {"capability": DYNAMIC_MODE_STOP_CAPABILITY},
"source_span": utterance,
}
)
continue
if action.get("action_type") == "dynamic_mode":
requests.append(
{
"domain": "seat",
"goal": (
f"Start bounded cyclic dynamic seat mode {action['mode_name']} "
f"on {action['target']}"
),
"target": TARGETS[action["target"]],
"parameters": {
"capability": DYNAMIC_MODE_CAPABILITY,
"mode_id": action["mode_id"],
"repeat": "until_stopped_or_timeout",
"max_runtime_ms": DYNAMIC_MODE_MAX_RUNTIME_MS,
},
"source_span": utterance,
}
)
continue
if action.get("action_type") == "position":
parameters = {
"axis": action["axis"],
@@ -158,6 +191,42 @@ def build_function_config(
"The supplied matrix is partial; this configuration is eligible for virtual execution only."
]
for index, action in enumerate(parsed["actions"], start=1):
if action.get("action_type") == "dynamic_mode_stop":
operation_id = f"{action['target']}_dynamic_mode_stop"
operations.append(
{
"id": operation_id,
"capability": DYNAMIC_MODE_STOP_CAPABILITY,
"target": TARGETS[action["target"]],
"parameters": {},
"timeout_ms": 5000,
"failure_policy": "abort",
}
)
assumptions.append(
f"Operation {operation_id} stops the firmware-owned loop and "
"requires 0x306 to return IDLE."
)
continue
if action.get("action_type") == "dynamic_mode":
operation_id = f"{action['target']}_dynamic_mode_{action['mode_id']}"
operations.append(
{
"id": operation_id,
"capability": DYNAMIC_MODE_CAPABILITY,
"target": TARGETS[action["target"]],
"parameters": {"mode_id": int(action["mode_id"])},
"timeout_ms": 5000,
"failure_policy": "abort",
}
)
assumptions.append(
f"Operation {operation_id} starts a firmware-owned cyclic loop with "
f"a {DYNAMIC_MODE_MAX_RUNTIME_MS} ms safety cap; "
"completion means that the controller acknowledged the mode and "
"reported it through 0x306, not that the loop terminated."
)
continue
if action.get("action_type") == "position":
operation_id = (
f"{action['target']}_{action['axis']}_{action['position_pct']}pct"
domains/seat/src/seat_domain/hardware_compiler.py
+67 -2
View File
@@ -16,7 +16,7 @@ STATE_VALUES = {"on": 1, "off": 2}
class T2GSeatCompiler:
"""Deterministic compiler for the HIL-verified T2G seat bench protocol."""
version = "2.0.0"
version = "2.1.0"
def __init__(self, mapping_path: Path, calibration_path: Path):
self.mapping = yaml.safe_load(mapping_path.read_text(encoding="utf-8"))
@@ -41,13 +41,29 @@ class T2GSeatCompiler:
for item in operations
if item["capability"] in {"seat.set_ventilation", "seat.set_heating"}
]
motion_operations = [item for item in operations if item not in thermal_operations]
dynamic_operations = [
item
for item in operations
if item["capability"] in {
"seat.start_dynamic_mode",
"seat.stop_dynamic_mode",
}
]
motion_operations = [
item
for item in operations
if item not in thermal_operations and item not in dynamic_operations
]
commands: list[dict[str, Any]] = []
if thermal_operations:
commands.extend(
self._compile_thermal(thermal_operations, actuator_state or {})
)
if dynamic_operations:
commands.extend(
self._compile_dynamic_mode(dynamic_operations, len(commands))
)
commands.extend(self._compile_motion(motion_operations, len(commands)))
canonical = json.dumps(
@@ -257,6 +273,55 @@ class T2GSeatCompiler:
commands.append(command)
return commands
def _compile_dynamic_mode(
self,
operations: list[dict[str, Any]],
command_offset: int,
) -> list[dict[str, Any]]:
if len(operations) != 1:
raise ValueError("one command may start or stop only one dynamic seat mode")
operation = operations[0]
self._require_supported(operation)
stopping = operation["capability"] == "seat.stop_dynamic_mode"
mode_id = 0 if stopping else int(operation["parameters"]["mode_id"])
if not stopping and mode_id not in {1, 2}:
raise ValueError("T2G dynamic mode_id must be 1 or 2")
return [
{
"id": f"t2g_dynamic_{operation['id']}",
"operation_ref": operation["id"],
"bus": "seat_can",
"message": "T2G_Seat_Move_Command_300",
"signals": {
"Direction": 0,
"Motor_Selector": 0,
"Opcode": int(
self.protocol[
"stop_dynamic_mode_opcode"
if stopping
else "start_dynamic_mode_opcode"
]
),
"Param_A": mode_id,
"Param_B": 0,
"Protocol_Version": int(self.protocol["version"]),
"Sequence": 0,
},
"send_offset_ms": (
command_offset * int(self.protocol["inter_axis_settle_ms"])
),
"completion": {
"message": "T2G_Seat_Dynamic_Mode_Status_306",
"signal": "Dynamic_State" if stopping else "Dynamic_Mode_ID",
"operator": "eq",
"value": 0 if stopping else mode_id,
"state_signal": "Dynamic_State",
"accepted_states": [0] if stopping else [1, 2],
"timeout_ms": max(int(operation.get("timeout_ms", 0)), 5000),
},
}
]
def _compile_timed_motion(
self,
operation: dict[str, Any],
domains/seat/src/seat_domain/hardware_executor.py
+99 -2
View File
@@ -156,7 +156,11 @@ class HardwareSeatExecutor:
}
results: list[dict[str, Any]] = []
commands = command_plan["spec"]["commands"]
has_motion = any(item["message"] == "T2G_Seat_Move_Command_300" for item in commands)
has_motion = any(
item["message"] == "T2G_Seat_Move_Command_300"
and int(item.get("signals", {}).get("Opcode", 0)) not in {0x09, 0x0F}
for item in commands
)
bus: CanBus | None = None
motion_armed = False
position_sync_required = False
@@ -249,7 +253,12 @@ class HardwareSeatExecutor:
elif command["message"] == "T2G_Seat_Thermal_Control_29E":
results.extend(self._execute_thermal(bus, command, referenced))
elif command["message"] == "T2G_Seat_Move_Command_300":
results.extend(self._execute_motion(bus, command, referenced))
if int(command["signals"]["Opcode"]) in {0x0E, 0x0F}:
results.extend(
self._execute_dynamic_mode(bus, command, referenced)
)
else:
results.extend(self._execute_motion(bus, command, referenced))
else:
raise HardwareExecutionError(
f"hardware executor refuses unknown message {command['message']}"
@@ -697,6 +706,94 @@ class HardwareSeatExecutor:
)
return output
def _execute_dynamic_mode(
self,
bus: CanBus,
command: dict[str, Any],
operations: list[dict[str, Any]],
) -> list[dict[str, Any]]:
if len(operations) != 1:
raise HardwareExecutionError(
"a T2G dynamic-mode frame must reference exactly one operation"
)
operation = operations[0]
stopping = operation["capability"] == "seat.stop_dynamic_mode"
mode_id = 0 if stopping else int(operation["parameters"]["mode_id"])
if not stopping and mode_id not in {1, 2}:
raise HardwareExecutionError("T2G dynamic mode_id must be 1 or 2")
signals = dict(command["signals"])
sequence = self._next_sequence()
signals["Sequence"] = sequence
command_frame = self._send_encoded(bus, command["message"], signals)
ack_values, ack_frame = self._wait_ack(
bus, sequence, 0x0F if stopping else 0x0E
)
completion = command["completion"]
def matches(values: dict[str, int], _: CanFrame) -> bool:
if stopping:
return (
values["Dynamic_Mode_ID"] == 0
and values["Dynamic_State"] == 0
)
return (
values["Dynamic_Mode_ID"] == mode_id
and values["Dynamic_State"] in set(
completion.get("accepted_states", [1, 2])
)
)
status_values, status_frame = self._receive_until(
bus,
0x306,
int(completion["timeout_ms"]),
matches,
)
mode_name = (
""
if stopping
else "木马模式"
if mode_id == 1
else "动态摇篮模式"
)
return [
{
"ref": operation["id"],
"domain": "seat",
"status": "succeeded",
"message": (
"physical 0x301/0x306 confirmed the dynamic seat mode is stopped"
if stopping
else (
f"physical 0x301/0x306 confirmed {mode_name} is active; "
"firmware will repeat it until stopped or the ten-minute cap"
)
),
"telemetry": {
"dynamic_mode_id": mode_id,
"dynamic_mode_name": mode_name,
"dynamic_state": status_values["Dynamic_State"],
"dynamic_phase": status_values["Dynamic_Phase"],
"dynamic_cycle_index": status_values["Dynamic_Cycle_Index"],
"dynamic_phase_elapsed_ms": status_values[
"Dynamic_Phase_Elapsed_Ms"
],
"repeat": (
"none" if stopping else "until_stopped_or_timeout"
),
"max_runtime_ms": 600_000,
"command_frame": command_frame.public(),
"ack_frame": ack_frame.public(),
"ack_values": ack_values,
"feedback_frame": status_frame.public(),
"hardware_feedback": True,
"simulated": False,
},
}
]
def _execute_motion(
self,
bus: CanBus,
domains/seat/src/seat_domain/validator.py
+15
View File
@@ -82,6 +82,21 @@ def validate_semantics(
"message": "当前所有座椅能力仅支持前排副驾。",
}
)
if action.get("action_type") == "dynamic_mode_stop":
has_motion = True
continue
if action.get("action_type") == "dynamic_mode":
has_motion = True
if action.get("mode_id") not in {1, 2}:
issues.append(
{
"severity": "error",
"code": "UNSUPPORTED_DYNAMIC_MODE",
"path": f"{path}.mode_id",
"message": "动态座椅模式编号必须是 1 或 2。",
}
)
continue
if action.get("action_type") == "position":
has_motion = True
axis = action.get("axis")
domains/seat/tests/integration/test_seat_service.py
+35
View File
@@ -32,6 +32,41 @@ class SeatServiceIntegrationTests(unittest.TestCase):
for event in result["trace_events"]:
self.assertEqual(self.service.contracts.validate(event), [])
def test_dynamic_mode_runs_until_explicit_natural_language_stop(self) -> None:
conversation_id = "conv-dynamic-stop"
started = self.service.process(
"\u5f00\u542f\u52a8\u6001\u6447\u7bee\u6a21\u5f0f",
provider="local",
conversation_id=conversation_id,
vehicle_id="denza_n7",
)
stopped = self.service.process(
"\u5173\u95ed\u52a8\u6001\u6447\u7bee\u6a21\u5f0f",
provider="local",
conversation_id=conversation_id,
vehicle_id="denza_n7",
)
self.assertEqual(started["status"], "succeeded")
self.assertEqual(
started["actuator_state"]["passenger"]["dynamic_mode"],
{
"mode_id": 2,
"name": "\u52a8\u6001\u6447\u7bee\u6a21\u5f0f",
"state": "running",
"repeat": "until_stopped_or_timeout",
"max_runtime_ms": 600_000,
},
)
self.assertEqual(stopped["status"], "succeeded")
operation = stopped["function_config"]["spec"]["operations"][0]
self.assertEqual(operation["capability"], "seat.stop_dynamic_mode")
self.assertEqual(operation["parameters"], {})
self.assertEqual(
stopped["actuator_state"]["passenger"]["dynamic_mode"]["state"],
"idle",
)
def test_gear_request_is_preserved_and_simulated(self) -> None:
result = self.service.process("副驾座椅加热开到三档", provider="local")
domains/seat/tests/unit/test_t2g_hardware.py
+145 -2
View File
@@ -16,6 +16,7 @@ if str(SEAT_SOURCE) not in sys.path:
from cognitive_agent.infrastructure.can.gateway import CanFrame, CanGatewayError # noqa: E402
from cognitive_agent.application.seat_service import SeatDemoService # noqa: E402
from cognitive_agent.runtime.contracts import ContractValidator # noqa: E402
from cognitive_agent.infrastructure.can.hardware import ( # noqa: E402
HardwareSafetyError,
SeatHardwareSettings,
@@ -242,6 +243,34 @@ class ReactiveFakeBus:
},
)
)
if opcode == 14:
self.received.append(
self._frame(
"T2G_Seat_Dynamic_Mode_Status_306",
{
"Dynamic_Mode_ID": command["Param_A"],
"Dynamic_State": 2,
"Dynamic_Phase": 2 if command["Param_A"] == 1 else 5,
"Dynamic_Active_Mask": 0x0F,
"Dynamic_Cycle_Index": 0,
"Dynamic_Phase_Elapsed_Ms": 100,
},
)
)
if opcode == 15:
self.received.append(
self._frame(
"T2G_Seat_Dynamic_Mode_Status_306",
{
"Dynamic_Mode_ID": 0,
"Dynamic_State": 0,
"Dynamic_Phase": 0,
"Dynamic_Active_Mask": 0,
"Dynamic_Cycle_Index": 0,
"Dynamic_Phase_Elapsed_Ms": 0,
},
)
)
if opcode == 0x0B and self.complete_motion:
motor = command["Motor_Selector"]
self.position_valid = True
@@ -447,7 +476,7 @@ class ProtocolSignatureFakeBus:
self.received: deque[CanFrame] = deque()
self.closed = False
if signature:
self.received.append(CanFrame(0x301, bytes.fromhex("10 00 00 00 00 00 F0 60")))
self.received.append(CanFrame(0x301, bytes.fromhex("10 00 00 00 00 00 F0 D0")))
for motor in range(4):
self.received.append(
CanFrame(0x302, bytes([motor, 0, 0x08, 0, 0, 0, 0, 0]))
@@ -464,6 +493,7 @@ class ProtocolSignatureFakeBus:
CanFrame(0x304, bytes([motor, 3, 0, 0, 0, 0, 0xF4, 0x01]))
)
self.received.append(CanFrame(0x305, bytes(8)))
self.received.append(CanFrame(0x306, bytes(8)))
self.received.append(CanFrame(0x4FA, bytes.fromhex("09 00 00 00 00 00 00 00")))
def send(self, arbitration_id: int, data: bytes) -> CanFrame:
@@ -485,6 +515,29 @@ class FailingProtocolSignatureBus(ProtocolSignatureFakeBus):
raise RuntimeError("CAN receive failed during passive protocol probe")
class LegacyProtocolSignatureFakeBus(ProtocolSignatureFakeBus):
def __init__(self):
super().__init__(False)
self.received.append(CanFrame(0x301, bytes.fromhex("10 00 00 00 00 00 F0 00")))
for motor in range(4):
self.received.append(
CanFrame(0x302, bytes([motor, 0, 0x08, 0, 0, 0, 0, 0]))
)
for motor, maximum in enumerate((480, 374, 917, 647)):
self.received.append(
CanFrame(
0x303,
bytes([motor, 0, 0, 0, maximum & 0xFF, maximum >> 8, 0, 0]),
)
)
for motor in range(4):
self.received.append(
CanFrame(0x304, bytes([motor, 3, 0, 0, 0, 0, 0xF4, 0x01]))
)
self.received.append(CanFrame(0x305, bytes(8)))
self.received.append(CanFrame(0x4FA, bytes(8)))
class FakeProgrammer:
def __init__(self):
self.calls = 0
@@ -511,6 +564,7 @@ class T2GHardwareTests(unittest.TestCase):
ROOT / "domains" / "seat" / "can" / "signal-mapping.hardware.yaml",
ROOT / "platforms" / "t2g_seat_bench" / "calibration" / "seat.yaml",
)
self.contracts = ContractValidator(ROOT)
def test_compiler_emits_verified_heating_payload(self) -> None:
config = function_config(
@@ -569,6 +623,34 @@ class T2GHardwareTests(unittest.TestCase):
self.assertEqual(fallback["signals"]["Param_B"], 900)
self.assertEqual(fallback["completion"]["signal"], "Task_State")
def test_compiler_emits_native_dynamic_mode_command(self) -> None:
config = function_config(
operation("seat.start_dynamic_mode", {"mode_id": 1})
)
plan = self.compiler.compile(config, {})
command = plan["spec"]["commands"][0]
frame = self.matrix.encode(command["message"], command["signals"])
self.assertEqual(frame["frame_id"], "0x300")
self.assertEqual(frame["data_hex"], "0E 10 00 00 01 00 00 00")
self.assertEqual(command["completion"]["message"], "T2G_Seat_Dynamic_Mode_Status_306")
self.assertEqual(command["completion"]["accepted_states"], [1, 2])
self.assertEqual(self.contracts.validate(plan), [])
def test_compiler_emits_native_dynamic_mode_stop_command(self) -> None:
config = function_config(operation("seat.stop_dynamic_mode", {}))
plan = self.compiler.compile(config, {})
command = plan["spec"]["commands"][0]
frame = self.matrix.encode(command["message"], command["signals"])
self.assertEqual(frame["frame_id"], "0x300")
self.assertEqual(frame["data_hex"], "0F 10 00 00 00 00 00 00")
self.assertEqual(command["completion"]["signal"], "Dynamic_State")
self.assertEqual(command["completion"]["accepted_states"], [0])
self.assertEqual(self.contracts.validate(plan), [])
def test_compiler_maps_large_relative_backrest_request_to_absolute_target(self) -> None:
config = function_config(
operation(
@@ -696,6 +778,46 @@ class T2GHardwareTests(unittest.TestCase):
self.assertEqual([item.data[0] for item in bus.sent], [9, 1])
self.assertTrue(bus.closed)
def test_executor_waits_for_native_dynamic_mode_feedback(self) -> None:
config = function_config(
operation("seat.start_dynamic_mode", {"mode_id": 2})
)
plan = self.compiler.compile(config, {})
bus = ReactiveFakeBus(self.matrix)
report = HardwareSeatExecutor(
self.matrix, lambda: bus, ack_timeout_ms=20
).execute(plan, config)
self.assertEqual(report["spec"]["status"], "succeeded")
telemetry = report["spec"]["results"][0]["telemetry"]
self.assertEqual(telemetry["dynamic_mode_id"], 2)
self.assertEqual(telemetry["dynamic_state"], 2)
self.assertEqual(telemetry["dynamic_phase"], 5)
self.assertEqual([item.data[0] for item in bus.sent], [9, 14])
self.assertTrue(bus.closed)
def test_executor_stops_dynamic_mode_without_homing_or_pre_stop(self) -> None:
config = function_config(operation("seat.stop_dynamic_mode", {}))
plan = self.compiler.compile(config, {})
bus = ReactiveFakeBus(self.matrix)
report = HardwareSeatExecutor(
self.matrix,
lambda: bus,
ack_timeout_ms=20,
synchronize_position_before_first_motion=True,
position_sync_state={"synchronized": False},
).execute(plan, config)
self.assertEqual(report["spec"]["status"], "succeeded")
telemetry = report["spec"]["results"][0]["telemetry"]
self.assertEqual(telemetry["dynamic_mode_id"], 0)
self.assertEqual(telemetry["dynamic_state"], 0)
self.assertEqual(telemetry["repeat"], "none")
self.assertEqual([item.data[0] for item in bus.sent], [15])
self.assertTrue(bus.closed)
def test_firmware_replacement_forces_four_axis_position_memory_sync(self) -> None:
config = function_config(
operation("seat.set_backrest_recline", {"position_pct": 50})
@@ -1453,6 +1575,9 @@ class T2GHardwareTests(unittest.TestCase):
settings.validate_function_config(
function_config(operation("seat.set_cushion_tilt", {"position_pct": 40}))
)
settings.validate_function_config(
function_config(operation("seat.stop_dynamic_mode", {}))
)
with self.assertRaises(HardwareSafetyError):
settings.validate_function_config(
function_config(
@@ -1559,7 +1684,7 @@ class T2GHardwareTests(unittest.TestCase):
self.assertIs(prepared, bus)
self.assertFalse(telemetry["programmed_this_request"])
self.assertEqual(telemetry["firmware_revision"], 6)
self.assertEqual(telemetry["firmware_revision"], 13)
self.assertEqual(bus.sent, [])
self.assertEqual(telemetry["max_hall_counts"], [480, 374, 917, 647])
self.assertFalse(bus.closed)
@@ -1617,6 +1742,24 @@ class T2GHardwareTests(unittest.TestCase):
guard.prepare_bus()
self.assertTrue(bus.closed)
def test_firmware_guard_reports_legacy_revision_and_missing_dynamic_status(self) -> None:
bus = LegacyProtocolSignatureFakeBus()
guard = T2GFirmwareGuard(
lambda: bus,
None,
auto_flash=False,
probe_timeout_ms=20,
)
with self.assertRaises(FirmwareProgrammingError) as raised:
guard.prepare_bus()
message = str(raised.exception)
self.assertIn("firmware_revision=0 (expected 13)", message)
self.assertIn("missing_frame_ids=0x306", message)
self.assertIn("max_hall_counts=480,374,917,647", message)
self.assertTrue(bus.closed)
def test_programmer_command_downloads_and_hands_off_to_running_target(self) -> None:
firmware_root = ROOT / "firmware" / "t2g" / "Seat_Controller_demo"
output_image = (