auto-pi-lot / autopilot / 747

pending completion

Build # 747

Build Type

Pull #177

travis-ci-com

Committed by

web-flow

Commit Message

<a href="https://github.com/auto-pi-lot/autopilot/commit/<a class=hub.com/auto-pi-lot/autopilot/commit/<a class="double-link" href="https://git"><a class=hub.com/auto-pi-lot/autopilot/commit/<a class="double-link" href="https://git"><a class=hub.com/auto-pi-lot/autopilot/commit/<a class="double-link" href="https://git"><a class=hub.com/auto-pi-lot/autopilot/commit/<a class="double-link" href="https://git"><a class=hub.com/auto-pi-lot/autopilot/commit/5e685f31c04f069dcdb71f9c7081097d58ae2ea2">5e685f31c">&lt;a href=&quot;https://github.com/auto-pi-lot/autopilot/commit/</a><a class="double-link" href="https://github.com/auto-pi-lot/autopilot/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/auto-pi-lot/autopilot/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/auto-pi-lot/autopilot/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/auto-pi-lot/autopilot/commit/5e685f31c04f069dcdb71f9c7081097d58ae2ea2">5e685f31c</a><a href="https://github.com/auto-pi-lot/autopilot/commit/5e685f31c04f069dcdb71f9c7081097d58ae2ea2">&lt;a href=&quot;https://github.com/auto-pi-lot/autopilot/commit/5e685f31c04f069dcdb71f9c7081097d58ae2ea2&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/auto-pi-lot/autopilot/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/auto-pi-lot/autopilot/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;https://git&amp;quot;&amp;gt;&amp;amp;lt;a class=&amp;lt;/a&amp;gt;hub.com/auto-pi-lot/autopilot/commit/5e685f31c04f069dcdb71f9c7081097d58ae2ea2&amp;quot;&amp;gt;5e685f31c&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/auto-pi-lot/autopilot/commit/5e685f31c04f069dcdb71f9c7081097d58ae2ea2&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/auto-pi-lot/autopilot/commit/5e685f31c04f069dcdb71f9c7081097d&amp;lt;/a&amp;gt;58ae2ea2&amp;</a>quot;&gt;&amp;quot;&amp;gt;Merge &amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/auto-pi-lot/autopilot/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/auto-pi-lot/autopilot/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/auto-pi-lot/autopi... (continued)

Pull Request Pull Request #177: v0.5.0a0 - The Data Modeling Edition

Run Details

3905 of 3905 new or added lines in 64 files covered. (100.0%)

3817 of 11409 relevant lines covered (33.46%)

1.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.88

/autopilot/data/models/protocol.py

"""
Representations of experimental protocols: multiple :class:`.Task` s grouped together with :class:`.Graduation` objects.
"""

import typing
from typing import Optional, Type

import tables
from pydantic import Field, create_model
import pandas as pd

import autopilot
from autopilot.root import Autopilot_Type
from autopilot.data.interfaces.tables import H5F_Group, H5F_Table
from autopilot.data.modeling.base import Table, Schema
# from autopilot.data.models.subject import Subject_Schema
from autopilot.stim.sound.sounds import STRING_PARAMS


class Task_Params(Autopilot_Type):
    """
    Metaclass for storing task parameters

    .. todo::

        Not yet used in GUI, terminal, and subject classes. Will replace the dictionary structure ASAP
    """

class Trial_Data(Table):
    """
    Base class for declaring trial data.

    Tasks should subclass this and add any additional parameters that are needed.
    The subject class will then use this to create a table in the hdf5 file.

    See :attr:`.Nafc.TrialData` for an example
    """
    group: Optional[str] = Field(None, description="Path of the parent step group")
    session: int = Field(..., description="Current training session, increments every time the task is started")
    session_uuid: Optional[str] = Field(None, description="Each session gets a unique uuid, regardless of the session integer, to enable independent addressing of sessions when session numbers might overlap (eg. reassignment)")
    trial_num: int = Field(..., description="Trial data is grouped within, well, trials, which increase (rather than resetting) across sessions within a task",
                           datajoint={"key":True})



class Step_Group(H5F_Group):
    """
    An hdf5 group for an individual step within a protocol.

    Typically this is populated by passing a step number and a dictionary of step parameters.
    """
    step_name: str
    step: int
    path: str
    trial_data: Optional[Type[Trial_Data]] = Trial_Data
    continuous_group: Optional[H5F_Group] = None

    def __init__(self,
                 step: int,
                 group_path: str,
                 step_dict: Optional[dict] = None,
                 step_name: Optional[str] = None,
                 trial_data: Optional[Type[Trial_Data]] = None,
                 **data):
        """

        Args:
            step (int): Step number within a protocol
            group_path (str): Path to the group within an HDF5 file
            step_dict (dict): Dictionary of step parameters. Either this or ``step_name`` must be passed
            step_name (str): Step name -- if ``step_dict`` is not present, use this to generate a name for the created hdf5 group
            trial_data (:class:`.Trial_Data`): Explicitly passed Trial_Data object. If not passed, get from the ``task_type`` parameter
                in the ``step_dict``
            **data: passed to superclass __init__ method
        """
        self._init_logger()
        if step_name is None and step_dict is None:
            raise ValueError('Need to give us something that will let us identify where to make this table!')

        if step_name is None:
            step_name = step_dict['step_name']

        if trial_data is None and step_dict:

            task_class = autopilot.get_task(step_dict['task_type'])
            if hasattr(task_class, 'TrialData'):
                trial_data = task_class.TrialData
                if issubclass(trial_data, tables.IsDescription):
                    self._logger.warning("Using pytables descriptions as TrialData is deprecated! Update to the pydantic TrialData model! Converting to TrialData class")
                    trial_data = Trial_Data.from_pytables_description(trial_data)

            else:
                trial_data = Trial_Data
        else:
            trial_data = Trial_Data

        # complete table decription with any stim parameters
        if step_dict and 'stim' in step_dict.keys():
            trial_data = self._make_stim_descriptors(step_dict['stim'], trial_data)

        # make group descriptions and children to prepare for ``make``
        group_name = f"S{step:02d}_{step_name}"
        path = '/'.join([group_path, group_name])

        continuous_group = H5F_Group(path='/'.join([path, 'continuous_data']))
        trial_table = H5F_Table(path='/'.join([path, 'trial_data']),
                                description=trial_data.to_pytables_description(),
                                title='Trial Data')

        super().__init__(
            path = path,
            step_name = step_name,
            step = step,
            trial_data = trial_data,
            continuous_group = continuous_group,
            children = [trial_table, continuous_group],
            **data
        )


    def _make_stim_descriptors(self, stim:dict, trial_data: Type[Trial_Data]) -> Type[Trial_Data]:
        if 'groups' in stim.keys():
            # managers have stim nested within groups, but this is still really ugly
            sound_params = {}
            for g in stim['groups']:
                for side, sounds in g['sounds'].items():
                    for sound in sounds:
                        for k, v in sound.items():
                            if k in STRING_PARAMS:
                                sound_params[k] = (str, ...)
                            else:
                                sound_params[k] = (float, ...)

        elif 'sounds' in stim.keys():
            # for now we just assume they're floats
            sound_params = {}
            for side, sounds in stim['sounds'].items():
                # each side has a list of sounds
                for sound in sounds:
                    for k, v in sound.items():
                        if k in STRING_PARAMS:
                            sound_params[k] = (str, ...)
                        else:
                            sound_params[k] = (float, ...)

        else:
            raise ValueError(f'Dont know how to handle stim like {stim}')

        trial_data = create_model('Trial_Data', __base__ = trial_data, **sound_params)
        return trial_data



class Protocol_Group(H5F_Group):
    """
    The group and subgroups for a given protocol.

    For each protocol, a main group is created that has the name of the protocol,
    and then subgroups are created for each of its steps.

    Within each step group, a table is made for TrialData, and tables are created
    as-needed for continuous data.

    For Example::

        / data
        |--- protocol_name
            |--- S##_step_name
            |   |--- trial_data
            |   |--- continuous_data
            |--- ... additional steps

    .. todo::

        Also make a Step group... what's the matter with ya.

    """
    protocol_name: str
    protocol: typing.List[dict]
    tabs: typing.List[H5F_Table]
    steps: typing.List[Step_Group]

    def __init__(self,
                 protocol_name: str,
                 protocol: typing.List[dict],
                 **data):
        """
        Override default __init__ method to populate a task's groups.

        .. todo::

            When finished, replace the implicit structure of the protocol dictionary with :class:`.Task_Params`

        Args:
            protocol_name (str): Name of a protocol (filename minus ``.json``)
            protocol (List[dict]): A list of dictionaries, one with the parameters for each task level.
            **data: passed to superclass init

        """
        path = f'/data/{protocol_name}'
        steps = []
        _tables = []

        for i, step in enumerate(protocol):
            step_group = Step_Group(step=i, group_path=path, step_dict=step)
            steps.append(step_group)

        super().__init__(
            path=path,
            protocol_name=protocol_name,
            protocol=protocol,
            tabs=_tables,
            steps=steps,
            children=steps,
            **data)

class Step_Data(Schema):
    """
    Schema for storing data for a single step of a protocol
    """
    task: Task_Params
    trial_data_table: Table
    trial_data: Trial_Data
    continuous_data: typing.Dict[str, list]

class Protocol_Data(Schema):
    steps: typing.List[Step_Data]


1	"""
2	Representations of experimental protocols: multiple :class:`.Task` s grouped together with :class:`.Graduation` objects.
3	"""
4
5	import typing	3✔
6	from typing import Optional, Type	3✔
7
8	import tables	3✔
9	from pydantic import Field, create_model	3✔
10	import pandas as pd	3✔
11
12	import autopilot	3✔
13	from autopilot.root import Autopilot_Type	3✔
14	from autopilot.data.interfaces.tables import H5F_Group, H5F_Table	3✔
15	from autopilot.data.modeling.base import Table, Schema	3✔
16	# from autopilot.data.models.subject import Subject_Schema
17	from autopilot.stim.sound.sounds import STRING_PARAMS	3✔
18
19
20	class Task_Params(Autopilot_Type):	3✔
21	"""
22	Metaclass for storing task parameters
23
24	.. todo::
25
26	Not yet used in GUI, terminal, and subject classes. Will replace the dictionary structure ASAP
27	"""
28
29	class Trial_Data(Table):	3✔
30	"""
31	Base class for declaring trial data.
32
33	Tasks should subclass this and add any additional parameters that are needed.
34	The subject class will then use this to create a table in the hdf5 file.
35
36	See :attr:`.Nafc.TrialData` for an example
37	"""
38	group: Optional[str] = Field(None, description="Path of the parent step group")	3✔
39	session: int = Field(..., description="Current training session, increments every time the task is started")	3✔
40	session_uuid: Optional[str] = Field(None, description="Each session gets a unique uuid, regardless of the session integer, to enable independent addressing of sessions when session numbers might overlap (eg. reassignment)")	3✔
41	trial_num: int = Field(..., description="Trial data is grouped within, well, trials, which increase (rather than resetting) across sessions within a task",	3✔
42	datajoint={"key":True})
43
44
45
46	class Step_Group(H5F_Group):	3✔
47	"""
48	An hdf5 group for an individual step within a protocol.
49
50	Typically this is populated by passing a step number and a dictionary of step parameters.
51	"""
52	step_name: str	3✔
53	step: int	3✔
54	path: str	3✔
55	trial_data: Optional[Type[Trial_Data]] = Trial_Data	3✔
56	continuous_group: Optional[H5F_Group] = None	3✔
57
58	def __init__(self,	3✔
59	step: int,
60	group_path: str,
61	step_dict: Optional[dict] = None,
62	step_name: Optional[str] = None,
63	trial_data: Optional[Type[Trial_Data]] = None,
64	**data):
65	"""
66
67	Args:
68	step (int): Step number within a protocol
69	group_path (str): Path to the group within an HDF5 file
70	step_dict (dict): Dictionary of step parameters. Either this or ``step_name`` must be passed
71	step_name (str): Step name -- if ``step_dict`` is not present, use this to generate a name for the created hdf5 group
72	trial_data (:class:`.Trial_Data`): Explicitly passed Trial_Data object. If not passed, get from the ``task_type`` parameter
73	in the ``step_dict``
74	**data: passed to superclass __init__ method
75	"""
76	self._init_logger()	3✔
77	if step_name is None and step_dict is None:	3✔
78	raise ValueError('Need to give us something that will let us identify where to make this table!')	×
79
80	if step_name is None:	3✔
81	step_name = step_dict['step_name']	3✔
82
83	if trial_data is None and step_dict:	3✔
84
85	task_class = autopilot.get_task(step_dict['task_type'])	3✔
86	if hasattr(task_class, 'TrialData'):	3✔
87	trial_data = task_class.TrialData	3✔
88	if issubclass(trial_data, tables.IsDescription):	3✔
89	self._logger.warning("Using pytables descriptions as TrialData is deprecated! Update to the pydantic TrialData model! Converting to TrialData class")	3✔
90	trial_data = Trial_Data.from_pytables_description(trial_data)	3✔
91
92	else:
93	trial_data = Trial_Data	×
94	else:
95	trial_data = Trial_Data	×
96
97	# complete table decription with any stim parameters
98	if step_dict and 'stim' in step_dict.keys():	3✔
99	trial_data = self._make_stim_descriptors(step_dict['stim'], trial_data)	3✔
100
101	# make group descriptions and children to prepare for ``make``
102	group_name = f"S{step:02d}_{step_name}"	3✔
103	path = '/'.join([group_path, group_name])	3✔
104
105	continuous_group = H5F_Group(path='/'.join([path, 'continuous_data']))	3✔
106	trial_table = H5F_Table(path='/'.join([path, 'trial_data']),	3✔
107	description=trial_data.to_pytables_description(),
108	title='Trial Data')
109
110	super().__init__(	3✔
111	path = path,
112	step_name = step_name,
113	step = step,
114	trial_data = trial_data,
115	continuous_group = continuous_group,
116	children = [trial_table, continuous_group],
117	**data
118	)
119
120
121	def _make_stim_descriptors(self, stim:dict, trial_data: Type[Trial_Data]) -> Type[Trial_Data]:	3✔
122	if 'groups' in stim.keys():	3✔
123	# managers have stim nested within groups, but this is still really ugly
124	sound_params = {}	×
125	for g in stim['groups']:	×
126	for side, sounds in g['sounds'].items():	×
127	for sound in sounds:	×
128	for k, v in sound.items():	×
129	if k in STRING_PARAMS:	×
130	sound_params[k] = (str, ...)	×
131	else:
132	sound_params[k] = (float, ...)	×
133
134	elif 'sounds' in stim.keys():	3✔
135	# for now we just assume they're floats
136	sound_params = {}	3✔
137	for side, sounds in stim['sounds'].items():	3✔
138	# each side has a list of sounds
139	for sound in sounds:	3✔
140	for k, v in sound.items():	3✔
141	if k in STRING_PARAMS:	3✔
142	sound_params[k] = (str, ...)	3✔
143	else:
144	sound_params[k] = (float, ...)	3✔
145
146	else:
147	raise ValueError(f'Dont know how to handle stim like {stim}')	×
148
149	trial_data = create_model('Trial_Data', __base__ = trial_data, **sound_params)	3✔
150	return trial_data	3✔
151
152
153
154	class Protocol_Group(H5F_Group):	3✔
155	"""
156	The group and subgroups for a given protocol.
157
158	For each protocol, a main group is created that has the name of the protocol,
159	and then subgroups are created for each of its steps.
160
161	Within each step group, a table is made for TrialData, and tables are created
162	as-needed for continuous data.
163
164	For Example::
165
166	/ data
167	\|--- protocol_name
168	\|--- S##_step_name
169	\| \|--- trial_data
170	\| \|--- continuous_data
171	\|--- ... additional steps
172
173	.. todo::
174
175	Also make a Step group... what's the matter with ya.
176
177	"""
178	protocol_name: str	3✔
179	protocol: typing.List[dict]	3✔
180	tabs: typing.List[H5F_Table]	3✔
181	steps: typing.List[Step_Group]	3✔
182
183	def __init__(self,	3✔
184	protocol_name: str,
185	protocol: typing.List[dict],
186	**data):
187	"""
188	Override default __init__ method to populate a task's groups.
189
190	.. todo::
191
192	When finished, replace the implicit structure of the protocol dictionary with :class:`.Task_Params`
193
194	Args:
195	protocol_name (str): Name of a protocol (filename minus ``.json``)
196	protocol (List[dict]): A list of dictionaries, one with the parameters for each task level.
197	**data: passed to superclass init
198
199	"""
200	path = f'/data/{protocol_name}'	3✔
201	steps = []	3✔
202	_tables = []	3✔
203
204	for i, step in enumerate(protocol):	3✔
205	step_group = Step_Group(step=i, group_path=path, step_dict=step)	3✔
206	steps.append(step_group)	3✔
207
208	super().__init__(	3✔
209	path=path,
210	protocol_name=protocol_name,
211	protocol=protocol,
212	tabs=_tables,
213	steps=steps,
214	children=steps,
215	**data)
216
217	class Step_Data(Schema):	3✔
218	"""
219	Schema for storing data for a single step of a protocol
220	"""
221	task: Task_Params	3✔
222	trial_data_table: Table	3✔
223	trial_data: Trial_Data	3✔
224	continuous_data: typing.Dict[str, list]	3✔
225
226	class Protocol_Data(Schema):	3✔
227	steps: typing.List[Step_Data]	3✔
228

auto-pi-lot / autopilot / 747

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous