Connection broken if connected in an new thread

[xXxNIKIxXx]

Describe the bug

When running a new thread, and then connectiong to an device (HomePod) if you try to use stream_file the connection is apperently broken.

Error log

Traceback (most recent call last):
  File "C:\venv\python\3\venv human reaction\lib\site-packages\pyatv\protocols\raop\audio_source.py", line 263, in get_buffered_io_metadata
    return await get_metadata(buffer)
  File "C:\venv\python\3\venv human reaction\lib\site-packages\pyatv\support\metadata.py", line 25, in get_metadata
    in_file = await loop.run_in_executor(None, _open_file, file)
  File "C:\Users\niklas\AppData\Local\Programs\Python\Python39\lib\asyncio\base_events.py", line 814, in run_in_executor
    executor = concurrent.futures.ThreadPoolExecutor(
  File "C:\Users\niklas\AppData\Local\Programs\Python\Python39\lib\concurrent\futures\__init__.py", line 49, in __getattr__
    from .thread import ThreadPoolExecutor as te
  File "C:\Users\niklas\AppData\Local\Programs\Python\Python39\lib\concurrent\futures\thread.py", line 37, in <module>
    threading._register_atexit(_python_exit)
  File "C:\Users\niklas\AppData\Local\Programs\Python\Python39\lib\threading.py", line 1414, in _register_atexit
    raise RuntimeError("can't register atexit after shutdown")
RuntimeError: can't register atexit after shutdown
Exception in thread Thread-3:
Traceback (most recent call last):
  File "C:\Users\niklas\AppData\Local\Programs\Python\Python39\lib\threading.py", line 980, in _bootstrap_inner
    self.run()
  File "C:\Users\niklas\AppData\Local\Programs\Python\Python39\lib\threading.py", line 917, in run
    self._target(*self._args, **self._kwargs)
  File "c:\Users\niklas\Python Util\code snipets\AirPlay\test3.py", line 56, in async_thread
    asyncio.run(main(device))
  File "C:\Users\niklas\AppData\Local\Programs\Python\Python39\lib\asyncio\runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "C:\Users\niklas\AppData\Local\Programs\Python\Python39\lib\asyncio\base_events.py", line 647, in run_until_complete
    return future.result()
  File "c:\Users\niklas\Python Util\code snipets\AirPlay\test3.py", line 48, in main
    await atv.stream.stream_file(process.stdout)
  File "C:\venv\python\3\venv human reaction\lib\site-packages\pyatv\core\facade.py", line 371, in stream_file
    await self.relay("stream_file")(
  File "C:\venv\python\3\venv human reaction\lib\site-packages\pyatv\protocols\raop\__init__.py", line 360, in stream_file
    audio_file = await open_source(
  File "C:\venv\python\3\venv human reaction\lib\site-packages\pyatv\protocols\raop\audio_source.py", line 732, in open_source
    return await BufferedIOBaseSource.open(source, sample_rate, channels, sample_size)
  File "C:\venv\python\3\venv human reaction\lib\site-packages\pyatv\protocols\raop\audio_source.py", line 338, in open
    src = await loop.run_in_executor(
  File "C:\Users\niklas\AppData\Local\Programs\Python\Python39\lib\asyncio\base_events.py", line 814, in run_in_executor
    executor = concurrent.futures.ThreadPoolExecutor(
  File "C:\Users\niklas\AppData\Local\Programs\Python\Python39\lib\concurrent\futures\__init__.py", line 49, in __getattr__
    from .thread import ThreadPoolExecutor as te
  File "C:\Users\niklas\AppData\Local\Programs\Python\Python39\lib\concurrent\futures\thread.py", line 37, in <module>
    threading._register_atexit(_python_exit)
  File "C:\Users\niklas\AppData\Local\Programs\Python\Python39\lib\threading.py", line 1414, in _register_atexit
    raise RuntimeError("can't register atexit after shutdown")
RuntimeError: can't register atexit after shutdown

How to reproduce the bug?

Have an function call the async function were i connect to an device and then strem audio. The start an new thread with the function fich is calling the async funtion.

from pyatv import scan, connect
import asyncio
import threading
import asyncio.subprocess as asp

async def main(device):
    loop = asyncio.get_event_loop()

    atv = await connect(device, loop)

    process = await asp.create_subprocess_exec("C:\\Program Files\\ffmpeg\\ffmpeg.exe", "-f", "dshow", "-i", "audio=Home Pod (VB-Audio Virtual Cable)", "-acodec", "libmp3lame", "-f", "mp3", "-", stdout=asp.PIPE, stderr=None)
    await atv.stream.stream_file(process.stdout)

def async_thread(device):
    asyncio.run(main(device))

devices = await scan(asyncio.get_event_loop())

thread = threading.Thread(target=async_thread, args=(devices[0],))
thread.start()

What is expected behavior?

The stream of ffmpeg should be normaly streamed to the device like it is when i am not in an new thread.

Operating System

Windows

Python

3.9

pyatv

0.14.5

Device

HomePod

Additional context

-

[postlund]

I don't think you can run the asyncio event loop from another thread (at least not yet). So I would say that is incorrect usage of asyncio. What are you trying to accomplish?

[xXxNIKIxXx]

I want to have an list of all HomePods and the can select one or multiple HomePods, were an live audio stream of my system Audio is played. I have found an workaround, wich is not that nice, but it works. I create like an thread with another process in it. And then it works.

[postlund]

You can do that in asyncio, but you don't use threads. Instead use tasks. Something like this will stream to all HomePods and wait for it to finish:

from pyatv import scan, connect, const
import asyncio
import asyncio.subprocess as asp

TARGET_MODELS = [const.DeviceModel.HomePod, const.DeviceModel.HomePodGen2, const.DeviceModel.HomePodMini]

async def stream_to_device(device):
    loop = asyncio.get_event_loop()

    atv = await connect(device, loop)

    process = await asp.create_subprocess_exec("C:\\Program Files\\ffmpeg\\ffmpeg.exe", "-f", "dshow", "-i", "audio=Home Pod (VB-Audio Virtual Cable)", "-acodec", "libmp3lame", "-f", "mp3", "-", stdout=asp.PIPE, stderr=None)
    await atv.stream.stream_file(process.stdout)

async def main():
    devices = await scan(asyncio.get_event_loop())
    streams = []
    for device in devices:
        if device.device_info.model in TARGET_MODELS:
            streams.append(asyncio.create_task(stream_to_device(device)))

    await asyncio.gather(*streams)

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

I have not tested this, but should be mostly correct.