mirror of
https://github.com/dragonpilot/dragonpilot.git
synced 2026-07-17 02:42:10 +08:00
cleanup panda types everywhere (#26574)
* cleanup panda types everywhere * one more
This commit is contained in:
@@ -275,12 +275,11 @@ class Controls:
|
||||
# self.events.add(EventName.highCpuUsage)
|
||||
|
||||
# Alert if fan isn't spinning for 5 seconds
|
||||
if self.sm['peripheralState'].pandaType == PandaType.dos:
|
||||
if self.sm['peripheralState'].fanSpeedRpm == 0 and self.sm['deviceState'].fanSpeedPercentDesired > 50:
|
||||
if (self.sm.frame - self.last_functional_fan_frame) * DT_CTRL > 5.0:
|
||||
self.events.add(EventName.fanMalfunction)
|
||||
else:
|
||||
self.last_functional_fan_frame = self.sm.frame
|
||||
if self.sm['peripheralState'].fanSpeedRpm == 0 and self.sm['deviceState'].fanSpeedPercentDesired > 50:
|
||||
if (self.sm.frame - self.last_functional_fan_frame) * DT_CTRL > 5.0:
|
||||
self.events.add(EventName.fanMalfunction)
|
||||
else:
|
||||
self.last_functional_fan_frame = self.sm.frame
|
||||
|
||||
# Handle calibration status
|
||||
cal_status = self.sm['liveCalibration'].calStatus
|
||||
|
||||
@@ -250,10 +250,9 @@ def calibration_incomplete_alert(CP: car.CarParams, CS: car.CarState, sm: messag
|
||||
|
||||
|
||||
def no_gps_alert(CP: car.CarParams, CS: car.CarState, sm: messaging.SubMaster, metric: bool, soft_disable_time: int) -> Alert:
|
||||
gps_integrated = sm['peripheralState'].pandaType in (log.PandaState.PandaType.uno, log.PandaState.PandaType.dos)
|
||||
return Alert(
|
||||
"Poor GPS reception",
|
||||
"Hardware malfunctioning if sky is visible" if gps_integrated else "Check GPS antenna placement",
|
||||
"Hardware malfunctioning if sky is visible",
|
||||
AlertStatus.normal, AlertSize.mid,
|
||||
Priority.LOWER, VisualAlert.none, AudibleAlert.none, .2, creation_delay=300.)
|
||||
|
||||
|
||||
@@ -68,7 +68,7 @@ if __name__ == "__main__":
|
||||
CP = None
|
||||
for msg in lr:
|
||||
if msg.which() == "pandaStates":
|
||||
if msg.pandaStates[0].pandaType not in ('uno', 'blackPanda', 'dos'):
|
||||
if msg.pandaStates[0].pandaType in ('unknown', 'whitePanda', 'greyPanda', 'pedal'):
|
||||
print("wrong panda type")
|
||||
break
|
||||
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
import threading
|
||||
from typing import Optional
|
||||
|
||||
from cereal import log
|
||||
from common.params import Params, put_nonblocking
|
||||
from common.realtime import sec_since_boot
|
||||
from system.hardware import HARDWARE
|
||||
@@ -39,12 +38,12 @@ class PowerMonitoring:
|
||||
self.car_battery_capacity_uWh = max((CAR_BATTERY_CAPACITY_uWh / 10), int(car_battery_capacity_uWh))
|
||||
|
||||
# Calculation tick
|
||||
def calculate(self, peripheralState, ignition):
|
||||
def calculate(self, voltage: Optional[int], ignition: bool):
|
||||
try:
|
||||
now = sec_since_boot()
|
||||
|
||||
# If peripheralState is None, we're probably not in a car, so we don't care
|
||||
if peripheralState is None or peripheralState.pandaType == log.PandaState.PandaType.unknown:
|
||||
if voltage is None:
|
||||
with self.integration_lock:
|
||||
self.last_measurement_time = None
|
||||
self.next_pulsed_measurement_time = None
|
||||
@@ -52,8 +51,8 @@ class PowerMonitoring:
|
||||
return
|
||||
|
||||
# Low-pass battery voltage
|
||||
self.car_voltage_instant_mV = peripheralState.voltage
|
||||
self.car_voltage_mV = ((peripheralState.voltage * CAR_VOLTAGE_LOW_PASS_K) + (self.car_voltage_mV * (1 - CAR_VOLTAGE_LOW_PASS_K)))
|
||||
self.car_voltage_instant_mV = voltage
|
||||
self.car_voltage_mV = ((voltage * CAR_VOLTAGE_LOW_PASS_K) + (self.car_voltage_mV * (1 - CAR_VOLTAGE_LOW_PASS_K)))
|
||||
statlog.gauge("car_voltage", self.car_voltage_mV / 1e3)
|
||||
|
||||
# Cap the car battery power and save it in a param every 10-ish seconds
|
||||
|
||||
@@ -1,10 +1,7 @@
|
||||
#!/usr/bin/env python3
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
from parameterized import parameterized
|
||||
|
||||
from cereal import log
|
||||
import cereal.messaging as messaging
|
||||
from common.params import Params
|
||||
params = Params()
|
||||
|
||||
@@ -21,7 +18,8 @@ with patch("common.realtime.sec_since_boot", new=mock_sec_since_boot):
|
||||
CAR_CHARGING_RATE_W, VBATT_PAUSE_CHARGING
|
||||
|
||||
TEST_DURATION_S = 50
|
||||
ALL_PANDA_TYPES = [(log.PandaState.PandaType.dos,)]
|
||||
GOOD_VOLTAGE = 12 * 1e3
|
||||
VOLTAGE_BELOW_PAUSE_CHARGING = (VBATT_PAUSE_CHARGING - 1) * 1e3
|
||||
|
||||
def pm_patch(name, value, constant=False):
|
||||
if constant:
|
||||
@@ -34,12 +32,6 @@ class TestPowerMonitoring(unittest.TestCase):
|
||||
params.remove("CarBatteryCapacity")
|
||||
params.remove("DisablePowerDown")
|
||||
|
||||
def mock_peripheralState(self, hw_type, car_voltage=12):
|
||||
ps = messaging.new_message('peripheralState').peripheralState
|
||||
ps.pandaType = hw_type
|
||||
ps.voltage = car_voltage * 1e3
|
||||
return ps
|
||||
|
||||
# Test to see that it doesn't do anything when pandaState is None
|
||||
def test_pandaState_present(self):
|
||||
pm = PowerMonitoring()
|
||||
@@ -49,75 +41,68 @@ class TestPowerMonitoring(unittest.TestCase):
|
||||
self.assertEqual(pm.get_car_battery_capacity(), (CAR_BATTERY_CAPACITY_uWh / 10))
|
||||
|
||||
# Test to see that it doesn't integrate offroad when ignition is True
|
||||
@parameterized.expand(ALL_PANDA_TYPES)
|
||||
def test_offroad_ignition(self, hw_type):
|
||||
def test_offroad_ignition(self):
|
||||
pm = PowerMonitoring()
|
||||
for _ in range(10):
|
||||
pm.calculate(self.mock_peripheralState(hw_type), True)
|
||||
pm.calculate(GOOD_VOLTAGE, True)
|
||||
self.assertEqual(pm.get_power_used(), 0)
|
||||
|
||||
# Test to see that it integrates with discharging battery
|
||||
@parameterized.expand(ALL_PANDA_TYPES)
|
||||
def test_offroad_integration_discharging(self, hw_type):
|
||||
def test_offroad_integration_discharging(self):
|
||||
POWER_DRAW = 4
|
||||
with pm_patch("HARDWARE.get_current_power_draw", POWER_DRAW):
|
||||
pm = PowerMonitoring()
|
||||
for _ in range(TEST_DURATION_S + 1):
|
||||
pm.calculate(self.mock_peripheralState(hw_type), False)
|
||||
pm.calculate(GOOD_VOLTAGE, False)
|
||||
expected_power_usage = ((TEST_DURATION_S/3600) * POWER_DRAW * 1e6)
|
||||
self.assertLess(abs(pm.get_power_used() - expected_power_usage), 10)
|
||||
|
||||
# Test to check positive integration of car_battery_capacity
|
||||
@parameterized.expand(ALL_PANDA_TYPES)
|
||||
def test_car_battery_integration_onroad(self, hw_type):
|
||||
def test_car_battery_integration_onroad(self):
|
||||
POWER_DRAW = 4
|
||||
with pm_patch("HARDWARE.get_current_power_draw", POWER_DRAW):
|
||||
pm = PowerMonitoring()
|
||||
pm.car_battery_capacity_uWh = 0
|
||||
for _ in range(TEST_DURATION_S + 1):
|
||||
pm.calculate(self.mock_peripheralState(hw_type), True)
|
||||
pm.calculate(GOOD_VOLTAGE, True)
|
||||
expected_capacity = ((TEST_DURATION_S/3600) * CAR_CHARGING_RATE_W * 1e6)
|
||||
self.assertLess(abs(pm.get_car_battery_capacity() - expected_capacity), 10)
|
||||
|
||||
# Test to check positive integration upper limit
|
||||
@parameterized.expand(ALL_PANDA_TYPES)
|
||||
def test_car_battery_integration_upper_limit(self, hw_type):
|
||||
def test_car_battery_integration_upper_limit(self):
|
||||
POWER_DRAW = 4
|
||||
with pm_patch("HARDWARE.get_current_power_draw", POWER_DRAW):
|
||||
pm = PowerMonitoring()
|
||||
pm.car_battery_capacity_uWh = CAR_BATTERY_CAPACITY_uWh - 1000
|
||||
for _ in range(TEST_DURATION_S + 1):
|
||||
pm.calculate(self.mock_peripheralState(hw_type), True)
|
||||
pm.calculate(GOOD_VOLTAGE, True)
|
||||
estimated_capacity = CAR_BATTERY_CAPACITY_uWh + (CAR_CHARGING_RATE_W / 3600 * 1e6)
|
||||
self.assertLess(abs(pm.get_car_battery_capacity() - estimated_capacity), 10)
|
||||
|
||||
# Test to check negative integration of car_battery_capacity
|
||||
@parameterized.expand(ALL_PANDA_TYPES)
|
||||
def test_car_battery_integration_offroad(self, hw_type):
|
||||
def test_car_battery_integration_offroad(self):
|
||||
POWER_DRAW = 4
|
||||
with pm_patch("HARDWARE.get_current_power_draw", POWER_DRAW):
|
||||
pm = PowerMonitoring()
|
||||
pm.car_battery_capacity_uWh = CAR_BATTERY_CAPACITY_uWh
|
||||
for _ in range(TEST_DURATION_S + 1):
|
||||
pm.calculate(self.mock_peripheralState(hw_type), False)
|
||||
pm.calculate(GOOD_VOLTAGE, False)
|
||||
expected_capacity = CAR_BATTERY_CAPACITY_uWh - ((TEST_DURATION_S/3600) * POWER_DRAW * 1e6)
|
||||
self.assertLess(abs(pm.get_car_battery_capacity() - expected_capacity), 10)
|
||||
|
||||
# Test to check negative integration lower limit
|
||||
@parameterized.expand(ALL_PANDA_TYPES)
|
||||
def test_car_battery_integration_lower_limit(self, hw_type):
|
||||
def test_car_battery_integration_lower_limit(self):
|
||||
POWER_DRAW = 4
|
||||
with pm_patch("HARDWARE.get_current_power_draw", POWER_DRAW):
|
||||
pm = PowerMonitoring()
|
||||
pm.car_battery_capacity_uWh = 1000
|
||||
for _ in range(TEST_DURATION_S + 1):
|
||||
pm.calculate(self.mock_peripheralState(hw_type), False)
|
||||
pm.calculate(GOOD_VOLTAGE, False)
|
||||
estimated_capacity = 0 - ((1/3600) * POWER_DRAW * 1e6)
|
||||
self.assertLess(abs(pm.get_car_battery_capacity() - estimated_capacity), 10)
|
||||
|
||||
# Test to check policy of stopping charging after MAX_TIME_OFFROAD_S
|
||||
@parameterized.expand(ALL_PANDA_TYPES)
|
||||
def test_max_time_offroad(self, hw_type):
|
||||
def test_max_time_offroad(self):
|
||||
MOCKED_MAX_OFFROAD_TIME = 3600
|
||||
POWER_DRAW = 0 # To stop shutting down for other reasons
|
||||
with pm_patch("MAX_TIME_OFFROAD_S", MOCKED_MAX_OFFROAD_TIME, constant=True), pm_patch("HARDWARE.get_current_power_draw", POWER_DRAW):
|
||||
@@ -125,25 +110,22 @@ class TestPowerMonitoring(unittest.TestCase):
|
||||
pm.car_battery_capacity_uWh = CAR_BATTERY_CAPACITY_uWh
|
||||
start_time = ssb
|
||||
ignition = False
|
||||
peripheralState = self.mock_peripheralState(hw_type)
|
||||
while ssb <= start_time + MOCKED_MAX_OFFROAD_TIME:
|
||||
pm.calculate(peripheralState, ignition)
|
||||
pm.calculate(GOOD_VOLTAGE, ignition)
|
||||
if (ssb - start_time) % 1000 == 0 and ssb < start_time + MOCKED_MAX_OFFROAD_TIME:
|
||||
self.assertFalse(pm.should_shutdown(ignition, True, start_time, False))
|
||||
self.assertTrue(pm.should_shutdown(ignition, True, start_time, False))
|
||||
|
||||
# Test to check policy of stopping charging when the car voltage is too low
|
||||
@parameterized.expand(ALL_PANDA_TYPES)
|
||||
def test_car_voltage(self, hw_type):
|
||||
def test_car_voltage(self):
|
||||
POWER_DRAW = 0 # To stop shutting down for other reasons
|
||||
TEST_TIME = 100
|
||||
with pm_patch("HARDWARE.get_current_power_draw", POWER_DRAW):
|
||||
pm = PowerMonitoring()
|
||||
pm.car_battery_capacity_uWh = CAR_BATTERY_CAPACITY_uWh
|
||||
ignition = False
|
||||
peripheralState = self.mock_peripheralState(hw_type, car_voltage=(VBATT_PAUSE_CHARGING - 1))
|
||||
for i in range(TEST_TIME):
|
||||
pm.calculate(peripheralState, ignition)
|
||||
pm.calculate(VOLTAGE_BELOW_PAUSE_CHARGING, ignition)
|
||||
if i % 10 == 0:
|
||||
self.assertEqual(pm.should_shutdown(ignition, True, ssb, True), (pm.car_voltage_mV < VBATT_PAUSE_CHARGING*1e3))
|
||||
self.assertTrue(pm.should_shutdown(ignition, True, ssb, True))
|
||||
@@ -157,9 +139,8 @@ class TestPowerMonitoring(unittest.TestCase):
|
||||
pm = PowerMonitoring()
|
||||
pm.car_battery_capacity_uWh = CAR_BATTERY_CAPACITY_uWh
|
||||
ignition = False
|
||||
peripheralState = self.mock_peripheralState(log.PandaState.PandaType.uno, car_voltage=(VBATT_PAUSE_CHARGING - 1))
|
||||
for i in range(TEST_TIME):
|
||||
pm.calculate(peripheralState, ignition)
|
||||
pm.calculate(VOLTAGE_BELOW_PAUSE_CHARGING, ignition)
|
||||
if i % 10 == 0:
|
||||
self.assertFalse(pm.should_shutdown(ignition, True, ssb, False))
|
||||
self.assertFalse(pm.should_shutdown(ignition, True, ssb, False))
|
||||
@@ -172,9 +153,8 @@ class TestPowerMonitoring(unittest.TestCase):
|
||||
pm = PowerMonitoring()
|
||||
pm.car_battery_capacity_uWh = CAR_BATTERY_CAPACITY_uWh
|
||||
ignition = True
|
||||
peripheralState = self.mock_peripheralState(log.PandaState.PandaType.uno, car_voltage=(VBATT_PAUSE_CHARGING - 1))
|
||||
for i in range(TEST_TIME):
|
||||
pm.calculate(peripheralState, ignition)
|
||||
pm.calculate(VOLTAGE_BELOW_PAUSE_CHARGING, ignition)
|
||||
if i % 10 == 0:
|
||||
self.assertFalse(pm.should_shutdown(ignition, True, ssb, False))
|
||||
self.assertFalse(pm.should_shutdown(ignition, True, ssb, False))
|
||||
@@ -188,9 +168,8 @@ class TestPowerMonitoring(unittest.TestCase):
|
||||
pm.car_battery_capacity_uWh = CAR_BATTERY_CAPACITY_uWh
|
||||
|
||||
ignition = False
|
||||
peripheralState = self.mock_peripheralState(log.PandaState.PandaType.uno, car_voltage=(VBATT_PAUSE_CHARGING - 1))
|
||||
for i in range(TEST_TIME):
|
||||
pm.calculate(peripheralState, ignition)
|
||||
pm.calculate(VOLTAGE_BELOW_PAUSE_CHARGING, ignition)
|
||||
if i % 10 == 0:
|
||||
self.assertFalse(pm.should_shutdown(ignition, False, ssb, False))
|
||||
self.assertFalse(pm.should_shutdown(ignition, False, ssb, False))
|
||||
|
||||
@@ -347,7 +347,8 @@ def thermald_thread(end_event, hw_queue):
|
||||
off_ts = sec_since_boot()
|
||||
|
||||
# Offroad power monitoring
|
||||
power_monitor.calculate(peripheralState, onroad_conditions["ignition"])
|
||||
voltage = None if peripheralState.pandaType == log.PandaState.PandaType.unknown else peripheralState.voltage
|
||||
power_monitor.calculate(voltage, onroad_conditions["ignition"])
|
||||
msg.deviceState.offroadPowerUsageUwh = power_monitor.get_power_used()
|
||||
msg.deviceState.carBatteryCapacityUwh = max(0, power_monitor.get_car_battery_capacity())
|
||||
current_power_draw = HARDWARE.get_current_power_draw()
|
||||
|
||||
Reference in New Issue
Block a user