Event handling with WinRT for python

Question

The Task

I'm trying to make a tool that auto recognizes playback from a non Spotify source and pauses Spotify for the duration of the other audio. I've been using Microsoft's WinRT projection for Python (https://github.com/Microsoft/xlang/tree/master/src/package/pywinrt/projection). I can get the program to recognize audio sources and playback status but only from the moment the function was called. After that it stops updating.

The Problem

My main problem is understanding how WinRT implemented event handling in Python. The windows docs talk about a "changed playback event" (https://docs.microsoft.com/en-us/uwp/api/windows.media.control.globalsystemmediatransportcontrolssession.playbackinfochanged?view=winrt-22000) but I can't find any clue to how this is accomplished in the Python projection of the API.

Current State

Here is my code so far, any help is appreciated.

import asyncio
import time
from winrt.windows.media.control import \
    GlobalSystemMediaTransportControlsSessionManager as MediaManager
from winrt.windows.media.control import \
    GlobalSystemMediaTransportControlsSession as SessionMananger
from winrt.windows.media.control import \
    PlaybackInfoChangedEventArgs as PlaybackEventArgs

async def toggle_spotify():
    sessions = await MediaManager.request_async() #grab session manager instance

    all_sessions = sessions.get_sessions() #grab sequence of current instances
    for current_session in all_sessions: #iterate and grab desired instances
        if "chrome" in current_session.source_app_user_model_id.lower():
            chrome_info = current_session.get_playback_info()
        if "spotify" in current_session.source_app_user_model_id.lower():
            spotify_manager = current_session 
            spotify_info = current_session.get_playback_info()
    if chrome_info.playback_status == 4 and spotify_info.playback_status == 4: #status of 4 is playing, 5 is paused
        await spotify_manager.try_toggle_play_pause_async()
    elif chrome_info.playback_status == 5 and spotify_info.playback_status == 5:
        await spotify_manager.try_toggle_play_pause_async()
        # print(f"chrome playback status: {chrome_info.playback_status}")
        # print(f"chrome playback status: {spotify_info.playback_status}")
        # print("+++++++++++++++++")
        # time.sleep(2)

if __name__ == '__main__':
    #asyncio.run(get_media_info())
    while True: #mimicking event handling by looping
        asyncio.run(toggle_spotify())
        time.sleep(0.1)

Answer 1

First, FYI, I've started a community fork of PyWinRT and the winrt package at https://github.com/pywinrt and published a winsdk package on PyPI. This contains many, many fixes over the seemingly unmaintained winrt package, including fixing some serious problems like leaking a COM object on every await of an _async() method. The new winsdk package also includes type hints which make solving issues like this much easier.

The way event handlers are used is currently documented here.

So for this specific case, your code might look something like this:

import asyncio

from winsdk.windows.media.control import (
    GlobalSystemMediaTransportControlsSessionManager as SessionManager,
    GlobalSystemMediaTransportControlsSessionPlaybackStatus as PlaybackStatus,
    SessionsChangedEventArgs,
)


async def update_sessions(manager: SessionManager) -> None:
    for session in manager.get_sessions():
        if "chrome" in session.source_app_user_model_id.lower():
            chrome_info = session.get_playback_info()
            if chrome_info.playback_status == PlaybackStatus.PLAYING:
                ...


def handle_sessions_changed(manager: SessionManager, args: SessionsChangedEventArgs) -> None:
    asyncio.create_task(update_sessions(manager))


async def main():
    manager = await SessionManager.request_async()

    # add callback to handle any future changes
    sessions_changed_token = manager.add_sessions_changed(handle_sessions_changed)

    try:
        # handle current state
        asyncio.create_task(update_sessions(manager))

        event = asyncio.Event()
        # wait forever - a real app would call event.set() to end the app
        await event.wait()
    finally:
        # remove the callback
        manager.remove_sessions_changed(sessions_changed_token)


if __name__ == '__main__':
    asyncio.run(main())

For some reason, the event doesn't actually seem to be firing, but a quick web search reveals that this is a common problem for the GlobalSystemMediaTransportControlsSessionManager.SesssionsChanged and seems unrelated to the Python bindings.

Also note that asyncio.create_task() is only needed if update_sessions() actually awaits something, otherwise it could just be called directly.