minghanchen
247 lines
10 KiB
Python
247 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
import unittest
|
|
|
|
from cognitive_agent.application.native_can_feedback import (
|
|
authoritative_physical_ch1_feedback_state,
|
|
decode_feedback,
|
|
feedback_matches,
|
|
merge_state,
|
|
)
|
|
from cognitive_agent.infrastructure.can.gateway import CanFrame
|
|
|
|
|
|
class NativeCanFeedbackTests(unittest.TestCase):
|
|
def frame(self, frame_id: int, data_hex: str) -> CanFrame:
|
|
return CanFrame(frame_id, bytes.fromhex(data_hex), channel=1, is_fd=True)
|
|
|
|
def test_passive_ch1_samples_decode_stationary_vehicle(self) -> None:
|
|
gear = decode_feedback(self.frame(0x342, "00001fb1c1fff778"))
|
|
speed = decode_feedback(self.frame(0x4EB, "0700f00001516766"))
|
|
|
|
self.assertEqual(gear["vehicle_safety"]["gear"], "P")
|
|
self.assertEqual(speed["vehicle_safety"]["speed_kmh"], 0)
|
|
self.assertEqual(gear["source"], "physical_can_ch1")
|
|
self.assertEqual(gear["channel"], 1)
|
|
|
|
def test_climate_feedback_is_matched_by_signal_value(self) -> None:
|
|
feedback = self.frame(0x404, "4369116c1818a2dd")
|
|
decoded = decode_feedback(feedback)["domain_state"]["climate"]
|
|
|
|
self.assertEqual(decoded["power"], "on")
|
|
self.assertEqual(decoded["zones"]["driver"]["temperature_c"], 24)
|
|
self.assertTrue(feedback_matches("climate_power:on", feedback))
|
|
self.assertTrue(feedback_matches("climate_temperature:driver:24", feedback))
|
|
self.assertFalse(feedback_matches("climate_temperature:driver:25", feedback))
|
|
|
|
def test_climate_feedback_decodes_front_defrost(self) -> None:
|
|
decoded = decode_feedback(
|
|
self.frame(0x404, "43A9116c1818a2dd")
|
|
)["domain_state"]["climate"]
|
|
|
|
self.assertEqual(decoded["front_defrost"], "on")
|
|
|
|
def test_window_position_match_uses_authoritative_0x388(self) -> None:
|
|
feedback = self.frame(0x388, "0000320000005500")
|
|
self.assertTrue(feedback_matches("window_position:front_left:50", feedback))
|
|
self.assertFalse(feedback_matches("window_position:front_left:80", feedback))
|
|
|
|
def test_individual_outlet_switch_matches_observed_4bb_signatures(self) -> None:
|
|
data = bytearray(64)
|
|
data[41:45] = bytes((0x26, 0x80, 0x80, 0x26))
|
|
data[45:49] = bytes((0x2D, 0x85, 0x9E, 0xD3))
|
|
feedback = CanFrame(0x4BB, bytes(data), channel=1, is_fd=True)
|
|
switches = decode_feedback(feedback)["domain_state"]["outlet"][
|
|
"switches"
|
|
]
|
|
|
|
self.assertEqual(switches["driver_left"]["state"], "close")
|
|
self.assertEqual(switches["driver_right"]["state"], "open")
|
|
self.assertEqual(switches["passenger_right"]["state"], "close")
|
|
self.assertTrue(
|
|
feedback_matches("outlet_switch:driver_left:close", feedback)
|
|
)
|
|
self.assertFalse(
|
|
feedback_matches("outlet_switch:driver_left:open", feedback)
|
|
)
|
|
|
|
def test_individual_outlet_switch_supports_direct_37b_feedback(self) -> None:
|
|
data = bytearray(64)
|
|
data[0] = 0x50
|
|
data[8] = 0b00001001 # driver-left open, driver-right close
|
|
data[14] = 0b10010000 # passenger-left open, passenger-right close
|
|
feedback = CanFrame(0x37B, bytes(data), channel=1, is_fd=True)
|
|
|
|
self.assertTrue(
|
|
feedback_matches("outlet_switch:driver_left:open", feedback)
|
|
)
|
|
self.assertTrue(
|
|
feedback_matches("outlet_switch:driver_right:close", feedback)
|
|
)
|
|
self.assertTrue(
|
|
feedback_matches("outlet_switch:passenger_left:open", feedback)
|
|
)
|
|
self.assertTrue(
|
|
feedback_matches("outlet_switch:passenger_right:close", feedback)
|
|
)
|
|
|
|
def test_light_feedback_requires_target_value_not_only_frame_id(self) -> None:
|
|
passive_feedback = self.frame(0x42E, "11140a03880200c6")
|
|
self.assertEqual(
|
|
decode_feedback(passive_feedback)["domain_state"]["interior_light"]["power"],
|
|
"unknown",
|
|
)
|
|
feedback = self.frame(0x42E, "11140a03200200c6")
|
|
decoded = decode_feedback(feedback)["domain_state"]
|
|
self.assertEqual(decoded["interior_light"]["power"], "off")
|
|
self.assertEqual(decoded["ambient_light"]["power"], "off")
|
|
self.assertTrue(feedback_matches("indoor_light:off", feedback))
|
|
self.assertFalse(feedback_matches("indoor_light:on", feedback))
|
|
self.assertTrue(feedback_matches("ambient_power:off", feedback))
|
|
|
|
def test_can1_matrix_rejects_reserved_ambient_color_values(self) -> None:
|
|
decoded = decode_feedback(
|
|
self.frame(0x42E, "11 64 32 03 90 02 00 C6")
|
|
)["domain_state"]["ambient_light"]
|
|
|
|
self.assertIsNone(decoded["front_color_index"])
|
|
self.assertIsNone(decoded["rear_color_index"])
|
|
|
|
def test_3ff_matrix_decodes_lighting_and_illuminance(self) -> None:
|
|
light = decode_feedback(
|
|
self.frame(0x3FF, "5B 09 FF 0F 32 61 04 12")
|
|
)["domain_state"]["interior_light"]
|
|
|
|
self.assertEqual(light["power"], "unknown")
|
|
self.assertEqual(light["footwell_power"], "off")
|
|
self.assertEqual(light["ambient_illuminance_lux"], 510)
|
|
|
|
def test_3f3_matrix_decodes_wall_and_ambient_switches(self) -> None:
|
|
decoded = decode_feedback(
|
|
self.frame(0x3F3, "00 00 00 00 00 00 00 84")
|
|
)["domain_state"]
|
|
|
|
self.assertEqual(decoded["interior_light"]["wall_light_power"], "on")
|
|
self.assertEqual(decoded["ambient_light"]["power"], "on")
|
|
|
|
def test_driver_seat_feedback_uses_0x4fa_not_0x42a(self) -> None:
|
|
passive_feedback = self.frame(0x4FA, "0980aa0000810000")
|
|
passive_driver = decode_feedback(passive_feedback)["actuator_state"]["driver"]
|
|
self.assertEqual(passive_driver["heating"], {"state": "off", "level": 0})
|
|
feedback = self.frame(0x4FA, "1180aa0000810000")
|
|
driver = decode_feedback(feedback)["actuator_state"]["driver"]
|
|
self.assertEqual(driver["ventilation"], {"state": "off", "level": 0})
|
|
self.assertEqual(driver["heating"], {"state": "on", "level": 1})
|
|
self.assertTrue(feedback_matches("driver_seat_thermal:heating:low", feedback))
|
|
self.assertFalse(feedback_matches("driver_seat_thermal:ventilation:high", feedback))
|
|
|
|
def test_driver_seat_feedback_supports_all_three_thermal_levels(self) -> None:
|
|
driver = decode_feedback(
|
|
self.frame(0x4FA, "1C00000000010000")
|
|
)["actuator_state"]["driver"]
|
|
|
|
self.assertEqual(driver["ventilation"], {"state": "on", "level": 3})
|
|
self.assertEqual(driver["heating"], {"state": "on", "level": 2})
|
|
|
|
def test_38a_matrix_decodes_front_lighting_feedback(self) -> None:
|
|
lights = decode_feedback(
|
|
self.frame(0x38A, "1969000000000000")
|
|
)["domain_state"]["front_light"]
|
|
|
|
self.assertEqual(lights["position_light"]["state"], "on")
|
|
self.assertEqual(lights["low_beam"]["state"], "off")
|
|
self.assertEqual(lights["high_beam"]["state"], "on")
|
|
self.assertEqual(lights["fog_light"]["state"], "on")
|
|
self.assertEqual(lights["brake_light"]["state"], "flashing")
|
|
self.assertEqual(lights["reverse_light"]["state"], "off")
|
|
self.assertEqual(lights["all_weather_light"]["state"], "on")
|
|
|
|
def test_driver_position_feedback_never_claims_absolute_position(self) -> None:
|
|
decoded = decode_feedback(
|
|
self.frame(0x494, "151515058c28203c")
|
|
)["actuator_state"]["driver"]["position_feedback"]
|
|
self.assertFalse(decoded["absolute_position_available"])
|
|
self.assertEqual(
|
|
set(decoded["initialized"]),
|
|
{"horizontal", "backrest", "height", "basin"},
|
|
)
|
|
|
|
def test_recursive_state_merge_preserves_other_devices(self) -> None:
|
|
target = {"driver": {"heating": {"state": "off"}}, "passenger": {"safe": True}}
|
|
update = {"driver": {"heating": {"state": "on", "level": 1}}}
|
|
merge_state(target, update)
|
|
self.assertEqual(target["driver"]["heating"]["state"], "on")
|
|
self.assertEqual(target["passenger"], {"safe": True})
|
|
|
|
def test_authoritative_state_requires_one_complete_consistent_ch1_frame(
|
|
self,
|
|
) -> None:
|
|
operation = {
|
|
"status": "succeeded",
|
|
"telemetry": {
|
|
"channel": 1,
|
|
"simulated": False,
|
|
"feedback_required": True,
|
|
"feedback_observed": True,
|
|
"feedback_match": True,
|
|
"feedback_frame": {
|
|
"frame_id": "0x404",
|
|
"payload_size": 8,
|
|
"data_hex": "5B 65 35 6C 18 18 A2 5D",
|
|
"timestamp": 12.5,
|
|
"channel": 1,
|
|
},
|
|
"feedback_state": {
|
|
"source": "physical_can_ch1",
|
|
"channel": 1,
|
|
"frame_id": "0x404",
|
|
"feedback_frame_id": "0x404",
|
|
"valid": True,
|
|
"domain_state": {"climate": {"power": "on"}},
|
|
},
|
|
},
|
|
}
|
|
|
|
state = authoritative_physical_ch1_feedback_state(operation)
|
|
self.assertIsNotNone(state)
|
|
self.assertTrue(state["verification"]["authoritative"])
|
|
self.assertEqual(state["verification"]["frame_id"], "0x404")
|
|
|
|
for label, mutate in (
|
|
(
|
|
"wrong frame channel",
|
|
lambda item: item["telemetry"]["feedback_frame"].update(
|
|
{"channel": 0}
|
|
),
|
|
),
|
|
(
|
|
"short payload",
|
|
lambda item: item["telemetry"]["feedback_frame"].update(
|
|
{"payload_size": 7}
|
|
),
|
|
),
|
|
(
|
|
"mismatched decoded id",
|
|
lambda item: item["telemetry"]["feedback_state"].update(
|
|
{"feedback_frame_id": "0x4FA"}
|
|
),
|
|
),
|
|
(
|
|
"empty decoded patch",
|
|
lambda item: item["telemetry"]["feedback_state"].pop(
|
|
"domain_state"
|
|
),
|
|
),
|
|
):
|
|
invalid = deepcopy(operation)
|
|
mutate(invalid)
|
|
with self.subTest(label=label):
|
|
self.assertIsNone(
|
|
authoritative_physical_ch1_feedback_state(invalid)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|