mawei/MoreTore
1
0
Fork 0
mirror of https://github.com/MoreTore/openpilot.git synced 2026-08-24 01:23:45 +08:00
Code Issues Packages Projects Releases Wiki Activity

Micd: update sound levels in callback (#26674)

Browse Source 
* update once reached 4096

* update once reached 4096

* reduce

* debug & cmt

* fix

* fifo again

* fix

* clean that up

* update filter on demand

Co-authored-by: Cameron Clough <cameronjclough@gmail.com>
This commit is contained in:
Shane Smiskol
2022-12-02 18:38:04 -08:00
committed by GitHub
parent 10085d1e3f
commit 20dba2f33d
1 changed files with 29 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
system/micd.py
+29 -25
View File
@@ -8,9 +8,10 @@ from common.realtime import Ratekeeper
from system.swaglog import cloudlog
RATE = 10
DT_MIC = 1. / RATE
FFT_SAMPLES = 4096
REFERENCE_SPL = 2e-5 # newtons/m^2
SAMPLE_RATE = 44100
FILTER_DT = 1. / (SAMPLE_RATE / FFT_SAMPLES)
def calculate_spl(measurements):
@@ -45,9 +46,25 @@ class Mic:
self.rk = Ratekeeper(RATE)
self.measurements = np.empty(0)
self.spl_filter_weighted = FirstOrderFilter(0, 2.5, DT_MIC, initialized=False)
self.sound_pressure = 0
self.sound_pressure_weighted = 0
self.sound_pressure_level_weighted = 0
self.spl_filter_weighted = FirstOrderFilter(0, 2.5, FILTER_DT, initialized=False)
def update(self):
msg = messaging.new_message('microphone')
msg.microphone.soundPressure = float(self.sound_pressure)
msg.microphone.soundPressureWeighted = float(self.sound_pressure_weighted)
msg.microphone.soundPressureWeightedDb = float(self.sound_pressure_level_weighted)
msg.microphone.filteredSoundPressureWeightedDb = float(self.spl_filter_weighted.x)
self.pm.send('microphone', msg)
self.rk.keep_time()
def callback(self, indata, frames, time, status):
"""
Using amplitude measurements, calculate an uncalibrated sound pressure and sound pressure level.
Then apply A-weighting to the raw amplitudes and run the same calculations again.
@@ -55,31 +72,18 @@ class Mic:
Logged A-weighted equivalents are rough approximations of the human-perceived loudness.
"""
if len(self.measurements) > 0:
sound_pressure, _ = calculate_spl(self.measurements)
measurements_weighted = apply_a_weighting(self.measurements)
sound_pressure_weighted, sound_pressure_level_weighted = calculate_spl(measurements_weighted)
self.spl_filter_weighted.update(sound_pressure_level_weighted)
else:
sound_pressure = 0
sound_pressure_weighted = 0
sound_pressure_level_weighted = 0
self.measurements = np.empty(0)
msg = messaging.new_message('microphone')
msg.microphone.soundPressure = float(sound_pressure)
msg.microphone.soundPressureWeighted = float(sound_pressure_weighted)
msg.microphone.soundPressureWeightedDb = float(sound_pressure_level_weighted)
msg.microphone.filteredSoundPressureWeightedDb = float(self.spl_filter_weighted.x)
self.pm.send('microphone', msg)
self.rk.keep_time()
def callback(self, indata, frames, time, status):
self.measurements = np.concatenate((self.measurements, indata[:, 0]))
while self.measurements.size >= FFT_SAMPLES:
measurements = self.measurements[:FFT_SAMPLES]
self.sound_pressure, _ = calculate_spl(measurements)
measurements_weighted = apply_a_weighting(measurements)
self.sound_pressure_weighted, self.sound_pressure_level_weighted = calculate_spl(measurements_weighted)
self.spl_filter_weighted.update(self.sound_pressure_level_weighted)
self.measurements = self.measurements[FFT_SAMPLES:]
def micd_thread(self, device=None):
if device is None:
device = "sysdefault"