ITEM:
<BLANKLINE>
"""
- def __init__(self, name, aliases=None, arguments=[], help=''):
+ def __init__(self, name, aliases=None, arguments=[], help='',
+ plugin=None):
# TODO: see_also=[other,command,instances,...]
self.name = name
if aliases == None:
help='Print a help message.'),
] + arguments
self._help = help
+ self.plugin = None
def run(self, hooke, inqueue=None, outqueue=None, **kwargs):
"""`Normalize inputs and handle <Argument help> before punting
def __init__(self, section, option=None, value=None, help=None, wrap=True):
self.section = section
self.option = option
- self.value = value
+ self.value = str(value)
self.help = help
self.wrap = wrap
PLUGIN_MODULES = [
# ('autopeak', True),
# ('curvetools', True),
+ ('convfilt', True),
('cut', True),
# ('fit', True),
# ('flatfilts-rolf', True),
"""
return []
- def _setup_commands(self):
- """Perform internal setup on stored commands.
-
- Currently:
-
- * Adds a `plugin` attribute to each command so they can access
- the plugin configuration with `.plugin.config`.
- """
- for command in self._commands:
- command.plugin = self
-
def commands(self):
"""Return a list of :class:`hooke.command.Command`\s provided.
"""
pass
items.append(item)
return items
+
+def setting_to_argument(self, setting):
+ """Convert a :class:`~hooke.conf.Setting` to an
+ :class:`~hooke.command.Argument`.
+ """
+ TODO
class ConfigPlugin (Builtin):
def __init__(self):
super(ConfigPlugin, self).__init__(name='config')
- self._commands = [GetCommand(), SetCommand(), PrintCommand()]
- self._setup_commands()
+ self._commands = [GetCommand(self), SetCommand(self),
+ PrintCommand(self)]
# Define common or complicated arguments
class GetCommand (Command):
"""Get the current value of a configuration option.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(GetCommand, self).__init__(
name='get config',
arguments=[SectionArgument, OptionArgument],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
outqueue.put(hooke.config.get(params['section'], params['option']))
noticed by the target classes unless the configuration is reloaded.
This reloading may cause problems in poorly written UIs.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(SetCommand, self).__init__(
name='set config',
arguments=[
name='value', type='string', optional=False,
help='Value to set.'),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
hooke.config.set(params['section'], params['option'], params['value'])
class PrintCommand (Command):
"""Get the current value of a configuration option.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(PrintCommand, self).__init__(
- name='print config', help=self.__doc__)
+ name='print config', help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
out = StringIO()
def __init__(self):
super(CurvePlugin, self).__init__(name='curve')
self._commands = [
- InfoCommand(), ExportCommand(),
- DifferenceCommand(), DerivativeCommand(),
- PowerSpectrumCommand()]
- self._setup_commands()
+ InfoCommand(self), ExportCommand(self),
+ DifferenceCommand(self), DerivativeCommand(self),
+ PowerSpectrumCommand(self)]
# Define common or complicated arguments
class InfoCommand (Command):
"""Get selected information about a :class:`hooke.curve.Curve`.
"""
- def __init__(self):
+ def __init__(self, plugin):
args = [
CurveArgument,
Argument(name='all', type='bool', default=False, count=1,
name=field, type='bool', default=False, count=1,
help='Get curve %s' % field))
super(InfoCommand, self).__init__(
- name='curve info', arguments=args, help=self.__doc__)
+ name='curve info', arguments=args,
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
fields = {}
"""Export a :class:`hooke.curve.Curve` data block as TAB-delimeted
ASCII text.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(ExportCommand, self).__init__(
name='export block',
arguments=[
File name for the output data. Defaults to 'curve.dat'
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
data = params['curve'].data[params['block']]
See :func:`hooke.util.calculus.derivative` for implementation
details.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(DifferenceCommand, self).__init__(
name='block difference',
arguments=[
Column of data block to differentiate.
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
a = params['curve'].data[params['block one']]
class DerivativeCommand (Command):
"""Calculate the difference between two blocks of data.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(DerivativeCommand, self).__init__(
name='block derivative',
arguments=[
central differencing.
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
data = params['curve'].data[params['block']]
class PowerSpectrumCommand (Command):
"""Calculate the power spectrum of a data block.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(PowerSpectrumCommand, self).__init__(
name='block power spectrum',
arguments=[
help="""
Number of samples per chunk. Use a power of two.
""".strip()),
- Argument(name='overlap', type='bool', default=False,
- help="""
+ Argument(name='overlap', type='bool', default=False,
+ help="""
If `True`, each chunk overlaps the previous chunk by half its length.
Otherwise, the chunks are end-to-end, and not overlapping.
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
data = params['curve'].data[params['block']]
class CutPlugin (Plugin):
def __init__(self):
super(CutPlugin, self).__init__(name='cut')
- self._commands = [CutCommand()]
- self._setup_commands()
+ self._commands = [CutCommand(self)]
# Define common or complicated arguments
The data is saved in TAB-delimited ASCII text, where the first column
is "x" and the second is "y". There is no header row.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(CutCommand, self).__init__(
name='cut',
arguments=[
File name for the output data.
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
if params['curve'] == None:
class DebugPlugin (Builtin):
def __init__(self):
super(DebugPlugin, self).__init__(name='debug')
- self._commands = [VersionCommand(), DebugCommand()]
- self._setup_commands()
+ self._commands = [VersionCommand(self), DebugCommand(self)]
class VersionCommand (Command):
"""Get the Hooke version, as well as versions for important Python
packages. Useful for debugging.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(VersionCommand, self).__init__(
- name='version', help=self.__doc__)
+ name='version', help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
lines = [
class DebugCommand (Command):
"""Get Hooke attributes. Useful for debugging.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(DebugCommand, self).__init__(
name='debug',
arguments=[
Hooke attribute to print.
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
if params['attribute'] == None:
class MultifitPlugin (Plugin):
def __init__(self):
super(MultifitPlugin, self).__init__(name='multifit')
- self._commands = [MultifitCommand()]
- self._setup_commands()
+ self._commands = [MultifitCommand(self)]
class MultifitCommand (Command):
"""
#NOTE: centerzero plot modifier should be activated (set centerzero
#1).
- def __init__(self):
+ def __init__(self, plugin):
super(MultifitCommand, self).__init__(
name='multifit',
arguments=[
Perform the fits on the current curve instead of iterating.
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
counter=0
def __init__(self):
super(NotePlugin, self).__init__(name='note')
self._commands = [
- AddNoteCommand(), ClearNoteCommand(), GetNoteCommand()]
- self._setup_commands()
+ AddNoteCommand(self), ClearNoteCommand(self), GetNoteCommand(self)]
class AddNoteCommand (Command):
"""Add a note to one of several Hooke objects.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(AddNoteCommand, self).__init__(
name='add note',
arguments=[
The note text.
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
params['target'].info['note'].append(params['note'])
class ClearNoteCommand (Command):
"""Remove a note or notes from one of several Hooke objects.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(ClearNoteCommand, self).__init__(
name='clear note',
arguments=[
typing mistakes ;).
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
num_notes = len(params['target'].info['note'])
class GetNoteCommand (Command):
"""Retrieve notes from one of several Hooke objects.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(GetNoteCommand, self).__init__(
name='get note',
arguments=[
Target object for the note. Defaults to the current playlist.
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
outqueue.put(params['target'].info['note'])
def __init__(self):
super(PlaylistPlugin, self).__init__(name='playlist')
self._commands = [
- NextCommand(), PreviousCommand(), JumpCommand(),
- IndexCommand(), CurveListCommand(),
- SaveCommand(), LoadCommand(),
- AddCommand(), AddGlobCommand(),
- RemoveCommand(), FilterCommand(), NoteFilterCommand()]
- self._setup_commands()
+ NextCommand(self), PreviousCommand(self), JumpCommand(self),
+ IndexCommand(self), CurveListCommand(self),
+ SaveCommand(self), LoadCommand(self),
+ AddCommand(self), AddGlobCommand(self),
+ RemoveCommand(self), FilterCommand(self), NoteFilterCommand(self)]
# Define common or complicated arguments
class NextCommand (Command):
"""Move playlist to the next curve.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(NextCommand, self).__init__(
name='next curve',
arguments=[PlaylistArgument],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
params['playlist'].next()
class PreviousCommand (Command):
"""Move playlist to the previous curve.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(PreviousCommand, self).__init__(
name='previous curve',
arguments=[PlaylistArgument],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
params['playlist'].previous()
class JumpCommand (Command):
"""Move playlist to a given curve.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(JumpCommand, self).__init__(
name='jump to curve',
arguments=[
Index of target curve.
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
params['playlist'].jump(params['index'])
The first curve has index 0.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(IndexCommand, self).__init__(
name='curve index',
arguments=[
PlaylistArgument,
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
outqueue.put(params['playlist']._index)
class CurveListCommand (Command):
"""Get the curves in a playlist.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(CurveListCommand, self).__init__(
name='playlist curves',
arguments=[
PlaylistArgument,
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
outqueue.put([c for c in params['playlist']])
class SaveCommand (Command):
"""Save a playlist.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(SaveCommand, self).__init__(
name='save playlist',
arguments=[
playlist.
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
params['playlist'].save(params['output'])
class LoadCommand (Command):
"""Load a playlist.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(LoadCommand, self).__init__(
name='load playlist',
arguments=[
Drivers for loading curves.
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
p = FilePlaylist(drivers=params['drivers'], path=params['input'])
class AddCommand (Command):
"""Add a curve to a playlist.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(AddCommand, self).__init__(
name='add curve to playlist',
arguments=[
Additional information for the input :class:`hooke.curve.Curve`.
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
params['playlist'].append_curve_by_path(params['input'],
command you can use globs (`data/curves/*.dat`) to add curves
for all matching files at once.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(AddGlobCommand, self).__init__(
name='glob curves to playlist',
arguments=[
Additional information for the input :class:`hooke.curve.Curve`.
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
for path in sorted(glob.glob(params['input'])):
class RemoveCommand (Command):
"""Remove a curve from a playlist.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(RemoveCommand, self).__init__(
name='remove curve from playlist',
arguments=[
Index of target curve.
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
params['playlist'].pop(params['index'])
command you can use a function `filter` to select the curves you
wish to keep.
"""
- def __init__(self):
+ def __init__(self, plugin, name='filter playlist', filter_fn=None):
super(FilterCommand, self).__init__(
- name='filter playlist',
+ name=name,
arguments=[
PlaylistArgument,
PlaylistNameArgument,
+ ],
+ help=self.__doc__, plugin=plugin)
+ self.filter_fn = filter_fn
+ if filter_fn == None:
+ self.arguments.append(
Argument(name='filter', type='function', optional=False,
help="""
Function returning `True` for "good" curves. `filter(curve) -> True/False`.
-""".strip()),
- ],
- help=self.__doc__)
+""".strip()))
def _run(self, hooke, inqueue, outqueue, params):
- p = params['playlist'].filter(params['filter'])
+ if self.filter_fn == None:
+ filter_fn = params['filter']
+ else:
+ filter_fn = self.filter_fn
+ p = params['playlist'].filter(filter_fn)
hooke.playlists.add(p)
outqueue.put(p)
class NoteFilterCommand (FilterCommand):
"""Create a subset playlist of curves with `.info['note'] != None`.
"""
- def __init__(self):
- super(NoteFilterCommand, self).__init__()
- self.name = 'note filter playlist'
- self.arguments = [a for a in self.arguments if a.name != 'filter']
-
- def _run(self, hooke, inqueue, outqueue, params):
- params['filter'] = lambda curve : \
- 'note' in curve.info and curve.info['note'] != None
- return super(NoteFilterCommand, self)._run(
- hooke, inqueue, outqueue, params)
+ def __init__(self, plugin):
+ super(NoteFilterCommand, self).__init__(
+ plugin, name='note filter playlist',
+ filter_fn=lambda curve : \
+ 'note' in curve.info and curve.info['note'] != None)
def __init__(self):
super(SystemPlugin, self).__init__(name='system')
self._commands = [
- ListDirectoryCommand(), GetWorkingDirectoryCommand(),
- ChangeDirectoryCommand(), SystemCommand()]
- self._setup_commands()
+ ListDirectoryCommand(self), GetWorkingDirectoryCommand(self),
+ ChangeDirectoryCommand(self), SystemCommand(self)]
class ListDirectoryCommand (Command):
"""List the files in a directory.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(ListDirectoryCommand, self).__init__(
name='ls', aliases=['dir'],
arguments=[
current working directory.
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
outqueue.put('\n'.join(sorted(os.listdir(params['path']))))
class GetWorkingDirectoryCommand (Command):
"""Get the current working directory.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(GetWorkingDirectoryCommand, self).__init__(
- name='cwd', aliases=['pwd'], help=self.__doc__)
+ name='cwd', aliases=['pwd'], help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
outqueue.put(os.getcwd())
class ChangeDirectoryCommand (Command):
"""Change the current working directory.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(ChangeDirectoryCommand, self).__init__(
name='cd',
arguments=[
directory.
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
os.chdir(os.path.expanduser(params['path']))
class SystemCommand (Command):
"""Execute a system command and report the output.
"""
- def __init__(self):
+ def __init__(self, plugin):
super(SystemCommand, self).__init__(
name='system',
arguments=[
Command line to execute.
""".strip()),
],
- help=self.__doc__)
+ help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
p = subprocess.Popen(
projects / hooke.git / commitdiff