import contextlib, threading, time
from . import adaptor, animation_threading, runner
from .. util import deprecated, log
from .. colors import palettes, legacy_palette
from .. project import attributes, fields
[docs]class Animation(object):
free_run = False
pre_recursion = fields.default_converter
subframes = 1
top_level = True
if deprecated.allowed():
_step = 0
FAIL_ON_EXCEPTION = False
COLOR_DEFAULTS = ()
[docs] @classmethod
def construct(cls, project, *, run=None, name=None, data=None, **desc):
"""
Construct an animation, set the runner, and add in the two
"reserved fields" `name` and `data`.
"""
from . failed import Failed
exception = desc.pop('_exception', None)
if exception:
a = Failed(project.layout, desc, exception)
else:
try:
a = cls(project.layout, **desc)
a._set_runner(run or {})
except Exception as e:
if cls.FAIL_ON_EXCEPTION:
raise
a = Failed(project.layout, desc, e)
a.name = name
a.data = data
return a
def __init__(self, layout, *, preclear=True, fail_on_exception=None, **kwds):
"""
Arguments:
preclear: If True, clear the layout before rendering the frame;
otherwise, the results of the previous frame are preserved
fail_on_exception: If False, exceptions thrown in the animation frame are
caught and reported;
if True, exceptions are are raised, potentially ending the
animation cycle and the program;
if None or not set, the value of Animation.FAIL_ON_EXCEPTION is used
"""
self.palette = legacy_palette.pop_legacy_palette(
kwds, *self.COLOR_DEFAULTS)
self.palette.length = layout.numLEDs
attributes.set_reserved(self, 'animation', **kwds)
self.layout = layout
assert layout
self.internal_delay = None
self.on_completion = None
self.state = runner.STATE.ready
self.preclear = preclear
self.runner = None
self.time = time.time
self.preframe_callbacks = []
self.fail_on_exception = self.FAIL_ON_EXCEPTION if fail_on_exception is None else fail_on_exception
[docs] def set_project(self, project):
self.time = project.clock.time
self.runner.set_project(project)
self.threading.set_project(project)
@property
def _led(self):
"""Many BiblioPixelAnimations use the "protected" variable _led."""
return self.layout
@_led.setter
def _led(self, layout):
self.layout = layout
@property
def color_list(self):
return self.layout.color_list
@color_list.setter
def color_list(self, cl):
self.layout.color_list = cl
@property
def completed(self):
"""Many BiblioPixelAnimations use the old `completed` variable."""
return self.state == runner.STATE.complete
@completed.setter
def completed(self, state):
self.state = runner.STATE.complete if state else runner.STATE.running
[docs] def pre_run(self):
pass
@property
def title(self):
return self.__class__.__name__
[docs] def step(self, amt=1):
pass
[docs] def cleanup(self, clean_layout=True):
self.threading.cleanup()
# Some cases we may not want to clear the screen
# Like with the remote, it would flash between anims
if clean_layout:
self.layout.cleanup()
[docs] def add_preframe_callback(self, callback):
"""
The preframe_callbacks are called right before the start of a
frame rendering pass.
To avoid race conditions when editing values, the ``Project``
adds a callback here for the top-level animation, to drain the
edit_queue at a moment where no rendering is
happening.
"""
self.preframe_callbacks.append(callback)
[docs] def start(self):
self.threading.start()
[docs] def stop(self):
self.threading.stop_event.set()
[docs] def join(self, timeout=None):
self.threading.join(timeout)
[docs] def run_all_frames(self):
for i in self.generate_frames():
pass
[docs] def generate_frames(self, clean_layout=True):
self._pre_run()
try:
if self.runner.repeats != 0:
while self.state == runner.STATE.running:
self._run_one_frame()
yield
finally:
self.cleanup(clean_layout)
self.on_completion and self.on_completion(self.state)
self.state = runner.STATE.ready
def _run_one_frame(self):
if self.top_level:
timestamps = [self.time()]
for cb in self.preframe_callbacks:
cb()
self.step(self.runner.amt)
if self.top_level:
timestamps.append(self.time())
self.layout.frame_render_time = timestamps[1] - timestamps[0]
self.layout.push_to_driver()
timestamps.append(self.time())
_report_framerate(timestamps)
self.cur_step += 1
if self.state == runner.STATE.complete and self.runner.max_cycles > 0:
if self.cycle_count < self.runner.max_cycles - 1:
self.cycle_count += 1
self.state = runner.STATE.running
if self.top_level:
self.threading.wait(self.sleep_time / self.subframes, timestamps)
if self.threading.stop_event.isSet():
self.state = runner.STATE.canceled
else:
self.state = self.runner.compute_state(self.cur_step, self.state)
def _pre_run(self):
self.state = runner.STATE.running
self.runner.run_start_time = self.time()
self.threading.stop_event.clear()
self.cur_step = 0
self.cycle_count = 0
if self.free_run:
self.sleep_time = 0
elif self.internal_delay:
self.sleep_time = self.internal_delay
else:
self.sleep_time = self.runner.sleep_time
adaptor.adapt_animation_layout(self)
self.preclear and self.layout.all_off()
self.pre_run()
def _set_runner(self, run):
self.runner = runner.Runner(**(run or {}))
self.threading = animation_threading.AnimationThreading(
self.runner, self.run_all_frames)
[docs] def run(self, **kwds):
deprecated.deprecated('BaseAnimation.run')
self._set_runner(kwds)
self.layout.start()
self.start()
if deprecated.allowed(): # pragma: no cover
BaseAnimation = Animation
def _report_framerate(timestamps):
total_time = timestamps[-1] - timestamps[0]
fps = int(1.0 / max(total_time, 0.001))
log.frame("%dms/%dfps / Frame: %dms / Update: %dms",
1000 * total_time,
fps,
1000 * (timestamps[1] - timestamps[0]),
1000 * (timestamps[2] - timestamps[1]))