import threading, time
from .. util import log
[docs]class AnimationThreading:
"""
AnimationThreading handles threading - and eventually multiprocessing - for
Animation.
"""
def __init__(self, runner, run):
self.runner = runner
self.run = run
self.stop_event = threading.Event()
self.thread = None
self.frame_overrun = False
self.sleep = time.sleep
[docs] def set_project(self, project):
self.sleep = project.clock.sleep
[docs] def join(self, timeout=None):
self.thread and self.thread.join(timeout)
[docs] def is_alive(self):
return self.thread and self.thread.is_alive()
[docs] def cleanup(self):
# if current thread is animation thread this was called
# by the context manager and thread is therefore already stopped
self.stop_event.set()
[docs] def target(self):
is_main = threading.current_thread() is threading.main_thread()
log.debug('Animation starts on %s thread', 'main' if is_main else 'new')
self.run()
log.debug('Animation complete')
[docs] def wait(self, wait_time, timestamps):
if not wait_time:
return
elapsed_time = timestamps[-1] - timestamps[0]
if elapsed_time >= wait_time + 0.001:
logger = log.frame if self.frame_overrun else log.warning
logger('Frame-time of %dms set, but took %dms',
1000 * wait_time, 1000 * elapsed_time)
self.frame_overrun = True
elif self.runner.threaded:
self.stop_event.wait(wait_time - elapsed_time)
else:
self.sleep(wait_time - elapsed_time)
[docs] def start(self):
self.stop_event.clear()
def start_thread(target):
self.thread = threading.Thread(target=target, daemon=True)
self.thread.start()
if self.runner.main:
if self.runner.threaded:
start_thread(self.target)
self.runner.main()
else:
start_thread(self.runner.main)
self.target()
elif self.runner.threaded:
start_thread(self.target)
else:
self.target()
self.thread and self.runner.main and self.thread.join()
[docs] def stopped(self):
# DEPRECATED
return not self.is_alive()