How to get audio (Mic) input working on Android with python/kivy

Question

EDIT from a visitor: The current consensus around this situation appears to be that Kivy does not support the microphone at this time, and we are begging people to help port the "audiostream" add-on forward, so that this can work again. Any tiny work to help this is greatly appreciated. More information below.

I try to get mic working on android, im using mainly kivy and buildozer I got working audio out with audiostream, however that module is so outdated it wont work anymore if use input "recording" GITHub Issue .well i was unable to get recording working on pc because it says "unsupported" as soon i use record functions, on documents it mentions only mobile devices, so that is ok. it can be replaced on those platforms anyways with pyaudio.

I have tried to search other options what i can use, so i came up across pyjnius and MediaRecorder, im very novice with java,(and trying to learn python atm, so novice there too) so i was unable to get it working. The problem lies, i need to get all mic data into bytes, this is easy with pyaudio, and it works. reason why im here, it is android where pyaudio not work(at least i havent yet tried compile libraries to android, i know this might be possible but alot work..)

Here is code what i have to try get it working:

            MediaRecorder = autoclass('android.media.MediaRecorder')
            AudioSource = autoclass('android.media.MediaRecorder$AudioSource')
            OutputFormat = autoclass('android.media.MediaRecorder$OutputFormat')
            AudioEncoder = autoclass('android.media.MediaRecorder$AudioEncoder')
            FileOutputStream = autoclass('java.io.FileOutputStream')
            gaindata = io.BytesIO()
    
            mRecorder = MediaRecorder()
            mRecorder.setAudioSource(AudioSource.MIC)
            mRecorder.setOutputFormat(OutputFormat.THREE_GPP)

            mRecorder.setOutputFile(gaindata.getBytes())
            mRecorder.setAudioEncoder(AudioEncoder.AMR_NB)
            mRecorder.prepare()

I know there is something about FileDescriptor, there is some examples, but all them have spaces on strings so i have no idea how to convert them to python.. all i want is setOutputFile -> gaindata

If there is another option would be nice, i need bytesIO data from microphone(prefer 8000,mono,raw wav without header OR GSM6.10) and i will convert it with soundfile(yes i did compiled libsndfile.so to arm) into gsm6.10 and put it into socket, its a VoIP app.

Answer 1

I have developed a module that implements VOIP in Kivy mobile applications, including handling microphone permissions, socket connections, and debugging. Currently in the process of being added to Kivy source code.

Video Demonstration: https://youtu.be/9GQdY-4Iuz4

Full source code and README.md explanation on GitHub: https://github.com/SanquezH/Kivy-VOIP

Inside of buildozer.spec file include the following permissions to allow VOIP functionality:

android.permissions = INTERNET,RECORD_AUDIO,ACCESS_NETWORK_STATE,WAKE_LOCK

For main.py and voip.py to work, pyjnius is required. Compiling code with Buildozer includes pyjnuis by default. Here is an alteration of main.py and voip.py from my GitHub source code to reduce the amount of code needed to answer your question:

main.py

from kivy.app import App
from kivy.uix.button import Button
from kivy.uix.boxlayout import BoxLayout
import threading
from voip import Client  # Import from voip.py module

class VOIPClientApp(App):
    # Initialize a client
    client = Client()

    # Configure connection from client to VOIP server
    client.dst_address = "192.168.1.67"  # Set to your server's IP address. Use root domain if using SSL (loopback by default)
    client.dst_port = 8080  # Set to your server's assigned port (port 8080 by default)
    client.timeout = 3  # Set 
    
    def build(self):
        # Create a call and end call button to alternate between to allow client control over VOIP call
        self.layout = BoxLayout(orientation='vertical')
        self.call_button = Button(text="Call")
        self.end_call_button = Button(text="End Call", disabled=True)
        self.call_button.bind(on_press=self.start_call)
        self.end_call_button.bind(on_press=self.end_call)
        self.layout.add_widget(self.call_button)
        self.layout.add_widget(self.end_call_button)
        return self.layout
    
    def auto_end_call(self):  # Automate ending call, including if connection closes externally
        if self.client.connected and self.client.hasPermission:
            # If client is connected and has permission, call is active
            while self.client.active_call:  # Loop that runs until call ends
                pass
        if not self.end_call_button.disabled:
            instance = VOIPClientApp()
            self.end_call(instance)

    def start_call(self, instance):
        # Disable call button after call button is pressed
        self.call_button.disabled = True
        self.end_call_button.disabled = False
        self.client.start_call()  # Initiate the VOIP call
        self.track_call_thread = threading.Thread(target=self.auto_end_call, daemon=True)
        self.track_call_thread.start()

    def end_call(self, instance):
        # Disable end call button after end call button is pressed
        self.end_call_button.disabled = True
        self.call_button.disabled = False
        self.client.end_call()  # End the VOIP call
        
if __name__ == "__main__":
    VOIPClientApp().run()

voip.py

import threading
from jnius import autoclass, JavaException
AudioRecord = autoclass("android.media.AudioRecord")
AudioSource = autoclass("android.media.MediaRecorder$AudioSource")
AudioFormat = autoclass("android.media.AudioFormat")
AudioTrack = autoclass("android.media.AudioTrack")
AudioManager = autoclass("android.media.AudioManager")
Socket = autoclass("java.net.Socket")
SocketTimer = autoclass("java.net.InetSocketAddress")

"""
MIT License

Copyright (c) 2024 Sanquez Heard

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software ...
"""

class Client:
    # Variables to be configured per client
    dst_address = "192.168.1.0"  # Set to VOIP server address
    dst_port = 8080
    timeout = 5  # Limits how long to wait for connection from server in seconds
    # Variables to adjust sound quality. Default settings recommended
    SAMPLE_RATE = 16000
    CHANNEL_CONFIG = AudioFormat.CHANNEL_IN_MONO
    AUDIO_FORMAT = AudioFormat.ENCODING_PCM_16BIT
    # Variables to be assigned dynamic values for VOIP services
    socket = None
    data_output_stream = None
    data_input_stream = None
    audio_record = None
    hasPermission = False
    connected = False
    active_call = False
    buffer_size = 640

    # Determine minimum buffer size, with default of 640
    def __init__(self):
        min_buffer_size = AudioRecord.getMinBufferSize(
            self.SAMPLE_RATE, self.CHANNEL_CONFIG, self.AUDIO_FORMAT
        )
        if min_buffer_size > self.buffer_size:
            self.buffer_size = min_buffer_size

    def start_call(self):
        self.verifyPermission()
        if self.hasPermission:  # If permission is granted, attempt server connection
            self.connected = False
            try:
                self.socket = Socket()
                self.socket.connect(
                    SocketTimer(self.dst_address, self.dst_port),
                    self.timeout * 1000
                )
                self.data_input_stream = self.socket.getInputStream()
                self.data_output_stream = self.socket.getOutputStream()
                self.connected = True
            except JavaException as e:
                print(e)
            if self.connected:
                # Establish VOIP session
                self.active_call = True
                self.record_thread = threading.Thread(target=self.send_audio, daemon=True)
                self.record_thread.start()
                self.listening_thread = threading.Thread(target=self.receive_audio, daemon=True)
                self.listening_thread.start()

    def end_call(self):
        # Ensure threads are closed before closing socket
        self.active_call = False
        if hasattr(self, "record_thread") and self.record_thread.is_alive():
            self.record_thread.join()
        if hasattr(self, "listening_thread") and self.listening_thread.is_alive():
            self.listening_thread.join()
        if self.socket != None:
            self.socket.close()
            self.socket = None

    def verifyPermission(self):
        self.hasPermission = False
        self.audio_record = AudioRecord(
            AudioSource.VOICE_COMMUNICATION,
            self.SAMPLE_RATE,
            self.CHANNEL_CONFIG,
            self.AUDIO_FORMAT,
            self.buffer_size,
        )
        # If the microphone state is initiatialized, allow VOIP call
        if self.audio_record.getState() != AudioRecord.STATE_UNINITIALIZED:
            self.hasPermission = True
        else:
            print(
                "Permission Error: "
                "Ensure RECORD_AUDIO (Mic) permission is enabled in app settings"
            )

    # Establish microphone stream
    def send_audio(self):
        audio_data = bytearray(self.buffer_size)
        self.audio_record.startRecording()
        while self.active_call == True:
            try:
                bytes_read = self.audio_record.read(
                    audio_data, 0, self.buffer_size
                )
                if (
                    bytes_read != AudioRecord.ERROR_INVALID_OPERATION
                    and bytes_read != AudioRecord.ERROR_BAD_VALUE
                ):
                    self.data_output_stream.write(audio_data, 0, bytes_read)
                elif bytes_read == AudioRecord.ERROR_INVALID_OPERATION:
                    print("ERROR_INVALID_OPERATION on microphone")
                else:
                    print("ERROR_BAD_VALUE on microphone")
            except JavaException as e:
                self.active_call = False
                print(e)

        self.audio_record.stop()

    # Establish speaker stream 
    def receive_audio(self):
        audio_track = AudioTrack(
            AudioManager.STREAM_VOICE_CALL,
            self.SAMPLE_RATE,
            AudioFormat.CHANNEL_OUT_MONO,
            self.AUDIO_FORMAT,
            self.buffer_size,
            AudioTrack.MODE_STREAM,
        )
        buffer = bytearray(self.buffer_size)
        audio_track.play()
        try:
            while self.active_call == True:
                bytes_received = self.data_input_stream.read(buffer)
                if bytes_received > 0:
                    audio_track.write(buffer, 0, bytes_received)
        except JavaException as e:
            self.active_call = False
            print(e)

        audio_track.stop()

Answer 2

I have looked into it, you can still make recording audio works in PC using SpeechRecognition which uses PyAudio. For mobile, I don't think Buildozer support that because I tried it and it didn't work. audiostream seems to be outdated, but I haven't tried it yet. edit: i just tried using audiostream, but i can't seem to make it work.

Answer 3

audiostream

audiostream uses pyjnius as well

https://github.com/kivy/audiostream/blob/master/audiostream/platform/plat_android.pyx

from jnius import autoclass
    AudioIn = autoclass('org.audiostream.AudioIn') 

I think best way would be fixing audiostream so other can use it as well because in the documentation of kivy it is mentioned to use it:

https://kivy.org/doc/master/api-kivy.core.audio.html

Note
The core audio library does not support recording audio. If you require this functionality, please refer to the audiostream extension.

Or you extract the core functionality of the project so you can use it.

pyaudio

Other project I found using microphone is https://pypi.org/project/SpeechRecognition/ which uses pyaudio

But I don't know if this works on android. Without your comment I would have thought it works because someone has created a kivy app to use it ...

https://github.com/jmercouris/speech_recognition