Konubinix' opinionated web of thoughts

An App to Measure the Volume of Ambiant Sound

Fleeting

We want to stay quiet when camping, and this is what will tell my kid when they are speaking too loud. The telling should reach everybody at once rather than tap one person on the shoulder, so it is not a phone anybody holds or pockets: it lies on the table, where a phone that starts to vibrate is heard by the whole group. That buzz is the message. hussshh on the screen is for whoever looks up.

The phone is an old Android already running my python runtime on android, and that is what makes this worth doing at all: a feature there is a python file dropped on the sdcard of a phone that is already running, with nothing to rebuild between an idea and it sitting on the table.

Hearing the room

What we are after is a room that stays loud, not the loudest instant in it. Android offers a shortcut to the microphone — MediaRecorder.getMaxAmplitude() — but it answers the other question: the largest sample since the previous call. Two moments with the same largest sample can hold wildly different amounts of noise, a single dropped fork or a table of shouting, and telling those two apart is the entire job here. It is also not free to ask: a recorder only answers once it has been prepared and started, which means handing it an output file and encoding a stream we would immediately throw away.

AudioRecord hands over the raw samples instead, and the amount of noise in a stretch of raw samples is exactly their root mean square. So we open the microphone in mono, sixteen bits a sample, at the one rate the API guarantees every device supports — and with several times the buffer the device says it needs, so the driver always has room to keep filling while we are busy with the samples it handed over last.

AudioRecord = autoclass("android.media.AudioRecord")
AudioFormat = autoclass("android.media.AudioFormat")
AudioSource = autoclass("android.media.MediaRecorder$AudioSource")

RATE = 44100          # the only rate AudioRecord guarantees everywhere
CHANNEL = AudioFormat.CHANNEL_IN_MONO
ENCODING = AudioFormat.ENCODING_PCM_16BIT


def _open_mic():
    size = AudioRecord.getMinBufferSize(RATE, CHANNEL, ENCODING)
    rec = AudioRecord(AudioSource.MIC, RATE, CHANNEL, ENCODING, size * 4)
    rec.startRecording()
    return rec

The apk was built asking for the microphone, but asking is not holding: recording is one of the permissions a user grants, and on this package it starts out ungranted — so the phone does have to change once, and granting it from the command line is the whole of the change. The vibrator needs no such ceremony; the apk asks for that one too, and it is handed over at install.

clk android adb shell pm grant eu.konubinix.konixpoc android.permission.RECORD_AUDIO

Those samples then have to be cut into stretches, and the length of a stretch decides what the number means. A tenth of a second is the compromise: long enough to hold ten or so cycles of the lowest pitch a human voice reaches, so the figure is an energy rather than an accident of where the wave happened to be, and short enough that speech is followed instead of averaged away.

WINDOW = 0.1                            # seconds of sound behind one figure
FRAMES = int(RATE * WINDOW)
SILENCE = -90.0                         # what we call a window with no signal


def _level(rec, buf):
    got = rec.read(buf, 0, len(buf))
    if got <= 0:
        return SILENCE
    rms = audioop.rms(bytes(buf[:got]), 2)
    return 20 * log10(rms / 32768.0) if rms else SILENCE

Under the hood, that reads back into a buffer Java filled: rec.read is handed a bytearray and pyjnius copies the java byte[] back into it when the call returns — the same write-back the thermal camera’s control transfers depend on.

A read can also come back with nothing at all — the microphone taken by something else, or failing mid-stream — and a window still has to be given some figure. It is given one far below anything a room ever reaches, so a microphone that has gone quiet on us reads as quiet rather than as noise.

A refused permission is the case I have not pinned down: it may look exactly like that, or it may stop the microphone opening at all. The two are worth telling apart, because one is silence that passes for a well-behaved family and the other is a microphone that refuses to start — a minute with the phone settles it.

One thing that number is not is a measurement of the room. It is a ratio against the largest value the converter can represent — a full-scale figure, dBFS — and there is no calibrated microphone anywhere in the chain to turn it into the decibels a sound level meter would show. That is fine. We never needed to know how loud the tent is in absolute terms; we need to know when it is louder than it was.

Too loud, and for long enough

Ten figures a second is far more opinion than we want. A fork on a plate clears any line we could draw and means nothing; a raised voice clears the same line and means everything. What separates them is not how loud they are but how long they last, so a level has to hold above the line for a while before it counts as the room being loud.

Coming back down needs a patience of its own, and a line of its own. The patience first: the quiet between two sentences is real quiet, so a room let off on its first calm window would have the same conversation tripping the buzz again on every breath. It only counts as settled once it has stayed quiet for a few seconds together.

And a level that has dropped a hair under the shouting line has not settled at all — it is still a table of people shouting, just barely less so. If that counted toward the few seconds, a room sitting steadily just under the line would spend them all and be declared calm with nobody lowering their voice. So the line a room has to fall under to start earning that patience sits well below the line it had to cross to be called loud in the first place.

So what we are building is a detector fed one level per window, carrying the upper line it must cross to call the room loud and the lower line it must drop back under to call it calm. In the promises that follow, a window at −20 dBFS is a voice well over the upper line and one at −50 is a room well under the lower one.

A single loud window is a noise, not a conversation, and must leave the room quiet.

d = Detector(upper_db=-30, lower_db=-36, attack=15, release=30)
assert not any(d.feed(lvl) for lvl in [-50, -20, -50, -50])

Held long enough, though, the same level is somebody shouting, and it should announce itself exactly once however long the shouting goes on — the buzz is a reminder, not an alarm.

assert [d.feed(-20) for _ in range(60)].count(True) == 1

Somebody drawing breath mid-rant is not the room calming down. A short gap followed by more shouting — more than enough on its own to set the thing off from cold — must earn no second buzz.

assert not any(d.feed(-50) for _ in range(10))
assert not any(d.feed(-20) for _ in range(20))

A group that actually settles, on the other hand, gets its slate wiped, and the next outburst is worth a fresh buzz.

assert not any(d.feed(-50) for _ in range(30))
assert [d.feed(-20) for _ in range(15)].count(True) == 1

The detector behind those four promises is a handful of counters and no clock of its own. It is fed one figure per window and counts windows, which is what lets the promises be made about a stream of numbers rather than about the passage of an afternoon.

class Detector(object):
    def __init__(self, upper_db, lower_db, attack, release):
        self.upper_db, self.lower_db = upper_db, lower_db
        self.attack, self.release = attack, release
        self.loud = False
        self.run = 0

    def feed(self, db):
        """One window's level in; True on the window the room turns loud."""
        if self.loud:
            self.run = self.run + 1 if db < self.lower_db else 0
            if self.run >= self.release:
                self.loud, self.run = False, 0
            return False
        self.run = self.run + 1 if db >= self.upper_db else 0
        if self.run >= self.attack:
            self.loud, self.run = True, 0
            return True
        return False
the four promises hold

The four numbers those promises are run with are placeholders. Where the lines actually sit depends on the phone, the table it lies on and how far away the voices are, so they are to be found by trial on the spot; a second and a half of shouting before the buzz, and three seconds of calm to forget it, are the starting points.

UPPER_DB = -30.0                        # to trial: the room is too loud
LOWER_DB = -36.0                        # to trial: and the room is calm again
ATTACK = int(1.5 / WINDOW)
RELEASE = int(3.0 / WINDOW)

Two halves, one message

The runtime this rides on comes in two halves. One is a foreground service that keeps running whether or not anything is on screen; the other is the kivy app that owns the screen. They are separate processes and talk to each other by sending named messages over a local socket, which is how every feature of the phone is already wired.

That split decides where the listening goes. A phone lying on a camping table has its screen off and the app long since backgrounded, so anything that has to keep hearing belongs in the service — and the service is also what shakes the phone, which is the part of the message that matters. The word on the screen is the app’s job, and it has to still be there whenever somebody gets round to looking.

The read is what paces that loop. It does not return until the window is full, so the service needs neither a sleep nor a clock to keep to ten judgements a second. Both edges of the verdict are worth passing on: the buzz goes out when the room turns loud, and the app is told when it settles so the screen can stop saying otherwise. The settling has no window of its own to announce it, though, so the loop watches the state across the judgement to catch it. The loop runs for as long as the listening lasts.

And the end of the listening has to be believed on the spot. Nobody ends it in a silent tent: the listening gets stopped because the table is mid-shout and the buzz is coming, which is the very moment the loop is a window away from firing. So the window it was already reading when the listening ended may still be judged, but it must not be allowed to speak.

listening = threading.Event()


def _listen():
    rec = None
    try:
        rec = _open_mic()
        listening.set()
        buf = bytearray(FRAMES * 2)
        detector = Detector(UPPER_DB, LOWER_DB, ATTACK, RELEASE)
        while listening.is_set():
            was_loud = detector.loud
            turned_loud = detector.feed(_level(rec, buf))
            if not listening.is_set():
                break
            if turned_loud:
                logger.info("soundmeter: too loud")
                plyer.vibrator.vibrate(VIBRATION)
                to_app("hussshh")
            elif was_loud and not detector.loud:
                to_app("hussshh:over")
    finally:
        listening.clear()
        if rec is not None:
            rec.stop()
            rec.release()

A buzz has to be long enough to be a sound in the room rather than a click, and short enough not to become the noise it is complaining about. Six tenths of a second is the guess to start from.

VIBRATION = 0.6                         # seconds of buzz, to trial

None of this should run all day — the microphone is open the whole time it is on, and we only want it during a meal. Whoever switches it on is sitting at that table with nothing but the phone in reach, so the switch lives in the app, and the app asks the service to start or stop listening.

The listening is a single flag, and the thread owns it: it goes up only once the microphone is in hand, and comes down on the way out however the loop ends, so it never claims a loop that is not running. That is what makes it worth answering a request with — starting waits a couple of seconds for it to go up, so a microphone that cannot be opened at all answers no rather than yes to a loop that died on its first line.

OPENING = 2.0                           # seconds to wait for the microphone


def start():
    if not listening.is_set():
        thread = threading.Thread(target=_listen)
        thread.daemon = True
        thread.start()
        listening.wait(OPENING)
    return listening.is_set()


def stop():
    listening.clear()
    return listening.is_set()

On the app’s side there is one screen, and it does both jobs. The word is read at arm’s length by somebody who has just been buzzed at, so it is one word as large as the screen allows, and it is put there and taken away by the two messages rather than by a timer — the screen then says what the room is doing now, and nobody has to wonder whether the hussshh they are looking at is still true. Below it sits the switch. Which way to toggle, it reads off its own label; what to say afterwards, it takes from the service’s answer. A screen built while the service is already listening therefore opens saying listen, and one press puts the two back in step.

OFF, ON = "listen", "stop listening"


class SoundMeterScreen(Screen):
    def __init__(self, *args, **kwargs):
        super(SoundMeterScreen, self).__init__(*args, **kwargs)
        self.word = Label(text="", font_size="120sp")
        self.switch = Button(text=OFF, font_size="40sp", size_hint=(1, 0.2))
        self.switch.bind(on_press=self.toggle)
        box = BoxLayout(orientation="vertical")
        box.add_widget(self.word)
        box.add_widget(self.switch)
        self.add_widget(box)

    def toggle(self, *_):
        starting = self.switch.text == OFF
        now_listening = to_service("sound:start" if starting else "sound:stop",
                                   wait_for_answer=True)
        self.switch.text = ON if now_listening else OFF

The word can arrive while the app is showing something else entirely, or while it has never shown this screen at all, so receiving it also means building the screen if need be and bringing it to the front. The room settling only touches the word: whoever has wandered off to another screen since should not be yanked back for good news.

def populate(sm):
    try:
        return sm.get_screen("soundmeter")
    except ScreenManagerException:
        screen = SoundMeterScreen(name="soundmeter")
        sm.add_widget(screen)
        return screen


@oschandler("hussshh")
def _(_):
    app = App.get_running_app()
    populate(app.sm).word.text = "hussshh"
    app.goto("soundmeter")


@oschandler("hussshh:over")
def _(_):
    populate(App.get_running_app().sm).word.text = ""

Landing it on the phone

The listener and the screen go onto the sdcard as soundmeter.py and soundmeterscreen.py, tangled straight over adb. The two messages are answered by the service, which reaches for the listener only when one of them arrives — so a device given no microphone module still boots, and only refuses when asked to listen. The app’s front screen grows one more button, which opens the screen above.

What is left is to sit at the table with it and find the two lines. Speak normally and nothing should happen; talk over each other and the table should buzz, once, a second and a half in.