API Reference

class CommonParams

Annotate **kwargs to improve auto-completion support.

from typing import Unpack

async def some_func(**kwargs: Unpack[CommonParams]):
    ...

clock: Clock

draw_target: Surface

executor: PriorityExecutor

pygame_clock: Clock

sdlevent: SDLEvent

class PriorityExecutor

Calls registered functions in the order of their priorities.

executor = PriorityExecutor()

values = []
executor.register(lambda: values.append('A'), priority=2)
executor.register(lambda: values.append('B'), priority=0)
executor()
assert values == ['B', 'A', ]
values.clear()

executor.register(lambda: values.append('C'), priority=1)
executor()
assert values == ['B', 'C', 'A', ]

The asyncpygame.run() creates an instance of this class and calls its __call__() every frame.

register(func, priority) → ExecutionRequest

class SDLEvent

# Waits for any mouse button to be pressed.
e = await sdlevent.wait(MOUSEBUTTONDOWN)

# Waits for any mouse button to be pressed or released.
e = await sdlevent.wait(MOUSEBUTTONDOWN, MOUSEBUTTONUP)

# Waits for the left mouse button to be pressed.
e = await sdlevent.wait(MOUSEBUTTONDOWN, filter=lambda e: e.button == 1)

dispatch(event: Event)

イベントの発生を待っているタスクにイベントを通知する。 asyncpygame.run() のみがこれを呼ぶべきでありアプリ側からは呼ぶべきではない。

subscribe(topics, callback, priority) → Subscriber

async型APIの礎となっているコールバック型API。直接触るべきではない。

wait(*event_types, filter=<function _accept_any>, priority, consume=False) → Awaitable[Event]

Waits for any of the specified types of events to occur.

wait_freq(*event_types, filter=<function _accept_any>, priority, consume=False)

MOUSEMOTION や FINGERMOTION などの頻りに起こりうるイベントを効率良く捌けるかもしれない機能。 以下のようなコードは

while True:
    e = await sdlevent.wait(FINGERMOTION)
    ...

以下のように書き換える事で効率が良くなるかもしれません。

async with sdlevent.wait_freq(FINGERMOTION) as finger_motion:
    while True:
        e = await finger_motion()
        ...

ただ代償としてコードが読みにくくなるうえ逆に効率が悪くなる可能性すらあるので人間が感知できる程の改善がみられない限りは使用を控えてください。 また多くの状況では FINGERUP が起きた時にループを抜けたいでしょうから典型的な用例は以下のようになります。

async with (
    asyncpygame.move_on_when(sdlevent.wait(FINGERUP)),
    sdlevent.wait_freq(FINGERMOTION) as finger_motion,
):
    while True:
        e = await finger_motion()
        ...

注意点としてas節で得た関数(上のコードでいう finger_motion)は必ずそのwithブロック内で直に用いるようにしてください。 以下の様に別の関数に渡したり、戻り値を別の関数に渡してはいけません。

async with sdlevent.wait_freq(FINGERMOTION) as finger_motion:
    another_func(finger_motion)  # NOT ALLOWED
    another_func(finger_motion())   # NOT ALLOWED

block_input_events(sdlevent: SDLEvent, priority) → ContextManager[Subscriber]

Returns a context manager that blocks input events.

with block_input_events(sdlevent, priority):
    ...

Warning

The context manager starts having the effect when it is created, not when its __enter__() method is called.

async capture_current_frame(executor: PriorityExecutor, priority, source: Surface) → Awaitable[Surface]

surface = await capture_current_frame(executor, priority, source)

quit(*args)

run(main_func, *, fps=30, auto_quit=True)

run_and_record(main_func, *, fps=30, auto_quit=True, outfile='./output.mkv', overwrite=False, outfile_options: Iterator[str] = ['-codec:v', 'libx265', '-qscale:v', '0'])

Runs the program while recording the screen to a video file using ffmpeg. Requires numpy.

# H.265/HEVC, maximum quality (default)
run_and_record(..., outfile_options=r"-codec:v libx265 -qscale:v 0".split())

# H.265/HEVC, lossless compression
run_and_record(..., outfile_options=r"-codec:v libx265 -x265-params lossless=1".split())

# WebP, lossless compression, infinite loop
run_and_record(..., outfile_options=r"-codec:v libwebp -lossless 1 -loop 0".split(), outfile="./output.webp")

(sub module) scene_switcher

class FadeTransition(*, overlay_color='black', out_duration=300, in_duration=300, interval=100)

scene_switcher.switch_to(next_scene, FadeTransition())

Parameters:

• out_duration – The duration of the fadeout animation.

• in_duration – The duration of the fadein animation.

• interval – The interval between the fadeout animation and the fadein animation.

class SceneSwitcher

async run(first_scene, *, userdata: Any = None, priority, sdlevent, **kwargs)

Parameters:

userdata – Use this to share data between scenes without relying on global variables.

switch_to(next_scene, transition: ~collections.abc.Callable[[...], ~collections.abc.AsyncGenerator[None, None]] = <function no_transition>)

Instructs the scene switcher to transition to another scene. Calling this method during an ongoing transition will have no effect.

class SlideTransition(*, direction='right', duration=800)

scene_switcher.switch_to(next_scene, SlideTransition())

Parameters:

• direction – ‘right’, ‘left’, ‘up’ or ‘down’.

• duration – The duration of the sliding animation.

Transition

Transition is a collections.abc.Callable that returns an async generator that yields twice. It defines a transition between two scenes.

async def my_transition(**common_params):
    # 1st part
    yield
    # 2nd part
    yield
    # 3rd part

scene_switcher.switch_to(next_scene, my_transition)

When your app switches from one scene to another, the transition between them will proceed as follows:

• The 1st part of the transition and the current scene run concurrently.

• The current scene gets cancelled.

• The 2nd part of the transition runs.

• The next scene starts.

• The 3rd part of the transition and the next scene run concurrently.

time ------------------------------------------------------>

----------------------|          |------------------------
    current scene     |          |      next scene
----------------------|          |------------------------

           |----------|----------|----------|
           | 1st part | 2nd part | 3rd part |
           |----------|----------|----------|

alias of Callable[[…], AsyncGenerator[None, None]]

async no_transition(**common_params)

scene_switcher.switch_to(next_scene, no_transition)