15073834064

Committed 16 May 2025 05:16PM UTC coverage: 49.414% (+2.6%) from 46.79%

Build # 15073834064

Build Type

Pull #750

github

Specific Base c98309

Committed by

web-flow

Commit Message

Merge 8e475a77c into e481532ae

Pull Request Pull Request #750: Online plots

Run Details

538 of 720 new or added lines in 3 files covered. (74.72%)

1000 existing lines in 20 files now uncovered.

4677 of 9465 relevant lines covered (49.41%)

0.49 hits per line

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

50.0

/iblrig_tasks/_iblrig_tasks_passiveChoiceWorld/task.py

import logging
import sys
import time
from datetime import timedelta
from pathlib import Path

import pandas as pd
import yaml

import iblrig.misc
from iblrig.base_choice_world import ChoiceWorldSession

log = logging.getLogger('iblrig.task')

# read defaults from task_parameters.yaml
with open(Path(__file__).parent.joinpath('task_parameters.yaml')) as f:
    DEFAULTS = yaml.safe_load(f)


class Session(ChoiceWorldSession):
    protocol_name = '_iblrig_tasks_passiveChoiceWorld'

    def __init__(
        self,
        *args,
        session_template_id=0,
        duration_spontaneous: int = DEFAULTS['SPONTANEOUS_ACTIVITY_SECONDS'],
        skip_event_replay: bool = DEFAULTS['SKIP_EVENT_REPLAY'],
        **kwargs,
    ):
        self.extractor_tasks = ['PassiveRegisterRaw', 'PassiveTask']
        super(ChoiceWorldSession, self).__init__(**kwargs)
        self.task_params.SESSION_TEMPLATE_ID = session_template_id
        all_trials = pd.read_parquet(Path(__file__).parent.joinpath('passiveChoiceWorld_trials_fixtures.pqt'))
        self.trials_table = all_trials[all_trials['session_id'] == self.task_params.SESSION_TEMPLATE_ID].copy()
        self.trials_table['reward_valve_time'] = self.compute_reward_time(amount_ul=self.trials_table['reward_amount'])
        assert duration_spontaneous < 60 * 60 * 24
        self.task_params['SPONTANEOUS_ACTIVITY_SECONDS'] = duration_spontaneous
        self.task_params['SKIP_EVENT_REPLAY'] = skip_event_replay
        if self.hardware_settings['MAIN_SYNC']:
            log.error('PassiveChoiceWorld extraction not supported for Bpod-only sessions!')

    @staticmethod
    def extra_parser():
        """:return: argparse.parser()"""
        parser = super(Session, Session).extra_parser()
        parser.add_argument(
            '--session_template_id',
            option_strings=['--session_template_id'],
            dest='session_template_id',
            default=0,
            type=int,
            help='pre-generated session index (zero-based)',
        )
        parser.add_argument(
            '--duration_spontaneous',
            option_strings=['--duration_spontaneous'],
            dest='duration_spontaneous',
            default=DEFAULTS['SPONTANEOUS_ACTIVITY_SECONDS'],
            type=int,
            help=f'duration of spontaneous activity in seconds (default: {DEFAULTS["SPONTANEOUS_ACTIVITY_SECONDS"]} s)',
        )
        parser.add_argument(
            '--skip_event_replay',
            option_strings=['--skip_event_replay'],
            action='store_true',
            dest='skip_event_replay',
            help='skip replay of events',
        )
        return parser

    def start_hardware(self):
        if not self.is_mock:
            self.start_mixin_frame2ttl()
            self.start_mixin_bpod()
            self.start_mixin_valve()
            self.start_mixin_sound()
            self.start_mixin_bonsai_cameras()
            self.start_mixin_bonsai_microphone()
            self.start_mixin_rotary_encoder()

    def get_state_machine_trial(self, *args, **kwargs):
        pass

    def next_trial(self):
        pass

    def _run(self):
        """Run the task with the actual state machine."""
        self.trigger_bonsai_cameras()

        # Run the passive part i.e. spontaneous activity and RFMapping stim
        self.run_passive_visual_stim(sa_time=timedelta(seconds=self.task_params['SPONTANEOUS_ACTIVITY_SECONDS']))

        if self.task_params['SKIP_EVENT_REPLAY'] is True:
            log.info('Skipping replay of task events')
            return

        # run the replay of task events: V for valve, T for tone, N for noise, G for gratings
        log.info('Starting replay of task events')
        action_show_stim = self.bpod.actions['bonsai_show_stim'][1]
        action_hide_stim = self.bpod.actions['bonsai_hide_stim'][1]
        byte_show_stim = self.bpod.serial_messages[action_show_stim]['message'][-1]
        byte_hide_stim = self.bpod.serial_messages[action_hide_stim]['message'][-1]

        if not self.is_mock:
            self.start_mixin_bonsai_visual_stimulus()
        for trial_num, trial in self.trials_table.iterrows():
            self.trial_num = trial_num
            log.info(f'Delay: {trial.stim_delay}; ID: {trial.stim_type}; Count: {self.trial_num}/300')
            sys.stdout.flush()
            time.sleep(trial.stim_delay)
            if trial.stim_type == 'V':
                self.valve_open(self.reward_time)
            elif trial.stim_type == 'T':
                self.sound_play_tone(state_timer=0.102)
            elif trial.stim_type == 'N':
                self.sound_play_noise(state_timer=0.510)
            elif trial.stim_type == 'G':
                # this will send the current trial info to the visual stim
                # we need to make sure Bonsai is in a state to display stimuli
                self.send_trial_info_to_bonsai()
                self.bonsai_visual_udp_client.send_message(r'/re', byte_show_stim)
                time.sleep(0.3)  # todo: this is a very inaccurate way of controlling stim duration!
                self.bonsai_visual_udp_client.send_message(r'/re', byte_hide_stim)
            if self.paths.SESSION_FOLDER.joinpath('.stop').exists():
                self.paths.SESSION_FOLDER.joinpath('.stop').unlink()
                break


if __name__ == '__main__':  # pragma: no cover
    # python .\iblrig_tasks\_iblrig_tasks_spontaneous\task.py --subject mysubject
    kwargs = iblrig.misc.get_task_arguments(parents=[Session.extra_parser()])
    sess = Session(**kwargs)
    sess.run()

1	import logging	1✔
2	import sys	1✔
3	import time	1✔
4	from datetime import timedelta	1✔
5	from pathlib import Path	1✔
6
7	import pandas as pd	1✔
8	import yaml	1✔
9
10	import iblrig.misc	1✔
11	from iblrig.base_choice_world import ChoiceWorldSession	1✔
12
13	log = logging.getLogger('iblrig.task')	1✔
14
15	# read defaults from task_parameters.yaml
16	with open(Path(__file__).parent.joinpath('task_parameters.yaml')) as f:	1✔
17	DEFAULTS = yaml.safe_load(f)	1✔
18
19
20	class Session(ChoiceWorldSession):	1✔
21	protocol_name = '_iblrig_tasks_passiveChoiceWorld'	1✔
22
23	def __init__(	1✔
24	self,
25	*args,
26	session_template_id=0,
27	duration_spontaneous: int = DEFAULTS['SPONTANEOUS_ACTIVITY_SECONDS'],
28	skip_event_replay: bool = DEFAULTS['SKIP_EVENT_REPLAY'],
29	**kwargs,
30	):
31	self.extractor_tasks = ['PassiveRegisterRaw', 'PassiveTask']	1✔
32	super(ChoiceWorldSession, self).__init__(**kwargs)	1✔
33	self.task_params.SESSION_TEMPLATE_ID = session_template_id	1✔
34	all_trials = pd.read_parquet(Path(__file__).parent.joinpath('passiveChoiceWorld_trials_fixtures.pqt'))	1✔
35	self.trials_table = all_trials[all_trials['session_id'] == self.task_params.SESSION_TEMPLATE_ID].copy()	1✔
36	self.trials_table['reward_valve_time'] = self.compute_reward_time(amount_ul=self.trials_table['reward_amount'])	1✔
37	assert duration_spontaneous < 60 * 60 * 24	1✔
38	self.task_params['SPONTANEOUS_ACTIVITY_SECONDS'] = duration_spontaneous	1✔
39	self.task_params['SKIP_EVENT_REPLAY'] = skip_event_replay	1✔
40	if self.hardware_settings['MAIN_SYNC']:	1✔
41	log.error('PassiveChoiceWorld extraction not supported for Bpod-only sessions!')	1✔
42
43	@staticmethod	1✔
44	def extra_parser():	1✔
45	""":return: argparse.parser()"""
46	parser = super(Session, Session).extra_parser()	1✔
47	parser.add_argument(	1✔
48	'--session_template_id',
49	option_strings=['--session_template_id'],
50	dest='session_template_id',
51	default=0,
52	type=int,
53	help='pre-generated session index (zero-based)',
54	)
55	parser.add_argument(	1✔
56	'--duration_spontaneous',
57	option_strings=['--duration_spontaneous'],
58	dest='duration_spontaneous',
59	default=DEFAULTS['SPONTANEOUS_ACTIVITY_SECONDS'],
60	type=int,
61	help=f'duration of spontaneous activity in seconds (default: {DEFAULTS["SPONTANEOUS_ACTIVITY_SECONDS"]} s)',
62	)
63	parser.add_argument(	1✔
64	'--skip_event_replay',
65	option_strings=['--skip_event_replay'],
66	action='store_true',
67	dest='skip_event_replay',
68	help='skip replay of events',
69	)
70	return parser	1✔
71
72	def start_hardware(self):	1✔
UNCOV 73	if not self.is_mock:	×
UNCOV 74	self.start_mixin_frame2ttl()	×
UNCOV 75	self.start_mixin_bpod()	×
UNCOV 76	self.start_mixin_valve()	×
UNCOV 77	self.start_mixin_sound()	×
UNCOV 78	self.start_mixin_bonsai_cameras()	×
UNCOV 79	self.start_mixin_bonsai_microphone()	×
UNCOV 80	self.start_mixin_rotary_encoder()	×
81
82	def get_state_machine_trial(self, args, *kwargs):	1✔
83	pass	1✔
84
85	def next_trial(self):	1✔
86	pass	1✔
87
88	def _run(self):	1✔
89	"""Run the task with the actual state machine."""
UNCOV 90	self.trigger_bonsai_cameras()	×
91
92	# Run the passive part i.e. spontaneous activity and RFMapping stim
UNCOV 93	self.run_passive_visual_stim(sa_time=timedelta(seconds=self.task_params['SPONTANEOUS_ACTIVITY_SECONDS']))	×
94
UNCOV 95	if self.task_params['SKIP_EVENT_REPLAY'] is True:	×
UNCOV 96	log.info('Skipping replay of task events')	×
UNCOV 97	return	×
98
99	# run the replay of task events: V for valve, T for tone, N for noise, G for gratings
UNCOV 100	log.info('Starting replay of task events')	×
UNCOV 101	action_show_stim = self.bpod.actions['bonsai_show_stim'][1]	×
UNCOV 102	action_hide_stim = self.bpod.actions['bonsai_hide_stim'][1]	×
UNCOV 103	byte_show_stim = self.bpod.serial_messages[action_show_stim]['message'][-1]	×
UNCOV 104	byte_hide_stim = self.bpod.serial_messages[action_hide_stim]['message'][-1]	×
105
UNCOV 106	if not self.is_mock:	×
UNCOV 107	self.start_mixin_bonsai_visual_stimulus()	×
UNCOV 108	for trial_num, trial in self.trials_table.iterrows():	×
UNCOV 109	self.trial_num = trial_num	×
UNCOV 110	log.info(f'Delay: {trial.stim_delay}; ID: {trial.stim_type}; Count: {self.trial_num}/300')	×
UNCOV 111	sys.stdout.flush()	×
UNCOV 112	time.sleep(trial.stim_delay)	×
UNCOV 113	if trial.stim_type == 'V':	×
UNCOV 114	self.valve_open(self.reward_time)	×
UNCOV 115	elif trial.stim_type == 'T':	×
UNCOV 116	self.sound_play_tone(state_timer=0.102)	×
UNCOV 117	elif trial.stim_type == 'N':	×
UNCOV 118	self.sound_play_noise(state_timer=0.510)	×
UNCOV 119	elif trial.stim_type == 'G':	×
120	# this will send the current trial info to the visual stim
121	# we need to make sure Bonsai is in a state to display stimuli
UNCOV 122	self.send_trial_info_to_bonsai()	×
UNCOV 123	self.bonsai_visual_udp_client.send_message(r'/re', byte_show_stim)	×
UNCOV 124	time.sleep(0.3) # todo: this is a very inaccurate way of controlling stim duration!	×
UNCOV 125	self.bonsai_visual_udp_client.send_message(r'/re', byte_hide_stim)	×
UNCOV 126	if self.paths.SESSION_FOLDER.joinpath('.stop').exists():	×
UNCOV 127	self.paths.SESSION_FOLDER.joinpath('.stop').unlink()	×
UNCOV 128	break	×
129
130
131	if __name__ == '__main__': # pragma: no cover
132	# python .\iblrig_tasks\_iblrig_tasks_spontaneous\task.py --subject mysubject
133	kwargs = iblrig.misc.get_task_arguments(parents=[Session.extra_parser()])
134	sess = Session(**kwargs)
135	sess.run()

int-brain-lab / iblrig / 15073834064

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