# License along with Hooke. If not, see
# <http://www.gnu.org/licenses/>.
-"""The ``command_stack`` module provides tools for managing stacks of
-:class:`~hooke.engine.CommandMessage`\s and applying those stacks to
-:class:`~hooke.curve.Curve`\s.
+"""The ``command_stack`` module provides tools for managing and
+executing stacks of :class:`~hooke.engine.CommandMessage`\s.
"""
-from .command import Command, Argument, Failure
-from .engine import CommandMessage
-from .playlist import NoteIndexList
+
+class CommandStack (list):
+ """Store a stack of commands.
+
+ Examples
+ --------
+ >>> from .engine import CommandMessage
+
+ Define two dummy commands for testing.
+
+ >>> class CommandA (object):
+ ... name = 'CommandA'
+ >>> ca = CommandA()
+ >>> class CommandB (CommandA):
+ ... name = 'CommandB'
+ >>> cb = CommandB()
+
+ Show off `CommandStack`\s functionality.
+
+ >>> c = CommandStack([CommandMessage(ca, {'param':'A'})])
+ >>> c.append(CommandMessage(cb, {'param':'B'}))
+ >>> c.append(CommandMessage(ca, {'param':'C'}))
+ >>> c.append(CommandMessage(cb, {'param':'D'}))
+
+ Implement a dummy :meth:`_execute` for testing. This would
+ usually call :meth:`hooke.command.Command.run` with appropriate
+ arguments.
+
+ >>> def execute(command_message):
+ ... cm = command_message
+ ... print 'EXECUTE', cm.command.name, cm.arguments
+ >>> c._execute = execute
+
+ >>> c.execute() # doctest: +ELLIPSIS
+ EXECUTE CommandA {'param': 'A'}
+ EXECUTE CommandB {'param': 'B'}
+ EXECUTE CommandA {'param': 'C'}
+ EXECUTE CommandB {'param': 'D'}
+
+ :meth:`filter` allows you to select which commands get executed.
+ If, for example, you are applying a set of commands to the current
+ :class:`~hooke.curve.Curve`, you may only want to execute
+ instances of :class:`~hooke.plugin.curve.CurveCommand`. Here we
+ only execute instances of `CommandB`.
+
+ >>> def filter(command_message):
+ ... return isinstance(command_message.command, CommandB)
+ >>> c.filter = filter
+
+ Apply the stack to the current curve
+ >>> c.execute() # doctest: +ELLIPSIS
+ EXECUTE CommandB {'param': 'B'}
+ EXECUTE CommandB {'param': 'D'}
+ """
+ def execute(self, *args, **kwargs):
+ """Execute a stack of commands.
+
+ See Also
+ --------
+ _execute, filter
+ """
+ for command_message in self:
+ if self.filter(command_message) == True:
+ self._execute(command_message, *args, **kwargs)
+
+ def filter(self, command_message):
+ """Any commands in the stack that are not subclasses of
+ :class:`~hooke.plugin.curve.CurveCommand` are ignored.
+ """
+ return True
+
+ def _execute(self, command_message):
+ raise NotImplementedError()
import os.path
import numpy
+from .command_stack import CommandStack
+
class NotRecognized (ValueError):
def __init__(self, curve):
`experiment`. These are two strings that can be used by Hooke
commands/plugins to understand what they are looking at.
- * `.info['filetype']` should contain the name of the exact
+ * :attr:`info['filetype']` should contain the name of the exact
filetype defined by the driver (so that filetype-speciofic
commands can know if they're dealing with the correct filetype).
- * `.info['experiment']` should contain an instance of a
+ * :attr:`info['experiment']` should contain an instance of a
:class:`hooke.experiment.Experiment` subclass to identify the
experiment type. For example, various
:class:`hooke.driver.Driver`\s can read in force-clamp data, but
Hooke commands could like to know if they're looking at force
clamp data, regardless of their origin.
+
+ Another important attribute is :attr:`command_stack`, which holds
+ a :class:`~hooke.command_stack.CommandStack` listing the commands
+ that have been applied to the `Curve` since loading.
"""
def __init__(self, path, info=None):
#the data dictionary contains: {name of data: list of data sets [{[x], [y]}]
info = {}
self.info = info
self.name = os.path.basename(path)
+ self.command_stack = CommandStack()
def identify(self, drivers):
"""Identify the appropriate :class:`hooke.driver.Driver` for
return
raise NotRecognized(self)
- def load(self):
+ def load(self, *args, **kwargs):
"""Use the driver to read the curve into memory.
+
+ Also runs any commands in :attr:`command_stack`. All
+ arguments are passed through to
+ :meth:`hooke.command_stack.CommandStack.execute`.
"""
data,info = self.driver.read(self.path, self.info)
self.data = data
for key,value in info.items():
self.info[key] = value
+ self.command_stack.execute(*args, **kwargs)
def unload(self):
"""Release memory intensive :attr:`.data`.
c._index = 0
return c
+
class Playlist (NoteIndexList):
"""A :class:`NoteIndexList` of :class:`hooke.curve.Curve`\s.
self.drivers = drivers
self._loaded = [] # List of loaded curves, see :meth:`._setup_item`.
self._max_loaded = 100 # curves to hold in memory simultaneously.
+ self._curve_load_args = ()
+ self._curve_load_kwargs = {}
def append_curve_by_path(self, path, info=None, identify=True):
path = os.path.normpath(path)
self.append(c)
return c
+ def set_curve_load_args(self, *args, **kwargs):
+ self._curve_load_args = args
+ self._curve_load_kwargs = kwargs
+
def _setup_item(self, curve):
if curve != None and curve not in self._loaded:
if curve not in self:
if curve.driver == None:
c.identify(self.drivers)
if curve.data == None:
- curve.load()
+ curve.load(*self._curve_load_args, **self._curve_load_kwargs)
self._loaded.append(curve)
if len(self._loaded) > self._max_loaded:
oldest = self._loaded.pop(0)
oldest.unload()
+
class FilePlaylist (Playlist):
version = '0.1'
projects / hooke.git / commitdiff