"""
A very simple task: Blink an LED
Written by @mikewehr in the ``mike`` branch: https://github.com/wehr-lab/autopilot/blob/mike/autopilot/tasks/blink.py
Demonstrates the basic structure of a task with one stage,
described in the comments throughout the task.
See the main tutorial for more detail: https://docs.auto-pi-lot.com/en/latest/guide.task.html#
This page is rendered in the docs here in order to provide links to the mentioned objects/classes/etc., but
this example was intended to be read as source code, as some comments will only be visible there.
.. note::
Currently, after completion, the task needs to be explicitly imported and added to :data:`.tasks.TASK_LIST` ,
this will be remedied shortly (see the ``registries`` branch).
"""
import itertools
import tables
import time
from datetime import datetime
from autopilot.hardware import gpio
from autopilot.tasks import Task
from collections import OrderedDict as odict
[docs]class Blink(Task):
"""
Blink an LED.
Note that we subclass the :class:`.tasks.Task` class (``Blink(Task)``) to provide us with some methods
useful for all Tasks.
Args:
pulse_duration (int, float): Duration the LED should be on, in ms
pulse_interval (int, float): Duration the LED should be off, in ms
"""
# Tasks need to have a few class attributes defined to be integrated into the rest of the system
# See here for more about class vs. instance attributes https://www.toptal.com/python/python-class-attributes-an-overly-thorough-guide
STAGE_NAMES = ["pulse"] # type: list
"""
An (optional) list or tuple of names of methods that will be used as stages for the task.
See :attr:`.Blink.stages` for more information
"""
PARAMS = odict()
"""
A dictionary that specifies the parameters that control the operation of the task -- each task presumably has some
range of options that allow slight variations (eg. different stimuli, etc.) on a shared task structure. This
dictionary specifies each ``PARAM`` as a human-readable ``tag`` and a ``type`` that is used by the gui to
create an appropriate input object. For example::
PARAMS['pulse_duration'] = {'tag': 'LED Pulse Duration (ms)', 'type': 'int'}
When instantiated, these params are passed to the ``__init__`` method.
A :class:`collections.OrderedDict` is used so that parameters can be presented in a predictable way to users.
"""
PARAMS['pulse_duration'] = {'tag': 'LED Pulse Duration (ms)', 'type': 'int'}
PARAMS['pulse_interval'] = {'tag': 'LED Pulse Interval (ms)', 'type': 'int'}
class TrialData(tables.IsDescription):
"""
This class declares the data that will be returned for each "trial" -- or complete set of executed task
stages. It is used by the :class:`.subject.Subject` object to make a data table with the
correct data types. Declare each piece of data using a pytables Column descriptor
(see https://www.pytables.org/usersguide/libref/declarative_classes.html#col-sub-classes for available
data types, and the pytables guide: https://www.pytables.org/usersguide/tutorials.html for more information)
For each trial, we'll return two timestamps, the time we turned the LED on, the time we turned it off,
and the trial number. Note that we use a 26-character :class:`tables.StringCol` for the timestamps,
which are given as an isoformatted string like ``'2021-02-16T18:11:35.752110'``
"""
trial_num = tables.Int32Col()
timestamp_on = tables.StringCol(26)
timestamp_off = tables.StringCol(26)
HARDWARE = {
'LEDS': {
'dLED': gpio.Digital_Out
}
}
"""
Declare the hardware that will be used in the task. Each hardware object is specified with a ``group`` and
an ``id`` as nested dictionaries. These descriptions require a set of hardware parameters in the autopilot
``prefs.json`` (typically generated by :mod:`autopilot.setup.setup_autopilot` ) with a matching ``group`` and
``id`` structure. For example, an LED declared like this in the :attr:`.Blink.HARDWARE` attribute::
HARDWARE = {'LEDS': {'dLED': gpio.Digital_Out}}
requires an entry in ``prefs.json`` like this::
"HARDWARE": {"LEDS": {"dLED": {
"pin": 1,
"polarity": 1
}}}
that will be used to instantiate the :class:`.hardware.gpio.Digital_Out` object, which is then available for use
in the task like::
self.hardware['LEDS']['dLED'].set(1)
"""
def __init__(self, stage_block=None, pulse_duration=100, pulse_interval=500, *args, **kwargs):
# first we call the superclass ('Task')'s initialization method. All tasks should accept ``*args``
# and ``**kwargs`` to pass parameters not explicitly specified by subclass up to the superclass.
super(Blink, self).__init__(*args, **kwargs)
# store parameters given on instantiation as instance attributes
self.pulse_duration = int(pulse_duration)
self.pulse_interval = int(pulse_interval)
self.stage_block = stage_block # type: "threading.Event"
# This allows us to cycle through the task by just repeatedly calling self.stages.next()
self.stages = itertools.cycle([self.pulse])
"""
Some generator that returns the stage methods that define the operation of the task.
To run a task, the :class:`.pilot.Pilot` object will call each stage function, which can return some dictionary
of data (see :meth:`.Blink.pulse` ) and wait until some flag (:attr:`.Blink.stage_block` ) is set to compute the
next stage. Since in this case we want to call the same method (:meth:`.Blink.pulse` ) over and over again,
we use an :class:`itertools.cycle` object (if we have more than one stage to call in a cycle, we could provide
them like ``itertools.cycle([self.stage_method_1, self.stage_method_2])`` . More complex tasks can define a custom
generator for finer control over stage progression.
"""
self.trial_counter = itertools.count()
"""
Some counter to keep track of the trial number
"""
self.init_hardware()
"""
Hardware is initialized by the superclass's :meth:`.Task.init_hardware` method, which creates all the
hardware objects defined in :attr:`.Blink.HARDWARE` according to their parameterization in
``prefs.json`` , and makes them available in the :attr:`.Blink.hardware` dictionary.
"""
self.logger.debug('Hardware initialized')
"""
All task subclass objects have an :attr:`.Task.logger` -- a :class:`logging.Logger` that allows
users to easily debug their tasks and see feedback about their operation. To prevent stdout from
getting clogged, logging messages are printed and stored according to the ``LOGLEVEL`` pref -- so this
message would only appear if ``LOGLEVEL == "DEBUG"``
"""
self.stage_block.set()
"""
We set the stage block and never clear it so that the :class:`.Pilot` doesn't wait for a trigger
to call the next stage -- it just does it as soon as the previous one completes.
See :meth:`.Pilot.run_task` for more detail on this loop.
"""
##################################################################################
# Stage Functions
##################################################################################
[docs] def pulse(self, *args, **kwargs):
"""
Turn an LED on and off according to :attr:`.Blink.pulse_duration` and :attr:`.Blink.pulse_interval`
Returns:
dict: A dictionary containing the trial number and two timestamps.
"""
# -------------
# turn light on
# use :meth:`.hardware.gpio.Digital_Out.set` method to turn the LED on
self.hardware['LEDS']['dLED'].set(1)
# store the timestamp
timestamp_on = datetime.now().isoformat()
# log status as a debug message
self.logger.debug('light on')
# sleep for the pulse_duration
time.sleep(self.pulse_duration / 1000)
# ------------
# turn light off, same as turning it on.
self.hardware['LEDS']['dLED'].set(0)
timestamp_off = datetime.now().isoformat()
self.logger.debug('light off')
time.sleep(self.pulse_interval / 1000)
# count and store the number of the current trial
self.current_trial = next(self.trial_counter)
data = {
'trial_num': self.current_trial,
'timestamp_on': timestamp_on,
'timestamp_off': timestamp_off
}
"""
Create the data dictionary to be returned from the stage. Note that each of the keys in the dictionary
must correspond to the names of the columns declared in the :attr:`.Blink.TrialData` descriptor.
At the conclusion of running the task, we will be able to access the data from the run with
:meth:`.Subject.get_trial_data`, which will be a :class:`pandas.DataFrame` with a row for each trial, and
a column for each of the fields here.
"""
# return the data dictionary from the stage method and yr done :)
return data