Source code for allensdk.brain_observatory.behavior.schemas

import numpy as np
from marshmallow import RAISE, Schema, fields

STYPE_DICT = {fields.Float: 'float', fields.Int: 'int',
              fields.String: 'text', fields.List: 'text',
              fields.DateTime: 'text', fields.UUID: 'text'}
TYPE_DICT = {fields.Float: float, fields.Int: int, fields.String: str,
             fields.List: np.ndarray, fields.DateTime: str, fields.UUID: str}


class RaisingSchema(Schema):
    class Meta:
        unknown = RAISE


class SubjectMetadataSchema(RaisingSchema):
    """This schema contains metadata pertaining to a subject in either a
    behavior or behavior + ophys experiment.
    """

    neurodata_type = 'BehaviorSubject'
    neurodata_type_inc = 'Subject'
    neurodata_doc = "Metadata for an AIBS behavior or behavior + ophys subject"
    # Fields to skip converting to extension
    # In this case they already exist in the 'Subject' builtin pyNWB class
    neurodata_skip = {"age_in_days", "genotype", "sex", "subject_id"}

    age_in_days = fields.String(
        doc='Age of the specimen donor/subject (in days)',
        required=True,
    )
    driver_line = fields.List(
        fields.String,
        doc="Driver line of subject",
        required=False,
        allow_none=True,
        shape=(None,),
    )
    # 'full_genotype' will be stored in pynwb Subject 'genotype' attr
    genotype = fields.String(
        doc='full genotype of subject',
        required=True,
    )
    # 'mouse_id' will be stored in pynwb Subject 'subject_id' attr
    subject_id = fields.Int(
        doc='Mouse ID of subject',
        required=True,
    )
    reporter_line = fields.String(
        doc="Reporter line of subject",
        required=True,
    )
    sex = fields.String(
        doc='Sex of the specimen donor/subject',
        required=True,
    )


class BehaviorMetadataSchema(RaisingSchema):
    """This schema contains metadata pertaining to behavior.
    """
    neurodata_type = 'BehaviorMetadata'
    neurodata_type_inc = 'LabMetaData'
    neurodata_doc = "Metadata for behavior and behavior + ophys experiments"
    neurodata_skip = {"date_of_acquisition"}

    behavior_session_id = fields.Int(
        doc='The unique ID for the behavior session',
        required=True
    )
    behavior_session_uuid = fields.UUID(
        doc='MTrain record for session, also called foraging_id',
        required=True,
    )
    stimulus_frame_rate = fields.Float(
        doc=('Frame rate (frames/second) of the '
             'visual_stimulus from the monitor'),
        required=True,
    )
    session_type = fields.String(
        doc='Experimental session description',
        allow_none=True,
        required=True,
    )
    # 'date_of_acquisition' will be stored in
    # pynwb NWBFile 'session_start_time' attr
    date_of_acquisition = fields.DateTime(
        doc='Date of the experiment (UTC, as string)',
        required=True,
    )
    equipment_name = fields.String(
        doc='Name of behavior or optical physiology experiment rig',
        required=True,
    )
    project_code = fields.String(
        doc='String Id of project associated with session.',
        allow_none=True,
        required=True,
    )


class NwbOphysMetadataSchema(RaisingSchema):
    """This schema contains fields that will be stored in pyNWB base classes
    pertaining to optical physiology."""
    # 'emission_lambda' will be stored in
    # pyNWB OpticalChannel 'emission_lambda' attr
    emission_lambda = fields.Float(
        doc='Emission lambda of fluorescent indicator',
        required=True,
    )
    # 'excitation_lambda' will be stored in the pyNWB ImagingPlane
    # 'excitation_lambda' attr
    excitation_lambda = fields.Float(
        doc='Excitation lambda of fluorescent indicator',
        required=True,
    )
    # 'indicator' will be stored in the pyNWB ImagingPlane 'indicator' attr
    indicator = fields.String(
        doc='Name of optical physiology fluorescent indicator',
        required=True,
    )
    # 'targeted_structure' will be stored in the pyNWB
    # ImagingPlane 'location' attr
    targeted_structure = fields.String(
        doc='Anatomical structure targeted for two-photon acquisition',
        required=True,
    )
    # 'ophys_frame_rate' will  be stored in the pyNWB ImagingPlane
    # 'imaging_rate' attr
    ophys_frame_rate = fields.Float(
        doc='Frame rate (frames/second) of the two-photon microscope',
        required=True,
    )


class OphysMetadataSchema(NwbOphysMetadataSchema):
    """This schema contains metadata pertaining to optical physiology (ophys).
    """
    ophys_experiment_id = fields.Int(
        doc='Unique ID for the ophys experiment (aka imaging plane)',
        required=True
    )
    ophys_session_id = fields.Int(
        doc='Unique ID for the ophys session',
        required=True
    )
    ophys_container_id = fields.Int(
        doc='Container ID for the container that contains this ophys session',
        required=True,
    )
    imaging_depth = fields.Int(
        doc=('Depth (microns) below the cortical surface '
             'targeted for two-photon acquisition'),
        required=True,
    )
    targeted_imaging_depth = fields.Int(
        doc=(
            "Average Depth (microns) below the cortical surface "
            "across experiments of a container."
            "targeted for two-photon acquisition"
        ),
        required=True,
    )
    field_of_view_width = fields.Int(
        doc='Width of optical physiology imaging plane in pixels',
        required=True,
    )
    field_of_view_height = fields.Int(
        doc='Height of optical physiology imaging plane in pixels',
        required=True,
    )
    imaging_plane_group = fields.Int(
        doc=('A numeric index which indicates the order that an imaging plane '
             'was acquired for a mesoscope experiment. Will be -1 for '
             'non-mesoscope data'),
        required=True
    )
    imaging_plane_group_count = fields.Int(
        doc=('The total number of plane groups collected in a session '
             'for a mesoscope experiment. Will be 0 if the scope did not '
             'capture multiple concurrent imaging planes.'),
        required=True
    )
    project_code = fields.String(
        doc='String Id of project associated with session.',
        allow_none=True,
        required=True,
    )


class OphysBehaviorMetadataSchema(BehaviorMetadataSchema, OphysMetadataSchema):
    """ This schema contains fields pertaining to ophys+behavior. It is used
    as a template for generating our custom NWB behavior + ophys extension.
    """
    neurodata_type = 'OphysBehaviorMetadata'
    neurodata_type_inc = 'BehaviorMetadata'
    neurodata_doc = "Metadata for behavior + ophys experiments"
    # Fields to skip converting to extension
    # They already exist as attributes for the following pyNWB classes:
    # OpticalChannel, ImagingPlane, NWBFile
    neurodata_skip = {"emission_lambda", "excitation_lambda", "indicator",
                      "targeted_structure", "date_of_acquisition",
                      "ophys_frame_rate"}


class CompleteOphysBehaviorMetadataSchema(OphysBehaviorMetadataSchema,
                                          SubjectMetadataSchema):
    """This schema combines fields from behavior, ophys, and subject schemas.
    Metadata info is passed by the behavior+ophys session in a combined lump
    containing all the field types.
    """
    pass


class BehaviorTaskParametersSchema(RaisingSchema):
    """This schema encompasses task parameters used for behavior or
    ophys + behavior.
    """
    neurodata_type = 'BehaviorTaskParameters'
    neurodata_type_inc = 'LabMetaData'
    neurodata_doc = "Metadata for behavior or behavior + ophys task parameters"

    blank_duration_sec = fields.List(
        fields.Float,
        doc=('The lower and upper bound (in seconds) for a randomly chosen '
             'inter-stimulus interval duration for a trial'),
        required=True,
        shape=(2,),
    )
    stimulus_duration_sec = fields.Float(
        doc='Duration of each stimulus presentation in seconds',
        required=True,
        allow_nan=True
    )
    omitted_flash_fraction = fields.Float(
        doc='Fraction of flashes/image presentations that were omitted',
        required=True,
        allow_nan=True,
    )
    response_window_sec = fields.List(
        fields.Float,
        doc=('The lower and upper bound (in seconds) for a randomly chosen '
             'time window where subject response influences trial outcome'),
        required=True,
        shape=(2,),
    )
    reward_volume = fields.Float(
        doc='Volume of water (in mL) delivered as reward',
        required=True,
    )
    auto_reward_volume = fields.Float(
        doc='Volume of water (in mL) delivered as an automatic reward',
        required=True,
    )
    session_type = fields.String(
        doc='Stage of behavioral task',
        required=True,
    )
    stimulus = fields.String(
        doc='Stimulus type',
        required=True,
    )
    stimulus_distribution = fields.String(
        doc=("Distribution type of drawing change times "
             "(e.g. 'geometric', 'exponential')"),
        required=True,
    )
    task = fields.String(
        doc='The name of the behavioral task',
        required=True,
    )
    n_stimulus_frames = fields.Int(
        doc='Total number of stimuli frames',
        required=True,
    )
    stimulus_name = fields.String(
        doc="Name of the stimulus file set presented",
        required=False
    )


class EyeTrackingRigGeometry(RaisingSchema):
    """Eye tracking rig geometry"""
    values = fields.Float(
        doc='position/rotation with respect to (x, y, z)',
        required=True,
        shape=(3,)
    )
    unit_of_measurement = fields.Str(
        doc='Unit of measurement for the data',
        required=True
    )


class OphysEyeTrackingRigMetadataSchema(RaisingSchema):
    """This schema encompasses metadata for ophys experiment rig
    """
    neurodata_type = 'OphysEyeTrackingRigMetadata'
    neurodata_type_inc = 'NWBDataInterface'
    neurodata_doc = "Metadata for ophys experiment rig"

    equipment = fields.Str(
        doc='Description of rig',
        required=True
    )
    monitor_position = fields.Nested(
        EyeTrackingRigGeometry,
        doc='position of monitor (x, y, z)',
        required=True
    )
    camera_position = fields.Nested(
        EyeTrackingRigGeometry,
        doc='position of camera (x, y, z)',
        required=True
    )
    led_position = fields.Nested(
        EyeTrackingRigGeometry,
        doc='position of LED (x, y, z)',
        required=True
    )
    monitor_rotation = fields.Nested(
        EyeTrackingRigGeometry,
        doc='rotation of monitor (x, y, z)',
        required=True
    )
    camera_rotation = fields.Nested(
        EyeTrackingRigGeometry,
        doc='rotation of camera (x, y, z)',
        required=True
    )