vee/Cognitive_Agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
4aded5c21dddd781cd67cba60d95ef0bf28a8a05
Cognitive_Agent/domains/seat/tests/unit/test_t2g_hardware.py
T

1504 lines
58 KiB
Python
Executable File
Raw Blame History

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.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,
):
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.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 == 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 cyclic status. Keep one more
# sample queued for the executor's post-home preflight read.
self.received.append(
self._frame("T2G_Seat_Move_Motor_Status_302", home_status)
)
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": 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 30")))
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.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 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 88 13 30 75")
self.assertEqual(command["signals"]["Param_B"], 30_000)
self.assertEqual(command["completion"]["timeout_ms"], 33_000)
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"], 1320)
self.assertEqual(command["signals"]["Param_B"], 900)
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"], 6400)
self.assertEqual(command["completion"]["signal"], "Value_X100")
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_passive_snapshot_reads_all_passenger_axes_and_thermal_state(self) -> None:
bus = ReactiveFakeBus(self.matrix)
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.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"]["Param_A"] == 0
for command in plan["spec"]["commands"]
)
)
bus = ReactiveFakeBus(self.matrix)
report = HardwareSeatExecutor(
self.matrix,
lambda: bus,
ack_timeout_ms=20,
stop_all_motor_selector=int(
self.compiler.protocol["stop_all_motor_selector"]
),
).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), (1, 0), (1, 1), (1, 2), (1, 3)],
)
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, 1, 3, 0x0B],
)
telemetry = report["spec"]["results"][0]["telemetry"]
recovery = telemetry["position_recovery"]
self.assertEqual(recovery["mode"], "verified_endpoint_drift_rehome")
self.assertGreater(recovery["away_probe"]["hall_delta"], 2)
self.assertEqual(recovery["home"]["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, 1, 3, 0x0B],
)
recovery = report["spec"]["results"][0]["telemetry"]["position_recovery"]
self.assertEqual(recovery["failed_absolute_hall_before"], 374)
self.assertEqual(recovery["failed_absolute_hall_after"], 4)
self.assertEqual(recovery["home"]["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, 1, 3, 0x0B],
)
recovery = report["spec"]["results"][0]["telemetry"]["position_recovery"]
self.assertEqual(recovery["failed_absolute_hall_before"], 480)
self.assertEqual(recovery["failed_absolute_hall_after"], 52)
self.assertEqual(recovery["endpoint_recovery_max_near_hall"], 72)
self.assertGreater(recovery["away_probe"]["hall_delta"], 2)
self.assertEqual(recovery["home"]["hall_after"], 0)
self.assertEqual(
report["spec"]["results"][0]["telemetry"]["effective_position_pct"],
0,
)
def test_zero_does_not_rehome_a_midrange_stall_as_an_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})
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"], "failed")
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, 1, 9])
self.assertIn("STALL", report["spec"]["results"][-1]["message"])
current = report["spec"]["results"][-1]["telemetry"][
"current_diagnostic"
]
self.assertEqual(current["sample_count"], 1)
self.assertEqual(current["max_current_delta_raw"], 700)
self.assertTrue(current["stall_overlap_seen"])
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["failed_absolute_hall_before"], 605)
self.assertEqual(recovery["failed_absolute_hall_after"], 3)
self.assertEqual(recovery["home"]["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_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"], "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_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.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}))
)
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)
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)
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"], 3)
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_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()
Reference in New Issue View Git Blame Copy Permalink