Source code for axopy.task.base

"""Base task implementation."""

from axopy import util
from import Design
from axopy.messaging import Transmitter, TransmitterBase

[docs]class Task(TransmitterBase): """Base class for tasks. This base class handles iteration through the trials of the task in blocks. Most task implementations will want to override the `prepare` and `run_trial` methods, while the rest can be left to default behavior. If you need to implement a custom constructor (``__init__``), you *must* call the base task ``__init__``:: class CustomTask(Task): def __init__(self, custom_param): super(CustomTask, self).__init__() Attributes ---------- trial : dict Dictionary containing the current trial's attributes. advance_block_key : str Key for the user to press in order to advance to the next block. Can set to ``None`` to disable the feature (next block starts immediately after one finishes). finished : Transmitter Emitted when the last trial of the last block has run. This is primarily for the :class:`axopy.experiment.Experiment` to know when the task has finished so it can run the next one. You shouldn't need to use this transmitter at all. """ advance_block_key = util.key_return finished = Transmitter() def __init__(self): super(Task, self).__init__() self._connections = {} design = Design() self.iter = _TaskIter(design) self.prepare_design(design)
[docs] def connect(self, transmitter, receiver): """Connect a transmitter to a receiver. This method helps the task keep track of connections so that all of the manually specified connections can be torn down by the :class:`axopy.experiment.Experiment`. """ name = _connection_name(transmitter, receiver) self._connections[name] = (transmitter, receiver) transmitter.connect(receiver)
[docs] def disconnect(self, transmitter, receiver): """Disconnect a transmitter from a receiver.""" name = _connection_name(transmitter, receiver) try: del self._connections[name] transmitter.disconnect(receiver) except KeyError: # tx/rx pair already removed/disconnected pass
[docs] def disconnect_all(self): """Disconnect all of the task's manually-created connections.""" for name, (tx, rx) in self._connections.items(): tx.disconnect(rx) self._connections.clear()
[docs] def prepare_design(self, design): """Callback for setting up the task design. See :class:`` for details on how to design the task. By default, nothing is added to the design. Parameters ---------- design : Design The task design object you can use to add blocks and trials. """ pass
[docs] def prepare_graphics(self, container): """Initialize graphical elements and messaging connections. This method should be overridden if the task uses any graphics (which most do). It is important to defer initializing any graphical elements until this method is called so that the graphical backend has a chance to start. Parameters ---------- container : axopy.gui.Container The graphical container you can add objects to. """ pass
[docs] def prepare_daq(self, daqstream): """Set up the input device, if applicable. Parameters ---------- daqstream : DaqStream Interface to the data acquisition device. """ pass
[docs] def prepare_storage(self, storage): """Initialize data storage. Override to read or write task data. A :class:`` object is given, which can be used to create a new :class:`` for storing new data or a :class:`` for reading in existing data. Note that the subject ID has already been set. Parameters ---------- storage : Storage The top-level storage object with which new storage can be allocated and existing data can be read. """ pass
[docs] def run(self): """Start running the task. Simply calls `next_block` to start running trials in the first block. This method is called automatically if the task is added to an :class:`~axopy.experiment.Experiment`. Tasks that have a block design shouldn't normally need to override this method. Tasks that are "free-running" for experimenter interaction (e.g. a plot visualization task that the experimenter controls) should override. """ self.next_block()
[docs] def next_block(self): """Get the next block of trials and starts running them. Before starting the block, a prompt is shown to verify that the user is ready to proceed. If there are no more blocks to run, the `finish` method is called. You usually do not need to override this method. """ block = self.iter.next_block() if block is None: self.finish() return self.block = block # wait for confirmation between blocks if self.advance_block_key is None: self.next_trial() else: self._awaiting_key = True
[docs] def next_trial(self): """Get the next trial in the block and starts running it. If there are no more trials in the block, the `finish_block` method is called. """ trial = self.iter.next_trial() if trial is None: self.finish_block() return self.trial = trial self.run_trial(trial)
[docs] def run_trial(self, trial): """Initiate a trial. By default, this method does nothing. Override to implement what happens in a trial. When a trial is complete, use `next_trial` to start the next. Parameters ---------- trial : object Trial data. This is whatever data is put into the `design` passed in. """ pass
[docs] def finish_block(self): """Finishes the block and starts the next one. Override if you need to do some cleanup between blocks. """ self.next_block()
[docs] def finish(self): """Clean up at the end of the task. Override if you need to clean up once the task is completely finished. If you do override this method, you should call the base :meth:`Task.finish()` method or call the ``finished`` transmitter yourself. """ self.finished.emit()
[docs] def key_press(self, key): """Handle key press events. Override this method to receive key press events. Available keys can be found in :mod:`axopy.util` (named `key_<keyname>`, e.g. `key_k`). Important note: if relying on the ``advance_block_key`` to advance the task, make sure to call this super implementation. """ if getattr(self, '_awaiting_key', False) and \ key == self.advance_block_key: self._awaiting_key = False self.next_trial()
def _connection_name(tx, rx): return str(tx) + str(rx) class _TaskIter(object): """Cleanly retrieves blocks of a task design and the trials within them. A task design is a sequence of sequences: a sequence of blocks which are themselves sequences of trials. The ``TaskIter`` iterates over blocks, returning the block data when a new block is available. Nested in the blocks are trials, which the ``TaskIter`` further iterates over, returning them when available. The ``TaskIter`` accepts either a :class:``` or a manually-created list of blocks, where each block is a list containing dictionaries corresponding to trials. """ def __init__(self, design): = design self.block_iter = iter(design) def next_block(self): """Get the next block in the task if available. If no more blocks are available, `None` is returned. Once this occurs, the task is complete and should be finished. """ try: block = next(self.block_iter) except StopIteration: return None self.trial_iter = iter(block) return block def next_trial(self): """Get the next trial in the current block, if available. If there are no more trials in the block, `None` is returned. Once this occurs, you should call `next_block` to get the next block of trials. """ try: trial = next(self.trial_iter) except StopIteration: return None return trial