mirror of
https://github.com/sunnypilot/sunnypilot.git
synced 2026-02-18 20:03:53 +08:00
d7b0a5fa7e
* record feedback with LKAS button * fix alert test * slightly simplify feedbackd * "Audio Feedback Saved" upon time expiration or early stop * earlySend --> earlyStop * userFlag --> userBookmark * RecordAudioFeedback param/toggle * add audioFeedback test * simplify feedbackd * send bookmark regardless of toggle, show feedback event with higher priority * add userBookmark to selfdrived sm * fix mispelled param name * default off and move to main * segmentNum --> blockNum, earlyStop --> lastBlock * preserve audioFeedback * get rid of lastBlock and just send bookmark saved at the end * update raylib side * update toggle description and add raylib toggle --------- Co-authored-by: Adeeb Shihadeh <adeebshihadeh@gmail.com>
53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
import pytest
|
|
import cereal.messaging as messaging
|
|
from cereal import car
|
|
from openpilot.common.params import Params
|
|
from openpilot.system.manager.process_config import managed_processes
|
|
|
|
|
|
class TestFeedbackd:
|
|
def setup_method(self):
|
|
self.pm = messaging.PubMaster(['carState', 'rawAudioData'])
|
|
self.sm = messaging.SubMaster(['audioFeedback'])
|
|
|
|
def _send_lkas_button(self, pressed: bool):
|
|
msg = messaging.new_message('carState')
|
|
msg.carState.canValid = True
|
|
msg.carState.buttonEvents = [{'type': car.CarState.ButtonEvent.Type.lkas, 'pressed': pressed}]
|
|
self.pm.send('carState', msg)
|
|
|
|
def _send_audio_data(self, count: int = 5):
|
|
for _ in range(count):
|
|
audio_msg = messaging.new_message('rawAudioData')
|
|
audio_msg.rawAudioData.data = bytes(1600) # 800 samples of int16
|
|
audio_msg.rawAudioData.sampleRate = 16000
|
|
self.pm.send('rawAudioData', audio_msg)
|
|
self.sm.update(timeout=100)
|
|
|
|
@pytest.mark.parametrize("record_feedback", [False, True])
|
|
def test_audio_feedback(self, record_feedback):
|
|
Params().put_bool("RecordAudioFeedback", record_feedback)
|
|
|
|
managed_processes["feedbackd"].start()
|
|
assert self.pm.wait_for_readers_to_update('carState', timeout=5)
|
|
assert self.pm.wait_for_readers_to_update('rawAudioData', timeout=5)
|
|
|
|
self._send_lkas_button(pressed=True)
|
|
self._send_audio_data()
|
|
self._send_lkas_button(pressed=False)
|
|
self._send_audio_data()
|
|
|
|
if record_feedback:
|
|
assert self.sm.updated['audioFeedback'], "audioFeedback should be published when enabled"
|
|
else:
|
|
assert not self.sm.updated['audioFeedback'], "audioFeedback should not be published when disabled"
|
|
|
|
self._send_lkas_button(pressed=True)
|
|
self._send_audio_data()
|
|
self._send_lkas_button(pressed=False)
|
|
self._send_audio_data()
|
|
|
|
assert not self.sm.updated['audioFeedback'], "audioFeedback should not be published after second press"
|
|
|
|
managed_processes["feedbackd"].stop()
|