executing stacks of :class:`~hooke.engine.CommandMessage`\s.
"""
+import os
+import os.path
+
+import yaml
+
+from .engine import CommandMessage
+
class CommandStack (list):
"""Store a stack of commands.
Examples
--------
- >>> from .engine import CommandMessage
>>> c = CommandStack([CommandMessage('CommandA', {'param':'A'})])
>>> c.append(CommandMessage('CommandB', {'param':'B'}))
>>> c.append(CommandMessage('CommandA', {'param':'C'}))
EXECUTE CommandC {'param': 'E'}
>>> c.append(cm)
>>> print [repr(cm) for cm in c] # doctest: +NORMALIZE_WHITESPACE
- ["<CommandMessage CommandA {'param': 'A'}>",
- "<CommandMessage CommandB {'param': 'B'}>",
- "<CommandMessage CommandA {'param': 'C'}>",
- "<CommandMessage CommandB {'param': 'D'}>",
- "<CommandMessage CommandC {'param': 'E'}>"]
+ ['<CommandMessage CommandA {param: A}>',
+ '<CommandMessage CommandB {param: B}>',
+ '<CommandMessage CommandA {param: C}>',
+ '<CommandMessage CommandB {param: D}>',
+ '<CommandMessage CommandC {param: E}>']
+
+ There is also a convenience function for clearing the stack.
+
+ >>> c.clear()
+ >>> print [repr(cm) for cm in c]
+ []
"""
def execute(self, hooke):
"""Execute a stack of commands.
def execute_command(self, hooke, command_message):
hooke.run_command(command=command_message.command,
arguments=command_message.arguments)
+
+ def clear(self):
+ while len(self) > 0:
+ self.pop()
+
+
+class FileCommandStack (CommandStack):
+ """A file-backed :class:`CommandStack`.
+ """
+ version = '0.1'
+
+ def __init__(self, *args, **kwargs):
+ super(FileCommandStack, self).__init__(*args, **kwargs)
+ self.name = None
+ self.path = None
+
+ def set_path(self, path):
+ """Set the path (and possibly the name) of the command stack.
+
+ Examples
+ --------
+ >>> c = FileCommandStack([CommandMessage('CommandA', {'param':'A'})])
+
+ :attr:`name` is set only if it starts out equal to `None`.
+ >>> c.name == None
+ True
+ >>> c.set_path(os.path.join('path', 'to', 'my', 'command', 'stack'))
+ >>> c.path
+ 'path/to/my/command/stack'
+ >>> c.name
+ 'stack'
+ >>> c.set_path(os.path.join('another', 'path'))
+ >>> c.path
+ 'another/path'
+ >>> c.name
+ 'stack'
+ """
+ if path != None:
+ self.path = path
+ if self.name == None:
+ self.name = os.path.basename(path)
+
+ def save(self, path=None, makedirs=True):
+ """Saves the command stack to `path`.
+ """
+ self.set_path(path)
+ dirname = os.path.dirname(self.path)
+ if makedirs == True and not os.path.isdir(dirname):
+ os.makedirs(dirname)
+ with open(self.path, 'w') as f:
+ f.write(self.flatten())
+
+ def load(self, path=None):
+ """Load a command stack from `path`.
+ """
+ self.set_path(path)
+ with open(self.path, 'r') as f:
+ text = f.read()
+ self.from_string(text)
+
+ def flatten(self):
+ """Create a string representation of the command stack.
+
+ A playlist is a YAML document with the following syntax::
+
+ - arguments: {param: A}
+ command: CommandA
+ - arguments: {param: B, ...}
+ command: CommandB
+ ...
+
+ Examples
+ --------
+ >>> c = FileCommandStack([CommandMessage('CommandA', {'param':'A'})])
+ >>> c.append(CommandMessage('CommandB', {'param':'B'}))
+ >>> c.append(CommandMessage('CommandA', {'param':'C'}))
+ >>> c.append(CommandMessage('CommandB', {'param':'D'}))
+ >>> print c.flatten()
+ - arguments: {param: A}
+ command: CommandA
+ - arguments: {param: B}
+ command: CommandB
+ - arguments: {param: C}
+ command: CommandA
+ - arguments: {param: D}
+ command: CommandB
+ <BLANKLINE>
+ """
+ return yaml.dump([{'command':cm.command,'arguments':cm.arguments}
+ for cm in self])
+
+ def from_string(self, string):
+ """Load a playlist from a string.
+
+ .. warning:: This is *not safe* with untrusted input.
+
+ Examples
+ --------
+
+ >>> string = '''- arguments: {param: A}
+ ... command: CommandA
+ ... - arguments: {param: B}
+ ... command: CommandB
+ ... - arguments: {param: C}
+ ... command: CommandA
+ ... - arguments: {param: D}
+ ... command: CommandB
+ ... '''
+ >>> c = FileCommandStack()
+ >>> c.from_string(string)
+ >>> print [repr(cm) for cm in c] # doctest: +NORMALIZE_WHITESPACE
+ ['<CommandMessage CommandA {param: A}>',
+ '<CommandMessage CommandB {param: B}>',
+ '<CommandMessage CommandA {param: C}>',
+ '<CommandMessage CommandB {param: D}>']
+ """
+ for x in yaml.load(string):
+ self.append(CommandMessage(command=x['command'],
+ arguments=x['arguments']))
import copy
import hashlib
+import os
import os.path
import types
import xml.dom.minidom
class FilePlaylist (Playlist):
+ """A file-backed :class:`Playlist`.
+ """
version = '0.1'
def __init__(self, drivers, name=None, path=None):
Relative paths are interpreted relative to the location of the
playlist file.
-
+
Examples
--------
self._digest = self.digest()
for curve in self:
curve.set_hooke(hooke)
-
- def save(self, path=None):
+
+ def save(self, path=None, makedirs=True):
"""Saves the playlist in a XML file.
"""
self.set_path(path)
+ dirname = os.path.dirname(self.path)
+ if makedirs == True and not os.path.isdir(dirname):
+ os.makedirs(dirname)
with open(self.path, 'w') as f:
f.write(self.flatten())
self._digest = self.digest()
"""
import logging
+import os.path
from Queue import Queue
from ..command import Command, Argument, Success, Failure
-from ..command_stack import CommandStack
+from ..command_stack import FileCommandStack
+from ..config import Setting
from ..engine import CloseEngine, CommandMessage
from . import Builtin
# The plugin itself
class CommandStackPlugin (Builtin):
+ """Commands for managing a command stack (similar to macros).
+ """
def __init__(self):
super(CommandStackPlugin, self).__init__(name='command_stack')
self._commands = [
PopCommand(self), GetCommand(self), GetStateCommand(self),
SaveCommand(self), LoadCommand(self),
]
- self.command_stack = CommandStack()
- self.path = None
+ self._settings = [
+ Setting(section=self.setting_section, help=self.__doc__),
+ Setting(section=self.setting_section, option='path',
+ value=os.path.join('resources', 'command_stack'),
+ help='Directory containing command stack files.'), # TODO: allow colon separated list, like $PATH.
+ ]
+ self.command_stack = FileCommandStack()
self.state = 'inactive'
# inactive <-> active.
self._valid_transitions = {
'active': ['inactive'],
}
+ def default_settings(self):
+ return self._settings
+
def log(self, msg):
log = logging.getLogger('hooke')
log.debug('%s %s' % (self.name, msg))
def _run(self, hooke, inqueue, outqueue, params):
self._set_state('active')
- self.plugin.command_stack = CommandStack() # clear command stack
+ self.plugin.command_stack = FileCommandStack() # clear command stack
super(StartCaptureCommand, self)._run(hooke, inqueue, outqueue, params)
"""
def __init__(self, plugin):
super(SaveCommand, self).__init__(
- name='save command stack', help=self.__doc__, plugin=plugin)
+ name='save command stack',
+ arguments=[
+ Argument(name='output', type='file',
+ help="""
+File name for the output command stack. Defaults to overwriting the
+input command stack. If the command stack does not have an input file
+(e.g. it is the default) then the file name defaults to `default`.
+""".strip()),
+ ],
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
- pass
+ params = self.__setup_params(hooke, params)
+ self.plugin.command_stack.save(params['output'])
+ def __setup_params(self, hooke, params):
+ if params['output'] == None and self.plugin.command_stack.path == None:
+ params['output'] = 'default'
+ if params['output'] != None:
+ params['output'] = os.path.join(
+ self.plugin.config['path'], params['output'])
+ return params
class LoadCommand (CommandStackCommand):
"""Load the command stack.
+
+ .. warning:: This is *not safe* with untrusted input.
"""
def __init__(self, plugin):
super(LoadCommand, self).__init__(
- name='load command stack', help=self.__doc__, plugin=plugin)
+ name='load command stack',
+ arguments=[
+ Argument(name='input', type='file',
+ help="""
+File name for the input command stack.
+""".strip()),
+ ],
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
- pass
+ params = self.__setup_params(hooke, params)
+ self.plugin.command_stack.clear()
+ self.plugin.command_stack.load(params['input'])
+
+ def __setup_params(self, hooke, params):
+ if params['input'] == None and self.plugin.command_stack.path == None:
+ params['input'] = 'default'
+ if params['input'] != None:
+ params['input'] = os.path.join(
+ self.plugin.config['path'], params['input'])
+ return params
--- /dev/null
+# Copyright (C) 2010 W. Trevor King <wking@drexel.edu>
+#
+# This file is part of Hooke.
+#
+# Hooke is free software: you can redistribute it and/or modify it
+# under the terms of the GNU Lesser General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# Hooke is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
+# or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General
+# Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with Hooke. If not, see
+# <http://www.gnu.org/licenses/>.
+
+"""
+We run this test in a temporary directory for easy cleanup.
+
+>>> import os
+>>> import shutil
+>>> import tempfile
+>>> temp_dir = tempfile.mkdtemp(prefix='tmp-hooke-')
+
+>>> from hooke.hooke import Hooke, HookeRunner
+>>> h = Hooke()
+>>> r = HookeRunner()
+
+Add add some commands to the stack.
+
+>>> h = r.run_lines(h, ['start_command_capture',
+... 'load_playlist test/data/test',
+... 'get_curve',
+... 'stop_command_capture']) # doctest: +REPORT_UDIFF
+Success
+<BLANKLINE>
+Success
+<BLANKLINE>
+Success
+<BLANKLINE>
+Success
+<BLANKLINE>
+>>> h = r.run_lines(h, ['get_command_stack']) # doctest: +NORMALIZE_WHITESPACE
+[<CommandMessage load playlist {input: test/data/test}>,
+ <CommandMessage get curve>]
+Success
+<BLANKLINE>
+
+Ensure we'll be saving in our temporary directory.
+
+>>> target_dir = os.path.join(temp_dir, 'resources', 'command_stack')
+>>> h = r.run_lines(h, ['set_config "command_stack plugin" path %s'
+... % target_dir])
+Success
+<BLANKLINE>
+>>> h = r.run_lines(h, ['get_command_stack']) # doctest: +NORMALIZE_WHITESPACE
+[<CommandMessage load playlist {input: test/data/test}>,
+ <CommandMessage get curve>]
+Success
+<BLANKLINE>
+
+Save the stack.
+
+>>> h = r.run_lines(h, ['get_command_stack']) # doctest: +NORMALIZE_WHITESPACE
+[<CommandMessage load playlist {input: test/data/test}>,
+ <CommandMessage get curve>]
+Success
+<BLANKLINE>
+>>> h = r.run_lines(h, ['save_command_stack'])
+Success
+<BLANKLINE>
+>>> os.listdir(temp_dir)
+['resources']
+>>> os.listdir(target_dir)
+['default']
+>>> with open(os.path.join(target_dir, 'default'), 'r') as f:
+... print f.read()
+- arguments: {input: !!python/unicode 'test/data/test'}
+ command: load playlist
+- arguments: {}
+ command: get curve
+<BLANKLINE>
+
+You can also specify the name explicitly.
+
+>>> h = r.run_lines(h, ['save_command_stack --output my_stack'])
+Success
+<BLANKLINE>
+>>> sorted(os.listdir(target_dir))
+['default', 'my_stack']
+>>> with open(os.path.join(target_dir, 'my_stack'), 'r') as f:
+... print f.read()
+- arguments: {input: !!python/unicode 'test/data/test'}
+ command: load playlist
+- arguments: {}
+ command: get curve
+<BLANKLINE>
+
+Further saves overwrite the last save/load path by default.
+
+>>> h = r.run_lines(h, ['restart_command_capture',
+... 'curve_info',
+... 'stop_command_capture']) # doctest: +REPORT_UDIFF
+Success
+<BLANKLINE>
+Success
+<BLANKLINE>
+Success
+<BLANKLINE>
+>>> h = r.run_lines(h, ['save_command_stack'])
+Success
+<BLANKLINE>
+>>> with open(os.path.join(target_dir, 'default'), 'r') as f:
+... print f.read()
+- arguments: {input: !!python/unicode 'test/data/test'}
+ command: load playlist
+- arguments: {}
+ command: get curve
+<BLANKLINE>
+>>> with open(os.path.join(target_dir, 'my_stack'), 'r') as f:
+... print f.read()
+- arguments: {input: !!python/unicode 'test/data/test'}
+ command: load playlist
+- arguments: {}
+ command: get curve
+- arguments: {}
+ command: curve info
+<BLANKLINE>
+
+But starting command capture (which clears the stack), reverts the
+default save name to `default`.
+
+>>> h = r.run_lines(h, ['start_command_capture',
+... 'version',
+... 'stop_command_capture']) # doctest: +REPORT_UDIFF
+Success
+<BLANKLINE>
+Success
+<BLANKLINE>
+Success
+<BLANKLINE>
+>>> h = r.run_lines(h, ['save_command_stack'])
+Success
+<BLANKLINE>
+>>> with open(os.path.join(target_dir, 'default'), 'r') as f:
+... print f.read()
+- arguments: {}
+ command: version
+<BLANKLINE>
+
+Clear the stack so loading behavior is more obvious.
+
+>>> h = r.run_lines(h, ['start_command_capture',
+... 'stop_command_capture']) # doctest: +REPORT_UDIFF
+Success
+<BLANKLINE>
+Success
+<BLANKLINE>
+
+Loading is just the inverse of saving.
+
+>>> h = r.run_lines(h, ['load_command_stack'])
+Success
+<BLANKLINE>
+>>> h = r.run_lines(h, ['get_command_stack'])
+[<CommandMessage version>]
+Success
+<BLANKLINE>
+>>> h = r.run_lines(h, ['load_command_stack --input my_stack'])
+Success
+<BLANKLINE>
+>>> h = r.run_lines(h, ['get_command_stack']) # doctest: +NORMALIZE_WHITESPACE
+[<CommandMessage load playlist {input: test/data/test}>,
+ <CommandMessage get curve>,
+ <CommandMessage curve info>]
+Success
+<BLANKLINE>
+
+Now that the current stack's name is `my_stack`, that will be the
+default stack loaded if you `load_command_stack` without `--input`.
+
+>>> with open(os.path.join(target_dir, 'my_stack'), 'w') as f:
+... f.write('\\n'.join([
+... '- arguments: {}',
+... ' command: debug',
+... '']))
+>>> h = r.run_lines(h, ['load_command_stack'])
+Success
+<BLANKLINE>
+>>> h = r.run_lines(h, ['get_command_stack']) # doctest: +NORMALIZE_WHITESPACE
+[<CommandMessage debug>]
+Success
+<BLANKLINE>
+
+Cleanup the temporary directory.
+
+>>> shutil.rmtree(temp_dir)
+"""
projects / hooke.git / commitdiff