CODEWITHSUNDEEP
pythoneventswindows-runtimepython-asynciowinrt-async
Wintermute
Wintermute

Reputation: 21

Event handling with WinRT for python

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://learn.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)

Upvotes: 2

Views: 1355

Answers (1)

David Lechner
David Lechner

Reputation: 1824

First, FYI, I've started a community fork of PyWinRT and the winrt package at https://github.com/pywinrt and published individual namespaces packages on PyPI, like winrt.Windows.Media.Control. This contains many, many fixes over the unmaintained winrt package, including fixing some serious problems like leaking a COM object on every await of an _async() method. The new packages also include 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
import contextlib

from winrt.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" not in session.source_app_user_model_id.lower():
            continue

        chrome_info = session.get_playback_info()
        if chrome_info.playback_status == PlaybackStatus.PLAYING:
            ...


async def main():
    async with contextlib.AsyncExitStack() as stack:
        loop = asyncio.get_running_loop()

        manager = await SessionManager.request_async()

        # by default, callbacks happen on a background thread, so we need
        # to use call_soon_threadsafe to run the update on the main thread
        def handle_sessions_changed(
            manager: SessionManager, args: SessionsChangedEventArgs
        ) -> None:
            loop.call_soon_threadsafe(update_sessions(manager))

        # add callback to handle any future changes
        sessions_changed_token = manager.add_sessions_changed(handle_sessions_changed)
        # ensure the callback is removed when the function exits
        stack.callback(manager.remove_sessions_changed, sessions_changed_token)

        # handle current state
        await update_sessions(manager)

        event = asyncio.Event()
        # wait forever - a real app would call event.set() to end the app
        await event.wait()


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.

Upvotes: 3

Related Questions