mirror of
https://github.com/infiniteCable2/openpilot.git
synced 2026-09-05 07:43:41 +08:00
Update disturbance_controller.py retry
This commit is contained in:
@@ -11,57 +11,56 @@ ALPHA_MAX = 0.4
|
||||
|
||||
class DisturbanceController:
|
||||
def __init__(self, CP):
|
||||
self.lowpass_filtered = 0.0
|
||||
self.lowpass_filtered_prev = 0.0
|
||||
self.alpha_prev = ALPHA_MIN
|
||||
self.desired_curvature_prev = 0.0
|
||||
self.pid = PIDController(1, 0, k_f=0, pos_limit=MAX_CURVATURE, neg_limit=-MAX_CURVATURE)
|
||||
self.reaction_hist = deque([0.0], maxlen=int(round(CP.steerActuatorDelay / DT_CTRL)) + 1)
|
||||
self.length_des_curv_hist = int(round(CP.steerActuatorDelay / DT_CTRL)) + 1
|
||||
self.desired_curvature_hist = deque([0.0], maxlen=self.length_des_curv_hist)
|
||||
|
||||
def reset(self):
|
||||
self.lowpass_filtered = 0.0
|
||||
self.lowpass_filtered_prev = 0.0
|
||||
self.alpha_prev = ALPHA_MIN
|
||||
self.desired_curvature_prev = 0.0
|
||||
self.pid.reset()
|
||||
self.reaction_hist.clear()
|
||||
self.reaction_hist.append(0.0)
|
||||
self.desired_curvature_hist.clear()
|
||||
self.desired_curvature_hist.append(0.0)
|
||||
|
||||
def compute_dynamic_alpha(self, desired_curvature, dt=DT_CTRL, A=0.02, n=2.0, beta=3.0, k=2.0):
|
||||
d_desired = abs(desired_curvature - self.desired_curvature_prev) / dt
|
||||
def compute_dynamic_alpha(self, desired_curvature_hist, dt=DT_CTRL, A=0.02, n=2.0, beta=3.0, k=2.0):
|
||||
if len(desired_curvature_hist) < self.length_des_curv_hist:
|
||||
return ALPHA_MAX
|
||||
d_desired = abs(desired_curvature_hist[0] - desired_curvature_hist[1]) / dt
|
||||
alpha_reactive = d_desired**n / (k * A) if A > 0 else 0.0
|
||||
alpha = np.clip(self.alpha_prev * np.exp(-beta * dt) + alpha_reactive, ALPHA_MIN, ALPHA_MAX)
|
||||
self.alpha_prev = alpha
|
||||
self.desired_curvature_prev = desired_curvature
|
||||
return alpha
|
||||
|
||||
def lowpass_filter(self, current_value, alpha):
|
||||
def lowpass_filter(self, current_value, lowpass_filtered_prev, alpha):
|
||||
alpha = min(alpha, ALPHA_MAX)
|
||||
if alpha >= ALPHA_MAX * 0.9:
|
||||
reset_factor = (alpha - ALPHA_MIN) / (ALPHA_MAX - ALPHA_MIN)
|
||||
self.lowpass_filtered = (1 - reset_factor) * self.lowpass_filtered + reset_factor * current_value
|
||||
lowpass_filtered_prev = (1 - reset_factor) * lowpass_filtered_prev + reset_factor * current_value
|
||||
else:
|
||||
self.lowpass_filtered = (1 - alpha) * self.lowpass_filtered + alpha * current_value
|
||||
return self.lowpass_filtered
|
||||
lowpass_filtered_prev = (1 - alpha) * lowpass_filtered_prev + alpha * current_value
|
||||
return lowpass_filtered_prev
|
||||
|
||||
def highpass_filter(self, current_value, lowpass_value):
|
||||
return current_value - lowpass_value
|
||||
|
||||
def compensate(self, CS, VM, params, calibrated_pose, desired_curvature, curvature_3dof):
|
||||
def compensate(self, CS, VM, params, calibrated_pose, desired_curvature):
|
||||
if calibrated_pose is None or CS.vEgo < 0.1:
|
||||
return desired_curvature
|
||||
|
||||
steering_angle_without_offset = math.radians(CS.steeringAngleDeg - params.angleOffsetDeg)
|
||||
#actual_curvature = -VM.calc_curvature_3dof(calibrated_pose.acceleration.y, calibrated_pose.acceleration.x,
|
||||
# calibrated_pose.angular_velocity.yaw, CS.vEgo, steering_angle_without_offset,
|
||||
# 0.)
|
||||
actual_curvature = curvature_3dof
|
||||
actual_curvature = -VM.calc_curvature_3dof(calibrated_pose.acceleration.y, calibrated_pose.acceleration.x,
|
||||
calibrated_pose.angular_velocity.yaw, CS.vEgo, steering_angle_without_offset, 0.)
|
||||
|
||||
alpha = self.compute_dynamic_alpha(desired_curvature)
|
||||
reaction = self.lowpass_filter(actual_curvature, alpha)
|
||||
self.reaction_hist.append(reaction)
|
||||
reaction_delayed = self.reaction_hist[0]
|
||||
alpha = self.compute_dynamic_alpha(self.desired_curvature_hist)
|
||||
reaction = self.lowpass_filter(actual_curvature, self.lowpass_filtered_prev, alpha)
|
||||
disturbance = self.highpass_filter(actual_curvature, reaction)
|
||||
|
||||
error = desired_curvature + disturbance
|
||||
error = disturbance
|
||||
output_curvature = self.pid.update(error, feedforward=desired_curvature, speed=CS.vEgo)
|
||||
|
||||
self.desired_curvature_hist.append(desired_curvature)
|
||||
self.alpha_prev = alpha
|
||||
self.lowpass_filtered_prev = reaction
|
||||
|
||||
return float(output_curvature)
|
||||
|
||||
Reference in New Issue
Block a user