Python: Need to call an API without interrupting loop

Question

I have a program I wrote in Python that uses the Spotipy library to call the Spotify API to get the user's currently playing song and the tempo of the song in question. It then uses that information, a serial connection, and an Arduino to run some lights.

The problem I'm running into is that I want the program to check the API, with frequency, to see if the song has changed, but each API call takes my network about 0.15 seconds, so after not too many calls it messes up the timing of the lights.

Here is the function that calls the API for the tempo and an example while loop. If you want to see the full project and the complete code, here is the link to the github - https://github.com/eboyce452/lightshow.git:

def check_bpm():

    global seconds_per_beat
    global current_track

    current_track = spotify.current_user_playing_track()
    if current_track is None:
        seconds_per_beat = 0.5
    else:
        current_track = json_normalize(current_track)
        current_track = str(current_track['item.id'].iloc[0])
        features = json_normalize(spotify.audio_features(current_track))
        tempo = float(features['tempo'].iloc[0])
        seconds_per_beat = 60/tempo

while True:

    check_bpm()
    pin2.write(1)
    time.sleep(seconds_per_beat)
    pin2.write(0)
    time.sleep(seconds_per_beat)

So what I'm looking for, in a perfect world, is a way to have the check_bpm() function running in the background so that the lights can stay on beat, and then when the song changes, have the loop be interrupted (with like continue or something) and the variable for seconds_per_beat be updated.

I genuinely have no idea if that's even possible or not, so feel free to weigh in on that. But what I'm most curious about is implementing a common form of concurrency so that while the check_bpm() function is waiting for the API call to finish, that it continues with the rest of the while loop so the lights don't get out of sync. I have been reading a lot about asyncio, but I am so unfamiliar with it that any help that can be offered is appreciated.

Thank you so much! Feel free to also check out the github for the project and leave any comments or criticism you'd like.

Answer 1

Yes - since you have a "while True" denoting a mainloop, and the changing variables are written on only one of the sides, and read on the other, this can be ported to use either asyncio or threads with ease.

In multi-threading, you'd have both the pin-sending function and the api-reading function working continuously in separate threads -as faras you are concerned they run in parallel - and sice ordinary variables are threadsafe in Python, that is all that is to it.

You can just start the API reading in a separate thread:

from time import sleep
from threading import Thread
...


def check_bpm():

    global seconds_per_beat
    global current_track

    while True:
        current_track = spotify.current_user_playing_track()
        if current_track is None:
            seconds_per_beat = 0.5
        else:
            current_track = json_normalize(current_track)
            current_track = str(current_track['item.id'].iloc[0])
            features = json_normalize(spotify.audio_features(current_track))
            tempo = float(features['tempo'].iloc[0])
            seconds_per_beat = 60/tempo

        # (if you don't care making another api call right away, just leave
        # here with no pause. Otherwise, insert a reasonable time.sleep here 
        # (inside the while loop))

def main():
    api_thread = Thread(target=check_bpm)
    api_thread.start()
    while True:

        pin2.write(1)
        time.sleep(seconds_per_beat)
        pin2.write(0)
        time.sleep(seconds_per_beat)

main()

Another path would be to use asyncio - it would take a different syntax - but on the end of the day, you'd have to delegate the API calls to another thread, unless the spotify library you are using also supports asyncio.
Even them, the "gain" would be more in "perceived code elegance" than any thing real - maybe it could work if you are running this code in an appliance PC with an stripped down OS/CPU setting that does not support multi-threading.

It would look like this:

from time import sleep
import asyncio

...


async def check_bpm():

    global seconds_per_beat
    global current_track

    loop = asyncio.get_event_loop()
    while True:
        current_track = await loop.run_in_executor(spotify.current_user_playing_track)
        if current_track is None:
            seconds_per_beat = 0.5
        else:
            current_track = json_normalize(current_track)
            current_track = str(current_track['item.id'].iloc[0])
            features = json_normalize(spotify.audio_features(current_track))
            tempo = float(features['tempo'].iloc[0])
            seconds_per_beat = 60/tempo
        # here we better leave some spacing for the other
        # functions to run, besides the 0.15s duration
        # of the API call
        await asyncio.sleep(1)

async def manage_pins():
    while True:
        pin2.write(1)
        await asyncio.sleep(seconds_per_beat)
        pin2.write(0)
        await asyncio.sleep(seconds_per_beat)


def main():
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.gather(manage_pins(), check_bpm()))

main()

So, here, the Python code is rewritten so that it voluntarily suspends execution to wait for "other things to happen" - this takes place on the "await" statements.
Unlike the code with threads, the global variables will only be changed while the pin-control awaits on "asyncio.sleep" - while in the threaded code, the variable change can take place at any time.

And since the call that blocks on the spotify api is not asynchronous (it is an ordinary function, not created with async def), we call it using the loop.run_in_executor command - that will make the thread creation and managing arrangements for us. The difference here is that the main code the pin code then is free to run while the api call is awaited.