Trying to read mute state of microphone

[MrEggy]

Firstly, thanks for your work on this. I have been trying to find a way unsuccessfully of determining if my microphone is muted whilst in a Teams call. This library looks like it might it might work. I've had a good play around with your example code. However I can't get it to work. Would it be too much to ask to request some sample code that might do this? Many thanks in advance.

[kdschlosser]

Do you want to know if it is muted or if there is any audio that is heard on the microphone?

[kdschlosser]

I also recommend using the develop branch as this branch works with python3

[kdschlosser]

if you are looking for the actual muting of the microphone and not if there is any audio being heard by the microphone you can use code like this.

you need to change the name of the device and endpoint to match what you have.

import signal as sig
import time
import pyWinCoreAudio
from pyWinCoreAudio import ON_ENDPOINT_VOLUME_CHANGED

last_mute = False

def on_session_volume_changed(
        signal,
        device_,
        endpoint_,
        session_,
        new_volume,
        new_mute
):
    global last_mute
    if new_mute != last_mute:
        print(device_.name)
        print(endpoint_.name)
        print('microphone mute:', new_mute)
        print()

        last_mute = new_mute

_on_session_volume_changed = None

for device in pyWinCoreAudio:
    print('device:', device.name)
    for endpoint in device:
        print('    endpoint:', endpoint.name)
        if (
            device.name == 'Realtek USB Audio' and 
            endpoint.name == 'Microphone (Realtek USB Audio)'
        ):
            _on_session_volume_changed = (
                ON_SESSION_VOLUME_CHANGED.register(
                    on_session_volume_changed,
                    device,
                    endpoint
                )
            )
    print()

class ServiceExit(Exception):
    """
    Custom exception which is used to trigger the clean exit
    of all running threads and the main program.
    """
    pass

def service_shutdown(signum, frame):
    print('Caught signal %d' % signum)
    raise ServiceExit

if _on_session_volume_changed is not None:
    sig.signal(sig.SIGTERM, service_shutdown)
    sig.signal(sig.SIGINT, service_shutdown)

    try:
        while True:
            time.sleep(0.5)

    except ServiceExit:
        pass

    _on_session_volume_changed.unregister()
else:
    print('device and endpoint not found')

That code that needs to be changed is this.

        if (
            device.name == 'Realtek USB Audio' and 
            endpoint.name == 'Microphone (Realtek USB Audio)'
        ):

If you run the script it will print off all of the names for you so you will know what you need to change the above to. You can also have it watch all devices that have microphones as well and if that is what you want let me know and I will tell you how to go about doing that.

If you want to know how to detect audio let me know and I can tell you how to do that as well.

[MrEggy]

Many thanks for your prompt and detailed response. I ran the script and got the following error.

Traceback (most recent call last): File "D:\test\pyWinCoreAudio-develop\mic.py", line 38, in ON_SESSION_VOLUME_CHANGED.register( ^^^^^^^^^^^^^^^^^^^^^^^^^ NameError: name 'ON_SESSION_VOLUME_CHANGED' is not defined. Did you mean: 'ON_ENDPOINT_VOLUME_CHANGED'?

So I changed line 38 to:

ON_ENDPOINT_VOLUME_CHANGED.register(

Now when I mute the microphone I get the following error:

Traceback (most recent call last): File "D:\test\pyWinCoreAudio-develop\pyWinCoreAudio\signal.py", line 332, in _loop func(*args, **kwargs) TypeError: on_session_volume_changed() got an unexpected keyword argument 'is_muted'

So this looks promising it is doing what I want it to do. However I note that if I mute the microphone externally to the Teams application it triggers the error. So this tells me that your script is working. Albeit with an error. However if I mute the microphone from within the Teams application nothing happens. So this would suggest that this mute is not being detected. Any ideas?

[kdschlosser]

Yeah your right. Sorry my bad I grabbed the wrong event.

[kdschlosser]

OK so muting the microphone from inside of the teams application may or may not mute the microphone at the system level. It could be muting it at the session level and if that is the case then you would need to use that session event that I had in there incorrectly.

If you paste the exact traceback you are getting I will be able to fix the issue. Ned that trraceback data to do that. What OS are you running? I know that the script works with Win 7, 8, 8.1 and 10. All of the features have not been tested with Win11 and if you are using Win11 it could be something that got changes in the Windows Core Audio API that I need to account for.

[kdschlosser]

Here is a revised version of the script.

This one will handle session mute events as well as the endpoint mute events.

The one thing that needs to really be paid attention to is holding a reference to the objects in the library. You can see my handling with sessions and not using the actual session object it's self as the identifier. I am using the name instead. This is because if the object gets stored and it doesn't get released when the session gets disconnected you end up with a memory leak. Everything needs to be cleaned up at the end.

import signal as sig
import threading
import pyWinCoreAudio
from pyWinCoreAudio import (
    ON_ENDPOINT_VOLUME_CHANGED,
    ON_SESSION_VOLUME_CHANGED,
    ON_SESSION_CREATED,
    ON_SESSION_DISCONNECT
)

# this lock object keeps the printed text from getting jumbled up with another 
# print that is being done, The callbacks are run in their own threads so 
# things can bump heads is not done properly.
print_lock = threading.Lock()

session_events_lock = threading.Lock()
session_events = {}

def on_session_created(
    _,
    device_,
    endpoint_,
    session_
):
    key = (device_, endpoint_, session_.name)
    with session_events_lock:
        if key not in session_events:
            session_events[key] = ON_SESSION_VOLUME_CHANGED(
                on_session_volume_changed,
                device_,
                endpoint_,
                session_
            )
            with print_lock:
                print('NEW SESSION:', device_.name, ':', endpoint_.name, ':', session_.name)

            with session_mutes_lock:
                session_mutes[key] = session_.volume.mute

def on_session_disconnect(
    _,
    device_,
    endpoint_,
    name,
    __
):
    key = (device_, endpoint_, name)
    with session_events_lock:

        if key in session_events:
            evnt = session_events.pop(key)
            evnt.unregister()

            with print_lock:
                print('SESSION DISCONNECT:', device_.name, ':', endpoint_.name, ':', name)

            with session_mutes_lock:
                del session_mutes[key]

session_mutes_lock = threading.Lock()
session_mutes = {}

def on_session_volume_changed(
        _,
        device_,
        endpoint_,
        session_,
        __,
        new_mute
):
    key = (device_, endpoint_, session_.name)
    with session_mutes_lock:
        if key in session_mutes:
            if session_mutes[key] != new_mute:

                with print_lock:
                    print(device_.name, ':', endpoint_.name, ':', session_.name, ': MUTE =', new_mute)

                session_mutes[key] = new_mute

endpoint_mute_lock = threading.Lock()
endpoint_mute = False

def on_endpoint_volume_changed(
        _,
        device_,
        endpoint_,
        is_muted,
        __,
        ___
):
    global endpoint_mute
    with endpoint_mute_lock:
        if is_muted != endpoint_mute:

            with print_lock:
                print(device_.name, ':', endpoint_.name, ': MUTE =', is_muted)

            endpoint_mute = is_muted

_on_session_volume_changed = None
_on_session_created = None
_on_session_disconnect = None

for device in pyWinCoreAudio:
    print('device:', device.name)
    for endpoint in device:
        print('    endpoint:', endpoint.name)
        if (
            device.name == 'Realtek USB Audio' and
            endpoint.name == 'Microphone (Realtek USB Audio)'
        ):
            endpoint_mute = endpoint.volume.mute

            _on_endpoint_volume_changed = (
                ON_ENDPOINT_VOLUME_CHANGED.register(
                    on_endpoint_volume_changed,
                    device,
                    endpoint
                )
            )

            _on_session_created = (
                ON_SESSION_CREATED.register(
                    on_session_created,
                    device,
                    endpoint
                )
            )

            _on_session_disconnect = ON_SESSION_DISCONNECT.register(
                on_session_disconnect,
                device,
                endpoint
            )

        # deletes reference
        del endpoint

    print()

    # deletes reference
    del device

class ServiceExit(Exception):
    """
    Custom exception which is used to trigger the clean exit
    of all running threads and the main program.
    """
    pass

def service_shutdown(signum, frame):
    print('Caught signal %d' % signum)
    raise ServiceExit

wait_event = threading.Event()

if _on_session_volume_changed is not None:
    sig.signal(sig.SIGTERM, service_shutdown)
    sig.signal(sig.SIGINT, service_shutdown)

    try:
        # we want to stop the main thread from exiting. This wait is a
        # blocking call and it has no timeout. That means it will wait forever
        # unless CTRL + C is pressed or the application is closed.
        # because of how I wrote the library you are able to run whatever it is
        # you need to inside the event callback. Each callback that is made
        # runs in it's own thread So it is possible to have collissions if
        # accessing the same data streucture (variable). This is why you see
        # me using the lock objects from the threading module. This is to
        # prevent any data corruption.
        wait_event.wait()
    except ServiceExit:
        pass

    _on_session_volume_changed.unregister()
    _on_session_created.unregister()
    _on_session_disconnect.unregister()

    with session_events_lock:
        for event in session_events.values():
            event.unregister()

        # deletes all references
        session_events.clear()        

    with session_mutes_lock:
        # deletes all references
        session_mutes.clear()

else:
    print('device and endpoint not found')

[MrEggy]

Firstly, can I say how amazed I am at how quickly you are able to rewrite code. Well done.

Windows 11 has this feature called Universal Mute. Where you can mute via an icon in the system tray. In theory it will mute for any application. Microsoft Teams interacts with this feature. So that if you mute within the Teams application is turns on the Universal Mute icon and visa versa. So I assumed this was happening at a system level. But obviously not.

With the second example it seems to be exiting with our entering the main wait loop, as I just see 'device and endpoint not found' after the device listing is output.

This is the full traceback from your first example. I think it is pretty much the same as what I posted earlier. It just repeats. Many thanks again in advance for your help.

File "D:\test\pyWinCoreAudio-develop\pyWinCoreAudio\signal.py", line 332, in _loop func(*args, *kwargs) TypeError: on_session_volume_changed() got an unexpected keyword argument 'is_muted' Traceback (most recent call last): File "D:\test\pyWinCoreAudio-develop\pyWinCoreAudio\signal.py", line 332, in _loop func(args, **kwargs) TypeError: on_session_volume_changed() got an unexpected keyword argument 'is_muted'

[kdschlosser]

give this one a shot...

import signal as sig
import threading
import pyWinCoreAudio
from pyWinCoreAudio import (
    ON_ENDPOINT_VOLUME_CHANGED,
    ON_SESSION_VOLUME_CHANGED,
    ON_SESSION_CREATED,
    ON_SESSION_DISCONNECT
)

# this lock object keeps the printed text from getting jumbled up with another
# print that is being done, The callbacks are run in their own threads so
# things can bump heads is not done properly.
print_lock = threading.Lock()

session_events_lock = threading.Lock()
session_events = {}

def on_session_created(
    signal,
    device,
    endpoint,
    session
):
    key = (device, endpoint, session.name)
    with session_events_lock:
        if key not in session_events:
            session_events[key] = ON_SESSION_VOLUME_CHANGED.register(
                on_session_volume_changed,
                device,
                endpoint,
                session
            )
            with print_lock:
                print('NEW SESSION:', device.name, ':', endpoint.name, ':', session.name)

            with session_mutes_lock:
                session_mutes[key] = session.volume.mute

def on_session_disconnect(
    signal,
    device,
    endpoint,
    name,
    reason
):
    key = (device, endpoint, name)
    with session_events_lock:

        if key in session_events:
            evnt = session_events.pop(key)
            evnt.unregister()

            with print_lock:
                print('SESSION DISCONNECT:', device.name, ':', endpoint.name, ':', name)

            with session_mutes_lock:
                del session_mutes[key]

session_mutes_lock = threading.Lock()
session_mutes = {}

def on_session_volume_changed(
        signal,
        device,
        endpoint,
        session,
        new_volume,
        new_mute
):
    key = (device, endpoint, session.name)
    with session_mutes_lock:
        if key in session_mutes:
            if session_mutes[key] != new_mute:

                with print_lock:
                    print(device.name, ':', endpoint.name, ':', session.name, ': MUTE =', new_mute)

                session_mutes[key] = new_mute

endpoint_mute_lock = threading.Lock()
endpoint_mute = False

def on_endpoint_volume_changed(
        signal,
        device,
        endpoint,
        is_muted,
        master_volume,
        channel_volumes
):
    global endpoint_mute
    with endpoint_mute_lock:
        if is_muted != endpoint_mute:

            with print_lock:
                print(device.name, ':', endpoint.name, ': MUTE =', is_muted)

            endpoint_mute = is_muted

_on_endpoint_volume_changed = None
_on_session_created = None
_on_session_disconnect = None

for device in pyWinCoreAudio:
    print('device:', device.name)
    for endpoint in device:
        print('    endpoint:', endpoint.name)

        if (
            device.name == 'DroidCam Virtual Audio' and
            endpoint.name == 'Microphone (DroidCam Virtual Audio)'
        ):
            endpoint_mute = endpoint.volume.mute

            for session in endpoint:
                key = (device, endpoint, session.name)

                with session_mutes_lock:
                    session_mutes[key] = session.volume.mute
                with session_events_lock:
                    session_events[key] = ON_SESSION_VOLUME_CHANGED.register(
                        on_session_volume_changed,
                        device,
                        endpoint,
                        session
                    )

            _on_endpoint_volume_changed = (
                ON_ENDPOINT_VOLUME_CHANGED.register(
                    on_endpoint_volume_changed,
                    device,
                    endpoint
                )
            )

            _on_session_created = (
                ON_SESSION_CREATED.register(
                    on_session_created,
                    device,
                    endpoint
                )
            )

            _on_session_disconnect = ON_SESSION_DISCONNECT.register(
                on_session_disconnect,
                device,
                endpoint
            )

        # deletes reference
        del endpoint

    print()

    # deletes reference
    del device

class ServiceExit(Exception):
    """
    Custom exception which is used to trigger the clean exit
    of all running threads and the main program.
    """
    pass

def service_shutdown(signum, frame):
    print('Caught signal %d' % signum)
    raise ServiceExit

wait_event = threading.Event()

if _on_endpoint_volume_changed is not None:
    sig.signal(sig.SIGTERM, service_shutdown)
    sig.signal(sig.SIGINT, service_shutdown)

    try:
        # we want to stop the main thread from exiting. This wait is a
        # blocking call and it has no timeout. That means it will wait forever
        # unless CTRL + C is pressed or the application is closed.
        # because of how I wrote the library you are able to run whatever it is
        # you need to inside the event callback. Each callback that is made
        # runs in it's own thread So it is possible to have collissions if
        # accessing the same data streucture (variable). This is why you see
        # me using the lock objects from the threading module. This is to
        # prevent any data corruption.
        wait_event.wait()
    except ServiceExit:
        pass

    _on_endpoint_volume_changed.unregister()
    _on_session_created.unregister()
    _on_session_disconnect.unregister()

    with session_events_lock:
        for event in session_events.values():
            event.unregister()

        # deletes all references
        session_events.clear()

    with session_mutes_lock:
        # deletes all references
        session_mutes.clear()

else:
    print('device and endpoint not found')