decode_threads example failing on latest from PyPi and on Develop Branch

[samuel-davis]

Hello, just started using this library and loving it. Unfortunately, I really need multithreaded decoding in my program in order for it to function correctly. Do you think you could look into this for me?

Input file path I have given was an absolute path to a video file.

Stack trace I am receiving.

Traceback (most recent call last):
  File "PyAV/examples/decode_threads.py", line 24, in <module>
    for f in pool.map(decode, 'test_video_30FPS.mp4'):
  File "/home/sam/.virtualenvs/pyav/local/lib/python2.7/site-packages/concurrent/futures/_base.py", line 641, in result_iterator
    yield fs.pop().result()
  File "/home/sam/.virtualenvs/pyav/local/lib/python2.7/site-packages/concurrent/futures/_base.py", line 455, in result
    return self.__get_result()
  File "/home/sam/.virtualenvs/pyav/local/lib/python2.7/site-packages/concurrent/futures/thread.py", line 63, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/home/sam/projects/dev-rta/integration_solution_ffmpeg/PyAV/examples/decode_threads.py", line 9, in decode
    container = av.open(input_file)
  File "av/container/core.pyx", line 262, in av.container.core.open
  File "av/container/core.pyx", line 218, in av.container.core.Container.__cinit__
  File "av/container/core.pyx", line 117, in av.container.core.ContainerProxy.__init__
  File "av/container/core.pyx", line 188, in av.container.core.ContainerProxy.err_check
  File "av/utils.pyx", line 105, in av.utils.err_check
av.AVError: [Errno 21] Is a directory: '/'

[samuel-davis]

Really struggling here guys, I'm trying to fix this on my end, but I have very little experience debugging Cython programs. Any help would be greatly appreciated.

[samuel-davis]

So I have tried, building from source, building from older commits. I have tried passing a variety of different inputs, both absolute and relative paths as well as file-like objects opened in a variety of modes. Still no luck.

It's worth noting, that opening regularly in this fashion ,
video = av.open(args.video_input)
works with no problems whatsoever.

But as soon as I run this test, it all fails on a errno_check function.

`import concurrent.futures
import sys

import av

def decode(input_file):

container = av.open(input_file)

stream = container.streams.video[0]
#stream.thread_count = 1

for _ in container.decode(stream):
    pass

return input_file

if name == 'main':

pool = concurrent.futures.ThreadPoolExecutor(4)
for f in pool.map(decode, "test_video_30FPS.mp4"):
    print(f)`

Its like the program only detects the first character of any string I pass in.

[mikeboers]

You need to show more code.

It looks like you've modified the example, which is using concurrent.futures.ThreadPoolExecutor to map a decode function to a list of paths which it decodes each on a thread.

You're passing a single string to map, which will pass each character of the path to the decode function.

If your goal is to have threaded decoding of a single file, and the default setting of inter-frame threading ('SLICE') isn't aggressive enough, you must set thread_type on a CodecContext (via a stream) to 'FRAME' or 'AUTO'.

[samuel-davis]

Well first, thanks for the help boss, really appreciate it. My goal is to have a single source, either video or url stream address decoded in multiple threads in parallel.

I didn't notice that it was sending the path object line by line lol.

Let me know if the following code is going to accomplish what I am trying to do.

import concurrent.futures

import av


def decode(input_file):
    container = av.open(input_file)

    stream = container.streams.video[0]
    # stream.thread_count = 1

    for _ in container.decode(stream):
        pass

    return input_file


if __name__ == '__main__':

    pool = concurrent.futures.ThreadPoolExecutor(4)
    paths = ["tf_test_video_30FPS.mp4", "tf_test_video_30FPS.mp4", "tf_test_video_30FPS.mp4"]
    for f in pool.map(decode, paths):
        print(f)

Seems like that wouldn't work because I would be opening the same file three times right? And then getting the same frames from three streams in parallel ?

[samuel-davis]

Update 1 : Well passing three paths definitely handles part one of the error that was due to me being dumb. So thanks for the help so far haha.

[mikeboers]

You are correct about how that will behave.

There are two types of threading FFmpeg provides: slice and frame. slice has multiple threads decoding one frame. frame has multiple threads decoding multiple frames. FFmpeg makes the threads; you never see them in Python land.

frame tends to be faster, but since it changes the API somewhat (by having more of a delay in decoding) I have not turned it on by default. slice is the default since it doesn't change anything.

You can set stream.codec.thread_type = 'AUTO' or 'FRAME' to try the others.

[samuel-davis]

Let me code something quickly , and if your still available, I would like to run it by you, I have been fighting this for about two weeks and finally came across this library and it was like a light at the end of a tunnel.

I was about to start writing my own filter to FFMPEG which has issues in and of itself for deployment of the program I'm making.

[samuel-davis]

Ok so write a small program to test the different threading pieces. The results look great for getting the frames back out of the video file quickly. the following code is what I did.

import av
import cv2

from utils import CustomFPS

MAX_FRAMES = 100


def do_play_av(input_file, thread_type=None):
    video = av.open(input_file)
    stream = next(s for s in video.streams if s.type == 'video')
    fps = CustomFPS()
    fps.start()
    if thread_type is not None:
        stream.thread_type = thread_type

    for packet in video.demux(stream):
        if fps._numFrames >= MAX_FRAMES:
            cv2.destroyAllWindows()
            break
        for frame in packet.decode():
            fps.update()
            # some other formats gray16be, bgr24, rgb24
            img = frame.to_nd_array(format='bgr24')
            cv2.imshow("Test", img)
        if cv2.waitKey(1) == 27:
            break
    cv2.destroyAllWindows()
    print("FPS for {} Frames {} ".format(MAX_FRAMES, fps.fps()))


if __name__ == '__main__':
    fps = CustomFPS()
    input_file = "supporting_files/tf_test_video_30FPS.mp4"
    do_play_av(input_file=input_file, thread_type=None)
    do_play_av(input_file=input_file, thread_type=1)
    do_play_av(input_file=input_file, thread_type=2)
    do_play_av(input_file=input_file, thread_type=3)

Output was.

FPS for 100 Frames 67.0385069184 
FPS for 100 Frames 96.7104894131 
FPS for 100 Frames 71.3774345951 
FPS for 100 Frames 97.599634587 

It's worth noting, the enums would not work and I had to use integers to specify the type.

[samuel-davis]

That leads me to the last step of really solving my problem, which necessarily isn't something you have to deal with but I would appreciate the help nonetheless. I need to run inference from a machine learning model from these frames as quickly as possible as they are taken out of the video file/stream and then do something with the corresponding results.

I was going to develop a filter for FFMPEG that handled doing this because attempting to do this in Python and Java becomes a parallelism/threading nightmare. Its very easy for me to make the requests, but performing the inference is an expensive operation and I needed to have multiple threads doing the same thing to get viewable playback on the video.

Any advice or maybe a point in the right direction? Is FFMPEG filters a good idea? Or is there a way to potentially solve this completely in Python possibly with your library?

[samuel-davis]

Oh and just for anyone curious later down the road, this is the speed without drawing with Opencv.

FPS for 100 Frames 254.679740227 
FPS for 100 Frames 642.636351368 
FPS for 100 Frames 259.448464454 
FPS for 100 Frames 945.680133152

[mikeboers]

Thank you for posting those results. They caught me by surprise. I had internalized that 3 meant 1 or 2, not 1 AND 2.

Can you tell me more about why the enums didn't work? I'd like to fix that.

As long as your machine learning model is not implemented in Python directly, and releases the GIL, I don't see any downside to doing it in Python. You'll want to be careful to copy data as little as possible (e.g. using the buffer protocols to convert frames into numpy arrays). I'm personally going through that aspect to try to get the copies down to zero, but I'm doing it in my spare time so it is weeks or more from being done.

In my personal projects, I tend to have one Python thread interacting with the video container and demuxing/decoding, and it pushes frames into a queue. Multiple consumer threads pull on that queue and do the processing.

[samuel-davis]

Sure thing, so when I attempt to set the thread type like this

stream.thread_type = 'AUTO'

I get the following output.

Traceback (most recent call last):
  File "/home/sam/tools/pycharm-2017.2.4/helpers/pydev/pydevd.py", line 1599, in <module>
    globals = debugger.run(setup['file'], None, None, is_module)
  File "/home/sam/tools/pycharm-2017.2.4/helpers/pydev/pydevd.py", line 1026, in run
    pydev_imports.execfile(file, globals, locals)  # execute the script
  File "/home/sam/projects/testbed.py", line 41, in <module>
    do_play_av(input_file=input_file, thread_type=1)
  File "/home/sam/projects/testbed.py", line 19, in do_play_av
    stream.thread_type = 'AUTO'
  File "av/stream.pyx", line 108, in av.stream.Stream.__setattr__
  File "av/codec/context.pyx", line 521, in av.codec.context.CodecContext.thread_type.__set__
TypeError: an integer is required

[samuel-davis]

And that's exactly the problems I was having with other solutions for grabbing frames from videos or streams, I would end up having to manipulate the data so much that it would actually slow things down to a crawl.

I'm way more comfortable doing threading and Queueing inside of Java, so it's been a learning experience here lately trying to get this ball rolling in Python but I'm getting a lot closer.

The machine learning model is not in Python perse, its housed within Tensorflow Serving, an external machine learning model server that serves up inference requests via GRPC calls and protobufs, so I'm not sure if it will release the GIL but here's hoping I can get it to play nice for me.