Source code for rcp.script
"""Objects related to performance scripting.
"""
################################################################
# Written in 2019 by Garth Zeglin <garthz@cmu.edu>
# To the extent possible under law, the author has dedicated all copyright
# and related and neighboring rights to this software to the public domain
# worldwide. This software is distributed without any warranty.
# You should have received a copy of the CC0 Public Domain Dedication along with this software.
# If not, see <http://creativecommons.org/publicdomain/zero/1.0/>.
################################################################
# standard Python libraries
import math, logging, functools, time, queue, threading
# set up logger for module.
log = logging.getLogger('script')
# N.B. the normal logger doesn't yet work from background threads since the
# default formatter writes to a Qt object only available on the main thread.
# filter out most logging; the default is NOTSET which passes along everything
log.setLevel(logging.INFO)
################################################################
[docs]class ScriptTimeoutException(Exception):
pass
[docs]class ScriptStopException(Exception):
pass
################################################################
[docs]class Script(object):
"""A script is a procedure run as a separate thread which is permitted to block.
This supports a simpler notation for linear action sequences as normal
procedures instead of callback state machines. The script thread
communicates with the main thread solely using message queues to avoid
synchronization problems.
:ivars input: unified input queue; each item is a tuple in which the first keyword identifies the message type
:ivars output: unified output queue; each item is a tuple in which the first keyword identifies the message type
"""
def __init__(self):
self.input = queue.Queue()
self.output = queue.Queue()
return
[docs] def start(self):
"""Start the script process in the background so will run asynchronously. This
function returns immediately."""
self.thread = threading.Thread(target=self.script_task)
self.thread.daemon = True
self.thread.start()
return
[docs] def write(self, string):
"""Internal method to send a console message to the main thread."""
self.output.put(('console', string))
[docs] def script_task(self):
"""Entry point for the script to run on a background thread. The default
implementation does nothing, this should be overridden in child
classes.
"""
pass
# ===============================================================
[docs] def event_wait(self, predicate, timeout):
"""Consume input messages until one is received which satisfies the predicate or
the timeout is reached. Returns the non-False value of the predicate or
the string 'timeout'.
"""
start = time.time()
remaining = timeout
while remaining > 0:
try:
command = self.input.get(block=True, timeout=remaining)
# apply the predicate and return any non-False value
value = predicate(command)
if value is not False:
return value
self.write("event_wait ignoring message " + repr(command))
remaining = start + timeout - time.time()
except queue.Empty:
# if the queue 'get' timed out, fall out of the loop
remaining = 0
return 'timeout'
# ===============================================================
def _user_stop_predicate(self, command):
if command[0] == 'console' and command[1] == 'stop':
return 'user_stop'
else:
return False
[docs] def sleep(self, duration):
"""Sleep for a given duration unless the user issues a stop command. Returns
nothing or raises a ScriptStopException."""
value = self.event_wait(self._user_stop_predicate, duration)
if value == 'user_stop':
raise ScriptStopException
else:
return
def _motion_stop_predicate(self, command):
if command[0] == 'status' and command[1] == 'stopped':
return 'motion_stop'
else:
return False
[docs] def wait_until_stopped(self, timeout = 20.0):
"""Sleep until the motion generator reports a stopped status or the timeout is reached."""
predicate = lambda command: self._user_stop_predicate(command) or self._motion_stop_predicate(command)
value = self.event_wait(predicate, timeout)
if value == 'user_stop':
raise ScriptStopException
elif value == 'motion_stop':
return
else:
raise ScriptTimeoutException
################################################################