storing force curves.
"""
+import logging
import os.path
+
import numpy
from .command_stack import CommandStack
self.command_stack = CommandStack()
self._hooke = None # Hooke instance for Curve.load()
+ def __str__(self):
+ return str(self.__unicode__())
+
+ def __unicode__(self):
+ return u'<%s %s>' % (self.__class__.__name__, self.name)
+
+ def __repr__(self):
+ return self.__str__()
+
def set_hooke(self, hooke=None):
if hooke != None:
self._hooke = hooke
:meth:`hooke.command_stack.CommandStack.execute`.
"""
self.set_hooke(hooke)
+ log = logging.getLogger('hooke')
+ log.debug('loading curve %s with driver %s' % (self.name, self.driver))
data,info = self.driver.read(self.path, self.info)
self.data = data
for key,value in info.items():
self.info[key] = value
- if self.hooke != None:
- # TODO: set 'curve' argument explicitly for CurveCommands
- self.command_stack.execute(self.hooke)
+ if self._hooke != None:
+ self.command_stack.execute(self._hooke)
+ elif len(self.command_stack) > 0:
+ log.warn(
+ 'could not execute command stack for %s without Hooke instance'
+ % self.name)
def unload(self):
"""Release memory intensive :attr:`.data`.
"""
+ log = logging.getLogger('hooke')
+ log.debug('unloading curve %s' % self.name)
self.data = None
"""
import logging
+from Queue import Queue, Empty
from .command import NullQueue
communication queues, so make sure they will not need user
interaction.
"""
+ log = logging.getLogger('hooke')
log.debug('engine running internal %s with %s'
% (command, arguments))
cmd = hooke.command_by_name[command]
- cmd.run(hooke, NullQueue(), NullQueue(), arguments)
+ outqueue = Queue()
+ cmd.run(hooke, NullQueue(), outqueue, **arguments)
+ while True:
+ try:
+ msg = outqueue.get(block=False)
+ except Empty:
+ break
+ log.debug('engine message from %s (%s): %s' % (command, type(msg), msg))
self._loaded = [] # List of loaded curves, see :meth:`._setup_item`.
self._max_loaded = 100 # curves to hold in memory simultaneously.
- def append_curve_by_path(self, path, info=None, identify=True):
+ def append_curve_by_path(self, path, info=None, identify=True, hooke=None):
path = os.path.normpath(path)
c = curve.Curve(path, info=info)
+ c.set_hooke(hooke)
if identify == True:
c.identify(self.drivers)
self.append(c)
doc = xml.dom.minidom.parseString(string)
self._from_xml_doc(doc, identify=identify)
- def load(self, path=None, identify=True):
+ def load(self, path=None, identify=True, hooke=None):
"""Load a playlist from a file.
"""
self.set_path(path)
doc = xml.dom.minidom.parse(self.path)
self._from_xml_doc(doc, identify=identify)
self._digest = self.digest()
-
+ for curve in self:
+ curve.set_hooke(hooke)
+
def save(self, path=None):
"""Saves the playlist in a XML file.
"""
from ..command import Command, Argument, Failure
from ..curve import Data
+from ..engine import CommandMessage
from ..util.calculus import derivative
from ..util.fft import unitary_avg_power_spectrum
from ..util.si import ppSI, join_data_label, split_data_label
super(CurveCommand, self).__init__(**kwargs)
def _curve(self, hooke, params):
+ """Get the selected curve.
+
+ Notes
+ -----
+ `hooke` is intended to attach the selected curve to the local
+ playlist, and the returned curve should not be effected by the
+ state of `hooke`. This is important for reliable
+ :class:`~hooke.command_stack.CommandStack`\s.
+ """
# HACK? rely on params['curve'] being bound to the local hooke
# playlist (i.e. not a copy, as you would get by passing a
# curve through the queue). Ugh. Stupid queues. As an
# queue...
return params['curve']
+ def _add_to_command_stack(self, params):
+ """Store the command name and current `params` values in the
+ curve's `.command_stack`.
+
+ If this would duplicate the command currently on top of the
+ stack, no action is taken. Call early on, or watch out for
+ repeated param processing.
+
+ Recommended practice is to *not* lock in argument values that
+ are loaded from the plugin's :attr:`.config`.
+
+ Notes
+ -----
+ Perhaps we should subclass :meth:`_run` and use :func:`super`,
+ or embed this in :meth:`run` to avoid subclasses calling this
+ method explicitly, with all the tedium and brittality that
+ implies. On the other hand, the current implemtnation allows
+ CurveCommands that don't effect the curve itself
+ (e.g. :class:`GetCommand`) to avoid adding themselves to the
+ stack entirely.
+ """
+ curve = self._curve(hooke=None, params=params)
+ if (len(curve.command_stack) > 0
+ and curve.command_stack[-1].command == self.name
+ and curve.command_stack[-1].arguments == params):
+ pass # no need to place duplicate calls on the stack.
+ else:
+ curve.command_stack.append(CommandMessage(
+ self.name, params))
+
class BlockCommand (CurveCommand):
"""A :class:`CurveCommand` operating on a :class:`~hooke.curve.Data` block.
help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
+ self._add_to_command_stack(params)
params = self.__setup_params(hooke=hooke, params=params)
data_A = self._get_column(hooke=hooke, params=params,
block_name='block A',
help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
+ self._add_to_command_stack(params)
params = self.__setup_params(hooke=hooke, params=params)
x_data = self._get_column(hooke=hooke, params=params,
column_name='x column')
help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
+ self._add_to_command_stack(params)
params = self.__setup_params(hooke=hooke, params=params)
data = self._get_column(hooke=hooke, params=params)
bounds = params['bounds']
help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
+ self._add_to_command_stack(params)
params = self.__setup_params(hooke=hooke, params=params)
block = self._block(hooke=hooke, params=params)
dist_data = self._get_column(hooke=hooke, params=params,
def _run(self, hooke, inqueue, outqueue, params):
p = FilePlaylist(drivers=params['drivers'], path=params['input'])
- p.load()
+ p.load(hooke=hooke)
hooke.playlists.append(p)
outqueue.put(p)
help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
- params['playlist'].append_curve_by_path(params['input'],
- params['info'])
+ params['playlist'].append_curve_by_path(
+ params['input'], params['info'], hooke=hooke)
class AddGlobCommand (Command):
"""Add curves to a playlist with file globbing.
def _run(self, hooke, inqueue, outqueue, params):
for path in sorted(glob.glob(params['input'])):
- params['playlist'].append_curve_by_path(path, params['info'])
+ params['playlist'].append_curve_by_path(
+ path, params['info'], hooke=hooke)
class RemoveCommand (Command):
"""Remove a curve from a playlist.
help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
+ self._add_to_command_stack(params)
params = self.__setup_params(hooke=hooke, params=params)
block = self._block(hooke=hooke, params=params)
dist_data = self._get_column(hooke=hooke, params=params,
help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
+ self._add_to_command_stack(params)
params = self.__setup_params(hooke=hooke, params=params)
def_data = self._get_column(hooke=hooke, params=params,
column_name='deflection column')
help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
+ self._add_to_command_stack(params)
params = self.__setup_params(hooke=hooke, params=params)
def_data = self._get_column(hooke=hooke, params=params,
column_name='deflection column')
help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
+ self._add_to_command_stack(params)
params = self.__setup_params(hooke=hooke, params=params)
block = self._block(hooke=hooke, params=params)
dist_data = self._get_column(hooke=hooke, params=params,
projects / hooke.git / commitdiff