API Reference

class Clock(initial_time=0)

anim_attrs(obj, *, duration, step=0, transition=<function Clock._linear>, output_seq_type=<class 'tuple'>, **animated_properties) → Awaitable

Animates attibutes of any object.

import types

obj = types.SimpleNamespace(x=0, size=(200, 300))
await clock.anim_attrs(obj, x=100, size=(400, 400), duration=2)

The output_seq_type parameter.

obj = types.SimpleNamespace(size=(200, 300))
await clock.anim_attrs(obj, size=(400, 400), duration=2, output_seq_type=list)
assert type(obj.size) is list

anim_attrs_abbr(obj, *, d, s=0, t=<function Clock._linear>, output_seq_type=<class 'tuple'>, **animated_properties) → Awaitable

anim_attrs() cannot animate attributes named step, duration and transition but this one can.

async anim_with_dt(*, step=0) → AsyncIterator[TimeUnit]

An async form of schedule_interval().

async for dt in clock.anim_with_dt(step=10):
    print(dt)
    if some_condition:
        break

The code above is quivalent to the below.

def callback(dt):
    print(dt)
    if some_condition:
        event.cancel()

event = clock.schedule_interval(callback, 10)

Restriction

You are not allowed to perform any kind of async operations during the loop.

async for dt in clock.anim_with_dt():
    await awaitable  # NOT ALLOWED
    async with async_context_manager:  # NOT ALLOWED
        ...
    async for __ in async_iterator:  # NOT ALLOWED
        ...

This is also true of other anim_with_xxx APIs.

async anim_with_dt_et(*, step=0) → AsyncIterator[tuple[TimeUnit, TimeUnit]]

anim_with_dt() and anim_with_et() combined.

async for dt, et in clock.anim_with_dt_et():
    ...

async anim_with_dt_et_ratio(*, base, step=0) → AsyncIterator[tuple[TimeUnit, TimeUnit, float]]

anim_with_dt(), anim_with_et() and anim_with_ratio() combined.

async for dt, et, p in clock.anim_with_dt_et_ratio(...):
    ...

Changed in version 0.5.0: The duration parameter was replaced with the base parameter. The loop no longer stops when the progression reaches 1.0.

async anim_with_et(*, step=0) → AsyncIterator[TimeUnit]

async for et in clock.anim_with_et():
    print(et)

The code above is equivalent to the below.

et = 0
async for dt in clock.anim_with_dt():
    et += dt
    print(et)

async anim_with_ratio(*, base, step=0) → AsyncIterator[float]

async for p in clock.anim_with_ratio(base=100):
    print(p * 100, "%")

The code above is equivalent to the below.

base = 100
async for et in clock.anim_with_et():
    print(et / base * 100, "%")

If you want to progress at a non-consistant rate, you may find the source code of the kivy.animation.AnimationTransition helpful.

async for p in clock.anim_with_ratio(base=...):
    p = p * p  # quadratic
    print(p * 100, "%")

Changed in version 0.5.0: The duration parameter was replaced with the base parameter. The loop no longer stops when the progression reaches 1.0.

property current_time: TimeUnit

async interpolate_scalar(start, end, *, duration, step=0, transition=<function Clock._linear>) → AsyncIterator

Interpolates between the values start and end in an async-manner.

async for v in clock.interpolate(0, 100, duration=100, step=30):
    print(int(v))

elapsed time	output
0	0
30	30
60	60
90	90
120	100

async interpolate_seq(start, end, *, duration, step=0, transition=<function Clock._linear>, output_type=<class 'tuple'>) → AsyncIterator

An alias for interpolate_sequence().

async interpolate_sequence(start, end, *, duration, step=0, transition=<function Clock._linear>, output_type=<class 'tuple'>) → AsyncIterator

Same as interpolate_scalar() except this one is for sequence type.

async for v in clock.interpolate_sequence([0, 50], [100, 100], duration=100, step=30):
    print(v)

elapsed time	output
0	(0, 50)
30	(30, 65)
60	(60, 80)
90	(90, 95)
120	(100, 100)

move_on_after(timeout) → AbstractAsyncContextManager[Task]

Returns an async context manager that applies a time limit to its code block, like trio.move_on_after() does.

async with clock.move_on_after(10) as timeout_tracker:
    ...

if timeout_tracker.finished:
    print("The code block was interrupted due to a timeout")
else:
    print("The code block exited gracefully.")

n_frames(n: int) → Awaitable

Waits for a specified number of times the tick() to be called.

await clock.n_frames(2)

If you want to wait for one time, sleep() is preferable for a performance reason.

await clock.sleep(0)

async run_in_executor(executer: ThreadPoolExecutor, func, *, polling_interval) → Awaitable

Runs a function within a concurrent.futures.ThreadPoolExecutor, and waits for the completion of the function.

executor = ThreadPoolExecutor()
return_value = await clock.run_in_executor(executor, func, polling_interval=...)

async run_in_thread(func, *, daemon=None, polling_interval) → Awaitable

Creates a new thread, runs a function within it, then waits for the completion of that function.

return_value = await clock.run_in_thread(func, polling_interval=...)

schedule_interval(func, interval) → ClockEvent

Schedules the func to be called repeatedly at a specified interval.

sleep(duration) → Awaitable

Waits for a specified period of time.

await clock.sleep(10)

tick(delta_time)

Advances the clock time and triggers scheduled events accordingly. The delta_time must be 0 or greater.

Warning

Don’t call this method recursively.

class ClockEvent

callback: Callable[[TimeUnit], None]

The callback function registered using the Clock.schedule_xxx() call that returned this instance. You can replace it with another one by simply assigning to this attribute.

event = clock.schedule_xxx(...)
event.callback = another_function

cancel()