Skip to content

trial.py

Events are linked to Trials

activate(trial_schema_name, event_schema_name, *, create_schema=True, create_tables=True, linking_module=None)

Activate this schema.

Parameters:

Name Type Description Default
trial_schema_name str

schema name on the database server to activate the trial element

 required
event_schema_name str

schema name on the database server to activate the event element

 required
create_schema bool

when True (default), create schema in the database if it does not yet exist.

 True
create_tables str

when True (default), create schema tables in the database if they do not yet exist.

 True
linking_module str

a module (or name) containing the required dependencies to activate the trial element

 None

Dependencies:

Upstream tables

Session: parent table to BehaviorRecording, identifying a recording session.

Functions

get_trialized_alignment_event_times(alignment_event_key: dict, trial_restriction: dict): For the trials identified by trial_restriction, identify recording times with respect to a given alignment_event. Returns pandas dataframe with trial_key, start (recording time), event (recording time), and end (recording time).

Source code in element_event/trial.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def activate(
    trial_schema_name,
    event_schema_name,
    *,
    create_schema=True,
    create_tables=True,
    linking_module=None,
):
    """Activate this schema.

    Args:
        trial_schema_name (str): schema name on the database server to activate the
                                 `trial` element
        event_schema_name (str): schema name on the database server to activate the
                                 `event` element
        create_schema (bool): when True (default), create schema in the database if it
                            does not yet exist.
        create_tables (str): when True (default), create schema tables in the database
                             if they do not yet exist.
        linking_module (str): a module (or name) containing the required dependencies
                              to activate the `trial` element

    Dependencies:
    Upstream tables:
        Session: parent table to BehaviorRecording, identifying a recording session.

    Functions:
        get_trialized_alignment_event_times(alignment_event_key: dict, trial_restriction:
                                            dict): For the trials identified by
                                            trial_restriction, identify recording times
                                            with respect to a given alignment_event.
                                            Returns pandas dataframe with trial_key,
                                            start (recording time), event (recording time),
                                            and end (recording time).
    """
    if isinstance(linking_module, str):
        linking_module = importlib.import_module(linking_module)
    assert inspect.ismodule(linking_module), (
        "The argument 'dependency' must" + " be a module or module name"
    )

    global _linking_module
    _linking_module = linking_module

    event.activate(
        event_schema_name,
        create_schema=create_schema,
        create_tables=create_tables,
        linking_module=_linking_module,
    )

    schema.activate(
        trial_schema_name,
        create_schema=create_schema,
        create_tables=create_tables,
        add_objects=_linking_module.__dict__,
    )

Block

Bases: dj.Imported

Set of experimental blocks within a recording session

Attributes event.BehaviorRecording (foreign key): event.BehaviorRecording primary key. block_id (smallint): block number (1-based indexing) block_start_time (float): Seconds relative to recording start block_stop_time (float): Seconds relative to recording stop

Source code in element_event/trial.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@schema
class Block(dj.Imported):
    """Set of experimental blocks within a recording session

    Attributes
        event.BehaviorRecording (foreign key): event.BehaviorRecording primary key.
        block_id (smallint): block number (1-based indexing)
        block_start_time (float): Seconds relative to recording start
        block_stop_time (float): Seconds relative to recording stop
    """

    definition = """ # Experimental blocks
    -> event.BehaviorRecording
    block_id               : smallint # block number (1-based indexing)
    ---
    block_start_time       : float     # (s) relative to recording start
    block_stop_time        : float     # (s) relative to recording stop
    """

    class Attribute(dj.Part):
        """Extra Block attributes to fully describe a block

        Attributes:
             Block (foreign key): Block table primary key.
             attribute_name ( varchar(32) ): Name of block attribute
             attribute_value ( varchar(2000) ): Optional. Block attribute value
             attribute_blob (longblob): Optional. Block attribute numerical numerical data
        """

        definition = """  # Additional block attributes to fully describe a block
        -> master
        attribute_name    : varchar(32)
        ---
        attribute_value='': varchar(2000)
        attribute_blob=null: longblob
        """

    def make(self, key):
        """Populate each unique entry in event.BehaviorRecording"""
        raise NotImplementedError("For `insert`, use `allow_direct_insert=True`")

Attribute

Bases: dj.Part

Extra Block attributes to fully describe a block

Attributes:

Name Type Description
Block foreign key

Block table primary key.
attribute_name varchar(32)

Name of block attribute
attribute_value varchar(2000)

Optional. Block attribute value
attribute_blob longblob

Optional. Block attribute numerical numerical data
Source code in element_event/trial.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class Attribute(dj.Part):
    """Extra Block attributes to fully describe a block

    Attributes:
         Block (foreign key): Block table primary key.
         attribute_name ( varchar(32) ): Name of block attribute
         attribute_value ( varchar(2000) ): Optional. Block attribute value
         attribute_blob (longblob): Optional. Block attribute numerical numerical data
    """

    definition = """  # Additional block attributes to fully describe a block
    -> master
    attribute_name    : varchar(32)
    ---
    attribute_value='': varchar(2000)
    attribute_blob=null: longblob
    """

make(key)

Populate each unique entry in event.BehaviorRecording

Source code in element_event/trial.py 
113
114
115
def make(self, key):
    """Populate each unique entry in event.BehaviorRecording"""
    raise NotImplementedError("For `insert`, use `allow_direct_insert=True`")

TrialType

Bases: dj.Lookup

Set of unique trial types present within a recording session

Attributes:

Name Type Description
trial_type varchar(16)

Name of trial type
trial_type_description varchar(256)

Optional. Long Description.
Source code in element_event/trial.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@schema
class TrialType(dj.Lookup):
    """Set of unique trial types present within a recording session

    Attributes:
        trial_type ( varchar(16) ): Name of trial type
        trial_type_description ( varchar(256) ): Optional. Long Description.
    """

    definition = """
    trial_type                : varchar(16)
    ---
    trial_type_description='' : varchar(256)
    """

Trial

Bases: dj.Imported

Set of all experimental trials from a behavioral recording

Attributes:

Name Type Description
event.BehaviorRecording foreign key

BehaviorRecording primary key
trial_id smallint

trial number (1-based indexing)
TrialType foreign key

Optional. TrialType primary key
trial_start_time float

Seconds relative to recording start
trial_stop_time float

Seconds relative to recording stop
Source code in element_event/trial.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
@schema
class Trial(dj.Imported):
    """Set of all experimental trials from a behavioral recording

    Attributes:
        event.BehaviorRecording (foreign key): BehaviorRecording primary key
        trial_id (smallint): trial number (1-based indexing)
        TrialType (foreign key): Optional. TrialType primary key
        trial_start_time (float): Seconds relative to recording start
        trial_stop_time (float): Seconds relative to recording stop
    """

    definition = """  # Experimental trials
    -> event.BehaviorRecording
    trial_id            : smallint # trial number (1-based indexing)
    ---
    -> [nullable] TrialType
    trial_start_time    : float  # (second) relative to recording start
    trial_stop_time     : float  # (second) relative to recording stop
    """

    class Attribute(dj.Part):
        """Extra trial attributes to fully describe a trial

        Attributes:
            Trial (foreign key): Trial table primary key.
            attribute_name ( varchar(32) ): Name of trial attribute
            attribute_value ( varchar(2000) ): Optional. Trial attribute value
            attribute_blob (longblob): Optional. Trial attribute numerical data
        """

        definition = """  # Additional trial attributes to fully describe a trial
        -> master
        attribute_name  : varchar(32)
        ---
        attribute_value='': varchar(2000)
        attribute_blob=null: longblob
        """

    def make(self, key):
        """Populate for each unique entry in event.BehaviorRecording"""
        raise NotImplementedError("For `insert`, use `allow_direct_insert=True`")

Attribute

Bases: dj.Part

Extra trial attributes to fully describe a trial

Attributes:

Name Type Description
Trial foreign key

Trial table primary key.
attribute_name varchar(32)

Name of trial attribute
attribute_value varchar(2000)

Optional. Trial attribute value
attribute_blob longblob

Optional. Trial attribute numerical data
Source code in element_event/trial.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
class Attribute(dj.Part):
    """Extra trial attributes to fully describe a trial

    Attributes:
        Trial (foreign key): Trial table primary key.
        attribute_name ( varchar(32) ): Name of trial attribute
        attribute_value ( varchar(2000) ): Optional. Trial attribute value
        attribute_blob (longblob): Optional. Trial attribute numerical data
    """

    definition = """  # Additional trial attributes to fully describe a trial
    -> master
    attribute_name  : varchar(32)
    ---
    attribute_value='': varchar(2000)
    attribute_blob=null: longblob
    """

make(key)

Populate for each unique entry in event.BehaviorRecording

Source code in element_event/trial.py 
173
174
175
def make(self, key):
    """Populate for each unique entry in event.BehaviorRecording"""
    raise NotImplementedError("For `insert`, use `allow_direct_insert=True`")

BlockTrial

Bases: dj.Imported

Set of trials associated with certain blocks

Attributes:

Name Type Description
Block foreign key

Block primary key
Trial foreign key

Trial primary key
Source code in element_event/trial.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
@schema
class BlockTrial(dj.Imported):
    """Set of trials associated with certain blocks

    Attributes:
        Block (foreign key): Block primary key
        Trial (foreign key): Trial primary key
    """

    definition = """
    -> Block
    -> Trial
    """

    def make(self, key):
        """Populate for each unique entry in Trial and Block"""
        raise NotImplementedError("For `insert`, use `allow_direct_insert=True`")

make(key)

Populate for each unique entry in Trial and Block

Source code in element_event/trial.py 
192
193
194
def make(self, key):
    """Populate for each unique entry in Trial and Block"""
    raise NotImplementedError("For `insert`, use `allow_direct_insert=True`")

TrialEvent

Bases: dj.Imported

Set of trials associated with certain events

Attributes:

Name Type Description
Block foreign key

Block primary key
event.Event foreign key

event.Event primary key
Source code in element_event/trial.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
@schema
class TrialEvent(dj.Imported):
    """Set of trials associated with certain events

    Attributes:
        Block (foreign key): Block primary key
        event.Event (foreign key): event.Event primary key
    """

    definition = """
    -> Trial
    -> event.Event
    """

    def make(self, key):
        """Populate for each unique entry in Trial and event.Event"""
        raise NotImplementedError("For `insert`, use `allow_direct_insert=True`")

make(key)

Populate for each unique entry in Trial and event.Event

Source code in element_event/trial.py 
211
212
213
def make(self, key):
    """Populate for each unique entry in Trial and event.Event"""
    raise NotImplementedError("For `insert`, use `allow_direct_insert=True`")

get_trialized_alignment_event_times(alignment_event_key, trial_restriction)

For the trials in trial_restriction, identify times WRT a given alignment_event.

WRT = With respect to

Parameters:

Name Type Description Default
alignment_event_key dict

key including information from event.AlignmentEvent

 required
trial_restriction dict

set or subset of trials from trial.Trial

 required

Returns:

Name Type Description
dataset pandas

Dataframe with each of the items listed below.
Dataset

trial_key (dict): key identifying a single trial

start (float): recording time (s) of the beginning of an alignment window

event (float): recording time (s) of an alignment event within the trial. If multiple events within a trial, select the last one

end (float): recording time (s) of the end of an alignment window

Source code in element_event/trial.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
def get_trialized_alignment_event_times(alignment_event_key, trial_restriction):
    """For the trials in trial_restriction, identify times WRT a given alignment_event.

    WRT = With respect to

    Args:
        alignment_event_key (dict): key including information from event.AlignmentEvent
        trial_restriction (dict): set or subset of trials from trial.Trial

    Returns:
        dataset (pandas): Dataframe with each of the items listed below.

    Dataset:
        trial_key (dict): key identifying a single trial \n
        start (float): recording time (s) of the beginning of an alignment window \n
        event (float): recording time (s) of an alignment event within the trial.
            If multiple events within a trial, select the last one\n
        end  (float): recording time (s) of the end of an alignment window
    """

    import pandas as pd

    session_key = (_linking_module.Session & trial_restriction).fetch1("KEY")
    trial_keys, trial_starts, trial_ends = (Trial ^ trial_restriction).fetch(
        "KEY", "trial_start_time", "trial_stop_time", order_by="trial_id"
    )
    alignment_spec = (event.AlignmentEvent & alignment_event_key).fetch1()

    alignment_times = []
    for trial_key, trial_start, trial_stop in zip(trial_keys, trial_starts, trial_ends):
        alignment_event_time = (
            event.Event
            & session_key
            & {"event_type": alignment_spec["alignment_event_type"]}
            & f"event_start_time BETWEEN {trial_start} AND {trial_stop}"
        )
        if alignment_event_time:
            # if  multiple alignment events, pick the last one in the trial
            alignment_event_time = alignment_event_time.fetch(
                "event_start_time", order_by="event_start_time DESC", limit=1
            )[0]
        else:
            alignment_times.append(
                {"trial_key": trial_key, "start": None, "event": None, "end": None}
            )
            continue

        alignment_start_time = (
            event.Event
            & session_key
            & {"event_type": alignment_spec["start_event_type"]}
            & f"event_start_time < {alignment_event_time}"
        )
        if alignment_start_time:
            # if multiple start events, pick most immediate prior alignment event
            alignment_start_time = alignment_start_time.fetch(
                "event_start_time", order_by="event_start_time DESC", limit=1
            )[0]
            alignment_start_time = max(alignment_start_time, trial_start)
        else:
            alignment_start_time = trial_start

        alignment_end_time = (
            event.Event
            & session_key
            & {"event_type": alignment_spec["end_event_type"]}
            & f"event_start_time > {alignment_event_time}"
        )
        if alignment_end_time:
            # if multiple of such start event, pick most immediate after alignment event
            alignment_end_time = alignment_end_time.fetch(
                "event_start_time", order_by="event_start_time", limit=1
            )[0]
            alignment_end_time = min(alignment_end_time, trial_stop)
        else:
            alignment_end_time = trial_stop

        alignment_start_time += alignment_spec["start_time_shift"]
        alignment_event_time += alignment_spec["alignment_time_shift"]
        alignment_end_time += alignment_spec["end_time_shift"]

        alignment_times.append(
            {
                "trial_key": trial_key,
                "start": alignment_start_time,
                "event": alignment_event_time,
                "end": alignment_end_time,
            }
        )

    return pd.DataFrame(alignment_times)