minghanchen
2033 lines
80 KiB
Python
Executable File
2033 lines
80 KiB
Python
Executable File
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import unittest
|
|
from collections import deque
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[4]
|
|
SEAT_SOURCE = ROOT / "domains" / "seat" / "src"
|
|
if str(SEAT_SOURCE) not in sys.path:
|
|
sys.path.insert(0, str(SEAT_SOURCE))
|
|
|
|
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,
|
|
)
|
|
from cognitive_agent.infrastructure.firmware import ( # noqa: E402
|
|
FirmwareProgrammingError,
|
|
T2GFirmwareGuard,
|
|
T2GSeatProgrammer,
|
|
)
|
|
from seat_domain import HardwareSeatExecutor, SeatCanMatrix, T2GSeatCompiler # noqa: E402
|
|
|
|
|
|
def function_config(*operations: dict[str, object]) -> dict[str, object]:
|
|
return {
|
|
"apiVersion": "cognitive-agent/v1alpha1",
|
|
"kind": "FunctionConfig",
|
|
"metadata": {
|
|
"id": "function-test-hardware",
|
|
"name": "T2G hardware test",
|
|
"version": "1.0.0",
|
|
"request_id": "req-hardware-test",
|
|
},
|
|
"spec": {"domain": "seat", "operations": list(operations)},
|
|
}
|
|
|
|
|
|
def operation(
|
|
capability: str,
|
|
parameters: dict[str, object],
|
|
*,
|
|
operation_id: str = "op-hardware-test",
|
|
instance: str = "passenger",
|
|
zone: str = "front_right",
|
|
motion: dict[str, object] | None = None,
|
|
) -> dict[str, object]:
|
|
result = {
|
|
"id": operation_id,
|
|
"capability": capability,
|
|
"target": {"instance": instance, "zone": zone},
|
|
"parameters": parameters,
|
|
"timeout_ms": 20,
|
|
}
|
|
if motion is not None:
|
|
result["motion"] = motion
|
|
return result
|
|
|
|
|
|
class ReactiveFakeBus:
|
|
def __init__(
|
|
self,
|
|
matrix: SeatCanMatrix,
|
|
*,
|
|
complete_motion: bool = True,
|
|
position_valid: bool = True,
|
|
absolute_stall_motors: set[int] | None = None,
|
|
absolute_timeout_motors: set[int] | None = None,
|
|
absolute_stall_hall_after: dict[int, int] | None = None,
|
|
absolute_failure_targets: dict[int, int] | None = None,
|
|
timed_failure_result: int | None = None,
|
|
timed_failure_event: int = 4,
|
|
timed_endpoint: str | None = None,
|
|
):
|
|
self.matrix = matrix
|
|
self.complete_motion = complete_motion
|
|
self.position_valid = position_valid
|
|
self.absolute_stall_motors = set(absolute_stall_motors or set())
|
|
self.absolute_timeout_motors = set(absolute_timeout_motors or set())
|
|
self.absolute_stall_hall_after = dict(absolute_stall_hall_after or {})
|
|
self.absolute_failure_targets = dict(absolute_failure_targets or {})
|
|
self.timed_failure_result = timed_failure_result
|
|
self.timed_failure_event = timed_failure_event
|
|
self.timed_endpoint = timed_endpoint
|
|
self.hall = [0, 0, 0, 0]
|
|
self.position = [0, 0, 0, 0]
|
|
self.max_hall = [480, 374, 917, 647]
|
|
self.sent: list[CanFrame] = []
|
|
self.received: deque[CanFrame] = deque()
|
|
self.closed = False
|
|
|
|
def _frame(self, message: str, values: dict[str, int]) -> CanFrame:
|
|
encoded = self.matrix.encode(message, values)
|
|
return CanFrame(int(encoded["frame_id"], 16), bytes.fromhex(encoded["data_hex"]))
|
|
|
|
def send(self, arbitration_id: int, data: bytes) -> CanFrame:
|
|
frame = CanFrame(arbitration_id, bytes(data))
|
|
self.sent.append(frame)
|
|
if arbitration_id == 0x300:
|
|
command = self.matrix.decode("T2G_Seat_Move_Command_300", data)
|
|
sequence = command["Sequence"]
|
|
opcode = command["Opcode"]
|
|
stop_all_selector_valid = (
|
|
opcode != 9 or command["Motor_Selector"] == 0
|
|
)
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_System_Status_301",
|
|
{
|
|
"Ack": (
|
|
4 if opcode == 9 and stop_all_selector_valid
|
|
else 2 if opcode == 9
|
|
else 1
|
|
),
|
|
"Reject_Reason": 0 if stop_all_selector_valid else 4,
|
|
"Last_Opcode": opcode,
|
|
"Last_Sequence": sequence,
|
|
"Safety_Inhibit": 0,
|
|
"Fault_Summary": 0,
|
|
},
|
|
)
|
|
)
|
|
if opcode == 9 and stop_all_selector_valid:
|
|
for motor in range(4):
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Motor_Status_302",
|
|
{
|
|
"Motor_Sub_ID": motor,
|
|
"Task_State": 0,
|
|
"Result": 0,
|
|
"Position_Valid": int(self.position_valid),
|
|
"Motor_Sequence": sequence,
|
|
"Value_X100": 0,
|
|
"Duty_Percent": 0,
|
|
},
|
|
)
|
|
)
|
|
for motor in range(4):
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Diagnostic_303",
|
|
{
|
|
"Motor_Sub_ID": motor,
|
|
"Hall_Count": self.hall[motor],
|
|
"Max_Hall_Count": self.max_hall[motor],
|
|
"Hall_Speed_Half_Pps": 180,
|
|
},
|
|
)
|
|
)
|
|
if opcode == 1 and self.complete_motion:
|
|
motor = command["Motor_Selector"]
|
|
if (
|
|
motor in self.absolute_stall_motors | self.absolute_timeout_motors
|
|
and command["Param_A"]
|
|
== self.absolute_failure_targets.get(motor, 0)
|
|
):
|
|
if motor in self.absolute_stall_hall_after:
|
|
self.hall[motor] = self.absolute_stall_hall_after[motor]
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Current_Diagnostic_304",
|
|
{
|
|
"Motor_Sub_ID": motor,
|
|
"Current_Sample_Valid": 1,
|
|
"Current_Baseline_Valid": 1,
|
|
"High_Current": int(
|
|
motor in self.absolute_stall_motors
|
|
),
|
|
"Hall_Activity_Low": 1,
|
|
"Current_Raw": 800,
|
|
"Current_Baseline_Raw": 100,
|
|
"Stall_Current_Delta_Threshold_Raw": 500,
|
|
},
|
|
)
|
|
)
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Motor_Status_302",
|
|
{
|
|
"Motor_Sub_ID": motor,
|
|
"Task_State": 7,
|
|
"Result": (
|
|
6 if motor in self.absolute_timeout_motors else 5
|
|
),
|
|
"Position_Valid": 1,
|
|
"Motor_Sequence": sequence,
|
|
"Value_X100": self.position[motor],
|
|
"Duty_Percent": 0,
|
|
},
|
|
)
|
|
)
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Diagnostic_303",
|
|
{
|
|
"Motor_Sub_ID": motor,
|
|
"Last_Motor_Event": (
|
|
5 if motor in self.absolute_timeout_motors else 4
|
|
),
|
|
"Hall_Count": self.hall[motor],
|
|
"Max_Hall_Count": self.max_hall[motor],
|
|
},
|
|
)
|
|
)
|
|
else:
|
|
self.position[motor] = command["Param_A"]
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Motor_Status_302",
|
|
{
|
|
"Motor_Sub_ID": motor,
|
|
"Task_State": 6,
|
|
"Result": 2,
|
|
"Position_Valid": 1,
|
|
"Motor_Sequence": sequence,
|
|
"Value_X100": command["Param_A"],
|
|
"Duty_Percent": 90,
|
|
},
|
|
)
|
|
)
|
|
# Absolute commands in a multi-axis request each perform
|
|
# a fresh Position_Valid preflight. Mirror cyclic 0x302.
|
|
for cyclic_motor in range(4):
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Motor_Status_302",
|
|
{
|
|
"Motor_Sub_ID": cyclic_motor,
|
|
"Task_State": 0,
|
|
"Result": 0,
|
|
"Position_Valid": 1,
|
|
"Motor_Sequence": sequence,
|
|
"Value_X100": self.position[cyclic_motor],
|
|
"Duty_Percent": 0,
|
|
},
|
|
)
|
|
)
|
|
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
|
|
self.hall[motor] = 0
|
|
self.position[motor] = 0
|
|
home_status = {
|
|
"Motor_Sub_ID": motor,
|
|
"Task_State": 6,
|
|
"Result": 2,
|
|
"Position_Valid": 1,
|
|
"Near_Endpoint": 1,
|
|
"Motor_Sequence": sequence,
|
|
"Value_X100": 0,
|
|
"Duty_Percent": 0,
|
|
}
|
|
self.received.append(
|
|
self._frame("T2G_Seat_Move_Motor_Status_302", home_status)
|
|
)
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Diagnostic_303",
|
|
{
|
|
"Motor_Sub_ID": motor,
|
|
"Hall_Count": 0,
|
|
"Max_Hall_Count": self.max_hall[motor],
|
|
},
|
|
)
|
|
)
|
|
# The real controller publishes all four cyclic motor status
|
|
# samples after a home completes. Keep a complete cycle queued
|
|
# for the next axis preflight in a four-axis zero request.
|
|
for cyclic_motor in range(4):
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Motor_Status_302",
|
|
{
|
|
"Motor_Sub_ID": cyclic_motor,
|
|
"Task_State": 0,
|
|
"Result": 0,
|
|
"Position_Valid": 1,
|
|
"Motor_Sequence": sequence,
|
|
"Value_X100": self.position[cyclic_motor],
|
|
"Duty_Percent": 0,
|
|
},
|
|
)
|
|
)
|
|
for cyclic_motor in range(4):
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Diagnostic_303",
|
|
{
|
|
"Motor_Sub_ID": cyclic_motor,
|
|
"Hall_Count": self.hall[cyclic_motor],
|
|
"Max_Hall_Count": self.max_hall[cyclic_motor],
|
|
"Hall_Speed_Half_Pps": 180,
|
|
},
|
|
)
|
|
)
|
|
if opcode == 3 and self.timed_endpoint is not None:
|
|
motor = command["Motor_Selector"]
|
|
at_far_endpoint = self.timed_endpoint == "far"
|
|
self.hall[motor] = self.max_hall[motor] if at_far_endpoint else 0
|
|
self.position[motor] = 10_000 if at_far_endpoint else 0
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Motor_Status_302",
|
|
{
|
|
"Motor_Sub_ID": motor,
|
|
"Task_State": 7,
|
|
"Result": 4,
|
|
"Direction": command["Direction"],
|
|
"Position_Valid": 1,
|
|
"Near_Endpoint": int(not at_far_endpoint),
|
|
"Far_Endpoint": int(at_far_endpoint),
|
|
"Motor_Sequence": sequence,
|
|
"Value_X100": self.position[motor],
|
|
"Duty_Percent": 0,
|
|
},
|
|
)
|
|
)
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Diagnostic_303",
|
|
{
|
|
"Motor_Sub_ID": motor,
|
|
"Last_Motor_Event": 3 if at_far_endpoint else 2,
|
|
"Hall_Count": self.hall[motor],
|
|
"Max_Hall_Count": self.max_hall[motor],
|
|
},
|
|
)
|
|
)
|
|
elif opcode == 3 and self.timed_failure_result is not None:
|
|
motor = command["Motor_Selector"]
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Motor_Status_302",
|
|
{
|
|
"Motor_Sub_ID": motor,
|
|
"Task_State": 7,
|
|
"Result": self.timed_failure_result,
|
|
"Direction": command["Direction"],
|
|
"Position_Valid": int(self.position_valid),
|
|
"Motor_Sequence": sequence,
|
|
"Value_X100": 0,
|
|
"Duty_Percent": command["Param_B"] // 10,
|
|
},
|
|
)
|
|
)
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Diagnostic_303",
|
|
{
|
|
"Motor_Sub_ID": motor,
|
|
"Last_Motor_Event": self.timed_failure_event,
|
|
"Hall_Count": self.hall[motor],
|
|
"Max_Hall_Count": self.max_hall[motor],
|
|
"Hall_Speed_Half_Pps": 180,
|
|
},
|
|
)
|
|
)
|
|
elif opcode == 3 and self.complete_motion:
|
|
motor = command["Motor_Selector"]
|
|
delta = max(1, command["Param_A"] // 35)
|
|
if command["Direction"] == 1:
|
|
self.hall[motor] = min(
|
|
self.max_hall[motor], self.hall[motor] + delta
|
|
)
|
|
else:
|
|
self.hall[motor] = max(0, self.hall[motor] - delta)
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Motor_Status_302",
|
|
{
|
|
"Motor_Sub_ID": motor,
|
|
"Task_State": 6,
|
|
"Result": 2,
|
|
"Direction": command["Direction"],
|
|
"Position_Valid": int(self.position_valid),
|
|
"Motor_Sequence": sequence,
|
|
"Value_X100": 0,
|
|
"Duty_Percent": command["Param_B"] // 10,
|
|
},
|
|
)
|
|
)
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Diagnostic_303",
|
|
{
|
|
"Motor_Sub_ID": motor,
|
|
"Hall_Count": self.hall[motor],
|
|
"Max_Hall_Count": self.max_hall[motor],
|
|
"Hall_Speed_Half_Pps": 180,
|
|
},
|
|
)
|
|
)
|
|
# Keep a cyclic sample for the endpoint re-home preflight
|
|
# that follows a successful opposite-direction probe.
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Diagnostic_303",
|
|
{
|
|
"Motor_Sub_ID": motor,
|
|
"Hall_Count": self.hall[motor],
|
|
"Max_Hall_Count": self.max_hall[motor],
|
|
"Hall_Speed_Half_Pps": 180,
|
|
},
|
|
)
|
|
)
|
|
elif arbitration_id == 0x29E:
|
|
control = self.matrix.decode("T2G_Seat_Thermal_Control_29E", data)
|
|
heating = control["Driver_Seat_Heating_On"]
|
|
ventilation = control["Driver_Seat_Ventilation_On"]
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Thermal_Status_4FA",
|
|
{
|
|
"Driver_Seat_Heating_Status": (
|
|
min(control["Driver_Seat_Heating_Gear"], 3) + 1
|
|
if heating == 1
|
|
else 1
|
|
),
|
|
"Driver_Seat_Ventilation_Status": (
|
|
min(control["Driver_Seat_Ventilation_Gear"], 3) + 1
|
|
if ventilation == 1
|
|
else 1
|
|
),
|
|
},
|
|
)
|
|
)
|
|
return frame
|
|
|
|
def receive(self, timeout_ms: int) -> CanFrame | None:
|
|
del timeout_ms
|
|
return self.received.popleft() if self.received else None
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
class ProtocolSignatureFakeBus:
|
|
def __init__(self, signature: bool):
|
|
self.sent: list[CanFrame] = []
|
|
self.received: deque[CanFrame] = deque()
|
|
self.closed = False
|
|
if signature:
|
|
self.received.append(CanFrame(0x301, bytes.fromhex("10 00 00 00 00 00 F0 F0")))
|
|
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(0x306, bytes(8)))
|
|
self.received.append(
|
|
CanFrame(0x307, bytes.fromhex("53 4D 56 38 0F 10 04 01"))
|
|
)
|
|
self.received.append(CanFrame(0x4FA, bytes.fromhex("09 00 00 00 00 00 00 00")))
|
|
|
|
def send(self, arbitration_id: int, data: bytes) -> CanFrame:
|
|
frame = CanFrame(arbitration_id, bytes(data))
|
|
self.sent.append(frame)
|
|
return frame
|
|
|
|
def receive(self, timeout_ms: int) -> CanFrame | None:
|
|
del timeout_ms
|
|
return self.received.popleft() if self.received else None
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
class FailingProtocolSignatureBus(ProtocolSignatureFakeBus):
|
|
def receive(self, timeout_ms: int) -> CanFrame | None:
|
|
del timeout_ms
|
|
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
|
|
|
|
def program(self) -> dict[str, object]:
|
|
self.calls += 1
|
|
return {
|
|
"backend": "fake",
|
|
"programmed": True,
|
|
"image": "Seat_Controller_demo.hex",
|
|
"image_sha256": "0" * 64,
|
|
}
|
|
|
|
|
|
class T2GHardwareTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.matrix = SeatCanMatrix(
|
|
[
|
|
ROOT / "domains" / "seat" / "can" / "seat-thermal-hardware-matrix.yaml",
|
|
ROOT / "domains" / "seat" / "can" / "seat-motion-hardware-matrix.yaml",
|
|
]
|
|
)
|
|
self.compiler = T2GSeatCompiler(
|
|
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(
|
|
operation("seat.set_heating", {"state": "on", "level": 3})
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
frames = [
|
|
self.matrix.encode(item["message"], item["signals"])
|
|
for item in plan["spec"]["commands"]
|
|
]
|
|
|
|
self.assertEqual(frames[0]["frame_id"], "0x3CE")
|
|
self.assertEqual(frames[0]["data_hex"], "10 00 00 00 00 00 00 00")
|
|
self.assertEqual(frames[1]["frame_id"], "0x29E")
|
|
self.assertEqual(frames[1]["data_hex"], "16 00 30 00 00 00 00 00")
|
|
|
|
def test_compiler_emits_verified_backrest_payload(self) -> None:
|
|
config = function_config(
|
|
operation("seat.set_backrest_recline", {"position_pct": 50})
|
|
)
|
|
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"], "01 12 00 00 88 13 30 75")
|
|
self.assertEqual(command["signals"]["Param_B"], 30_000)
|
|
self.assertEqual(command["completion"]["timeout_ms"], 33_000)
|
|
|
|
def test_compiler_prefers_absolute_relative_target_with_timed_fallback(self) -> None:
|
|
config = function_config(
|
|
operation(
|
|
"seat.set_backrest_recline",
|
|
{"position_pct": 31, "angle_deg": 109},
|
|
motion={
|
|
"mode": "relative",
|
|
"direction": "increase",
|
|
"start_position_pct": 20,
|
|
"delta_position_pct": 11,
|
|
"start_angle_deg": 104,
|
|
"delta_angle_deg": 5,
|
|
},
|
|
)
|
|
)
|
|
|
|
command = self.compiler.compile(config, {})["spec"]["commands"][0]
|
|
|
|
self.assertEqual(command["signals"]["Opcode"], 1)
|
|
self.assertEqual(command["signals"]["Direction"], 0)
|
|
self.assertEqual(command["signals"]["Param_A"], 3100)
|
|
self.assertEqual(command["completion"]["signal"], "Value_X100")
|
|
fallback = command["relative_fallback"]
|
|
self.assertEqual(fallback["signals"]["Opcode"], 3)
|
|
self.assertEqual(fallback["signals"]["Direction"], 1)
|
|
self.assertEqual(fallback["signals"]["Param_A"], 1320)
|
|
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(
|
|
"seat.set_backrest_recline",
|
|
{"position_pct": 64, "angle_deg": 124},
|
|
motion={
|
|
"mode": "relative",
|
|
"direction": "increase",
|
|
"start_position_pct": 20,
|
|
"delta_position_pct": 44,
|
|
"start_angle_deg": 104,
|
|
"delta_angle_deg": 20,
|
|
},
|
|
)
|
|
)
|
|
|
|
command = self.compiler.compile(config, {})["spec"]["commands"][0]
|
|
|
|
self.assertEqual(command["signals"]["Opcode"], 1)
|
|
self.assertEqual(command["signals"]["Direction"], 0)
|
|
self.assertEqual(command["signals"]["Param_A"], 6400)
|
|
self.assertEqual(command["completion"]["signal"], "Value_X100")
|
|
|
|
def test_compiler_emits_bounded_user_timed_motion(self) -> None:
|
|
config = function_config(
|
|
operation(
|
|
"seat.set_backrest_recline",
|
|
{"position_pct": 100, "angle_deg": 163},
|
|
motion={
|
|
"mode": "timed",
|
|
"direction": "increase",
|
|
"start_position_pct": 0,
|
|
"delta_position_pct": 100,
|
|
"requested_duration_ms": 70_000,
|
|
"applied_duration_ms": 30_000,
|
|
"max_duration_ms": 30_000,
|
|
"min_duration_ms": 300,
|
|
"duration_clamped": True,
|
|
"endpoint_expected": "far",
|
|
},
|
|
)
|
|
)
|
|
|
|
command = self.compiler.compile(config, {})["spec"]["commands"][0]
|
|
|
|
self.assertEqual(command["signals"]["Opcode"], 3)
|
|
self.assertEqual(command["signals"]["Direction"], 1)
|
|
self.assertEqual(command["signals"]["Param_A"], 30_000)
|
|
self.assertEqual(command["signals"]["Param_B"], 900)
|
|
self.assertEqual(command["completion"]["timeout_ms"], 35_000)
|
|
|
|
def test_compiler_uses_bench_safe_legrest_timed_duty(self) -> None:
|
|
config = function_config(
|
|
operation(
|
|
"seat.set_legrest_height",
|
|
{"position_pct": 30, "angle_deg": 117},
|
|
motion={
|
|
"mode": "timed",
|
|
"direction": "increase",
|
|
"start_position_pct": 0,
|
|
"delta_position_pct": 30,
|
|
"requested_duration_ms": 3_000,
|
|
"applied_duration_ms": 3_000,
|
|
"max_duration_ms": 25_000,
|
|
"min_duration_ms": 300,
|
|
"duration_clamped": False,
|
|
},
|
|
)
|
|
)
|
|
|
|
command = self.compiler.compile(config, {})["spec"]["commands"][0]
|
|
|
|
self.assertEqual(command["signals"]["Motor_Selector"], 3)
|
|
self.assertEqual(command["signals"]["Opcode"], 3)
|
|
self.assertEqual(command["signals"]["Param_A"], 3_000)
|
|
self.assertEqual(command["signals"]["Param_B"], 500)
|
|
|
|
def test_driver_target_is_rejected_by_hardware_compiler(self) -> None:
|
|
driver = function_config(
|
|
operation(
|
|
"seat.set_heating",
|
|
{"state": "on", "level": 2},
|
|
instance="driver",
|
|
zone="front_left",
|
|
)
|
|
)
|
|
|
|
with self.assertRaisesRegex(ValueError, "passenger/front_right"):
|
|
self.compiler.compile(driver, {})
|
|
|
|
def test_compiler_spaces_sequential_axes_around_work_flash_persistence(self) -> None:
|
|
config = function_config(
|
|
operation("seat.set_longitudinal_position", {"position_pct": 51}),
|
|
operation("seat.set_cushion_tilt", {"position_pct": 30}),
|
|
)
|
|
|
|
commands = self.compiler.compile(config, {})["spec"]["commands"]
|
|
|
|
self.assertEqual(
|
|
[command["send_offset_ms"] for command in commands],
|
|
[0, 2_000],
|
|
)
|
|
|
|
def test_duplicate_passenger_motion_is_sent_once(self) -> None:
|
|
config = function_config(
|
|
operation(
|
|
"seat.set_backrest_recline",
|
|
{"position_pct": 60},
|
|
operation_id="op-passenger-a",
|
|
),
|
|
operation(
|
|
"seat.set_backrest_recline",
|
|
{"position_pct": 60},
|
|
operation_id="op-passenger-b",
|
|
),
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
self.assertEqual(len(plan["spec"]["commands"]), 1)
|
|
self.assertEqual(
|
|
plan["spec"]["commands"][0]["operation_ref"],
|
|
"op-passenger-a,op-passenger-b",
|
|
)
|
|
|
|
bus = ReactiveFakeBus(self.matrix)
|
|
report = HardwareSeatExecutor(
|
|
self.matrix, lambda: bus, ack_timeout_ms=20
|
|
).execute(plan, config)
|
|
self.assertEqual([item.data[0] for item in bus.sent], [9, 1])
|
|
operation_results = [
|
|
item for item in report["spec"]["results"] if item["ref"].startswith("op-")
|
|
]
|
|
self.assertEqual(
|
|
{item["ref"] for item in operation_results},
|
|
{"op-passenger-a", "op-passenger-b"},
|
|
)
|
|
|
|
def test_executor_waits_for_ack_and_physical_motion_feedback(self) -> None:
|
|
config = function_config(
|
|
operation("seat.set_backrest_recline", {"position_pct": 50})
|
|
)
|
|
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"]["mode"], "hil")
|
|
self.assertEqual(report["spec"]["status"], "succeeded")
|
|
self.assertFalse(report["spec"]["results"][0]["telemetry"]["simulated"])
|
|
self.assertTrue(report["spec"]["results"][0]["telemetry"]["hardware_feedback"])
|
|
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})
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(self.matrix)
|
|
bus.hall = [220, 180, 600, 310]
|
|
bus.position = [4500, 4800, 6500, 4700]
|
|
position_sync_state = {"synchronized": True}
|
|
|
|
class ReprogrammedGuard:
|
|
def prepare_bus(self) -> tuple[ReactiveFakeBus, dict[str, object]]:
|
|
return bus, {
|
|
"firmware_verified": True,
|
|
"programmed_this_request": True,
|
|
}
|
|
|
|
with patch("seat_domain.hardware_executor.time.sleep"):
|
|
report = HardwareSeatExecutor(
|
|
self.matrix,
|
|
lambda: bus,
|
|
ack_timeout_ms=20,
|
|
motor_max_hall={0: 480, 1: 374, 2: 917, 3: 647},
|
|
firmware_guard=ReprogrammedGuard(), # type: ignore[arg-type]
|
|
synchronize_position_before_first_motion=True,
|
|
position_sync_state=position_sync_state,
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "succeeded")
|
|
sent_commands = [
|
|
self.matrix.decode("T2G_Seat_Move_Command_300", frame.data)
|
|
for frame in bus.sent
|
|
if frame.arbitration_id == 0x300
|
|
]
|
|
self.assertEqual(
|
|
[(item["Opcode"], item["Motor_Selector"]) for item in sent_commands],
|
|
[(9, 0), (0x0B, 0), (0x0B, 1), (0x0B, 2), (0x0B, 3), (1, 2)],
|
|
)
|
|
sync = next(
|
|
item
|
|
for item in report["spec"]["results"]
|
|
if item["ref"] == "seat-position-memory-sync"
|
|
)
|
|
self.assertEqual(
|
|
sync["telemetry"]["trigger"], "repository_firmware_reprogrammed"
|
|
)
|
|
self.assertEqual(len(sync["telemetry"]["axes"]), 4)
|
|
self.assertTrue(position_sync_state["synchronized"])
|
|
|
|
def test_first_process_motion_syncs_once_and_reuses_the_reference(self) -> None:
|
|
config = function_config(
|
|
operation("seat.set_longitudinal_position", {"position_pct": 40})
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
state = {"synchronized": False}
|
|
first_bus = ReactiveFakeBus(self.matrix)
|
|
with patch("seat_domain.hardware_executor.time.sleep"):
|
|
first = HardwareSeatExecutor(
|
|
self.matrix,
|
|
lambda: first_bus,
|
|
ack_timeout_ms=20,
|
|
motor_max_hall={0: 480, 1: 374, 2: 917, 3: 647},
|
|
synchronize_position_before_first_motion=True,
|
|
position_sync_state=state,
|
|
).execute(plan, config)
|
|
|
|
second_bus = ReactiveFakeBus(self.matrix)
|
|
second = HardwareSeatExecutor(
|
|
self.matrix,
|
|
lambda: second_bus,
|
|
ack_timeout_ms=20,
|
|
motor_max_hall={0: 480, 1: 374, 2: 917, 3: 647},
|
|
synchronize_position_before_first_motion=True,
|
|
position_sync_state=state,
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(first["spec"]["status"], "succeeded")
|
|
self.assertEqual(second["spec"]["status"], "succeeded")
|
|
self.assertEqual([item.data[0] for item in first_bus.sent], [9, 0x0B, 0x0B, 0x0B, 0x0B, 1])
|
|
self.assertEqual([item.data[0] for item in second_bus.sent], [9, 1])
|
|
|
|
def test_passive_snapshot_reads_all_passenger_axes_and_thermal_state(self) -> None:
|
|
bus = ReactiveFakeBus(self.matrix)
|
|
bus.received.extend(
|
|
[
|
|
bus._frame(
|
|
"T2G_Seat_Move_System_Status_301",
|
|
{
|
|
"Ack": 1,
|
|
"System_State": 2,
|
|
"Last_Sequence": 9,
|
|
"Last_Opcode": 1,
|
|
"Position_Valid_Mask": 0x0F,
|
|
"Firmware_Revision": 1,
|
|
},
|
|
),
|
|
bus._frame(
|
|
"T2G_Seat_Move_Diagnostic_303",
|
|
{
|
|
"Motor_Sub_ID": 0,
|
|
"Hall_Count": 120,
|
|
"Max_Hall_Count": 480,
|
|
"Hall_Speed_Half_Pps": 4,
|
|
},
|
|
),
|
|
bus._frame(
|
|
"T2G_Seat_Move_Current_Diagnostic_304",
|
|
{
|
|
"Motor_Sub_ID": 0,
|
|
"Current_Sample_Valid": 1,
|
|
"Current_Baseline_Valid": 1,
|
|
"Current_Raw": 240,
|
|
"Current_Baseline_Raw": 100,
|
|
"Stall_Current_Delta_Threshold_Raw": 500,
|
|
},
|
|
),
|
|
bus._frame(
|
|
"T2G_Seat_Thermal_Current_Diagnostic_305",
|
|
{
|
|
"Fan_Current_Sample_Valid": 1,
|
|
"Heater_Current_Sample_Valid": 1,
|
|
"Fan_Running": 1,
|
|
"Heater_Level": 2,
|
|
"Fan_Current_Raw": 300,
|
|
"Heater_Current_Raw": 450,
|
|
},
|
|
),
|
|
]
|
|
)
|
|
for motor, value_x100 in enumerate((1200, 3400, 5600, 7800)):
|
|
bus.received.append(
|
|
bus._frame(
|
|
"T2G_Seat_Move_Motor_Status_302",
|
|
{
|
|
"Motor_Sub_ID": motor,
|
|
"Task_State": 6,
|
|
"Result": 2,
|
|
"Position_Valid": 1,
|
|
"Motor_Sequence": 7,
|
|
"Value_X100": value_x100,
|
|
"Duty_Percent": 0,
|
|
},
|
|
)
|
|
)
|
|
bus.received.append(
|
|
bus._frame(
|
|
"T2G_Seat_Thermal_Status_4FA",
|
|
{
|
|
"Driver_Seat_Ventilation_Status": 1,
|
|
"Driver_Seat_Heating_Status": 3,
|
|
},
|
|
)
|
|
)
|
|
executor = HardwareSeatExecutor(
|
|
self.matrix,
|
|
lambda: bus,
|
|
ack_timeout_ms=20,
|
|
thermal_timeout_ms=20,
|
|
axis_maximum_x100={
|
|
capability: 10_000 for capability in self.compiler.axes
|
|
},
|
|
)
|
|
|
|
physical = executor.read_physical_state()
|
|
|
|
self.assertTrue(physical["complete"])
|
|
self.assertEqual(
|
|
{
|
|
axis: value["position_pct"]
|
|
for axis, value in physical["position"].items()
|
|
},
|
|
{
|
|
"longitudinal": 12,
|
|
"cushion_tilt": 34,
|
|
"backrest_recline": 56,
|
|
"legrest_height": 78,
|
|
},
|
|
)
|
|
self.assertEqual(physical["ventilation"]["state"], "off")
|
|
self.assertEqual(physical["heating"]["state"], "on")
|
|
self.assertEqual(physical["heating"]["level"], 2)
|
|
self.assertEqual(physical["channel"], 0)
|
|
self.assertEqual(
|
|
physical["protocol_profile"], "passenger-seat-native-can0-v1"
|
|
)
|
|
self.assertEqual(physical["system"]["position_valid_mask"], 0x0F)
|
|
self.assertEqual(
|
|
physical["diagnostics"]["motors"]["longitudinal"]["hall_count"],
|
|
120,
|
|
)
|
|
self.assertEqual(
|
|
physical["diagnostics"]["motors"]["longitudinal"]["current"]["raw"],
|
|
240,
|
|
)
|
|
self.assertEqual(
|
|
physical["diagnostics"]["thermal"]["heater_current_raw"], 450
|
|
)
|
|
self.assertEqual(
|
|
physical["observed_frame_ids"],
|
|
["0x301", "0x302", "0x303", "0x304", "0x305", "0x4FA"],
|
|
)
|
|
self.assertTrue(bus.closed)
|
|
|
|
def test_executor_sends_firmware_valid_stop_all_before_four_axis_zero(self) -> None:
|
|
config = function_config(
|
|
operation(
|
|
"seat.set_longitudinal_position",
|
|
{"position_pct": 0},
|
|
operation_id="zero-horizontal",
|
|
),
|
|
operation(
|
|
"seat.set_cushion_tilt",
|
|
{"position_pct": 0},
|
|
operation_id="zero-cushion",
|
|
),
|
|
operation(
|
|
"seat.set_backrest_recline",
|
|
{"position_pct": 0},
|
|
operation_id="zero-backrest",
|
|
),
|
|
operation(
|
|
"seat.set_legrest_height",
|
|
{"position_pct": 0},
|
|
operation_id="zero-legrest",
|
|
),
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
self.assertEqual(self.compiler.protocol["stop_all_motor_selector"], 0)
|
|
self.assertEqual(
|
|
[
|
|
command["signals"]["Motor_Selector"]
|
|
for command in plan["spec"]["commands"]
|
|
],
|
|
[0, 1, 2, 3],
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
command["signals"]["Opcode"] == 0x0B
|
|
and command["signals"]["Param_A"] == 0x4B52
|
|
and command["signals"]["Param_B"] == 120
|
|
for command in plan["spec"]["commands"]
|
|
)
|
|
)
|
|
|
|
bus = ReactiveFakeBus(self.matrix)
|
|
position_sync_state = {"synchronized": False}
|
|
report = HardwareSeatExecutor(
|
|
self.matrix,
|
|
lambda: bus,
|
|
ack_timeout_ms=20,
|
|
stop_all_motor_selector=int(
|
|
self.compiler.protocol["stop_all_motor_selector"]
|
|
),
|
|
synchronize_position_before_first_motion=True,
|
|
position_sync_state=position_sync_state,
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "succeeded")
|
|
sent_commands = [
|
|
self.matrix.decode("T2G_Seat_Move_Command_300", frame.data)
|
|
for frame in bus.sent
|
|
if frame.arbitration_id == 0x300
|
|
]
|
|
self.assertEqual(
|
|
[(item["Opcode"], item["Motor_Selector"]) for item in sent_commands],
|
|
[(9, 0), (0x0B, 0), (0x0B, 1), (0x0B, 2), (0x0B, 3)],
|
|
)
|
|
self.assertTrue(position_sync_state["synchronized"])
|
|
self.assertEqual(
|
|
sum(
|
|
item["ref"] == "seat-position-memory-sync"
|
|
for item in report["spec"]["results"]
|
|
),
|
|
1,
|
|
)
|
|
|
|
def test_zero_recovers_verified_stale_near_endpoint_offset(self) -> None:
|
|
config = function_config(
|
|
operation("seat.set_longitudinal_position", {"position_pct": 0})
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(self.matrix, absolute_stall_motors={0})
|
|
bus.hall[0] = 6
|
|
bus.position[0] = 125
|
|
|
|
report = HardwareSeatExecutor(
|
|
self.matrix,
|
|
lambda: bus,
|
|
ack_timeout_ms=20,
|
|
motor_max_hall={0: 480, 1: 374, 2: 917, 3: 647},
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "succeeded")
|
|
sent_commands = [
|
|
self.matrix.decode("T2G_Seat_Move_Command_300", frame.data)
|
|
for frame in bus.sent
|
|
if frame.arbitration_id == 0x300
|
|
]
|
|
self.assertEqual([item["Opcode"] for item in sent_commands], [9, 0x0B])
|
|
telemetry = report["spec"]["results"][0]["telemetry"]
|
|
recovery = telemetry["position_recovery"]
|
|
self.assertEqual(recovery["mode"], "explicit_known_range_home")
|
|
self.assertEqual(recovery["hall_before"], 6)
|
|
self.assertEqual(recovery["hall_after"], 0)
|
|
self.assertEqual(telemetry["effective_position_pct"], 0)
|
|
|
|
def test_zero_recovers_after_full_travel_reaches_near_endpoint_window(self) -> None:
|
|
config = function_config(
|
|
operation("seat.set_cushion_tilt", {"position_pct": 0})
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(
|
|
self.matrix,
|
|
absolute_stall_motors={1},
|
|
absolute_stall_hall_after={1: 4},
|
|
)
|
|
bus.hall[1] = 374
|
|
bus.position[1] = 10000
|
|
|
|
report = HardwareSeatExecutor(
|
|
self.matrix,
|
|
lambda: bus,
|
|
ack_timeout_ms=20,
|
|
motor_max_hall={0: 480, 1: 374, 2: 917, 3: 647},
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "succeeded")
|
|
sent_commands = [
|
|
self.matrix.decode("T2G_Seat_Move_Command_300", frame.data)
|
|
for frame in bus.sent
|
|
if frame.arbitration_id == 0x300
|
|
]
|
|
self.assertEqual([item["Opcode"] for item in sent_commands], [9, 0x0B])
|
|
recovery = report["spec"]["results"][0]["telemetry"]["position_recovery"]
|
|
self.assertEqual(recovery["hall_before"], 374)
|
|
self.assertEqual(recovery["hall_after"], 0)
|
|
|
|
def test_zero_recovers_real_m0_stall_at_calibrated_near_endpoint(self) -> None:
|
|
config = function_config(
|
|
operation("seat.set_longitudinal_position", {"position_pct": 0})
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(
|
|
self.matrix,
|
|
absolute_stall_motors={0},
|
|
absolute_stall_hall_after={0: 52},
|
|
)
|
|
bus.hall[0] = 480
|
|
bus.position[0] = 10_000
|
|
|
|
report = HardwareSeatExecutor(
|
|
self.matrix,
|
|
lambda: bus,
|
|
ack_timeout_ms=20,
|
|
motor_max_hall={0: 480, 1: 374, 2: 917, 3: 647},
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "succeeded")
|
|
sent_commands = [
|
|
self.matrix.decode("T2G_Seat_Move_Command_300", frame.data)
|
|
for frame in bus.sent
|
|
if frame.arbitration_id == 0x300
|
|
]
|
|
self.assertEqual([item["Opcode"] for item in sent_commands], [9, 0x0B])
|
|
recovery = report["spec"]["results"][0]["telemetry"]["position_recovery"]
|
|
self.assertEqual(recovery["hall_before"], 480)
|
|
self.assertEqual(recovery["hall_after"], 0)
|
|
self.assertEqual(
|
|
report["spec"]["results"][0]["telemetry"]["effective_position_pct"],
|
|
0,
|
|
)
|
|
|
|
def test_explicit_zero_homes_even_when_the_stored_position_is_midrange(self) -> None:
|
|
config = function_config(
|
|
operation("seat.set_longitudinal_position", {"position_pct": 0})
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(self.matrix, absolute_stall_motors={0})
|
|
bus.hall[0] = 116
|
|
bus.position[0] = 2417
|
|
|
|
report = HardwareSeatExecutor(
|
|
self.matrix,
|
|
lambda: bus,
|
|
ack_timeout_ms=20,
|
|
motor_max_hall={0: 480, 1: 374, 2: 917, 3: 647},
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "succeeded")
|
|
sent_commands = [
|
|
self.matrix.decode("T2G_Seat_Move_Command_300", frame.data)
|
|
for frame in bus.sent
|
|
if frame.arbitration_id == 0x300
|
|
]
|
|
self.assertEqual([item["Opcode"] for item in sent_commands], [9, 0x0B])
|
|
recovery = report["spec"]["results"][0]["telemetry"]["position_recovery"]
|
|
self.assertEqual(recovery["hall_before"], 116)
|
|
self.assertEqual(recovery["hall_after"], 0)
|
|
|
|
def test_zero_recovers_near_endpoint_timeout_after_full_travel(self) -> None:
|
|
config = function_config(
|
|
operation("seat.set_backrest_recline", {"position_pct": 0})
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(
|
|
self.matrix,
|
|
absolute_timeout_motors={2},
|
|
absolute_stall_hall_after={2: 3},
|
|
)
|
|
bus.hall[2] = 605
|
|
bus.position[2] = 4485
|
|
|
|
report = HardwareSeatExecutor(
|
|
self.matrix,
|
|
lambda: bus,
|
|
ack_timeout_ms=20,
|
|
motor_max_hall={0: 480, 1: 374, 2: 917, 3: 647},
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "succeeded")
|
|
recovery = report["spec"]["results"][0]["telemetry"]["position_recovery"]
|
|
self.assertEqual(recovery["hall_before"], 605)
|
|
self.assertEqual(recovery["hall_after"], 0)
|
|
|
|
def test_far_endpoint_timeout_within_position_tolerance_is_accepted(self) -> None:
|
|
config = function_config(
|
|
operation("seat.set_legrest_height", {"position_pct": 100})
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(
|
|
self.matrix,
|
|
absolute_timeout_motors={3},
|
|
absolute_stall_hall_after={3: 646},
|
|
absolute_failure_targets={3: 10_000},
|
|
)
|
|
bus.position[3] = 9_990
|
|
|
|
report = HardwareSeatExecutor(
|
|
self.matrix,
|
|
lambda: bus,
|
|
ack_timeout_ms=20,
|
|
axis_maximum_x100={"seat.set_legrest_height": 10_000},
|
|
motor_max_hall={0: 480, 1: 374, 2: 917, 3: 647},
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "succeeded")
|
|
telemetry = report["spec"]["results"][0]["telemetry"]
|
|
self.assertEqual(
|
|
telemetry["terminal_tolerance_acceptance"]["mode"],
|
|
"verified_far_endpoint_within_target_tolerance",
|
|
)
|
|
self.assertEqual(telemetry["effective_position_pct"], 100)
|
|
|
|
def test_midrange_timeout_within_tolerance_still_fails(self) -> None:
|
|
config = function_config(
|
|
operation("seat.set_legrest_height", {"position_pct": 50})
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(
|
|
self.matrix,
|
|
absolute_timeout_motors={3},
|
|
absolute_stall_hall_after={3: 323},
|
|
absolute_failure_targets={3: 5_000},
|
|
)
|
|
bus.position[3] = 4_990
|
|
|
|
report = HardwareSeatExecutor(
|
|
self.matrix,
|
|
lambda: bus,
|
|
ack_timeout_ms=20,
|
|
axis_maximum_x100={"seat.set_legrest_height": 10_000},
|
|
motor_max_hall={0: 480, 1: 374, 2: 917, 3: 647},
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "failed")
|
|
self.assertIn("TIMEOUT", report["spec"]["results"][-1]["message"])
|
|
|
|
def test_executor_blocks_absolute_motion_when_board_position_is_invalid(self) -> None:
|
|
config = function_config(
|
|
operation("seat.set_backrest_recline", {"position_pct": 50})
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(self.matrix, position_valid=False)
|
|
|
|
report = HardwareSeatExecutor(
|
|
self.matrix, lambda: bus, ack_timeout_ms=20
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "failed")
|
|
self.assertEqual([item.data[0] for item in bus.sent], [9, 9])
|
|
failure = report["spec"]["results"][-1]
|
|
self.assertIn("Position_Valid=0", failure["message"])
|
|
self.assertTrue(failure["telemetry"]["known_range_home_required"])
|
|
|
|
def test_large_relative_request_is_blocked_without_sending_absolute_opcode(self) -> None:
|
|
config = function_config(
|
|
operation(
|
|
"seat.set_backrest_recline",
|
|
{"position_pct": 64, "angle_deg": 124},
|
|
motion={
|
|
"mode": "relative",
|
|
"direction": "increase",
|
|
"start_position_pct": 20,
|
|
"delta_position_pct": 44,
|
|
"start_angle_deg": 104,
|
|
"delta_angle_deg": 20,
|
|
},
|
|
)
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(self.matrix, position_valid=False)
|
|
|
|
report = HardwareSeatExecutor(
|
|
self.matrix, lambda: bus, ack_timeout_ms=20
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "failed")
|
|
self.assertEqual([item.data[0] for item in bus.sent], [9, 9])
|
|
self.assertNotIn(1, [item.data[0] for item in bus.sent])
|
|
self.assertTrue(
|
|
report["spec"]["results"][-1]["telemetry"]["known_range_home_required"]
|
|
)
|
|
|
|
def test_executor_can_home_invalid_axis_when_separately_enabled(self) -> None:
|
|
config = function_config(
|
|
operation("seat.set_backrest_recline", {"position_pct": 50})
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(self.matrix, position_valid=False)
|
|
|
|
report = HardwareSeatExecutor(
|
|
self.matrix,
|
|
lambda: bus,
|
|
ack_timeout_ms=20,
|
|
motor_max_hall={0: 480, 1: 374, 2: 917, 3: 647},
|
|
auto_home=True,
|
|
home_timeout_s=10,
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "succeeded")
|
|
self.assertEqual([item.data[0] for item in bus.sent], [9, 0x0B, 1])
|
|
telemetry = report["spec"]["results"][0]["telemetry"]
|
|
self.assertEqual(telemetry["position_recovery"]["mode"], "home_known_range")
|
|
self.assertEqual(telemetry["position_recovery"]["expected_max_hall"], 917)
|
|
|
|
def test_executor_accepts_hall_confirmed_timed_motion_without_position(self) -> None:
|
|
config = function_config(
|
|
operation(
|
|
"seat.set_backrest_recline",
|
|
{"position_pct": 31, "angle_deg": 109},
|
|
motion={
|
|
"mode": "relative",
|
|
"direction": "increase",
|
|
"start_position_pct": 20,
|
|
"delta_position_pct": 11,
|
|
"start_angle_deg": 104,
|
|
"delta_angle_deg": 5,
|
|
},
|
|
)
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(self.matrix, position_valid=False)
|
|
|
|
report = HardwareSeatExecutor(
|
|
self.matrix, lambda: bus, ack_timeout_ms=20
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "succeeded")
|
|
self.assertEqual([item.data[0] for item in bus.sent], [9, 3])
|
|
telemetry = report["spec"]["results"][0]["telemetry"]
|
|
self.assertEqual(telemetry["motion_mode"], "relative_timed_duty")
|
|
self.assertFalse(telemetry["position_valid"])
|
|
self.assertGreater(telemetry["hall_delta"], 0)
|
|
self.assertTrue(telemetry["hall_direction_verified"])
|
|
|
|
def test_executor_accepts_user_timed_motion_that_reaches_endpoint_early(self) -> None:
|
|
config = function_config(
|
|
operation(
|
|
"seat.set_backrest_recline",
|
|
{"position_pct": 100, "angle_deg": 163},
|
|
motion={
|
|
"mode": "timed",
|
|
"direction": "increase",
|
|
"start_position_pct": 0,
|
|
"delta_position_pct": 100,
|
|
"requested_duration_ms": 70_000,
|
|
"applied_duration_ms": 30_000,
|
|
"max_duration_ms": 30_000,
|
|
"min_duration_ms": 300,
|
|
"duration_clamped": True,
|
|
"endpoint_expected": "far",
|
|
},
|
|
)
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(self.matrix, timed_endpoint="far")
|
|
|
|
report = HardwareSeatExecutor(
|
|
self.matrix,
|
|
lambda: bus,
|
|
ack_timeout_ms=20,
|
|
axis_maximum_x100={"seat.set_backrest_recline": 10_000},
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "succeeded")
|
|
self.assertEqual([item.data[0] for item in bus.sent], [9, 3])
|
|
telemetry = report["spec"]["results"][0]["telemetry"]
|
|
self.assertEqual(telemetry["motion_mode"], "timed_user_request")
|
|
self.assertEqual(telemetry["endpoint_reached"], "far")
|
|
self.assertTrue(telemetry["endpoint_terminal_acceptance"])
|
|
self.assertEqual(telemetry["effective_position_pct"], 100)
|
|
self.assertIn("最大允许时长为30秒", telemetry["user_notice"])
|
|
self.assertIn("到达最后方", telemetry["user_notice"])
|
|
|
|
def test_executor_does_not_hide_non_endpoint_user_timed_stall(self) -> None:
|
|
config = function_config(
|
|
operation(
|
|
"seat.set_backrest_recline",
|
|
{"position_pct": 100, "angle_deg": 163},
|
|
motion={
|
|
"mode": "timed",
|
|
"direction": "increase",
|
|
"start_position_pct": 0,
|
|
"delta_position_pct": 100,
|
|
"requested_duration_ms": 2_000,
|
|
"applied_duration_ms": 2_000,
|
|
"max_duration_ms": 30_000,
|
|
"min_duration_ms": 300,
|
|
"duration_clamped": False,
|
|
},
|
|
)
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(
|
|
self.matrix,
|
|
timed_failure_result=5,
|
|
timed_failure_event=4,
|
|
)
|
|
|
|
report = HardwareSeatExecutor(
|
|
self.matrix,
|
|
lambda: bus,
|
|
ack_timeout_ms=20,
|
|
axis_maximum_x100={"seat.set_backrest_recline": 10_000},
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "failed")
|
|
self.assertEqual(
|
|
report["spec"]["results"][-1]["telemetry"]["failure_category"],
|
|
"stall_no_hall_activity",
|
|
)
|
|
|
|
def test_executor_uses_closed_loop_target_for_calibrated_relative_motion(self) -> None:
|
|
config = function_config(
|
|
operation(
|
|
"seat.set_backrest_recline",
|
|
{"position_pct": 31, "angle_deg": 109},
|
|
motion={
|
|
"mode": "relative",
|
|
"direction": "increase",
|
|
"start_position_pct": 20,
|
|
"delta_position_pct": 11,
|
|
"start_angle_deg": 104,
|
|
"delta_angle_deg": 5,
|
|
},
|
|
)
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(self.matrix, position_valid=True)
|
|
|
|
report = HardwareSeatExecutor(
|
|
self.matrix,
|
|
lambda: bus,
|
|
ack_timeout_ms=20,
|
|
axis_maximum_x100={"seat.set_backrest_recline": 10_000},
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "succeeded")
|
|
self.assertEqual([item.data[0] for item in bus.sent], [9, 1])
|
|
telemetry = report["spec"]["results"][0]["telemetry"]
|
|
self.assertEqual(telemetry["motion_mode"], "relative_absolute_target")
|
|
self.assertEqual(telemetry["effective_position_pct"], 31)
|
|
|
|
def test_executor_reports_stall_with_post_failure_hall_diagnostics(self) -> None:
|
|
config = function_config(
|
|
operation(
|
|
"seat.set_backrest_recline",
|
|
{"position_pct": 31, "angle_deg": 109},
|
|
motion={
|
|
"mode": "relative",
|
|
"direction": "increase",
|
|
"start_position_pct": 20,
|
|
"delta_position_pct": 11,
|
|
"start_angle_deg": 104,
|
|
"delta_angle_deg": 5,
|
|
},
|
|
)
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(
|
|
self.matrix,
|
|
position_valid=False,
|
|
timed_failure_result=5,
|
|
timed_failure_event=4,
|
|
)
|
|
|
|
report = HardwareSeatExecutor(
|
|
self.matrix, lambda: bus, ack_timeout_ms=20
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "failed")
|
|
self.assertEqual([item.data[0] for item in bus.sent], [9, 3, 9])
|
|
failure = report["spec"]["results"][-1]
|
|
self.assertIn("STALL", failure["message"])
|
|
self.assertEqual(failure["telemetry"]["motor_result_name"], "STALL")
|
|
self.assertEqual(failure["telemetry"]["last_motor_event_name"], "STALL")
|
|
self.assertEqual(
|
|
failure["telemetry"]["failure_category"], "stall_no_hall_activity"
|
|
)
|
|
self.assertEqual(failure["telemetry"]["hall_before"], 0)
|
|
self.assertEqual(failure["telemetry"]["hall_after"], 0)
|
|
self.assertEqual(
|
|
failure["telemetry"]["diagnostic_frame"]["frame_id"], "0x303"
|
|
)
|
|
|
|
def test_executor_sends_stop_all_after_motion_feedback_timeout(self) -> None:
|
|
config = function_config(
|
|
operation("seat.set_cushion_tilt", {"position_pct": 50})
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(self.matrix, complete_motion=False)
|
|
report = HardwareSeatExecutor(
|
|
self.matrix, lambda: bus, ack_timeout_ms=20
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "succeeded")
|
|
self.assertEqual(
|
|
report["spec"]["verification_status"],
|
|
"submitted_unverified",
|
|
)
|
|
self.assertEqual([item.data[0] for item in bus.sent], [9, 1, 9])
|
|
telemetry = report["spec"]["results"][-1]["telemetry"]
|
|
self.assertTrue(telemetry["stop_all_attempted"])
|
|
self.assertEqual(telemetry["delivery_status"], "submitted")
|
|
self.assertEqual(telemetry["command_frame"]["frame_id"], "0x300")
|
|
self.assertEqual(telemetry["verification_status"], "submitted_unverified")
|
|
self.assertFalse(telemetry["physical_effect_confirmed"])
|
|
|
|
def test_executor_waits_for_real_thermal_status(self) -> None:
|
|
config = function_config(
|
|
operation("seat.set_ventilation", {"state": "on", "level": 2})
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
bus = ReactiveFakeBus(self.matrix)
|
|
report = HardwareSeatExecutor(
|
|
self.matrix, lambda: bus, thermal_timeout_ms=20
|
|
).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "succeeded")
|
|
telemetry = report["spec"]["results"][0]["telemetry"]
|
|
self.assertEqual(telemetry["feedback_frame"]["frame_id"], "0x4FA")
|
|
self.assertEqual(telemetry["feedback_value"], 3)
|
|
self.assertEqual(telemetry["verification_status"], "verified_success")
|
|
|
|
def test_executor_preserves_can_adapter_open_failure(self) -> None:
|
|
config = function_config(
|
|
operation("seat.set_heating", {"state": "on", "level": 1})
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
|
|
def unavailable_bus():
|
|
raise CanGatewayError(
|
|
"ZCAN_OpenDevice failed; connect the adapter and close ZCANPRO first"
|
|
)
|
|
|
|
report = HardwareSeatExecutor(self.matrix, unavailable_bus).execute(plan, config)
|
|
|
|
self.assertEqual(report["spec"]["status"], "failed")
|
|
failure = report["spec"]["results"][-1]
|
|
self.assertEqual(failure["telemetry"]["error_type"], "CanGatewayError")
|
|
self.assertEqual(
|
|
failure["telemetry"]["verification_status"], "failed_or_blocked"
|
|
)
|
|
self.assertIn("ZCAN_OpenDevice failed", failure["message"])
|
|
self.assertFalse(failure["telemetry"]["stop_all_attempted"])
|
|
|
|
def test_settings_allow_only_front_passenger_target(self) -> None:
|
|
environment = {
|
|
"COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "true",
|
|
"COGNITIVE_AGENT_HARDWARE_APPROVAL_TOKEN_REQUIRED": "true",
|
|
"COGNITIVE_AGENT_HARDWARE_APPROVAL_TOKEN": "bench-secret",
|
|
"COGNITIVE_AGENT_SEAT_FIRMWARE_PROFILES": "combined",
|
|
"COGNITIVE_AGENT_SEAT_CAN_BACKEND": "python-can",
|
|
"COGNITIVE_AGENT_SEAT_CAN_BITRATE": "500000",
|
|
"COGNITIVE_AGENT_SEAT_CAN_INTERFACE": "socketcan",
|
|
"COGNITIVE_AGENT_SEAT_CAN_CHANNEL": "0",
|
|
"COGNITIVE_AGENT_SEAT_AUTO_FLASH_ENABLED": "false",
|
|
}
|
|
with patch.dict(os.environ, environment, clear=False):
|
|
settings = SeatHardwareSettings.from_env(ROOT)
|
|
settings = replace(
|
|
settings, firmware_images={"combined": ROOT / "pyproject.toml"}
|
|
)
|
|
|
|
settings.authorize("bench-secret")
|
|
settings.validate_function_config(
|
|
function_config(
|
|
operation(
|
|
"seat.set_heating",
|
|
{"state": "on", "level": 1},
|
|
instance="passenger",
|
|
zone="front_right",
|
|
)
|
|
)
|
|
)
|
|
with self.assertRaises(HardwareSafetyError):
|
|
settings.validate_function_config(
|
|
function_config(
|
|
operation(
|
|
"seat.set_heating",
|
|
{"state": "on", "level": 1},
|
|
instance="rear_left",
|
|
zone="rear_left",
|
|
)
|
|
)
|
|
)
|
|
settings.validate_function_config(
|
|
function_config(
|
|
operation("seat.set_longitudinal_position", {"position_pct": 40})
|
|
)
|
|
)
|
|
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(
|
|
operation("seat.set_legrest_extension", {"position_pct": 40})
|
|
)
|
|
)
|
|
with self.assertRaises(HardwareSafetyError):
|
|
settings.validate_function_config(
|
|
function_config(
|
|
operation(
|
|
"seat.set_heating",
|
|
{"state": "on", "level": 1},
|
|
instance="driver",
|
|
zone="front_left",
|
|
)
|
|
)
|
|
)
|
|
with self.assertRaises(HardwareSafetyError):
|
|
settings.authorize("wrong-secret")
|
|
|
|
def test_settings_enable_auto_flash_by_default_only_for_hardware_mode(self) -> None:
|
|
hardware_environment = {
|
|
"COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "true",
|
|
"COGNITIVE_AGENT_HARDWARE_APPROVAL_TOKEN": "bench-secret",
|
|
"COGNITIVE_AGENT_SEAT_FIRMWARE_PROFILES": "combined",
|
|
"COGNITIVE_AGENT_SEAT_CAN_BACKEND": "python-can",
|
|
"COGNITIVE_AGENT_SEAT_CAN_INTERFACE": "socketcan",
|
|
"COGNITIVE_AGENT_SEAT_CAN_CHANNEL": "0",
|
|
}
|
|
with patch.dict(os.environ, hardware_environment, clear=True):
|
|
automatic = SeatHardwareSettings.from_env(ROOT)
|
|
self.assertTrue(automatic.enabled)
|
|
self.assertTrue(automatic.auto_flash_enabled)
|
|
self.assertTrue(automatic.position_sync_enabled)
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{
|
|
**hardware_environment,
|
|
"COGNITIVE_AGENT_SEAT_AUTO_FLASH_ENABLED": "false",
|
|
},
|
|
clear=True,
|
|
):
|
|
opted_out = SeatHardwareSettings.from_env(ROOT)
|
|
self.assertFalse(opted_out.auto_flash_enabled)
|
|
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
dry_run = SeatHardwareSettings.from_env(ROOT)
|
|
self.assertFalse(dry_run.enabled)
|
|
self.assertFalse(dry_run.auto_flash_enabled)
|
|
self.assertFalse(dry_run.position_sync_enabled)
|
|
|
|
def test_settings_allow_explicit_local_debug_mode_without_token(self) -> None:
|
|
environment = {
|
|
"COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "true",
|
|
"COGNITIVE_AGENT_HARDWARE_APPROVAL_TOKEN_REQUIRED": "false",
|
|
"COGNITIVE_AGENT_SEAT_FIRMWARE_PROFILES": "combined",
|
|
"COGNITIVE_AGENT_SEAT_CAN_BACKEND": "python-can",
|
|
"COGNITIVE_AGENT_SEAT_CAN_INTERFACE": "socketcan",
|
|
"COGNITIVE_AGENT_SEAT_CAN_CHANNEL": "0",
|
|
"COGNITIVE_AGENT_SEAT_AUTO_FLASH_ENABLED": "false",
|
|
}
|
|
with patch.dict(os.environ, environment, clear=True):
|
|
settings = SeatHardwareSettings.from_env(ROOT)
|
|
|
|
self.assertFalse(settings.approval_required)
|
|
self.assertTrue(settings.configured)
|
|
settings.authorize(None)
|
|
public = settings.public_info()
|
|
self.assertFalse(public["approval_required"])
|
|
self.assertFalse(public["approval_configured"])
|
|
|
|
def test_settings_reject_driver_target(self) -> None:
|
|
environment = {
|
|
"COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "true",
|
|
"COGNITIVE_AGENT_HARDWARE_APPROVAL_TOKEN": "bench-secret",
|
|
"COGNITIVE_AGENT_SEAT_FIRMWARE_PROFILES": "combined",
|
|
}
|
|
with patch.dict(os.environ, environment, clear=False):
|
|
settings = SeatHardwareSettings.from_env(ROOT)
|
|
with self.assertRaises(HardwareSafetyError):
|
|
settings.validate_function_config(
|
|
function_config(
|
|
operation(
|
|
"seat.set_heating",
|
|
{"state": "on", "level": 1},
|
|
operation_id="op-driver",
|
|
instance="driver",
|
|
zone="front_left",
|
|
),
|
|
)
|
|
)
|
|
|
|
def test_firmware_guard_reuses_bus_when_protocol_signature_matches(self) -> None:
|
|
bus = ProtocolSignatureFakeBus(True)
|
|
guard = T2GFirmwareGuard(
|
|
lambda: bus,
|
|
None,
|
|
auto_flash=False,
|
|
probe_timeout_ms=20,
|
|
)
|
|
|
|
prepared, telemetry = guard.prepare_bus()
|
|
|
|
self.assertIs(prepared, bus)
|
|
self.assertFalse(telemetry["programmed_this_request"])
|
|
self.assertEqual(telemetry["firmware_revision"], 15)
|
|
self.assertEqual(telemetry["firmware_build_id"], 16)
|
|
self.assertEqual(bus.sent, [])
|
|
self.assertEqual(telemetry["max_hall_counts"], [480, 374, 917, 647])
|
|
self.assertFalse(bus.closed)
|
|
|
|
def test_firmware_guard_programs_then_reverifies(self) -> None:
|
|
before = ProtocolSignatureFakeBus(False)
|
|
after = ProtocolSignatureFakeBus(True)
|
|
buses = deque([before, after])
|
|
programmer = FakeProgrammer()
|
|
guard = T2GFirmwareGuard(
|
|
buses.popleft,
|
|
programmer, # type: ignore[arg-type]
|
|
auto_flash=True,
|
|
probe_timeout_ms=20,
|
|
boot_wait_ms=0,
|
|
)
|
|
|
|
prepared, telemetry = guard.prepare_bus()
|
|
|
|
self.assertIs(prepared, after)
|
|
self.assertEqual(programmer.calls, 1)
|
|
self.assertTrue(before.closed)
|
|
self.assertTrue(telemetry["programmed_this_request"])
|
|
|
|
def test_firmware_guard_programs_when_unflashed_target_does_not_ack(self) -> None:
|
|
before = FailingProtocolSignatureBus(False)
|
|
after = ProtocolSignatureFakeBus(True)
|
|
buses = deque([before, after])
|
|
programmer = FakeProgrammer()
|
|
guard = T2GFirmwareGuard(
|
|
buses.popleft,
|
|
programmer, # type: ignore[arg-type]
|
|
auto_flash=True,
|
|
probe_timeout_ms=20,
|
|
boot_wait_ms=0,
|
|
)
|
|
|
|
prepared, telemetry = guard.prepare_bus()
|
|
|
|
self.assertIs(prepared, after)
|
|
self.assertEqual(programmer.calls, 1)
|
|
self.assertTrue(before.closed)
|
|
self.assertTrue(telemetry["programmed_this_request"])
|
|
|
|
def test_firmware_guard_fails_closed_when_auto_flash_is_disabled(self) -> None:
|
|
bus = ProtocolSignatureFakeBus(False)
|
|
guard = T2GFirmwareGuard(
|
|
lambda: bus,
|
|
None,
|
|
auto_flash=False,
|
|
probe_timeout_ms=20,
|
|
)
|
|
|
|
with self.assertRaises(FirmwareProgrammingError):
|
|
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 15)", message)
|
|
self.assertIn("missing_frame_ids=0x306,0x307", 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 = (
|
|
firmware_root / "IAR" / "rev_d" / "Exe" / "Seat_Controller_demo.out"
|
|
)
|
|
program_macro = firmware_root / "IAR" / "Seat_Controller_program.mac"
|
|
programmer = T2GSeatProgrammer(
|
|
cspybat=ROOT / "fake-iar" / "common" / "bin" / "cspybat.exe",
|
|
debug_file=output_image,
|
|
hex_image=(
|
|
firmware_root / "IAR" / "rev_d" / "Exe" / "Seat_Controller_demo.hex"
|
|
),
|
|
macro_file=program_macro,
|
|
)
|
|
|
|
# Dependency discovery is covered by settings/configuration tests. This
|
|
# test only validates the command and never starts C-SPY or a probe.
|
|
with patch.object(T2GSeatProgrammer, "required_files", return_value=()):
|
|
command = programmer.command()
|
|
|
|
self.assertEqual(command[3], str(output_image))
|
|
self.assertIn(f"--macro={program_macro}", command)
|
|
self.assertNotIn("--download_only", command)
|
|
self.assertIn("--leave_target_running", command)
|
|
timeout_index = command.index("--timeout")
|
|
self.assertEqual(command[timeout_index + 1], "3000")
|
|
self.assertIn("--jet_probe=cmsisdap", command)
|
|
self.assertNotIn("Seat_Controller_demo.mac", " ".join(command))
|
|
|
|
def test_hardware_validation_replaces_reference_dry_run_warnings(self) -> None:
|
|
validation = [
|
|
{"severity": "warning", "code": code, "message": "dry-run"}
|
|
for code in (
|
|
"LEVEL_COMMAND_SIMULATED",
|
|
"SYNTHETIC_MOTION_CAN",
|
|
"DRIVER_MOTION_INTERLOCK_NOT_IMPLEMENTED",
|
|
"COMBINED_THERMAL_SIMULATED",
|
|
"VIRTUAL_EXECUTION_ONLY",
|
|
)
|
|
]
|
|
|
|
SeatDemoService._adapt_validation_for_hardware(validation)
|
|
|
|
codes = {issue["code"] for issue in validation}
|
|
self.assertEqual(
|
|
codes,
|
|
{
|
|
"LEVEL_COMMAND_T2G_MAPPED",
|
|
"T2G_MOTION_PROTOCOL_MAPPED",
|
|
"BENCH_ONLY_NO_VEHICLE_INTERLOCKS",
|
|
"COMBINED_THERMAL_T2G_MAPPED",
|
|
"REFERENCE_MATRIX_OVERRIDDEN_BY_T2G_BENCH",
|
|
},
|
|
)
|
|
messages = " ".join(issue["message"] for issue in validation)
|
|
self.assertNotIn("禁止实车下发", messages)
|
|
self.assertNotIn("只能虚拟执行", messages)
|
|
|
|
def test_physical_feedback_updates_only_front_passenger(self) -> None:
|
|
actions = [
|
|
{
|
|
"target": "passenger",
|
|
"function": "ventilation",
|
|
"state": "on",
|
|
"requested_level": 2,
|
|
}
|
|
]
|
|
config = function_config(
|
|
operation(
|
|
"seat.set_ventilation",
|
|
{"state": "on", "level": 2},
|
|
instance="passenger",
|
|
zone="front_right",
|
|
)
|
|
)
|
|
report = {
|
|
"spec": {
|
|
"results": [
|
|
{
|
|
"ref": "op-hardware-test",
|
|
"status": "succeeded",
|
|
"telemetry": {"feedback_value": 3},
|
|
}
|
|
]
|
|
}
|
|
}
|
|
|
|
projected = SeatDemoService._hardware_feedback_actions(
|
|
actions, config, report
|
|
)
|
|
|
|
self.assertEqual({item["target"] for item in projected}, {"passenger"})
|
|
self.assertTrue(
|
|
all(item["requested_level"] == 2 for item in projected)
|
|
)
|
|
|
|
def test_complete_physical_snapshot_projects_unrequested_passenger_axes(self) -> None:
|
|
actions = [
|
|
{
|
|
"domain": "seat",
|
|
"target": "passenger",
|
|
"action_type": "position",
|
|
"axis": "longitudinal",
|
|
"position_pct": 25,
|
|
}
|
|
]
|
|
config = function_config(
|
|
operation(
|
|
"seat.set_longitudinal_position",
|
|
{"position_pct": 25},
|
|
)
|
|
)
|
|
report = {
|
|
"spec": {
|
|
"results": [
|
|
{
|
|
"ref": "op-hardware-test",
|
|
"status": "succeeded",
|
|
"telemetry": {"effective_position_pct": 25},
|
|
}
|
|
],
|
|
"physical_state": {
|
|
"position": {
|
|
axis: {
|
|
"position_pct": 0,
|
|
"position_valid": True,
|
|
}
|
|
for axis in (
|
|
"longitudinal",
|
|
"cushion_tilt",
|
|
"backrest_recline",
|
|
"legrest_height",
|
|
)
|
|
},
|
|
"ventilation": {"state": "off", "level": 0},
|
|
"heating": {"state": "off", "level": 0},
|
|
},
|
|
}
|
|
}
|
|
|
|
projected = SeatDemoService._hardware_feedback_actions(
|
|
actions, config, report
|
|
)
|
|
|
|
positions = {
|
|
item["axis"]: item["position_pct"]
|
|
for item in projected
|
|
if item.get("action_type") == "position"
|
|
}
|
|
self.assertEqual(
|
|
positions,
|
|
{
|
|
"longitudinal": 0,
|
|
"cushion_tilt": 0,
|
|
"backrest_recline": 0,
|
|
"legrest_height": 0,
|
|
},
|
|
)
|
|
self.assertEqual({item["target"] for item in projected}, {"passenger"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|