Blink¶
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.
This page is rendered in the docs here in order to provide links to the mentioned objects/classes/etc., but it was written as source code initially and translated to .rst, so the narrative flow is often inverted: text follows code as comments, rather than text introducing and narrating code.
Preamble¶
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
class Blink(Task):
"""
Blink an LED.
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
"""
Note that we subclass the Task class (Blink(Task)) to provide us with some methods
useful for all Tasks, and to make it available to the task registry (see Plugins & The Wiki).
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
Params¶
STAGE_NAMES = ["pulse"] # type: list
"""
An (optional) list or tuple of names of methods that will be used as stages for the task.
See ``stages`` for more information
"""
PARAMS = odict()
PARAMS['pulse_duration'] = {'tag': 'LED Pulse Duration (ms)', 'type': 'int'}
PARAMS['pulse_interval'] = {'tag': 'LED Pulse Interval (ms)', 'type': 'int'}
PARAMS - 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 collections.OrderedDict is used so that parameters can be presented in a predictable way to users.
TrialData¶
class TrialData(tables.IsDescription):
trial_num = tables.Int32Col()
timestamp_on = tables.StringCol(26)
timestamp_off = tables.StringCol(26)
TrialData declares the data that will be returned for each “trial” – or complete set of executed task
stages. It is used by the 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 tables.StringCol for the timestamps,
Hardware¶
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 autopilot.setup.setup_autopilot ) with a matching group and
id structure. For example, an LED declared like this in the 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 hardware.gpio.Digital_Out object, which is then available for use
in the task like:
self.hardware['LEDS']['dLED'].set(1)
Initialization¶
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.:
def __init__(self, stage_block=None, pulse_duration=100, pulse_interval=500, *args, **kwargs):
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 pilot.Pilot object will call each stage function, which can return some dictionary
of data (see pulse() ) and wait until some flag (stage_block ) is set to compute the
next stage. Since in this case we want to call the same method (pulse() ) over and over again,
we use an 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
"""
Hardware is initialized by the superclass’s Task.init_hardware() method, which creates all the
hardware objects defined in HARDWARE according to their parameterization in
prefs.json , and makes them available in the hardware dictionary.:
self.init_hardware()
self.logger.debug('Hardware initialized')
All task subclass objects have an logger – a 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 Pilot doesn’t wait for a trigger
to call the next stage – it just does it as soon as the previous one completes.
See run_task() for more detail on this loop.
Stage Methods¶
def pulse(self, *args, **kwargs):
"""
Turn an LED on and off according to :attr:`~examples.tasks.Blink.pulse_duration` and :attr:`~examples.tasks.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
}
return data
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 TrialData descriptor.
At the conclusion of running the task, we will be able to access the data from the run with
Subject.get_trial_data(), which will be a pandas.DataFrame with a row for each trial, and
a column for each of the fields here.
Full Source¶
1"""
2A very simple task: Blink an LED
3
4Written by @mikewehr in the ``mike`` branch: https://github.com/wehr-lab/autopilot/blob/mike/autopilot/tasks/blink.py
5
6Demonstrates the basic structure of a task with one stage,
7described in the comments throughout the task.
8
9See the main tutorial for more detail: https://docs.auto-pi-lot.com/en/latest/guide.task.html#
10
11This page is rendered in the docs here in order to provide links to the mentioned objects/classes/etc., but
12this example was intended to be read as source code, as some comments will only be visible there.
13"""
14import itertools
15import tables
16import time
17from datetime import datetime
18
19from autopilot.hardware import gpio
20from autopilot.tasks import Task
21from collections import OrderedDict as odict
22
23class Blink(Task):
24 """
25 Blink an LED.
26
27 Note that we subclass the :class:`~autopilot.tasks.Task` class (``Blink(Task)``) to provide us with some methods
28 useful for all Tasks.
29
30 Args:
31 pulse_duration (int, float): Duration the LED should be on, in ms
32 pulse_interval (int, float): Duration the LED should be off, in ms
33
34 """
35 # Tasks need to have a few class attributes defined to be integrated into the rest of the system
36 # See here for more about class vs. instance attributes https://www.toptal.com/python/python-class-attributes-an-overly-thorough-guide
37
38 STAGE_NAMES = ["pulse"] # type: list
39 """
40 An (optional) list or tuple of names of methods that will be used as stages for the task.
41
42 See :attr:`~examples.tasks.Blink.stages` for more information
43 """
44
45 PARAMS = odict()
46 """
47 A dictionary that specifies the parameters that control the operation of the task -- each task presumably has some
48 range of options that allow slight variations (eg. different stimuli, etc.) on a shared task structure. This
49 dictionary specifies each ``PARAM`` as a human-readable ``tag`` and a ``type`` that is used by the gui to
50 create an appropriate input object. For example::
51
52 PARAMS['pulse_duration'] = {'tag': 'LED Pulse Duration (ms)', 'type': 'int'}
53
54 When instantiated, these params are passed to the ``__init__`` method.
55
56 A :class:`collections.OrderedDict` is used so that parameters can be presented in a predictable way to users.
57 """
58 PARAMS['pulse_duration'] = {'tag': 'LED Pulse Duration (ms)', 'type': 'int'}
59 PARAMS['pulse_interval'] = {'tag': 'LED Pulse Interval (ms)', 'type': 'int'}
60
61 class TrialData(tables.IsDescription):
62 """
63 This class declares the data that will be returned for each "trial" -- or complete set of executed task
64 stages. It is used by the :class:`~autopilot.core.subject.Subject` object to make a data table with the
65 correct data types. Declare each piece of data using a pytables Column descriptor
66 (see https://www.pytables.org/usersguide/libref/declarative_classes.html#col-sub-classes for available
67 data types, and the pytables guide: https://www.pytables.org/usersguide/tutorials.html for more information)
68
69 For each trial, we'll return two timestamps, the time we turned the LED on, the time we turned it off,
70 and the trial number. Note that we use a 26-character :class:`tables.StringCol` for the timestamps,
71 which are given as an isoformatted string like ``'2021-02-16T18:11:35.752110'``
72 """
73 trial_num = tables.Int32Col()
74 timestamp_on = tables.StringCol(26)
75 timestamp_off = tables.StringCol(26)
76
77
78 HARDWARE = {
79 'LEDS': {
80 'dLED': gpio.Digital_Out
81 }
82 }
83 """
84 Declare the hardware that will be used in the task. Each hardware object is specified with a ``group`` and
85 an ``id`` as nested dictionaries. These descriptions require a set of hardware parameters in the autopilot
86 ``prefs.json`` (typically generated by :mod:`autopilot.setup.setup_autopilot` ) with a matching ``group`` and
87 ``id`` structure. For example, an LED declared like this in the :attr:`~examples.tasks.Blink.HARDWARE` attribute::
88
89 HARDWARE = {'LEDS': {'dLED': gpio.Digital_Out}}
90
91 requires an entry in ``prefs.json`` like this::
92
93 "HARDWARE": {"LEDS": {"dLED": {
94 "pin": 1,
95 "polarity": 1
96 }}}
97
98 that will be used to instantiate the :class:`.hardware.gpio.Digital_Out` object, which is then available for use
99 in the task like::
100
101 self.hardware['LEDS']['dLED'].set(1)
102 """
103
104 def __init__(self, stage_block=None, pulse_duration=100, pulse_interval=500, *args, **kwargs):
105 # first we call the superclass ('Task')'s initialization method. All tasks should accept ``*args``
106 # and ``**kwargs`` to pass parameters not explicitly specified by subclass up to the superclass.
107 super(Blink, self).__init__(*args, **kwargs)
108
109 # store parameters given on instantiation as instance attributes
110 self.pulse_duration = int(pulse_duration)
111 self.pulse_interval = int(pulse_interval)
112 self.stage_block = stage_block # type: "threading.Event"
113
114 # This allows us to cycle through the task by just repeatedly calling self.stages.next()
115 self.stages = itertools.cycle([self.pulse])
116 """
117 Some generator that returns the stage methods that define the operation of the task.
118
119 To run a task, the :class:`.pilot.Pilot` object will call each stage function, which can return some dictionary
120 of data (see :meth:`~examples.tasks.Blink.pulse` ) and wait until some flag (:attr:`~examples.tasks.Blink.stage_block` ) is set to compute the
121 next stage. Since in this case we want to call the same method (:meth:`~examples.tasks.Blink.pulse` ) over and over again,
122 we use an :class:`itertools.cycle` object (if we have more than one stage to call in a cycle, we could provide
123 them like ``itertools.cycle([self.stage_method_1, self.stage_method_2])`` . More complex tasks can define a custom
124 generator for finer control over stage progression.
125 """
126
127 self.trial_counter = itertools.count()
128 """
129 Some counter to keep track of the trial number
130 """
131
132
133 self.init_hardware()
134
135 """
136 Hardware is initialized by the superclass's :meth:`.Task.init_hardware` method, which creates all the
137 hardware objects defined in :attr:`~examples.tasks.Blink.HARDWARE` according to their parameterization in
138 ``prefs.json`` , and makes them available in the :attr:`~examples.tasks.Blink.hardware` dictionary.
139 """
140
141 self.logger.debug('Hardware initialized')
142
143 """
144 All task subclass objects have an :attr:`~autopilot.tasks.Task.logger` -- a :class:`logging.Logger` that allows
145 users to easily debug their tasks and see feedback about their operation. To prevent stdout from
146 getting clogged, logging messages are printed and stored according to the ``LOGLEVEL`` pref -- so this
147 message would only appear if ``LOGLEVEL == "DEBUG"``
148 """
149
150 self.stage_block.set()
151
152 """
153 We set the stage block and never clear it so that the :class:`.Pilot` doesn't wait for a trigger
154 to call the next stage -- it just does it as soon as the previous one completes.
155
156 See :meth:`~autopilot.core.pilot.Pilot.run_task` for more detail on this loop.
157 """
158
159
160 ##################################################################################
161 # Stage Functions
162 ##################################################################################
163 def pulse(self, *args, **kwargs):
164 """
165 Turn an LED on and off according to :attr:`~examples.tasks.Blink.pulse_duration` and :attr:`~examples.tasks.Blink.pulse_interval`
166
167 Returns:
168 dict: A dictionary containing the trial number and two timestamps.
169 """
170 # -------------
171 # turn light on
172
173 # use :meth:`.hardware.gpio.Digital_Out.set` method to turn the LED on
174 self.hardware['LEDS']['dLED'].set(1)
175 # store the timestamp
176 timestamp_on = datetime.now().isoformat()
177 # log status as a debug message
178 self.logger.debug('light on')
179 # sleep for the pulse_duration
180 time.sleep(self.pulse_duration / 1000)
181
182 # ------------
183 # turn light off, same as turning it on.
184
185 self.hardware['LEDS']['dLED'].set(0)
186 timestamp_off = datetime.now().isoformat()
187 self.logger.debug('light off')
188 time.sleep(self.pulse_interval / 1000)
189
190 # count and store the number of the current trial
191 self.current_trial = next(self.trial_counter)
192
193
194 data = {
195 'trial_num': self.current_trial,
196 'timestamp_on': timestamp_on,
197 'timestamp_off': timestamp_off
198 }
199
200 """
201 Create the data dictionary to be returned from the stage. Note that each of the keys in the dictionary
202 must correspond to the names of the columns declared in the :attr:`~examples.tasks.Blink.TrialData` descriptor.
203
204 At the conclusion of running the task, we will be able to access the data from the run with
205 :meth:`.Subject.get_trial_data`, which will be a :class:`pandas.DataFrame` with a row for each trial, and
206 a column for each of the fields here.
207 """
208
209 # return the data dictionary from the stage method and yr done :)
210 return data