Skip to content

funcs

This module and original idea is by cid-chan (Sarah cid@cid-chan.moe)

Functions:

gather async

gather(*coroutines: AnyCoroutine[T0, T]) -> tuple[T, ...]
Source code
43
44
async def gather(*coroutines: AnyCoroutine[T0, T]) -> tuple[T, ...]:
    return await GatherRequests(coroutines)

get_frame async

get_frame(clip: VideoNode, frame_no: int) -> VideoFrame
Source code
21
22
async def get_frame(clip: vs.VideoNode, frame_no: int) -> vs.VideoFrame:
    return await SingleFrameRequest(clip, frame_no)

get_frames async

get_frames(*clips: VideoNode, frame_no: int) -> tuple[VideoFrame, ...]
Source code
25
26
async def get_frames(*clips: vs.VideoNode, frame_no: int) -> tuple[vs.VideoFrame, ...]:
    return await wait(get_frame(clip, frame_no) for clip in clips)

get_frames_shifted async

get_frames_shifted(
    clip: VideoNode,
    frame_no: int,
    shifts: int | tuple[int, int] | Iterable[int] = (-1, 1),
) -> tuple[VideoFrame, ...]
Source code
29
30
31
32
33
34
35
36
37
38
39
40
async def get_frames_shifted(
    clip: vs.VideoNode, frame_no: int, shifts: int | tuple[int, int] | Iterable[int] = (-1, 1)
) -> tuple[vs.VideoFrame, ...]:
    if isinstance(shifts, int):
        shifts = (-shifts, shifts)

    if isinstance(shifts, tuple):
        start, stop = shifts
        step = -1 if stop < start else 1
        shifts = range(start, stop + step, step)

    return await wait(get_frame(clip, frame_no + shift) for shift in shifts)

wait async

wait(coroutines: Iterable[AnyCoroutine[T0, T]]) -> tuple[T, ...]
Source code
47
48
async def wait(coroutines: Iterable[AnyCoroutine[T0, T]]) -> tuple[T, ...]:
    return await GatherRequests(tuple(coroutines))