mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-10-28 00:53:45 +08:00
Restore Forester first generation angle control
This commit is contained in:
firestar5683
@@ -13,7 +13,7 @@ from opendbc.car.vehicle_model import VehicleModel
|
||||
# involves the total steering angle change rather than rate, but these limits work well for now
|
||||
MAX_STEER_RATE = 25 # deg/s
|
||||
MAX_STEER_RATE_FRAMES = 7 # tx control frames needed before torque can be cut
|
||||
GEN2_ANGLE_PORTS = frozenset((CAR.SUBARU_CROSSTREK_2025, CAR.SUBARU_LEGACY_2025, CAR.SUBARU_ASCENT_2023))
|
||||
GEN2_ANGLE_PORTS = frozenset((CAR.SUBARU_CROSSTREK_2025, CAR.SUBARU_LEGACY_2025, CAR.SUBARU_ASCENT_2023, CAR.SUBARU_FORESTER_2022))
|
||||
|
||||
|
||||
class CarController(CarControllerBase):
|
||||
@@ -136,12 +136,13 @@ class CarController(CarControllerBase):
|
||||
CS.out.vEgoRaw, CS.out.steeringAngleDeg, angle_active,
|
||||
self.p.FIXED_ANGLE_LIMITS)
|
||||
else:
|
||||
assert self.VM is not None
|
||||
apply_angle = apply_steer_angle_limits_vm(actuators.steeringAngleDeg, self.apply_angle_last,
|
||||
CS.out.vEgoRaw, CS.out.steeringAngleDeg, angle_active,
|
||||
self.p, self.VM)
|
||||
self.apply_angle_last = apply_angle
|
||||
self.angle_was_active = angle_active
|
||||
angle_bus = CanBus.main if self.CP.carFingerprint == CAR.SUBARU_ASCENT_2023 else CanBus.alt
|
||||
angle_bus = CanBus.main if self.CP.carFingerprint in (CAR.SUBARU_ASCENT_2023, CAR.SUBARU_FORESTER_2022) else CanBus.alt
|
||||
can_sends.append(subarucan.create_steering_control_angle(self.packer, apply_angle, angle_active, angle_bus))
|
||||
else:
|
||||
apply_torque = int(round(actuators.torque * self.p.STEER_MAX))
|
||||
|
||||
@@ -99,7 +99,7 @@ class CarState(CarStateBase):
|
||||
cp_cruise = cp_alt if self.CP.flags & SubaruFlags.GLOBAL_GEN2 else cp
|
||||
cp_es_brake = cp_alt if self.CP.flags & SubaruFlags.GLOBAL_GEN2 else cp_cam
|
||||
|
||||
if self.CP.carFingerprint in (CAR.SUBARU_CROSSTREK_2025, CAR.SUBARU_LEGACY_2025, CAR.SUBARU_ASCENT_2023):
|
||||
if self.CP.carFingerprint in (CAR.SUBARU_CROSSTREK_2025, CAR.SUBARU_LEGACY_2025, CAR.SUBARU_ASCENT_2023, CAR.SUBARU_FORESTER_2022):
|
||||
ret.cruiseState.enabled = cp_es_brake.vl["ES_Status"]["Cruise_Activated"] != 0
|
||||
ret.cruiseState.available = cp_cam.vl["ES_DashStatus"]["Cruise_On"] != 0
|
||||
elif self.CP.flags & (SubaruFlags.HYBRID | SubaruFlags.LKAS_ANGLE):
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
"""Reject contradictory discovery evidence for the two supported Gen2 angle identities."""
|
||||
"""Reject contradictory discovery evidence for the scoped Subaru angle identities."""
|
||||
from opendbc.car.fw_query_definitions import LiveFwVersions, OfflineFwVersions
|
||||
from opendbc.car.structs import CarParams
|
||||
|
||||
|
||||
SCOPED_CANDIDATES = frozenset(("SUBARU_CROSSTREK_2025", "SUBARU_LEGACY_2025"))
|
||||
SCOPED_CANDIDATES = frozenset(("SUBARU_CROSSTREK_2025", "SUBARU_LEGACY_2025", "SUBARU_FORESTER_2022"))
|
||||
REQUIRED_ECUS = frozenset((CarParams.Ecu.abs, CarParams.Ecu.fwdCamera, CarParams.Ecu.engine))
|
||||
|
||||
|
||||
|
||||
@@ -7,7 +7,7 @@ from opendbc.car.subaru.carcontroller import CarController
|
||||
from opendbc.car.subaru.carstate import CarState
|
||||
from opendbc.car.subaru.values import CAR, GLOBAL_ES_ADDR, CanBus, SubaruFlags, SubaruSafetyFlags
|
||||
|
||||
GEN2_ANGLE_PORTS = frozenset((CAR.SUBARU_CROSSTREK_2025, CAR.SUBARU_LEGACY_2025, CAR.SUBARU_ASCENT_2023))
|
||||
GEN2_ANGLE_PORTS = frozenset((CAR.SUBARU_CROSSTREK_2025, CAR.SUBARU_LEGACY_2025, CAR.SUBARU_ASCENT_2023, CAR.SUBARU_FORESTER_2022))
|
||||
|
||||
|
||||
class CarInterface(CarInterfaceBase):
|
||||
|
||||
@@ -5,7 +5,7 @@ from opendbc.car.subaru.values import CAR, SubaruFlags, SubaruSafetyFlags
|
||||
AOL_EXPERIENCE = 32
|
||||
TORQUE_CARS = frozenset((CAR.SUBARU_ASCENT, CAR.SUBARU_OUTBACK, CAR.SUBARU_LEGACY,
|
||||
CAR.SUBARU_IMPREZA, CAR.SUBARU_IMPREZA_2020, CAR.SUBARU_FORESTER))
|
||||
ANGLE_CARS = frozenset((CAR.SUBARU_CROSSTREK_2025, CAR.SUBARU_LEGACY_2025, CAR.SUBARU_ASCENT_2023))
|
||||
ANGLE_CARS = frozenset((CAR.SUBARU_CROSSTREK_2025, CAR.SUBARU_LEGACY_2025, CAR.SUBARU_ASCENT_2023, CAR.SUBARU_FORESTER_2022))
|
||||
|
||||
|
||||
def qualified(cp, *, marked_only=False):
|
||||
@@ -26,7 +26,7 @@ def qualified(cp, *, marked_only=False):
|
||||
angle = cp.carFingerprint in ANGLE_CARS
|
||||
if angle:
|
||||
expected |= int(SubaruSafetyFlags.LKAS_ANGLE)
|
||||
if cp.carFingerprint != CAR.SUBARU_CROSSTREK_2025:
|
||||
if cp.carFingerprint in (CAR.SUBARU_LEGACY_2025, CAR.SUBARU_ASCENT_2023):
|
||||
expected |= int(SubaruSafetyFlags.FIXED_ANGLE_LIMITS)
|
||||
if cp.carFingerprint == CAR.SUBARU_ASCENT_2023:
|
||||
expected |= int(SubaruSafetyFlags.ANGLE_MAIN_BUS)
|
||||
|
||||
@@ -7,7 +7,7 @@ from opendbc.car import Bus, structs
|
||||
from opendbc.car.subaru import subarucan
|
||||
from opendbc.car.subaru.stock_aol import ANGLE_CARS, TORQUE_CARS
|
||||
from opendbc.car.subaru.tests.test_ascent_angle import setup_controller, step
|
||||
from opendbc.car.subaru.values import CAR, DBC
|
||||
from opendbc.car.subaru.values import DBC, SubaruSafetyFlags
|
||||
from opendbc.safety.tests import test_subaru_gen2_angle_pair as pair
|
||||
from opendbc.safety.tests.libsafety import libsafety_py
|
||||
|
||||
@@ -17,7 +17,7 @@ from opendbc.safety.tests.libsafety import libsafety_py
|
||||
@pytest.mark.parametrize('position', ('within', 'limit', 'beyond'))
|
||||
def test_inactive_sensor_angle_is_bounded_and_native_accepts(car, sign, position):
|
||||
cp, controller, command, state = setup_controller(car)
|
||||
limit = 650.0 if car == CAR.SUBARU_CROSSTREK_2025 else 545.0
|
||||
limit = 545.0 if cp.safetyConfigs[0].safetyParam & SubaruSafetyFlags.FIXED_ANGLE_LIMITS else 650.0
|
||||
measured = sign * {'within': 100.0, 'limit': limit, 'beyond': limit + 5.0}[position]
|
||||
packer, frames = pair.TestSubaruGen2AnglePair.sources(car, cruise=False)
|
||||
measurement = packer.make_can_msg('Steering_2', 0, {'Steering_Angle': measured})
|
||||
|
||||
@@ -77,6 +77,7 @@
|
||||
{.msg = {{MSG_SUBARU_ES_DashStatus, SUBARU_CAM_BUS, 8, 10U, .max_counter = 15U, .ignore_quality_flag = true}, { 0 }, { 0 }}}, \
|
||||
|
||||
static bool subaru_gen2 = false;
|
||||
static bool subaru_forester_angle = false;
|
||||
static bool subaru_lkas_angle = false;
|
||||
static bool subaru_fixed_angle_limits = false;
|
||||
static bool subaru_angle_main_bus = false;
|
||||
@@ -123,7 +124,7 @@ static void subaru_rx_hook(const CANPacket_t *msg) {
|
||||
update_sample(&angle_meas, angle_meas_new);
|
||||
}
|
||||
|
||||
if (subaru_lkas_angle && (msg->addr == MSG_SUBARU_ES_Status) && (msg->bus == SUBARU_ALT_BUS)) {
|
||||
if (subaru_lkas_angle && (msg->addr == MSG_SUBARU_ES_Status) && (msg->bus == (subaru_forester_angle ? SUBARU_CAM_BUS : SUBARU_ALT_BUS))) {
|
||||
pcm_cruise_check(GET_BIT(msg, 29U));
|
||||
subaru_aol_grant(GET_BIT(msg, 29U));
|
||||
}
|
||||
@@ -200,6 +201,12 @@ static bool subaru_tx_hook(const CANPacket_t *msg) {
|
||||
.wheelbase = 2.6700000762939453,
|
||||
};
|
||||
|
||||
const AngleSteeringParams SUBARU_FORESTER_2022_ANGLE_PARAMS = {
|
||||
.slip_factor = -0.0006281956135115129,
|
||||
.steer_ratio = 17.0,
|
||||
.wheelbase = 2.6700000762939453,
|
||||
};
|
||||
|
||||
const LongitudinalLimits SUBARU_LONG_LIMITS = {
|
||||
.min_gas = 808, // appears to be engine braking
|
||||
.max_gas = 3400, // approx 2 m/s^2 when maxing cruise_rpm and cruise_throttle
|
||||
@@ -244,7 +251,7 @@ static bool subaru_tx_hook(const CANPacket_t *msg) {
|
||||
if (subaru_fixed_angle_limits) {
|
||||
violation |= steer_angle_cmd_checks(desired_angle, lkas_request, SUBARU_FIXED_ANGLE_STEERING_LIMITS);
|
||||
} else {
|
||||
violation |= steer_angle_cmd_checks_vm(desired_angle, lkas_request, SUBARU_ANGLE_STEERING_LIMITS, SUBARU_CROSSTREK_2025_ANGLE_PARAMS);
|
||||
violation |= steer_angle_cmd_checks_vm(desired_angle, lkas_request, SUBARU_ANGLE_STEERING_LIMITS, subaru_forester_angle ? SUBARU_FORESTER_2022_ANGLE_PARAMS : SUBARU_CROSSTREK_2025_ANGLE_PARAMS);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -344,6 +351,11 @@ static safety_config subaru_init(uint16_t param) {
|
||||
{0x390U, SUBARU_ALT_BUS, 8, .check_relay = false},
|
||||
};
|
||||
|
||||
static const CanMsg SUBARU_FORESTER_ANGLE_TX_MSGS[] = {
|
||||
SUBARU_BASE_TX_MSGS(SUBARU_MAIN_BUS, MSG_SUBARU_ES_LKAS_ANGLE)
|
||||
SUBARU_COMMON_TX_MSGS(SUBARU_MAIN_BUS)
|
||||
};
|
||||
|
||||
const uint16_t SUBARU_PARAM_ASCENT_ALERT = 2048;
|
||||
const bool ascent_alert = param == (1U | 16U | 128U | 32U | SUBARU_PARAM_ASCENT_ALERT);
|
||||
const uint16_t SUBARU_PARAM_GEN2 = 1;
|
||||
@@ -353,12 +365,14 @@ static safety_config subaru_init(uint16_t param) {
|
||||
const uint16_t SUBARU_PARAM_ANGLE_MAIN_BUS = 32;
|
||||
subaru_angle_main_bus = GET_FLAG(param, SUBARU_PARAM_ANGLE_MAIN_BUS);
|
||||
|
||||
subaru_forester_angle = (param == 16U) &&
|
||||
(((unsigned int)alternative_experience == 0U) || ((unsigned int)alternative_experience == 32U));
|
||||
subaru_gen2 = GET_FLAG(param, SUBARU_PARAM_GEN2);
|
||||
subaru_lkas_angle = GET_FLAG(param, SUBARU_PARAM_LKAS_ANGLE);
|
||||
subaru_fixed_angle_limits = GET_FLAG(param, SUBARU_PARAM_FIXED_ANGLE);
|
||||
subaru_angle_invalid = (subaru_lkas_angle && (base_param != (SUBARU_PARAM_GEN2 | SUBARU_PARAM_LKAS_ANGLE)) &&
|
||||
(base_param != (SUBARU_PARAM_GEN2 | SUBARU_PARAM_LKAS_ANGLE | SUBARU_PARAM_FIXED_ANGLE)) &&
|
||||
(base_param != (SUBARU_PARAM_GEN2 | SUBARU_PARAM_LKAS_ANGLE | SUBARU_PARAM_FIXED_ANGLE | SUBARU_PARAM_ANGLE_MAIN_BUS)) && !ascent_alert) ||
|
||||
(base_param != (SUBARU_PARAM_GEN2 | SUBARU_PARAM_LKAS_ANGLE | SUBARU_PARAM_FIXED_ANGLE | SUBARU_PARAM_ANGLE_MAIN_BUS)) && !ascent_alert && !subaru_forester_angle) ||
|
||||
((subaru_fixed_angle_limits || subaru_angle_main_bus) && !subaru_lkas_angle);
|
||||
|
||||
// TODO: re-enable once more work is done on the limits
|
||||
@@ -372,7 +386,7 @@ static safety_config subaru_init(uint16_t param) {
|
||||
((unsigned int)alternative_experience == 32U)));
|
||||
subaru_aol_reset();
|
||||
subaru_aol_enabled = ((unsigned int)alternative_experience == 32U) &&
|
||||
((param == 0U) || (param == 1U) || (param == 17U) || (param == 145U) ||
|
||||
((param == 16U) || (param == 0U) || (param == 1U) || (param == 17U) || (param == 145U) ||
|
||||
avh || (param == 177U) || (param == 401U) || (param == 8U) || (param == 520U) || (param == 512U) || (param == 2225U));
|
||||
if (subaru_aol_enabled) {
|
||||
static const AolSafetyPolicy policy = {
|
||||
@@ -390,7 +404,17 @@ static safety_config subaru_init(uint16_t param) {
|
||||
static RxCheck subaru_gen2_angle_rx_checks[] = {
|
||||
SUBARU_GEN2_ANGLE_RX_CHECKS()
|
||||
};
|
||||
ret = avh ? (stop_start ? BUILD_SAFETY_CFG(subaru_gen2_angle_rx_checks, SUBARU_AVH_STOP_START_TX_MSGS) :
|
||||
static RxCheck subaru_forester_angle_rx_checks[] = {
|
||||
{.msg = {{MSG_SUBARU_Throttle, SUBARU_MAIN_BUS, 8, 100U, .max_counter = 15U, .ignore_quality_flag = true}, { 0 }, { 0 }}},
|
||||
{.msg = {{MSG_SUBARU_Steering_Torque, SUBARU_MAIN_BUS, 8, 50U, .max_counter = 15U, .ignore_quality_flag = true}, { 0 }, { 0 }}},
|
||||
{.msg = {{MSG_SUBARU_Steering_2, SUBARU_MAIN_BUS, 8, 50U, .max_counter = 15U, .ignore_quality_flag = true}, { 0 }, { 0 }}},
|
||||
{.msg = {{MSG_SUBARU_Wheel_Speeds, SUBARU_MAIN_BUS, 8, 50U, .max_counter = 15U, .ignore_quality_flag = true}, { 0 }, { 0 }}},
|
||||
{.msg = {{MSG_SUBARU_Brake_Status, SUBARU_MAIN_BUS, 8, 50U, .max_counter = 15U, .ignore_quality_flag = true}, { 0 }, { 0 }}},
|
||||
{.msg = {{MSG_SUBARU_ES_Status, SUBARU_CAM_BUS, 8, 20U, .max_counter = 15U, .ignore_quality_flag = true}, { 0 }, { 0 }}},
|
||||
{.msg = {{MSG_SUBARU_ES_DashStatus, SUBARU_CAM_BUS, 8, 10U, .max_counter = 15U, .ignore_quality_flag = true}, { 0 }, { 0 }}},
|
||||
};
|
||||
ret = subaru_forester_angle ? BUILD_SAFETY_CFG(subaru_forester_angle_rx_checks, SUBARU_FORESTER_ANGLE_TX_MSGS) :
|
||||
avh ? (stop_start ? BUILD_SAFETY_CFG(subaru_gen2_angle_rx_checks, SUBARU_AVH_STOP_START_TX_MSGS) :
|
||||
BUILD_SAFETY_CFG(subaru_gen2_angle_rx_checks, SUBARU_AVH_TX_MSGS)) :
|
||||
stop_start ? BUILD_SAFETY_CFG(subaru_gen2_angle_rx_checks, SUBARU_STOP_START_TX_MSGS) :
|
||||
ascent_alert ? BUILD_SAFETY_CFG(subaru_gen2_angle_rx_checks, SUBARU_ASCENT_ALERT_TX_MSGS) :
|
||||
|
||||
@@ -0,0 +1,294 @@
|
||||
import unittest
|
||||
|
||||
from opendbc.can import CANPacker
|
||||
from opendbc.car import Bus, structs
|
||||
from opendbc.car.fw_versions import match_fw_to_car
|
||||
from opendbc.car.subaru.carcontroller import CarController
|
||||
from opendbc.car.subaru.carstate import CarState
|
||||
from opendbc.car.subaru.interface import CarInterface
|
||||
from opendbc.car.subaru.values import CAR, DBC, SubaruSafetyFlags
|
||||
from opendbc.safety.tests.libsafety import libsafety_py
|
||||
|
||||
|
||||
CARS = (CAR.SUBARU_FORESTER_2022,)
|
||||
|
||||
|
||||
class TestSubaruForesterAngle(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.safety = libsafety_py.libsafety
|
||||
|
||||
@staticmethod
|
||||
def packet(frame):
|
||||
return libsafety_py.make_CANPacket(frame[0], frame[2], frame[1])
|
||||
|
||||
@staticmethod
|
||||
def params(car, *, alpha=False, release=False):
|
||||
return CarInterface.get_params(car, {0: {}, 1: {}, 2: {}}, [], alpha, release, False)
|
||||
|
||||
def mode(self, cp, *, extra_flags=0):
|
||||
cfg = cp.safetyConfigs[0]
|
||||
self.assertEqual(self.safety.set_safety_hooks(cfg.safetyModel.raw, cfg.safetyParam | extra_flags), 0)
|
||||
self.safety.init_tests()
|
||||
self.safety.set_timer(1_000_000)
|
||||
|
||||
@staticmethod
|
||||
def sources(car, *, brake=False, cruise=True, speed=30, main=True, torque=0):
|
||||
packer = CANPacker(DBC[car][Bus.pt])
|
||||
inputs: tuple[tuple[str, int, dict[str, float]], ...] = (
|
||||
('Throttle', 0, {}),
|
||||
('Steering_Torque', 0, {'Steer_Torque_Sensor': torque}),
|
||||
('Steering_2', 0, {'Steering_Angle': 0}),
|
||||
('Wheel_Speeds', 0, {'FL': speed, 'FR': speed, 'RL': speed, 'RR': speed}),
|
||||
('Brake_Status', 0, {'Brake': int(brake)}),
|
||||
('ES_Status', 2, {'Cruise_Activated': int(cruise)}),
|
||||
('ES_DashStatus', 2, {'Cruise_On': int(main)}),
|
||||
('Transmission', 0, {'Gear': 121}),
|
||||
('Dashlights', 0, {}),
|
||||
('BodyInfo', 0, {}),
|
||||
('ES_Distance', 2, {}),
|
||||
('ES_LKAS_State', 2, {}),
|
||||
('ES_Brake', 2, {}),
|
||||
)
|
||||
return packer, [packer.make_can_msg(name, bus, values) for name, bus, values in inputs]
|
||||
|
||||
def controller(self, car, cp, frames, *, active=True):
|
||||
state = CarState(cp)
|
||||
parsers = state.get_can_parsers(cp)
|
||||
state.update(parsers)
|
||||
for parser in parsers.values():
|
||||
parser.update([(1_000_000_000, frames)])
|
||||
state.out = state.update(parsers).as_reader()
|
||||
self.assertEqual(state.out.gearShifter, structs.CarState.GearShifter.drive)
|
||||
self.assertGreater(state.out.vEgoRaw, 0)
|
||||
|
||||
command = structs.CarControl()
|
||||
command.enabled = True
|
||||
command.latActive = active
|
||||
command.actuators.steeringAngleDeg = 1.0
|
||||
_, sends = CarController(DBC[car], cp).update(command.as_reader(), state, 1_100_000_000)
|
||||
steer = next(frame for frame in sends if frame[0] == 0x124)
|
||||
self.assertEqual(steer[2], 0)
|
||||
self.assertEqual(bool(steer[1][1] & 0x10), active and state.out.cruiseState.available and
|
||||
state.out.cruiseState.enabled and not state.out.brakePressed)
|
||||
return state, steer, sends
|
||||
|
||||
def arm(self, cp, frames):
|
||||
self.mode(cp)
|
||||
required = {0x40, 0x119, 0x11A, 0x13A, 0x13C, 0x222, 0x321}
|
||||
for frame in frames:
|
||||
if frame[0] in required:
|
||||
self.assertTrue(self.safety.safety_rx_hook(self.packet(frame)), hex(frame[0]))
|
||||
self.safety.safety_tick()
|
||||
self.assertTrue(self.safety.safety_config_valid())
|
||||
self.assertTrue(self.safety.get_controls_allowed())
|
||||
|
||||
def test_stock_packed_host_to_native_and_inactive(self):
|
||||
for car in CARS:
|
||||
for release in (False, True):
|
||||
with self.subTest(car=car, release=release):
|
||||
cp = self.params(car, alpha=True, release=release)
|
||||
self.assertFalse(cp.dashcamOnly)
|
||||
self.assertFalse(cp.openpilotLongitudinalControl)
|
||||
self.assertFalse(cp.alphaLongitudinalAvailable)
|
||||
self.assertEqual(cp.steerControlType, structs.CarParams.SteerControlType.angle)
|
||||
packer, frames = self.sources(car)
|
||||
state, steer, sends = self.controller(car, cp, frames)
|
||||
self.assertTrue(state.out.cruiseState.enabled)
|
||||
self.assertNotEqual(steer[1][5:7], b'\x00\x00')
|
||||
self.arm(cp, frames)
|
||||
self.assertTrue(self.safety.safety_tx_hook(self.packet(steer)))
|
||||
self.assertEqual({frame[0] for frame in sends}, {0x124, 0x321, 0x322})
|
||||
self.assertFalse(self.safety.safety_tx_hook(self.packet((0x122, bytes(8), 0))))
|
||||
self.assertFalse(self.safety.safety_tx_hook(self.packet((0x124, steer[1], 1))))
|
||||
self.assertFalse(self.safety.safety_tx_hook(self.packet((0x124, steer[1][:-1], 0))))
|
||||
self.assertFalse(self.safety.safety_tx_hook(self.packet(packer.make_can_msg('ES_Distance', 2, {}))))
|
||||
cancel = structs.CarControl()
|
||||
cancel.cruiseControl.cancel = True
|
||||
_, cancel_sends = CarController(DBC[car], cp).update(cancel.as_reader(), state, 1_100_000_000)
|
||||
distance = next(frame for frame in cancel_sends if frame[0] == 0x221)
|
||||
self.assertEqual(distance[2], 0)
|
||||
self.assertTrue(self.safety.safety_tx_hook(self.packet(distance)))
|
||||
|
||||
_, inactive, _ = self.controller(car, cp, frames, active=False)
|
||||
self.assertTrue(self.safety.safety_tx_hook(self.packet(inactive)))
|
||||
brake_frames = [packer.make_can_msg('Brake_Status', 0, {'Brake': 1}) if f[0] == 0x13C else f for f in frames]
|
||||
brake_state, brake_steer, _ = self.controller(car, cp, brake_frames)
|
||||
self.assertTrue(brake_state.out.brakePressed)
|
||||
self.assertFalse(brake_steer[1][1] & 0x10)
|
||||
self.assertTrue(self.safety.safety_rx_hook(self.packet(brake_frames[4])))
|
||||
self.assertFalse(self.safety.get_controls_allowed())
|
||||
self.assertTrue(self.safety.safety_tx_hook(self.packet(brake_steer)))
|
||||
self.assertFalse(self.safety.safety_tx_hook(self.packet(steer)))
|
||||
|
||||
def test_required_health_and_raw_long_conflict(self):
|
||||
for car in CARS:
|
||||
cp = self.params(car)
|
||||
_, frames = self.sources(car)
|
||||
required = [f for f in frames if f[0] in {0x40, 0x119, 0x11A, 0x13A, 0x13C, 0x222, 0x321}]
|
||||
for missing in required:
|
||||
with self.subTest(car=car, missing=hex(missing[0])):
|
||||
self.mode(cp)
|
||||
for frame in required:
|
||||
if frame is not missing:
|
||||
self.safety.safety_rx_hook(self.packet(frame))
|
||||
self.safety.safety_tick()
|
||||
self.assertFalse(self.safety.safety_config_valid())
|
||||
self.arm(cp, frames)
|
||||
_, steer, _ = self.controller(car, cp, frames)
|
||||
invalid_option = 0x200 if car == CAR.SUBARU_LEGACY_2025 else 0x100
|
||||
for flags in (SubaruSafetyFlags.LONG, 0x20, invalid_option):
|
||||
with self.subTest(car=car, flags=flags):
|
||||
self.mode(cp, extra_flags=flags)
|
||||
for frame in required:
|
||||
self.safety.safety_rx_hook(self.packet(frame))
|
||||
self.safety.safety_tick()
|
||||
self.assertFalse(self.safety.safety_tx_hook(self.packet(steer)))
|
||||
for raw_param in (0x50, 0x30, 0x31, 0x111):
|
||||
with self.subTest(car=car, raw_param=hex(raw_param)):
|
||||
self.assertEqual(self.safety.set_safety_hooks(structs.CarParams.SafetyModel.subaru, raw_param), 0)
|
||||
self.safety.init_tests()
|
||||
for frame in required:
|
||||
self.safety.safety_rx_hook(self.packet(frame))
|
||||
self.safety.safety_tick()
|
||||
self.assertFalse(self.safety.safety_tx_hook(self.packet(steer)))
|
||||
self.mode(cp)
|
||||
for frame in required:
|
||||
if frame[0] == 0x222:
|
||||
frame = (frame[0], frame[1][:-1] + bytes([frame[1][-1] ^ 1]), frame[2])
|
||||
self.safety.safety_rx_hook(self.packet(frame))
|
||||
self.safety.safety_tick()
|
||||
self.assertFalse(self.safety.safety_config_valid())
|
||||
|
||||
for source in required:
|
||||
for bad in ((source[0], source[1], (source[2] + 1) % 3),
|
||||
(source[0], source[1][:-1], source[2])):
|
||||
with self.subTest(car=car, source=hex(source[0]), bad=bad[2:]):
|
||||
self.mode(cp)
|
||||
for frame in required:
|
||||
self.safety.safety_rx_hook(self.packet(bad if frame is source else frame))
|
||||
self.safety.safety_tick()
|
||||
self.assertFalse(self.safety.safety_config_valid())
|
||||
|
||||
self.arm(cp, frames)
|
||||
self.safety.set_timer(10_000_000)
|
||||
self.safety.safety_tick()
|
||||
self.assertFalse(self.safety.safety_config_valid())
|
||||
|
||||
self.mode(cp)
|
||||
for frame in required:
|
||||
self.safety.safety_rx_hook(self.packet(frame))
|
||||
for _ in range(6):
|
||||
self.safety.safety_rx_hook(self.packet(next(f for f in required if f[0] == 0x222)))
|
||||
self.safety.safety_tick()
|
||||
self.assertFalse(self.safety.safety_config_valid())
|
||||
|
||||
def test_stock_main_driver_and_cruise_handoff(self):
|
||||
for car in CARS:
|
||||
cp = self.params(car)
|
||||
for override in ({'main': False}, {'cruise': False}, {'brake': True}, {'torque': 250}):
|
||||
with self.subTest(car=car, override=override):
|
||||
_, frames = self.sources(car, **override)
|
||||
self.mode(cp)
|
||||
for frame in frames:
|
||||
self.safety.safety_rx_hook(self.packet(frame))
|
||||
state, steer, _ = self.controller(car, cp, frames)
|
||||
if override.get('torque'):
|
||||
# The host deliberately waits for a second steering period before
|
||||
# cutting a request on persistent driver input.
|
||||
command = structs.CarControl()
|
||||
command.enabled = command.latActive = True
|
||||
command.actuators.steeringAngleDeg = 1.0
|
||||
controller = CarController(DBC[car], cp)
|
||||
controller.update(command.as_reader(), state, 1_100_000_000)
|
||||
controller.update(command.as_reader(), state, 1_110_000_000)
|
||||
_, sends = controller.update(command.as_reader(), state, 1_120_000_000)
|
||||
steer = next(frame for frame in sends if frame[0] == 0x124)
|
||||
self.assertFalse(steer[1][1] & 0x10)
|
||||
self.assertTrue(self.safety.safety_tx_hook(self.packet(steer)))
|
||||
|
||||
_, frames = self.sources(car)
|
||||
self.arm(cp, frames)
|
||||
_, steer, _ = self.controller(car, cp, frames)
|
||||
self.assertTrue(self.safety.safety_tx_hook(self.packet(steer)))
|
||||
_, main_off = self.sources(car, main=False)
|
||||
dash = next(frame for frame in main_off if frame[0] == 0x321)
|
||||
self.assertTrue(self.safety.safety_rx_hook(self.packet(dash)))
|
||||
self.assertFalse(self.safety.safety_tx_hook(self.packet(steer)))
|
||||
|
||||
def test_forwarding_and_relay(self):
|
||||
for car in CARS:
|
||||
cp = self.params(car)
|
||||
self.mode(cp)
|
||||
self.assertEqual(self.safety.safety_fwd_hook(0, 0x124), 2)
|
||||
self.assertEqual(self.safety.safety_fwd_hook(2, 0x124), -1)
|
||||
self.assertEqual(self.safety.safety_fwd_hook(2, 0x321), -1)
|
||||
self.assertEqual(self.safety.safety_fwd_hook(2, 0x322), -1)
|
||||
self.assertEqual(self.safety.safety_fwd_hook(2, 0x221), 0)
|
||||
self.assertEqual(self.safety.safety_fwd_hook(1, 0x124), -1)
|
||||
_, frames = self.sources(car)
|
||||
_, steer, _ = self.controller(car, cp, frames)
|
||||
self.arm(cp, frames)
|
||||
self.assertTrue(self.safety.safety_tx_hook(self.packet(steer)))
|
||||
self.safety.set_relay_malfunction(True)
|
||||
self.assertFalse(self.safety.safety_tx_hook(self.packet(steer)))
|
||||
self.assertEqual(self.safety.safety_fwd_hook(0, 0x124), -1)
|
||||
self.assertEqual(self.safety.safety_fwd_hook(2, 0x221), -1)
|
||||
|
||||
def test_host_native_angle_rate_matrix(self):
|
||||
for car in CARS:
|
||||
for speed in (5, 30, 80):
|
||||
for target in (-650., 650.):
|
||||
with self.subTest(car=car, speed=speed, target=target):
|
||||
cp = self.params(car)
|
||||
_, frames = self.sources(car, speed=speed)
|
||||
state = CarState(cp)
|
||||
parsers = state.get_can_parsers(cp)
|
||||
state.update(parsers)
|
||||
for parser in parsers.values():
|
||||
parser.update([(1_000_000_000, frames)])
|
||||
state.out = state.update(parsers).as_reader()
|
||||
self.arm(cp, frames)
|
||||
controller = CarController(DBC[car], cp)
|
||||
command = structs.CarControl()
|
||||
command.enabled = command.latActive = True
|
||||
command.actuators.steeringAngleDeg = target
|
||||
seen = []
|
||||
for i in range(120):
|
||||
self.safety.set_timer(1_000_000 + i * 10_000)
|
||||
_, sends = controller.update(command.as_reader(), state, 1_000_000_000 + i * 10_000_000)
|
||||
for frame in sends:
|
||||
if frame[0] == 0x124:
|
||||
self.assertTrue(self.safety.safety_tx_hook(self.packet(frame)),
|
||||
f'{car.value} {speed=} {target=} {i=} {frame[1].hex()}')
|
||||
seen.append(frame)
|
||||
self.assertGreater(len(seen), 50)
|
||||
self.assertNotEqual(seen[-1][1][5:7], b'\x00\x00')
|
||||
|
||||
def test_own_vehicle_model_and_exact_configuration(self):
|
||||
from opendbc.car.vehicle_model import VehicleModel, calc_slip_factor
|
||||
cp = self.params(CARS[0])
|
||||
self.assertEqual(cp.safetyConfigs[0].safetyParam, 16)
|
||||
self.assertAlmostEqual(cp.mass, 1704.)
|
||||
self.assertAlmostEqual(calc_slip_factor(VehicleModel(cp)), -0.0006281956135115129, places=13)
|
||||
self.assertFalse(cp.openpilotLongitudinalControl or cp.alphaLongitudinalAvailable)
|
||||
|
||||
def test_firmware_complete_partial_and_mixed(self):
|
||||
from opendbc.car.subaru.fingerprints import FW_VERSIONS
|
||||
car = CARS[0]
|
||||
rows = [(ecu, addr, sub, values[0]) for (ecu, addr, sub), values in FW_VERSIONS[car].items()]
|
||||
|
||||
def records(entries):
|
||||
return [structs.CarParams.CarFw(ecu=ecu, address=addr, subAddress=0 if sub is None else sub, fwVersion=value, brand='subaru')
|
||||
for ecu, addr, sub, value in entries]
|
||||
|
||||
exact, matches = match_fw_to_car(records(rows), '00000000000000000', log=False)
|
||||
self.assertTrue(exact)
|
||||
self.assertEqual(matches, {car})
|
||||
for partial in (rows[:1], rows[1:2]):
|
||||
_, matches = match_fw_to_car(records(partial), '00000000000000000', log=False)
|
||||
self.assertNotIn(car, matches)
|
||||
foreign = next((ecu, addr, sub, value) for sibling, firmware in FW_VERSIONS.items() if sibling != car
|
||||
for (ecu, addr, sub), values in firmware.items() for value in values
|
||||
if (ecu, addr, sub) in FW_VERSIONS[car] and value not in FW_VERSIONS[car][(ecu, addr, sub)])
|
||||
_, matches = match_fw_to_car(records([*rows, foreign]), '00000000000000000', log=False)
|
||||
self.assertNotIn(car, matches)
|
||||
@@ -34,20 +34,22 @@ class TestSubaruGen2AnglePair(unittest.TestCase):
|
||||
@staticmethod
|
||||
def sources(car, *, brake=False, cruise=True, speed=30, main=True, torque=0):
|
||||
packer = CANPacker(DBC[car][Bus.pt])
|
||||
inputs = (
|
||||
wheel_bus = 0 if car == CAR.SUBARU_FORESTER_2022 else 1
|
||||
status_bus = 2 if car == CAR.SUBARU_FORESTER_2022 else 1
|
||||
inputs: tuple[tuple[str, int, dict[str, float]], ...] = (
|
||||
('Throttle', 0, {}),
|
||||
('Steering_Torque', 0, {'Steer_Torque_Sensor': torque}),
|
||||
('Steering_2', 0, {'Steering_Angle': 0}),
|
||||
('Wheel_Speeds', 1, {'FL': speed, 'FR': speed, 'RL': speed, 'RR': speed}),
|
||||
('Brake_Status', 1, {'Brake': int(brake)}),
|
||||
('ES_Status', 1, {'Cruise_Activated': int(cruise)}),
|
||||
('Wheel_Speeds', wheel_bus, {'FL': speed, 'FR': speed, 'RL': speed, 'RR': speed}),
|
||||
('Brake_Status', wheel_bus, {'Brake': int(brake)}),
|
||||
('ES_Status', status_bus, {'Cruise_Activated': int(cruise)}),
|
||||
('ES_DashStatus', 2, {'Cruise_On': int(main)}),
|
||||
('Transmission', 0, {'Gear': 121}),
|
||||
('Dashlights', 0, {}),
|
||||
('BodyInfo', 0, {}),
|
||||
('ES_Distance', 1, {}),
|
||||
('ES_Distance', status_bus, {}),
|
||||
('ES_LKAS_State', 2, {}),
|
||||
('ES_Brake', 1, {}),
|
||||
('ES_Brake', status_bus, {}),
|
||||
)
|
||||
return packer, [packer.make_can_msg(name, bus, values) for name, bus, values in inputs]
|
||||
|
||||
|
||||
@@ -65,7 +65,7 @@ class TestSubaruStockAol(unittest.TestCase):
|
||||
self.assertTrue(self.safety.get_controls_allowed())
|
||||
|
||||
def test_all_stock_profiles_full_engagement_brake_reverse_drive_main_off(self):
|
||||
for param in (0, 1, 17, 145, 177, 2225):
|
||||
for param in (0, 1, 16, 17, 145, 177, 2225):
|
||||
with self.subTest(param=param):
|
||||
self.configure(param)
|
||||
self.arm()
|
||||
@@ -177,7 +177,7 @@ class TestSubaruStockAol(unittest.TestCase):
|
||||
self.assertEqual(self.feed(cruise=True), 1)
|
||||
|
||||
def test_unrecognized_words_cannot_acquire_independent_permission(self):
|
||||
for param in (2, 3, 16, 209, 65535):
|
||||
for param in (2, 3, 48, 80, 209, 65535):
|
||||
self.configure(param)
|
||||
for _ in range(12):
|
||||
self.feed(cruise=False)
|
||||
|
||||
@@ -94,7 +94,7 @@ int main() {
|
||||
subaru.safety_mode = 11U;
|
||||
subaru.safety_param = static_cast<uint16_t>(word);
|
||||
subaru.capability_flags = 1U;
|
||||
const bool expected = (word == 0U) || (word == 1U) || (word == 17U) ||
|
||||
const bool expected = (word == 16U) || (word == 0U) || (word == 1U) || (word == 17U) ||
|
||||
(word == 145U) || (word == 1169U) || (word == 1425U) || (word == 401U) || (word == 177U) || (word == 8U) || (word == 520U) || (word == 512U) || (word == 2225U);
|
||||
assert(aol_capable(subaru, 11U, VEHICLE_REGISTRY) == expected);
|
||||
subaru.capability_flags = 0U;
|
||||
|
||||
@@ -29,7 +29,7 @@ def policy_for(cp):
|
||||
|
||||
|
||||
def native_profile_supported(model, param):
|
||||
return model == int(structs.CarParams.SafetyModel.subaru) and param in (0, 1, 8, 17, 145, 177, 401, 512, 520, 1169, 1425, 2225)
|
||||
return model == int(structs.CarParams.SafetyModel.subaru) and param in (0, 1, 8, 16, 17, 145, 177, 401, 512, 520, 1169, 1425, 2225)
|
||||
|
||||
|
||||
def native_accepts_cp(cp, model, param):
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
#include "selfdrive/pandad/aol_protocol.h"
|
||||
|
||||
inline bool subaru_aol_param(uint16_t param) {
|
||||
return (param == 0U) || (param == 1U) || (param == 17U) || (param == 145U) ||
|
||||
return (param == 16U) || (param == 0U) || (param == 1U) || (param == 17U) || (param == 145U) ||
|
||||
(param == 1169U) || (param == 1425U) || (param == 401U) || (param == 177U) || (param == 8U) || (param == 520U) || (param == 512U) || (param == 2225U);
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
"""Actual automatic/manual factory, Card, SelfdriveD, Controls and native Forester angle joins."""
|
||||
import pytest
|
||||
from opendbc.car.subaru.values import CAR
|
||||
from openpilot.starpilot.car.subaru.tests.native_fixture import subaru_native_pair as subaru_native_pair
|
||||
|
||||
|
||||
@pytest.mark.parametrize('aol', [False, True])
|
||||
@pytest.mark.parametrize('manual', [False, True])
|
||||
def test_actual_forester_full_lifecycle(aol, manual, subaru_native_pair, tmp_path):
|
||||
from openpilot.starpilot.car.subaru.tests.test_stock_aol_callers import test_actual_full_engage_brake_reverse_temporary_recovery_mainoff
|
||||
test_actual_full_engage_brake_reverse_temporary_recovery_mainoff(CAR.SUBARU_FORESTER_2022, subaru_native_pair, tmp_path,
|
||||
aol=aol, manual=manual)
|
||||
@@ -22,20 +22,19 @@ from openpilot.starpilot.car.subaru.tests.native_fixture import subaru_native_pa
|
||||
|
||||
|
||||
@pytest.mark.parametrize('candidate', [CAR.SUBARU_ASCENT_2023, CAR.SUBARU_LEGACY_2025, CAR.SUBARU_CROSSTREK_2025, CAR.SUBARU_FORESTER, CAR.SUBARU_OUTBACK])
|
||||
def test_actual_full_engage_brake_reverse_temporary_recovery_mainoff(candidate, subaru_native_pair, tmp_path):
|
||||
aol = True
|
||||
def test_actual_full_engage_brake_reverse_temporary_recovery_mainoff(candidate, subaru_native_pair, tmp_path, aol=True, manual=False):
|
||||
fw = firmware_records(candidate, FW_VERSIONS, 'subaru')
|
||||
rows = []
|
||||
with actual_factory(candidate, gen_empty_fingerprint(), fw, '0' * 17, manual=False, aol=aol) as (card, params, matches):
|
||||
with actual_factory(candidate, gen_empty_fingerprint(), fw, '0' * 17, manual=manual, aol=aol) as (card, params, matches):
|
||||
cp = card.CP
|
||||
ci = card.CI
|
||||
safety = libsafety_py.libsafety
|
||||
assert cp.pcmCruise and not cp.openpilotLongitudinalControl and not cp.alphaLongitudinalAvailable
|
||||
assert cp.alternativeExperience == 32 and not cp.dashcamOnly
|
||||
assert cp.alternativeExperience == (32 if aol else 0) and not cp.dashcamOnly
|
||||
sd, controls = SelfdriveD(CP=cp), Controls()
|
||||
stream = CompleteCan(ci)
|
||||
safety.init_tests()
|
||||
safety.set_alternative_experience(32)
|
||||
safety.set_alternative_experience(cp.alternativeExperience)
|
||||
assert safety.set_safety_hooks(cp.safetyConfigs[0].safetyModel.raw, cp.safetyConfigs[0].safetyParam) == 0
|
||||
safety.set_aol_test_heartbeat(False)
|
||||
library, bridge = subaru_native_pair
|
||||
@@ -47,7 +46,7 @@ def test_actual_full_engage_brake_reverse_temporary_recovery_mainoff(candidate,
|
||||
receipt = None
|
||||
steering_name = 'ES_LKAS_ANGLE' if cp.steerControlType == structs.CarParams.SteerControlType.angle else 'ES_LKAS'
|
||||
steering_address = ci.CC.packer.dbc.name_to_msg[steering_name].address
|
||||
steering_bus = 0 if candidate in (CAR.SUBARU_ASCENT_2023, CAR.SUBARU_FORESTER, CAR.SUBARU_OUTBACK) else 1
|
||||
steering_bus = 0 if candidate in (CAR.SUBARU_ASCENT_2023, CAR.SUBARU_FORESTER, CAR.SUBARU_OUTBACK, CAR.SUBARU_FORESTER_2022) else 1
|
||||
decoder = CANParser(DBC[candidate][Bus.pt], [(steering_name, 0)], steering_bus)
|
||||
dash_decoder = CANParser(DBC[candidate][Bus.pt], [('ES_LKAS_State', 0)], 0)
|
||||
dash_emissions = 0
|
||||
@@ -79,7 +78,7 @@ def test_actual_full_engage_brake_reverse_temporary_recovery_mainoff(candidate,
|
||||
panda = messaging.new_message('pandaStates', 1, valid=True, logMonoTime=now)
|
||||
panda.pandaStates[0].safetyModel = cp.safetyConfigs[0].safetyModel
|
||||
panda.pandaStates[0].safetyParam = cp.safetyConfigs[0].safetyParam
|
||||
panda.pandaStates[0].alternativeExperience = 32
|
||||
panda.pandaStates[0].alternativeExperience = cp.alternativeExperience
|
||||
panda.pandaStates[0].ignitionLine = True
|
||||
panda.pandaStates[0].controlsAllowed = bool(safety.get_controls_allowed())
|
||||
panda.pandaStates[0].safetyRxChecksInvalid = not bool(safety.safety_config_valid())
|
||||
@@ -91,7 +90,7 @@ def test_actual_full_engage_brake_reverse_temporary_recovery_mainoff(candidate,
|
||||
peers.append(panda.as_reader())
|
||||
if 'carControl' in latest_controls:
|
||||
peers.append(latest_controls['carControl'])
|
||||
peers.extend(driven_new[s] for s in ('onroadEvents', 'selfdriveState', 'aolAxisState') if s in driven_new)
|
||||
peers.extend(driven_new[s] for s in ('onroadEvents', 'selfdriveState', 'aolAxisState') if s in driven_new and s in card.sm.data)
|
||||
if receipt is not None and tick % 10 == 1:
|
||||
peers.append(receipt.as_reader())
|
||||
card.sm.update_msgs((now + 1000) / 1e9, peers)
|
||||
@@ -128,13 +127,13 @@ def test_actual_full_engage_brake_reverse_temporary_recovery_mainoff(candidate,
|
||||
driven_new = selfdrive_step(sd, published, new_native, previous_controls, cp, safety, now, calibration='calibrated')
|
||||
driven.update(driven_new)
|
||||
order = None
|
||||
if tick % 10 == 0:
|
||||
if aol and tick % 10 == 0:
|
||||
receipt, order = pandad.step(driven, published, now)
|
||||
feed_perception_only(controls, now, tick, enabled=sd.enabled, active=sd.active)
|
||||
plan = messaging.new_message('longitudinalPlan', valid=True, logMonoTime=now)
|
||||
plan.longitudinalPlan.aTarget = 0.5
|
||||
control_peers = [published['carState'], published['carOutput'], plan.as_reader()]
|
||||
control_peers.extend(driven_new[s] for s in ('selfdriveState', 'onroadEvents', 'aolAxisState') if s in driven_new)
|
||||
control_peers.extend(driven_new[s] for s in ('selfdriveState', 'onroadEvents', 'aolAxisState') if s in driven_new and s in controls.sm.data)
|
||||
if receipt is not None and tick % 10 == 0:
|
||||
control_peers.append(receipt.as_reader())
|
||||
controls.sm.update_msgs((now + 1000) / 1e9, control_peers)
|
||||
@@ -176,12 +175,12 @@ def test_actual_full_engage_brake_reverse_temporary_recovery_mainoff(candidate,
|
||||
active_emissions.append(tick)
|
||||
if 165 <= tick < 180 or 240 <= tick < 255 or not aol and (125 <= tick < 145 or 205 <= tick < 220 or 280 <= tick < 300):
|
||||
assert not request and output == 0, (tick, decoded_steering)
|
||||
intent = decode_intent(published['aolIntentWire'].aolIntentWire)
|
||||
intent = decode_intent(published['aolIntentWire'].aolIntentWire) if aol else None
|
||||
row = {
|
||||
'tick': tick,
|
||||
'intent': intent.allowedLatch,
|
||||
'armed': intent.lateralArmed,
|
||||
'cardLatch': card.aol_card_intent.allowed_latch,
|
||||
'intent': bool(intent and intent.allowedLatch),
|
||||
'armed': bool(intent and intent.lateralArmed),
|
||||
'cardLatch': bool(card.aol_card_intent and card.aol_card_intent.allowed_latch),
|
||||
'lat': command.latActive,
|
||||
'long': command.longActive,
|
||||
'enabled': sd.enabled,
|
||||
@@ -194,25 +193,26 @@ def test_actual_full_engage_brake_reverse_temporary_recovery_mainoff(candidate,
|
||||
'order': order,
|
||||
'tx': tx,
|
||||
'steering': decoded_steering,
|
||||
'fault': str(driven['aolAxisState'].aolAxisState.faultReason),
|
||||
'fault': str(driven['aolAxisState'].aolAxisState.faultReason) if aol else 'unsupported',
|
||||
'events': [str(e.name) for e in driven['onroadEvents'].onroadEvents],
|
||||
}
|
||||
rows.append(row)
|
||||
assert not command.longActive, row
|
||||
if 10 <= tick < 30:
|
||||
assert not command.latActive and not intent.allowedLatch and row['permission'] == 0, row
|
||||
assert not command.latActive and not bool(intent and intent.allowedLatch) and row['permission'] == 0, row
|
||||
if 65 <= tick < 90:
|
||||
assert command.latActive and intent.allowedLatch == aol and row['permission'] == 1, row
|
||||
assert command.latActive and bool(intent and intent.allowedLatch) == aol and row['permission'] == int(aol), row
|
||||
if 125 <= tick < 145 or 205 <= tick < 220 or 280 <= tick < 300:
|
||||
assert command.latActive == aol and intent.allowedLatch == aol and row['permission'] == int(aol) and not row['controls'], row
|
||||
assert command.latActive == aol and bool(intent and intent.allowedLatch) == aol and row['permission'] == int(aol) and not row['controls'], row
|
||||
if 165 <= tick < 180:
|
||||
assert not command.latActive and not intent.allowedLatch and intent.lateralArmed == aol and row['cardLatch'] == aol and row['permission'] == 0, row
|
||||
assert (not command.latActive and not bool(intent and intent.allowedLatch) and
|
||||
bool(intent and intent.lateralArmed) == aol and row['cardLatch'] == aol and row['permission'] == 0), row
|
||||
if 240 <= tick < 255:
|
||||
assert not command.latActive and intent.lateralArmed == aol and row['cardLatch'] == aol and row['permission'] == 0, row
|
||||
assert not command.latActive and bool(intent and intent.lateralArmed) == aol and row['cardLatch'] == aol and row['permission'] == 0, row
|
||||
if 330 <= tick < 400:
|
||||
assert not command.latActive and not intent.allowedLatch, row
|
||||
assert not command.latActive and not bool(intent and intent.allowedLatch), row
|
||||
if tick >= 440:
|
||||
assert command.latActive and intent.allowedLatch == aol and row['permission'] == 1, row
|
||||
assert command.latActive and bool(intent and intent.allowedLatch) == aol and row['permission'] == int(aol), row
|
||||
assert len(active_emissions) >= 4, rows[65:90]
|
||||
assert dash_emissions >= 40
|
||||
finally:
|
||||
|
||||
@@ -99,6 +99,8 @@ AOL_LIFECYCLE_TARGETS = (
|
||||
"openpilot/starpilot/car/subaru/tests/test_avh_callers.py",
|
||||
"opendbc_repo/opendbc/safety/tests/test_subaru_avh_startup.py",
|
||||
"openpilot/starpilot/car/subaru/tests/test_angle_tune.py",
|
||||
"openpilot/starpilot/car/subaru/tests/test_forester_angle_callers.py",
|
||||
"opendbc_repo/opendbc/safety/tests/test_subaru_forester_angle.py",
|
||||
)
|
||||
|
||||
AOL_LIFECYCLE_SCOPE = {
|
||||
@@ -374,6 +376,8 @@ RELEASE_TARGETS = (
|
||||
"opendbc_repo/opendbc/safety/tests/test_subaru_sng.py",
|
||||
"opendbc_repo/opendbc/safety/tests/test_subaru_stop_start.py",
|
||||
"opendbc_repo/opendbc/safety/tests/test_subaru_avh_startup.py",
|
||||
"opendbc_repo/opendbc/safety/tests/test_subaru_forester_angle.py",
|
||||
"opendbc_repo/opendbc/safety/tests/test_subaru_receive_guards.py",
|
||||
)
|
||||
|
||||
# Vehicle tests import both the vendored car code and these host control owners.
|
||||
|
||||
Reference in New Issue
Block a user