mirror of
https://github.com/sunnypilot/sunnypilot.git
synced 2026-08-28 04:53:43 +08:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 04a8ad189e | |||
| e3cd22b765 |
+1
-1
Submodule opendbc_repo updated: ef0c35496b...06743dfb39
@@ -353,6 +353,7 @@ struct OnroadEventSP @0xda96579883444c35 {
|
||||
speedLimitPending @22;
|
||||
e2eChime @23;
|
||||
laneChangeRoadEdge @24;
|
||||
bigModelReady @25;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -382,7 +383,6 @@ struct CarControlSP @0xa5cd762cd951a455 {
|
||||
leadOne @2 :LeadData;
|
||||
leadTwo @3 :LeadData;
|
||||
intelligentCruiseButtonManagement @4 :IntelligentCruiseButtonManagement;
|
||||
fordLateralPath @5 :FordLateralPath;
|
||||
|
||||
struct Param {
|
||||
key @0 :Text;
|
||||
@@ -403,14 +403,6 @@ struct CarControlSP @0xa5cd762cd951a455 {
|
||||
}
|
||||
}
|
||||
|
||||
struct FordLateralPath {
|
||||
pathOffset @0 :Float32; # c0 [m]
|
||||
pathAngle @1 :Float32; # c1 [rad]
|
||||
curvature @2 :Float32; # c2 [1/m]
|
||||
curvatureRate @3 :Float32; # c3 [1/m^2]
|
||||
valid @4 :Bool;
|
||||
}
|
||||
|
||||
struct BackupManagerSP @0xf98d843bfd7004a3 {
|
||||
backupStatus @0 :Status;
|
||||
restoreStatus @1 :Status;
|
||||
|
||||
@@ -246,7 +246,6 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> keys = {
|
||||
|
||||
// mapd
|
||||
{"MapAdvisorySpeedLimit", {CLEAR_ON_ONROAD_TRANSITION, FLOAT}},
|
||||
{"Mapd_ClearCache", {CLEAR_ON_MANAGER_START, BOOL}},
|
||||
{"MapdVersion", {PERSISTENT, STRING}},
|
||||
{"MapSpeedLimit", {CLEAR_ON_ONROAD_TRANSITION, FLOAT, "0.0"}},
|
||||
{"NextMapSpeedLimit", {CLEAR_ON_ONROAD_TRANSITION, JSON}},
|
||||
|
||||
@@ -63,6 +63,5 @@ def convert_carControlSP(struct: capnp.lib.capnp._DynamicStructReader) -> struct
|
||||
struct_dataclass.intelligentCruiseButtonManagement = structs.IntelligentCruiseButtonManagement(
|
||||
**remove_deprecated(struct_dict.get('intelligentCruiseButtonManagement', {}))
|
||||
)
|
||||
struct_dataclass.fordLateralPath = structs.FordLateralPath(**remove_deprecated(struct_dict.get('fordLateralPath', {})))
|
||||
|
||||
return struct_dataclass
|
||||
|
||||
@@ -13,7 +13,6 @@ from openpilot.common.swaglog import cloudlog
|
||||
from opendbc.car.car_helpers import interfaces
|
||||
from opendbc.car.vehicle_model import VehicleModel
|
||||
from openpilot.selfdrive.controls.lib.drive_helpers import clip_curvature
|
||||
from openpilot.selfdrive.controls.lib.ford_path import FordPath, FordPathController
|
||||
from openpilot.selfdrive.controls.lib.latcontrol import LatControl
|
||||
from openpilot.selfdrive.controls.lib.latcontrol_pid import LatControlPID
|
||||
from openpilot.selfdrive.controls.lib.latcontrol_angle import LatControlAngle, STEER_ANGLE_SATURATION_THRESHOLD
|
||||
@@ -53,8 +52,6 @@ class Controls(ControlsExt):
|
||||
self.steer_limited_by_safety = False
|
||||
self.curvature = 0.0
|
||||
self.desired_curvature = 0.0
|
||||
self.ford_path_controller = FordPathController()
|
||||
self.ford_path = FordPath()
|
||||
|
||||
self.pose_calibrator = PoseCalibrator()
|
||||
self.calibrated_pose: Pose | None = None
|
||||
@@ -158,12 +155,6 @@ class Controls(ControlsExt):
|
||||
actuators.curvature = float(lateral_output)
|
||||
else:
|
||||
actuators.steeringAngleDeg = float(lateral_output)
|
||||
if self.CP.brand == "ford":
|
||||
self.ford_path = self.ford_path_controller.update(model_v2 if self.sm.valid['modelV2'] else None,
|
||||
self.desired_curvature, v_ego=CS.vEgo, active=CC.latActive,
|
||||
current_curvature=self.curvature, yaw_rate=CS.yawRate,
|
||||
actuator_delay=lat_delay)
|
||||
actuators.curvature = float(self.ford_path.curvature)
|
||||
# Ensure no NaNs/Infs
|
||||
for p in ACTUATOR_FIELDS:
|
||||
attr = getattr(actuators, p)
|
||||
|
||||
@@ -1,163 +0,0 @@
|
||||
from dataclasses import dataclass, fields
|
||||
import math
|
||||
|
||||
import numpy as np
|
||||
|
||||
|
||||
DBC_OFFSET = (-5.12, 5.11)
|
||||
DBC_ANGLE = (-0.5, 0.5235)
|
||||
DBC_CURVATURE = (-0.02, 0.02)
|
||||
DBC_CURVATURE_RATE = (-0.001024, 0.001023)
|
||||
|
||||
_PATH_OFFSET_DISTANCE = 7.0
|
||||
_PATH_MIN_LOOKAHEAD = 7.0
|
||||
_CURVATURE_RATE_HORIZONS = (3.5, 5.0, 7.0)
|
||||
_CENTERING_CURVATURE_BASEBAND = (0.003, 0.006)
|
||||
_TRACKING_ERROR_DEADZONE = 0.0005
|
||||
_TRACKING_ERROR_LIMIT = 0.012
|
||||
_PATH_RATES = (4.0, 1.0, math.inf, 0.002)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FordPath:
|
||||
valid: bool = False
|
||||
path_offset: float = 0.0
|
||||
path_angle: float = 0.0
|
||||
curvature: float = 0.0
|
||||
curvature_rate: float = 0.0
|
||||
|
||||
|
||||
def _finite(value: float) -> float:
|
||||
return float(value) if math.isfinite(value) else 0.0
|
||||
|
||||
|
||||
def _sample(distance: float, distances: list[float], values: list[float]) -> float:
|
||||
return float(np.interp(distance, distances, values))
|
||||
|
||||
|
||||
def _model_path(model) -> tuple[list[float], list[float], list[float]] | None:
|
||||
try:
|
||||
x = [float(value) for value in model.position.x]
|
||||
y = [float(value) for value in model.position.y]
|
||||
heading = [float(value) for value in model.orientation.z]
|
||||
except (AttributeError, TypeError, ValueError):
|
||||
return None
|
||||
if len(x) < 2 or len(x) != len(y) or len(x) != len(heading):
|
||||
return None
|
||||
if not all(math.isfinite(value) for values in (x, y, heading) for value in values):
|
||||
return None
|
||||
|
||||
distance = [0.0]
|
||||
for i in range(1, len(x)):
|
||||
distance.append(distance[-1] + math.hypot(x[i] - x[i - 1], y[i] - y[i - 1]))
|
||||
if distance[-1] <= 0.0:
|
||||
return None
|
||||
|
||||
unwrapped_heading = [heading[0]]
|
||||
for value in heading[1:]:
|
||||
delta = (value - unwrapped_heading[-1] + math.pi) % (2.0 * math.pi) - math.pi
|
||||
unwrapped_heading.append(unwrapped_heading[-1] + delta)
|
||||
return distance, y, unwrapped_heading
|
||||
|
||||
|
||||
def _curvature_rate(path: tuple[list[float], list[float], list[float]]) -> float:
|
||||
distance, _, heading = path
|
||||
rates = []
|
||||
for requested_horizon in _CURVATURE_RATE_HORIZONS:
|
||||
horizon = min(requested_horizon, distance[-1])
|
||||
start = _sample(0.0, distance, heading)
|
||||
midpoint = _sample(0.5 * horizon, distance, heading)
|
||||
end = _sample(horizon, distance, heading)
|
||||
rates.append(4.0 * (start - 2.0 * midpoint + end) / horizon ** 2)
|
||||
|
||||
magnitude = sum(abs(rate) for rate in rates)
|
||||
if magnitude == 0.0:
|
||||
return 0.0
|
||||
return sorted(rates)[1] * abs(sum(rates)) / magnitude
|
||||
|
||||
|
||||
def _curvature(path: tuple[list[float], list[float], list[float]]) -> float:
|
||||
distance, _, heading = path
|
||||
horizon = min(_PATH_MIN_LOOKAHEAD, distance[-1])
|
||||
return (_sample(horizon, distance, heading) - _sample(0.0, distance, heading)) / horizon
|
||||
|
||||
|
||||
def _encode_path(model, desired_curvature: float | None, v_ego: float, current_curvature: float | None) -> FordPath:
|
||||
path = _model_path(model)
|
||||
if path is None:
|
||||
return FordPath()
|
||||
|
||||
distance, offset, heading = path
|
||||
lookahead = max(_finite(v_ego), _PATH_MIN_LOOKAHEAD)
|
||||
path_offset = _sample(_PATH_OFFSET_DISTANCE, distance, offset)
|
||||
path_angle = _sample(lookahead, distance, heading)
|
||||
model_curvature = _curvature(path)
|
||||
action_curvature = model_curvature if desired_curvature is None else _finite(desired_curvature)
|
||||
requested_curvature = max((model_curvature, action_curvature), key=abs)
|
||||
maneuver_residual = requested_curvature - model_curvature
|
||||
path_offset += 0.5 * maneuver_residual * _PATH_OFFSET_DISTANCE ** 2
|
||||
path_angle += maneuver_residual * lookahead
|
||||
offset_curvature = 2.0 * path_offset / _PATH_OFFSET_DISTANCE ** 2
|
||||
angle_curvature = path_angle / lookahead
|
||||
geometry_demand = max(abs(offset_curvature), abs(angle_curvature))
|
||||
correction = 0.0
|
||||
if current_curvature is not None:
|
||||
target_curvature = requested_curvature
|
||||
tracking_error = target_curvature - _finite(current_curvature)
|
||||
correction = math.copysign(max(abs(tracking_error) - _TRACKING_ERROR_DEADZONE, 0.0), tracking_error)
|
||||
correction_limit = _TRACKING_ERROR_LIMIT
|
||||
if correction * target_curvature < 0.0:
|
||||
correction_limit = 0.5 * abs(target_curvature)
|
||||
correction = float(np.clip(correction, -correction_limit, correction_limit))
|
||||
path_offset += 0.5 * correction * _PATH_OFFSET_DISTANCE ** 2
|
||||
path_angle += correction * lookahead
|
||||
|
||||
control_demand = max(geometry_demand, abs(requested_curvature), abs(correction))
|
||||
maneuver_share = float(np.interp(control_demand, _CENTERING_CURVATURE_BASEBAND, (0.0, 1.0)))
|
||||
path_offset *= maneuver_share
|
||||
path_angle *= maneuver_share
|
||||
centering_curvature = model_curvature if model_curvature * requested_curvature >= 0.0 else 0.0
|
||||
|
||||
return FordPath(
|
||||
valid=True,
|
||||
path_offset=float(np.clip(path_offset, *DBC_OFFSET)),
|
||||
path_angle=float(np.clip(path_angle, *DBC_ANGLE)),
|
||||
curvature=float(np.clip(centering_curvature * (1.0 - maneuver_share), *DBC_CURVATURE)),
|
||||
curvature_rate=float(np.clip(_curvature_rate(path), *DBC_CURVATURE_RATE)),
|
||||
)
|
||||
|
||||
|
||||
class FordPathController:
|
||||
"""Convert the model path directly into one vehicle-independent Ford path command."""
|
||||
|
||||
def __init__(self, dt: float = 0.01):
|
||||
self.dt = dt
|
||||
self._last_path = FordPath(valid=True)
|
||||
|
||||
def reset(self) -> None:
|
||||
self._last_path = FordPath(valid=True)
|
||||
|
||||
def _limit(self, target: FordPath) -> FordPath:
|
||||
values = []
|
||||
for field, rate in zip(fields(FordPath)[1:], _PATH_RATES, strict=True):
|
||||
previous = getattr(self._last_path, field.name)
|
||||
value = getattr(target, field.name)
|
||||
values.append(float(np.clip(value, previous - rate * self.dt, previous + rate * self.dt)))
|
||||
self._last_path = FordPath(True, *values)
|
||||
return self._last_path
|
||||
|
||||
def update(self, model, desired_curvature: float | None = None, *, v_ego: float = 0.0, active: bool = True,
|
||||
current_curvature: float | None = None, yaw_rate: float = 0.0, actuator_delay: float = 0.0) -> FordPath:
|
||||
del yaw_rate, actuator_delay
|
||||
if not active:
|
||||
self.reset()
|
||||
return FordPath()
|
||||
if model is None:
|
||||
return self._limit(FordPath(valid=True))
|
||||
return self._limit(_encode_path(model, desired_curvature, v_ego, current_curvature))
|
||||
|
||||
|
||||
def encode_ford_path(model, t_prev: float, desired_curvature: float | None = None, *, v_ego: float = 0.0,
|
||||
current_curvature: float | None = None, yaw_rate: float = 0.0, actuator_delay: float = 0.0) -> FordPath:
|
||||
del t_prev, yaw_rate, actuator_delay
|
||||
return _encode_path(model, desired_curvature, v_ego, current_curvature)
|
||||
@@ -1,297 +0,0 @@
|
||||
import math
|
||||
from types import SimpleNamespace
|
||||
|
||||
import numpy as np
|
||||
|
||||
from openpilot.cereal import custom
|
||||
from openpilot.selfdrive.car.helpers import convert_carControlSP
|
||||
from openpilot.selfdrive.controls.lib.ford_path import DBC_CURVATURE, FordPathController, encode_ford_path
|
||||
|
||||
|
||||
def _path(curvature: float, curvature_rate: float = 0.0, speed: float = 8.0):
|
||||
t = np.linspace(0.0, 3.0, 61)
|
||||
distance = speed * t
|
||||
heading = curvature * distance + 0.5 * curvature_rate * distance ** 2
|
||||
x = np.zeros_like(distance)
|
||||
y = np.zeros_like(distance)
|
||||
for i in range(1, len(distance)):
|
||||
ds = distance[i] - distance[i - 1]
|
||||
average_heading = 0.5 * (heading[i] + heading[i - 1])
|
||||
x[i] = x[i - 1] + ds * math.cos(average_heading)
|
||||
y[i] = y[i - 1] + ds * math.sin(average_heading)
|
||||
return SimpleNamespace(
|
||||
position=SimpleNamespace(t=t.tolist(), x=x.tolist(), y=y.tolist()),
|
||||
orientation=SimpleNamespace(z=heading.tolist()),
|
||||
)
|
||||
|
||||
|
||||
def _offset_path(offset: float, speed: float = 8.0):
|
||||
t = np.linspace(0.0, 3.0, 61)
|
||||
distance = speed * t
|
||||
return SimpleNamespace(
|
||||
position=SimpleNamespace(t=t.tolist(), x=distance.tolist(), y=np.full_like(distance, offset).tolist()),
|
||||
orientation=SimpleNamespace(z=np.zeros_like(distance).tolist()),
|
||||
)
|
||||
|
||||
|
||||
def _equivalent_curvature(path, distance: float = 7.0) -> float:
|
||||
offset = path.path_offset + path.path_angle * distance + 0.5 * path.curvature * distance ** 2 + \
|
||||
path.curvature_rate * distance ** 3 / 6.0
|
||||
return 2.0 * offset / distance ** 2
|
||||
|
||||
|
||||
def test_gentle_arc_uses_forward_pose_fields():
|
||||
path = encode_ford_path(_path(0.008), 0.0, v_ego=8.0)
|
||||
|
||||
assert path.valid
|
||||
assert path.path_offset > 0.1
|
||||
assert path.path_angle > 0.03
|
||||
assert path.curvature == 0.0
|
||||
assert _equivalent_curvature(path) > 0.008
|
||||
assert abs(path.curvature_rate) < 1e-5
|
||||
|
||||
|
||||
def test_sunnypilot_path_message_round_trip():
|
||||
message = custom.CarControlSP.new_message()
|
||||
message.fordLateralPath.pathOffset = 0.3
|
||||
message.fordLateralPath.pathAngle = -0.2
|
||||
message.fordLateralPath.curvature = 0.008
|
||||
message.fordLateralPath.curvatureRate = -0.0004
|
||||
message.fordLateralPath.valid = True
|
||||
|
||||
path = convert_carControlSP(message.as_reader()).fordLateralPath
|
||||
assert np.isclose(path.pathOffset, 0.3)
|
||||
assert np.isclose(path.pathAngle, -0.2)
|
||||
assert np.isclose(path.curvature, 0.008)
|
||||
assert np.isclose(path.curvatureRate, -0.0004)
|
||||
assert path.valid
|
||||
|
||||
|
||||
def test_tight_arc_uses_signed_forward_pose_without_slow_c2():
|
||||
left = encode_ford_path(_path(0.04), 0.0, v_ego=8.0)
|
||||
right = encode_ford_path(_path(-0.04), 0.0, v_ego=8.0)
|
||||
|
||||
assert left.curvature == 0.0
|
||||
assert right.curvature == 0.0
|
||||
assert abs(left.curvature_rate) < 1e-4
|
||||
assert abs(right.curvature_rate) < 1e-4
|
||||
assert left.path_angle > 0.06
|
||||
assert right.path_angle < -0.06
|
||||
assert left.path_offset > 0.5
|
||||
assert right.path_offset < -0.5
|
||||
|
||||
|
||||
def test_c2_does_not_increase_while_tight_curve_unwinds():
|
||||
curvatures = (0.04, 0.018, 0.016, 0.014, 0.012, 0.010, 0.008, 0.006, 0.0)
|
||||
commands = [encode_ford_path(_path(curvature), 0.0, v_ego=8.0).curvature for curvature in curvatures]
|
||||
|
||||
assert np.all(np.diff(commands) <= 1e-9)
|
||||
|
||||
|
||||
def test_lateral_delay_does_not_change_the_reference_polynomial():
|
||||
early = FordPathController().update(_path(0.012, 0.0003), v_ego=10.0, current_curvature=-0.01, actuator_delay=0.1)
|
||||
late = FordPathController().update(_path(0.012, 0.0003), v_ego=10.0, current_curvature=-0.01, actuator_delay=0.9)
|
||||
|
||||
assert early == late
|
||||
|
||||
|
||||
def test_fresh_model_replaces_previous_path_without_hidden_state():
|
||||
controller = FordPathController(dt=1.0)
|
||||
initial = controller.update(_offset_path(0.4), v_ego=8.0)
|
||||
replanned = controller.update(_path(0.0), v_ego=8.0)
|
||||
|
||||
assert initial.path_offset > 0.39
|
||||
assert replanned == FordPathController().update(_path(0.0), v_ego=8.0)
|
||||
|
||||
|
||||
def test_s_turn_reverses_fast_fields_while_c2_is_bounded():
|
||||
controller = FordPathController(dt=0.05)
|
||||
controller.update(_path(0.04), v_ego=8.0, yaw_rate=0.0)
|
||||
controller.update(_path(0.04), v_ego=8.0, yaw_rate=0.16)
|
||||
|
||||
outputs = []
|
||||
for frame_id in range(5):
|
||||
model = _path(-0.02)
|
||||
model.frameId = frame_id + 1
|
||||
model.timestampEof = frame_id + 1
|
||||
outputs.append(controller.update(model, -0.02, v_ego=8.0, current_curvature=0.02, yaw_rate=0.32))
|
||||
|
||||
assert all(path.valid for path in outputs)
|
||||
assert all(DBC_CURVATURE[0] <= path.curvature <= DBC_CURVATURE[1] for path in outputs)
|
||||
assert all(path.curvature <= 0.0 for path in outputs)
|
||||
assert outputs[-1].path_angle < -0.03
|
||||
assert outputs[-1].path_offset < 0.0
|
||||
|
||||
|
||||
def test_reversal_does_not_add_software_persistence_to_centering_c2():
|
||||
controller = FordPathController()
|
||||
assert controller.update(_path(0.002), 0.0, v_ego=8.0).curvature > 0.0
|
||||
|
||||
reversing = controller.update(_path(-0.02), -0.02, v_ego=8.0)
|
||||
|
||||
assert reversing.curvature <= 0.0
|
||||
|
||||
|
||||
def test_requested_turn_is_not_cancelled_by_previous_path():
|
||||
controller = FordPathController(dt=1.0)
|
||||
previous = _path(-0.02, speed=3.0)
|
||||
previous.frameId = 1
|
||||
previous.timestampEof = 1
|
||||
controller.update(previous, -0.02, v_ego=3.0, current_curvature=-0.01)
|
||||
|
||||
requested = _path(0.02, speed=3.0)
|
||||
requested.frameId = 2
|
||||
requested.timestampEof = 2
|
||||
command = controller.update(requested, 0.02, v_ego=3.0, current_curvature=0.007)
|
||||
|
||||
assert command.path_offset >= 0.0
|
||||
assert command.path_angle >= 0.0
|
||||
assert command.curvature >= 0.0
|
||||
assert _equivalent_curvature(command) >= 0.02
|
||||
|
||||
|
||||
def test_short_low_speed_model_uses_available_path_endpoint():
|
||||
command = FordPathController().update(_path(0.02, speed=1.0), 0.02, v_ego=1.0, current_curvature=0.0)
|
||||
|
||||
assert command.valid
|
||||
assert command.path_offset > 0.0
|
||||
assert command.path_angle > 0.0
|
||||
|
||||
|
||||
def test_action_demand_exposes_forward_path_authority():
|
||||
command = FordPathController().update(_path(0.002), 0.0055, v_ego=8.0, current_curvature=0.0005)
|
||||
|
||||
assert _equivalent_curvature(command) >= 0.004
|
||||
|
||||
|
||||
def test_action_curvature_corrects_stale_opposing_model_at_low_speed():
|
||||
command = FordPathController().update(_path(-0.001, speed=1.0), 0.005, v_ego=1.0, current_curvature=0.001)
|
||||
|
||||
assert command.path_offset > 0.0
|
||||
assert command.path_angle > 0.0
|
||||
assert command.curvature >= 0.0
|
||||
assert _equivalent_curvature(command) >= 0.004
|
||||
|
||||
|
||||
def test_small_action_sign_noise_does_not_reverse_a_strong_model_path():
|
||||
command = FordPathController(dt=1.0).update(_path(0.04), -0.0005, v_ego=6.0, current_curvature=0.02)
|
||||
|
||||
assert command.path_offset > 0.0
|
||||
assert command.path_angle > 0.0
|
||||
assert _equivalent_curvature(command) > 0.02
|
||||
|
||||
|
||||
def test_measured_curvature_after_path_exit_commands_countersteer():
|
||||
command = FordPathController().update(_path(0.0), 0.0, v_ego=8.0, current_curvature=0.006)
|
||||
|
||||
assert command.curvature == 0.0
|
||||
assert command.path_offset < 0.0
|
||||
assert command.path_angle < 0.0
|
||||
|
||||
|
||||
def test_measured_curvature_does_not_cancel_a_modeled_arc():
|
||||
controller = FordPathController(dt=1.0)
|
||||
command = controller.update(_path(0.004), 0.003, v_ego=8.0, current_curvature=0.012)
|
||||
tracking = FordPathController(dt=1.0).update(_path(0.004), 0.003, v_ego=8.0, current_curvature=0.004)
|
||||
|
||||
assert command.path_offset > 0.0
|
||||
assert command.path_angle > 0.0
|
||||
assert command.path_angle < tracking.path_angle
|
||||
assert _equivalent_curvature(command) > 0.0
|
||||
|
||||
|
||||
def test_model_reversal_suppresses_old_c2_and_countersteers():
|
||||
reversing = FordPathController().update(_path(-0.02), -0.0005, v_ego=8.0, current_curvature=0.01)
|
||||
|
||||
assert reversing.curvature <= 0.0
|
||||
assert reversing.path_angle < 0.0
|
||||
|
||||
|
||||
def test_reversal_noise_band_is_continuous():
|
||||
inside = FordPathController(dt=1.0).update(_path(0.02), -0.000099, v_ego=8.0, current_curvature=0.01)
|
||||
outside = FordPathController(dt=1.0).update(_path(0.02), -0.000101, v_ego=8.0, current_curvature=0.01)
|
||||
|
||||
assert abs(outside.path_angle - inside.path_angle) < 0.005
|
||||
|
||||
|
||||
def test_yaw_rate_does_not_create_a_second_path_source():
|
||||
model = _offset_path(0.25)
|
||||
controller = FordPathController(dt=1.0)
|
||||
before = controller.update(model, v_ego=8.0, current_curvature=0.0)
|
||||
after = controller.update(model, v_ego=8.0, yaw_rate=0.16)
|
||||
|
||||
assert before.valid and after.valid
|
||||
assert after == before
|
||||
|
||||
|
||||
def test_invalid_ford_yaw_rate_does_not_rotate_the_reference():
|
||||
model = _offset_path(0.25)
|
||||
valid = FordPathController(dt=0.01)
|
||||
invalid = FordPathController(dt=0.01)
|
||||
valid.update(model, v_ego=8.0)
|
||||
invalid.update(model, v_ego=8.0)
|
||||
|
||||
expected = valid.update(model, v_ego=8.0, yaw_rate=0.0)
|
||||
sentinel = invalid.update(model, v_ego=8.0, yaw_rate=6.6066)
|
||||
|
||||
assert sentinel == expected
|
||||
|
||||
|
||||
def test_curvature_error_increases_forward_pose_command_while_behind():
|
||||
behind = FordPathController(dt=1.0).update(_path(0.008), 0.008, v_ego=15.0, current_curvature=0.0)
|
||||
tracking = FordPathController(dt=1.0).update(_path(0.008), 0.008, v_ego=15.0, current_curvature=0.008)
|
||||
|
||||
assert behind.path_offset > tracking.path_offset + 0.01
|
||||
assert behind.path_angle > tracking.path_angle + 0.015
|
||||
assert np.isclose(behind.curvature, tracking.curvature)
|
||||
assert np.isclose(behind.curvature_rate, tracking.curvature_rate)
|
||||
|
||||
|
||||
def test_rolling_arc_stays_active_while_vehicle_unwinds():
|
||||
controller = FordPathController()
|
||||
controller.update(_path(0.02), 0.02, v_ego=15.0, current_curvature=0.02, yaw_rate=0.3)
|
||||
outputs = [controller.update(_path(0.02), 0.003, v_ego=15.0, current_curvature=0.01, yaw_rate=0.15) for _ in range(4)]
|
||||
unwinding = outputs[-1]
|
||||
|
||||
assert 0.0 <= unwinding.curvature <= 0.003
|
||||
assert unwinding.path_angle > 0.03
|
||||
|
||||
|
||||
def test_geometric_c2_remains_active_for_centering():
|
||||
centering = FordPathController(dt=1.0).update(_path(0.002), 0.0, v_ego=15.0, current_curvature=0.0)
|
||||
|
||||
assert centering.curvature > 0.001
|
||||
assert abs(centering.path_offset) < 1e-9
|
||||
assert abs(centering.path_angle) < 1e-9
|
||||
|
||||
|
||||
def test_tight_turn_from_stop_builds_bounded_forward_pose_authority():
|
||||
controller = FordPathController()
|
||||
outputs = [controller.update(_path(0.04), 0.04, v_ego=0.0, current_curvature=0.0) for _ in range(20)]
|
||||
path = outputs[-1]
|
||||
|
||||
assert path.curvature == 0.0
|
||||
assert path.path_offset > 0.7
|
||||
assert path.path_angle > 0.15
|
||||
assert np.max(np.abs(np.diff([output.path_offset for output in outputs]))) <= 0.04 + 1e-9
|
||||
assert np.max(np.abs(np.diff([output.path_angle for output in outputs]))) <= 0.01 + 1e-9
|
||||
|
||||
|
||||
def test_curvature_feedback_is_bounded_for_bad_measurement():
|
||||
bounded = FordPathController().update(_path(0.008), 0.008, v_ego=15.0, current_curvature=-0.02)
|
||||
corrupted = FordPathController().update(_path(0.008), 0.008, v_ego=15.0, current_curvature=-1.0)
|
||||
|
||||
assert np.isclose(corrupted.path_angle, bounded.path_angle)
|
||||
|
||||
|
||||
def test_invalid_model_ramps_pose_to_zero_while_remaining_in_extended_mode():
|
||||
controller = FordPathController()
|
||||
for _ in range(10):
|
||||
active = controller.update(_offset_path(0.4), v_ego=12.0)
|
||||
missing = controller.update(None, v_ego=12.0)
|
||||
|
||||
assert active.path_offset > 0.0
|
||||
assert missing.valid
|
||||
assert np.isclose(active.path_offset - missing.path_offset, 0.04)
|
||||
assert missing.curvature == 0.0
|
||||
assert not controller.update(_path(0.0), v_ego=12.0, active=False).valid
|
||||
@@ -198,6 +198,7 @@ class SelfdriveD(CruiseHelper):
|
||||
loading = self.params.get_bool("UsbGpuLoading")
|
||||
if self.big_model_loading and not loading:
|
||||
self.big_model_ready_t = time.monotonic()
|
||||
self.events_sp.add(custom.OnroadEventSP.EventName.bigModelReady)
|
||||
self.big_model_loading = loading
|
||||
if self.big_model_loading:
|
||||
self.events.add(EventName.bigModelLoading)
|
||||
|
||||
@@ -8,6 +8,7 @@ import datetime
|
||||
import os
|
||||
import platform
|
||||
import requests
|
||||
import shutil
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from time import monotonic
|
||||
@@ -74,12 +75,22 @@ class OSMLayout(Widget):
|
||||
def _update_map_size(self):
|
||||
threading.Thread(target=self.calculate_size, daemon=True).start()
|
||||
|
||||
def _on_confirm_delete_maps(self):
|
||||
ui_state.params.put_bool("Mapd_ClearCache", True)
|
||||
def _do_delete_maps(self):
|
||||
if MAP_PATH.exists():
|
||||
shutil.rmtree(MAP_PATH)
|
||||
|
||||
for param in ("OsmDownloadedDate", "OsmLocal", "OsmLocationName", "OsmLocationTitle", "OsmStateName", "OsmStateTitle"):
|
||||
ui_state.params.remove(param)
|
||||
|
||||
self._delete_maps_btn.action_item.set_enabled(True)
|
||||
self._delete_maps_btn.action_item.set_text(tr("DELETE"))
|
||||
self._update_map_size()
|
||||
|
||||
def _on_confirm_delete_maps(self):
|
||||
self._delete_maps_btn.action_item.set_enabled(False)
|
||||
self._delete_maps_btn.action_item.set_text("DELETING...")
|
||||
threading.Thread(target=self._do_delete_maps).start()
|
||||
|
||||
def _delete_maps(self):
|
||||
self._show_confirm(tr("This will delete ALL downloaded maps\n\nAre you sure you want to delete all maps?"),
|
||||
tr("Yes, delete all maps"), self._on_confirm_delete_maps)
|
||||
|
||||
@@ -55,19 +55,6 @@ def cleanup_old_osm_data(files_to_remove: list[str]) -> None:
|
||||
shutil.rmtree(file, ignore_errors=False)
|
||||
|
||||
|
||||
def clear_downloaded_maps() -> None:
|
||||
"""Deletes downloaded OSM map data and resets params."""
|
||||
path = f"{Paths.mapd_root()}/offline"
|
||||
if os.path.exists(path):
|
||||
shutil.rmtree(path, ignore_errors=True)
|
||||
|
||||
for param in ("OsmDownloadedDate", "OsmLocal", "OsmLocationName", "OsmLocationTitle",
|
||||
"OsmStateName", "OsmStateTitle"):
|
||||
params.remove(param)
|
||||
|
||||
cloudlog.info("mapd: downloaded maps cleared")
|
||||
|
||||
|
||||
def request_refresh_osm_location_data(nations: list[str], states: list[str] | None = None) -> None:
|
||||
params.put("OsmDownloadedDate", str(datetime.now().timestamp()), block=True)
|
||||
params.put_bool("OsmDbUpdatesCheck", False, block=True)
|
||||
@@ -144,10 +131,6 @@ def main_thread():
|
||||
show_alert = bool(get_files_for_cleanup() and params.get_bool("OsmLocal"))
|
||||
set_offroad_alert("Offroad_OSMUpdateRequired", show_alert, "This alert will be cleared when new maps are downloaded.")
|
||||
|
||||
if params.get("Mapd_ClearCache"):
|
||||
clear_downloaded_maps()
|
||||
params.remove("Mapd_ClearCache")
|
||||
|
||||
update_osm_db()
|
||||
live_map_sp.tick()
|
||||
rk.keep_time()
|
||||
|
||||
@@ -104,14 +104,6 @@ class ControlsExt(ModelStateBase):
|
||||
CC_SP.intelligentCruiseButtonManagement.sendButton = icbm_src.sendButton
|
||||
CC_SP.intelligentCruiseButtonManagement.vTarget = icbm_src.vTarget
|
||||
|
||||
ford_path = getattr(self, 'ford_path', None)
|
||||
if ford_path is not None:
|
||||
CC_SP.fordLateralPath.valid = ford_path.valid
|
||||
CC_SP.fordLateralPath.pathOffset = ford_path.path_offset
|
||||
CC_SP.fordLateralPath.pathAngle = ford_path.path_angle
|
||||
CC_SP.fordLateralPath.curvature = ford_path.curvature
|
||||
CC_SP.fordLateralPath.curvatureRate = ford_path.curvature_rate
|
||||
|
||||
return CC_SP
|
||||
|
||||
@staticmethod
|
||||
|
||||
@@ -252,4 +252,12 @@ EVENTS_SP: dict[int, dict[str, Alert | AlertCallbackType]] = {
|
||||
AlertStatus.userPrompt, AlertSize.small,
|
||||
Priority.LOW, VisualAlert.none, AudibleAlert.prompt, 0.1),
|
||||
},
|
||||
|
||||
EventNameSP.bigModelReady: {
|
||||
ET.PERMANENT: Alert(
|
||||
"Big Model Ready",
|
||||
"",
|
||||
AlertStatus.normal, AlertSize.small,
|
||||
Priority.LOW, VisualAlert.none, AudibleAlert.prompt, 2.),
|
||||
},
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user