orin
841 lines
32 KiB
Python
Executable File
841 lines
32 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 # noqa: E402
|
|
from cognitive_agent.application.seat_service import SeatDemoService # noqa: E402
|
|
from cognitive_agent.infrastructure.can.hardware import ( # noqa: E402
|
|
HardwareSafetyError,
|
|
SeatHardwareSettings,
|
|
)
|
|
from cognitive_agent.infrastructure.firmware import ( # noqa: E402
|
|
EXPECTED_IDENTITY_PAYLOAD,
|
|
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 = "driver",
|
|
zone: str = "front_left",
|
|
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,
|
|
timed_failure_result: int | None = None,
|
|
timed_failure_event: int = 4,
|
|
):
|
|
self.matrix = matrix
|
|
self.complete_motion = complete_motion
|
|
self.position_valid = position_valid
|
|
self.timed_failure_result = timed_failure_result
|
|
self.timed_failure_event = timed_failure_event
|
|
self.hall = [0, 0, 0, 0]
|
|
self.max_hall = [480, 790, 1590, 960]
|
|
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"]
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_System_Status_301",
|
|
{
|
|
"Ack": 1,
|
|
"Reject_Reason": 0,
|
|
"Last_Opcode": opcode,
|
|
"Last_Sequence": sequence,
|
|
"Safety_Inhibit": 0,
|
|
"Safety_Fault": 0,
|
|
},
|
|
)
|
|
)
|
|
if opcode == 9:
|
|
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],
|
|
"Max_Speed_Pps_X100": 9000,
|
|
},
|
|
)
|
|
)
|
|
if opcode == 1 and self.complete_motion:
|
|
self.received.append(
|
|
self._frame(
|
|
"T2G_Seat_Move_Motor_Status_302",
|
|
{
|
|
"Motor_Sub_ID": command["Motor_Selector"],
|
|
"Task_State": 8,
|
|
"Result": 2,
|
|
"Position_Valid": 1,
|
|
"Motor_Sequence": sequence,
|
|
"Value_X100": command["Param_A"],
|
|
"Duty_Percent": command["Param_B"],
|
|
},
|
|
)
|
|
)
|
|
if 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": 9,
|
|
"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"],
|
|
},
|
|
)
|
|
)
|
|
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],
|
|
"Max_Speed_Pps_X100": 9000,
|
|
},
|
|
)
|
|
)
|
|
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": 8,
|
|
"Result": 2,
|
|
"Direction": command["Direction"],
|
|
"Position_Valid": int(self.position_valid),
|
|
"Motor_Sequence": sequence,
|
|
"Value_X100": 0,
|
|
"Duty_Percent": command["Param_B"],
|
|
},
|
|
)
|
|
)
|
|
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],
|
|
"Max_Speed_Pps_X100": 9000,
|
|
},
|
|
)
|
|
)
|
|
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 IdentityFakeBus:
|
|
def __init__(self, identity: bytes | None):
|
|
self.identity = identity
|
|
self.sent: list[CanFrame] = []
|
|
self.closed = False
|
|
|
|
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
|
|
if self.identity is None:
|
|
return None
|
|
identity = self.identity
|
|
self.identity = None
|
|
return CanFrame(0x305, identity)
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
class FailingIdentityBus(IdentityFakeBus):
|
|
def send(self, arbitration_id: int, data: bytes) -> CanFrame:
|
|
del arbitration_id, data
|
|
raise RuntimeError("CAN target did not ACK the identity probe")
|
|
|
|
|
|
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",
|
|
)
|
|
|
|
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 94 11 32 00")
|
|
|
|
def test_compiler_maps_small_relative_backrest_request_to_timed_duty(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"], 3)
|
|
self.assertEqual(command["signals"]["Direction"], 1)
|
|
self.assertEqual(command["signals"]["Param_A"], 5500)
|
|
self.assertEqual(command["signals"]["Param_B"], 50)
|
|
self.assertEqual(command["completion"]["signal"], "Task_State")
|
|
|
|
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"], 5760)
|
|
self.assertEqual(command["completion"]["signal"], "Value_X100")
|
|
|
|
def test_passenger_alias_emits_the_same_physical_payload(self) -> None:
|
|
driver = function_config(
|
|
operation("seat.set_heating", {"state": "on", "level": 2})
|
|
)
|
|
passenger = function_config(
|
|
operation(
|
|
"seat.set_heating",
|
|
{"state": "on", "level": 2},
|
|
instance="passenger",
|
|
zone="front_right",
|
|
)
|
|
)
|
|
|
|
driver_plan = self.compiler.compile(driver, {})
|
|
passenger_plan = self.compiler.compile(passenger, {})
|
|
self.assertEqual(
|
|
driver_plan["spec"]["commands"], passenger_plan["spec"]["commands"]
|
|
)
|
|
|
|
def test_driver_and_passenger_identical_motion_is_sent_once(self) -> None:
|
|
config = function_config(
|
|
operation(
|
|
"seat.set_backrest_recline",
|
|
{"position_pct": 60},
|
|
operation_id="op-driver",
|
|
),
|
|
operation(
|
|
"seat.set_backrest_recline",
|
|
{"position_pct": 60},
|
|
operation_id="op-passenger",
|
|
instance="passenger",
|
|
zone="front_right",
|
|
),
|
|
)
|
|
plan = self.compiler.compile(config, {})
|
|
self.assertEqual(len(plan["spec"]["commands"]), 1)
|
|
self.assertEqual(
|
|
plan["spec"]["commands"][0]["operation_ref"],
|
|
"op-driver,op-passenger",
|
|
)
|
|
|
|
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-driver", "op-passenger"})
|
|
|
|
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_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"]["calibration_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"]["calibration_required"]
|
|
)
|
|
|
|
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_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=4,
|
|
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_legrest_extension", {"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"], "failed")
|
|
self.assertEqual([item.data[0] for item in bus.sent], [9, 1, 9])
|
|
self.assertTrue(
|
|
report["spec"]["results"][-1]["telemetry"]["stop_all_attempted"]
|
|
)
|
|
|
|
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)
|
|
|
|
def test_settings_allow_shared_target_aliases_but_fail_closed_for_other_targets(self) -> None:
|
|
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_BITRATE": "500000",
|
|
"COGNITIVE_AGENT_SEAT_CAN_INTERFACE": "socketcan",
|
|
"COGNITIVE_AGENT_SEAT_CAN_CHANNEL": "0",
|
|
"COGNITIVE_AGENT_SEAT_AUTO_FLASH_ENABLED": "false",
|
|
"COGNITIVE_AGENT_ENABLE_UNVERIFIED_LONGITUDINAL": "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",
|
|
)
|
|
)
|
|
)
|
|
with self.assertRaises(HardwareSafetyError):
|
|
settings.validate_function_config(
|
|
function_config(
|
|
operation(
|
|
"seat.set_longitudinal_position", {"position_pct": 40}
|
|
)
|
|
)
|
|
)
|
|
with self.assertRaises(HardwareSafetyError):
|
|
settings.authorize("wrong-secret")
|
|
|
|
def test_settings_reject_conflicting_shared_alias_requests(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",
|
|
),
|
|
operation(
|
|
"seat.set_heating",
|
|
{"state": "on", "level": 3},
|
|
operation_id="op-passenger",
|
|
instance="passenger",
|
|
zone="front_right",
|
|
),
|
|
)
|
|
)
|
|
|
|
def test_firmware_guard_reuses_bus_when_identity_matches(self) -> None:
|
|
bus = IdentityFakeBus(EXPECTED_IDENTITY_PAYLOAD)
|
|
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(bus.sent[0].arbitration_id, 0x304)
|
|
self.assertFalse(bus.closed)
|
|
|
|
def test_firmware_guard_programs_then_reverifies(self) -> None:
|
|
before = IdentityFakeBus(None)
|
|
after = IdentityFakeBus(EXPECTED_IDENTITY_PAYLOAD)
|
|
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 = FailingIdentityBus(None)
|
|
after = IdentityFakeBus(EXPECTED_IDENTITY_PAYLOAD)
|
|
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 = IdentityFakeBus(None)
|
|
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_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_is_mirrored_to_both_logical_seats(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}, {"driver", "passenger"})
|
|
self.assertTrue(
|
|
all(item["requested_level"] == 2 for item in projected)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|