diff --git a/selfdrive/ui/soundd.py b/selfdrive/ui/soundd.py index 08e2314f8..9ad40c4bb 100644 --- a/selfdrive/ui/soundd.py +++ b/selfdrive/ui/soundd.py @@ -180,26 +180,32 @@ class Soundd: sounds_path = standard_path if random_events_path.exists(): - wavefile = wave.open(str(random_events_path), 'r') + path = random_events_path elif sounds_path.exists(): - wavefile = wave.open(str(sounds_path), 'r') + path = sounds_path else: if filename == "startup.wav": filename = "engage.wav" - wavefile = wave.open(BASEDIR + "/selfdrive/assets/sounds/" + filename, 'r') + path = Path(BASEDIR) / "selfdrive" / "assets" / "sounds" / filename + if not path.exists(): + cloudlog.warning(f"soundd: missing {filename}, skipping") + continue - assert wavefile.getnchannels() == 1 - assert wavefile.getsampwidth() == 2 - assert wavefile.getframerate() == SAMPLE_RATE - - length = wavefile.getnframes() - self.loaded_sounds[sound] = np.frombuffer(wavefile.readframes(length), dtype=np.int16).astype(np.float32) / (2**16/2) + try: + with wave.open(str(path), 'r') as wavefile: + if wavefile.getnchannels() != 1 or wavefile.getsampwidth() != 2 or wavefile.getframerate() != SAMPLE_RATE: + cloudlog.warning(f"soundd: invalid format {path}, skipping") + continue + length = wavefile.getnframes() + self.loaded_sounds[sound] = np.frombuffer(wavefile.readframes(length), dtype=np.int16).astype(np.float32) / (2**16/2) + except (FileNotFoundError, OSError, wave.Error): + cloudlog.exception(f"soundd: failed to load {path}") def get_sound_data(self, frames): # get "frames" worth of data from the current alert sound, looping when required ret = np.zeros(frames, dtype=np.float32) - if self.current_alert != AudibleAlert.none: + if self.current_alert != AudibleAlert.none and self.current_alert in self.loaded_sounds: num_loops = sound_list[self.current_alert][1] sound_data = self.loaded_sounds[self.current_alert] written_frames = 0 @@ -239,7 +245,10 @@ class Soundd: sink.close() def update_alert(self, new_alert): - current_alert_played_once = self.current_alert == AudibleAlert.none or self.current_sound_frame > len(self.loaded_sounds[self.current_alert]) + if new_alert != AudibleAlert.none and new_alert not in self.loaded_sounds: + new_alert = AudibleAlert.none + loaded = self.loaded_sounds.get(self.current_alert) + current_alert_played_once = self.current_alert == AudibleAlert.none or loaded is None or self.current_sound_frame > len(loaded) if self.current_alert != new_alert and (new_alert != AudibleAlert.none or current_alert_played_once): self.current_alert = new_alert self.current_sound_frame = 0 diff --git a/selfdrive/ui/tests/test_soundd.py b/selfdrive/ui/tests/test_soundd.py index e4f3ccc15..214ce3f45 100644 --- a/selfdrive/ui/tests/test_soundd.py +++ b/selfdrive/ui/tests/test_soundd.py @@ -1,4 +1,4 @@ -from cereal import log +from cereal import custom, log from cereal import messaging from cereal.messaging import SubMaster, PubMaster from openpilot.selfdrive.ui.soundd import ( @@ -8,12 +8,14 @@ from openpilot.selfdrive.ui.soundd import ( check_selfdrive_timeout_alert, is_turn_steering_limit_alert, should_mute_turn_steering_limit_alert, + starpilot_alert_key, ) import numpy as np import time AudibleAlert = log.SelfdriveState.AudibleAlert +StarPilotAudibleAlert = custom.StarPilotCarControl.HUDControl.AudibleAlert class TestSoundd: @@ -34,6 +36,22 @@ class TestSoundd: assert not should_mute_turn_steering_limit_alert("steerSaturated/warning", 10.0, 0.0) assert not should_mute_turn_steering_limit_alert("laneChangeBlocked/warning", 10.0, 25.0) + def test_load_sounds_skips_missing_custom_clips(self, tmp_path): + soundd = Soundd.__new__(Soundd) + soundd.sound_directory = tmp_path / "sounds" + soundd.sound_directory.mkdir() + soundd.random_events_directory = tmp_path / "random_events" + soundd.random_events_directory.mkdir() + + soundd.load_sounds() + + assert AudibleAlert.engage in soundd.loaded_sounds + assert starpilot_alert_key(StarPilotAudibleAlert.angry) not in soundd.loaded_sounds + soundd.current_alert = starpilot_alert_key(StarPilotAudibleAlert.angry) + soundd.current_volume = 1.0 + soundd.current_sound_frame = 0 + np.testing.assert_array_equal(soundd.get_sound_data(4), np.zeros(4, dtype=np.float32)) + def test_bluetooth_audio_mutes_local_only_while_healthy(self): soundd = Soundd.__new__(Soundd) samples = np.array([0.25, -0.5], dtype=np.float32)