14196118657

Committed 01 Apr 2025 12:52PM UTC coverage: 47.634% (+0.8%) from 46.79%

Build # 14196118657

Build Type

Pull #795

github

Specific Base cfb5bd

Committed by

web-flow

Commit Message

Merge 5ba5d5f25 into 58cf64236

Pull Request Pull Request #795: fixes for habituation CW

Run Details

11 of 12 new or added lines in 1 file covered. (91.67%)

1083 existing lines in 22 files now uncovered.

4288 of 9002 relevant lines covered (47.63%)

0.95 hits per line

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

70.73

/iblrig/misc.py

"""
Provides collection of functionality used throughout the iblrig repository.

Assortment of functions, frequently used, but without a great deal of commonality. Functions can,
and should, be broken out into their own files and/or classes as the organizational needs of this
repo change over time.
"""

import argparse
import datetime
import logging
from collections.abc import Sequence
from pathlib import Path
from typing import Literal

import numpy as np

FLAG_FILE_NAMES = ['transfer_me.flag', 'create_me.flag', 'poop_count.flag', 'passive_data_for_ephys.flag']

log = logging.getLogger(__name__)


def get_task_argument_parser(parents: Sequence[argparse.ArgumentParser] = None):
    """
    Return the task's argument parser.

    This function is kept separate from parsing for purposes of unit testing.
    """
    parser = argparse.ArgumentParser(parents=parents or [])
    parser.add_argument('-s', '--subject', required=True, help='Subject name')
    parser.add_argument('-u', '--user', required=False, default=None, help='Alyx username to register the session')
    parser.add_argument(
        '-p',
        '--projects',
        nargs='+',
        default=[],
        help="project name(s), something like 'psychedelics' or 'ibl_neuropixel_brainwide_01'; if specify "
        'multiple projects, use a space to separate them',
    )
    parser.add_argument(
        '-c',
        '--procedures',
        nargs='+',
        default=[],
        help="long description of what is occurring, something like 'Ephys recording with acute probe(s)'; "
        'be sure to use the double quote characters to encapsulate the description and a space to separate '
        'multiple procedures',
    )
    parser.add_argument('-w', '--weight', type=float, dest='subject_weight_grams', required=False, default=None)
    parser.add_argument('--no-interactive', dest='interactive', action='store_false')
    parser.add_argument('--append', dest='append', action='store_true')
    parser.add_argument('--stub', type=Path, help='Path to _ibl_experiment.description.yaml stub file.')
    parser.add_argument(
        '--log-level',
        default='INFO',
        help='verbosity of the console logger (default: INFO)',
        choices=['NOTSET', 'DEBUG', 'INFO', 'WARN', 'ERROR', 'CRITICAL'],
    )
    parser.add_argument('--wizard', dest='wizard', action='store_true', help=argparse.SUPPRESS)
    return parser


def _post_parse_arguments(**kwargs):
    """
    Post-process arguments after parsing.

    This function is used to force the interactive mode to True (as it is a call from a user) and to override the
    settings file value for the user. This function is split for the purpos of unit-testing.

    Parameters
    ----------
    kwargs : dict
        Keyword arguments passed to argparse.ArgumentParser.

    Returns
    -------
    kwargs : dict
        Keyword arguments passed to argparse.ArgumentParser.
    """
    # override the settings file value if the user is specified
    user = kwargs.pop('user')
    if user is not None:
        kwargs['iblrig_settings'] = {'ALYX_USER': user}
    return kwargs


def get_task_arguments(parents: Sequence[argparse.ArgumentParser] = None):
    """
    Parse input to run the tasks. All the variables are fed to the Session instance
    task.py -s subject_name -p projects_name -c procedures_name --no-interactive
    :param extra_args: list of dictionaries of additional argparse arguments to add to the parser
        For example, to add a new toto and titi arguments, use:
        get_task_arguments({'--toto', type=str, default='toto'}, {'--titi', action='store_true', default=False})
    :return:
    """
    parser = get_task_argument_parser(parents=parents)
    kwargs = vars(parser.parse_args())
    return _post_parse_arguments(**kwargs)


def _is_datetime(x: str) -> bool:
    """
    Check if a string is a date in the format YYYY-MM-DD.

    Parameters
    ----------
    x : str
        The string to check.

    Returns
    -------
    bool or None
        True if the string matches the date format, False otherwise, or None if there's an exception.
    """
    try:
        datetime.strptime(x, '%Y-%m-%d')
        return True
    except ValueError:
        return False


def get_session_path(path: str | Path) -> Path | None:
    """Returns the session path from any filepath if the date/number pattern is found."""
    if path is None:
        return
    if isinstance(path, str):
        path = Path(path)
    sess = None
    for i, p in enumerate(path.parts):
        if p.isdigit() and _is_datetime(path.parts[i - 1]):
            sess = Path().joinpath(*path.parts[: i + 1])

    return sess


def get_port_events(events: dict, name: str = '') -> list:
    out: list = []
    for key, value in events.items():
        if name in key:
            out.extend(value)
    return sorted(out)


def truncated_exponential(scale: float = 0.35, min_value: float = 0.2, max_value: float = 0.5) -> float:
    """
    Generate a truncated exponential random variable within a specified range.

    Parameters
    ----------
    scale : float, optional
        Scale of the exponential distribution (inverse of rate parameter). Defaults to 0.35.
    min_value : float, optional
        Minimum value for the truncated range. Defaults to 0.2.
    max_value : float, optional
        Maximum value for the truncated range. Defaults to 0.5.

    Returns
    -------
    float
        Truncated exponential random variable.

    Notes
    -----
    This function generates a random variable from an exponential distribution
    with the specified `scale`. It then checks if the generated value is within
    the specified range `[min_value, max_value]`. If it is within the range, it returns
    the generated value; otherwise, it recursively generates a new value until it falls
    within the specified range.

    The `scale` should typically be greater than or equal to the `min_value` to avoid
    potential issues with infinite recursion.
    """
    x = np.random.exponential(scale)
    if min_value <= x <= max_value:
        return x
    else:
        return truncated_exponential(scale, min_value, max_value)


def get_biased_probs(n: int, idx: int = -1, p_idx: float = 0.5) -> list[float]:
    """
    Calculate biased probabilities for all elements of an array such that the
    `i`th value has probability `p_i` for being drawn relative to the remaining
    values.

    See: https://github.com/int-brain-lab/iblrig/issues/74

    Parameters
    ----------
    n : int
        The length of the array, i.e., the number of probabilities to generate.
    idx : int, optional
        The index of the value that has the biased probability. Defaults to -1.
    p_idx : float, optional
        The probability of the `idx`-th value relative to the rest. Defaults to 0.5.

    Returns
    -------
    List[float]
        List of biased probabilities.

    Raises
    ------
    IndexError
        If `idx` is out of range
    ValueError
        If `p_idx` is 0.
    """
    if n == 1:
        return [1.0]
    if idx not in range(-n, n):
        raise IndexError('`idx` is out of range.')
    if p_idx == 0:
        raise ValueError('Probability must be larger than 0.')
    z = n - 1 + p_idx
    p = [1 / z] * n
    p[idx] *= p_idx
    return p


def draw_contrast(
    contrast_set: list[float],
    probability_type: Literal['skew_zero', 'biased', 'uniform'] = 'biased',
    idx: int = -1,
    idx_probability: float = 0.5,
) -> float:
    """
    Draw a contrast value from a given iterable based to the specified probability type.

    Parameters
    ----------
    contrast_set : list[float]
        The set of contrast values from which to draw.
    probability_type : Literal["skew_zero", "biased", "uniform"], optional
        The type of probability distribution to use.
        - "skew_zero" or "biased": Draws with a biased probability distribution based on idx and idx_probability,
        - "uniform": Draws with a uniform probability distribution.
        Defaults to "biased".
    idx : int, optional
        Index for probability manipulation (with "skew_zero" or "biased"), default: -1.
    idx_probability : float, optional
        Probability for the specified index (with "skew_zero" or "biased"), default: 0.5.

    Returns
    -------
    float
        The drawn contrast value.

    Raises
    ------
    ValueError
        If an unsupported `probability_type` is provided.
    """
    if probability_type in ['skew_zero', 'biased']:
        p = get_biased_probs(n=len(contrast_set), idx=idx, p_idx=idx_probability)
        return np.random.choice(contrast_set, p=p)
    elif probability_type == 'uniform':
        return np.random.choice(contrast_set)
    else:
        raise ValueError("Unsupported probability_type. Use 'skew_zero', 'biased', or 'uniform'.")


def online_std(new_sample: float, new_count: int, old_mean: float, old_std: float) -> tuple[float, float]:
    """
    Update the mean and standard deviation of a group of values after a sample update.

    Parameters
    ----------
    new_sample : float
        The new sample to be included.
    new_count : int
        The new count of samples (including new_sample).
    old_mean : float
        The previous mean (N - 1).
    old_std : float
        The previous standard deviation (N - 1).

    Returns
    -------
    tuple[float, float]
        Updated mean and standard deviation.
    """
    if new_count == 1:
        return new_sample, 0.0
    new_mean = (old_mean * (new_count - 1) + new_sample) / new_count
    new_std = np.sqrt((old_std**2 * (new_count - 1) + (new_sample - old_mean) * (new_sample - new_mean)) / new_count)
    return new_mean, new_std

1	"""
2	Provides collection of functionality used throughout the iblrig repository.
3
4	Assortment of functions, frequently used, but without a great deal of commonality. Functions can,
5	and should, be broken out into their own files and/or classes as the organizational needs of this
6	repo change over time.
7	"""
8
9	import argparse	2✔
10	import datetime	2✔
11	import logging	2✔
12	from collections.abc import Sequence	2✔
13	from pathlib import Path	2✔
14	from typing import Literal	2✔
15
16	import numpy as np	2✔
17
18	FLAG_FILE_NAMES = ['transfer_me.flag', 'create_me.flag', 'poop_count.flag', 'passive_data_for_ephys.flag']	2✔
19
20	log = logging.getLogger(__name__)	2✔
21
22
23	def get_task_argument_parser(parents: Sequence[argparse.ArgumentParser] = None):	2✔
24	"""
25	Return the task's argument parser.
26
27	This function is kept separate from parsing for purposes of unit testing.
28	"""
29	parser = argparse.ArgumentParser(parents=parents or [])	2✔
30	parser.add_argument('-s', '--subject', required=True, help='Subject name')	2✔
31	parser.add_argument('-u', '--user', required=False, default=None, help='Alyx username to register the session')	2✔
32	parser.add_argument(	2✔
33	'-p',
34	'--projects',
35	nargs='+',
36	default=[],
37	help="project name(s), something like 'psychedelics' or 'ibl_neuropixel_brainwide_01'; if specify "
38	'multiple projects, use a space to separate them',
39	)
40	parser.add_argument(	2✔
41	'-c',
42	'--procedures',
43	nargs='+',
44	default=[],
45	help="long description of what is occurring, something like 'Ephys recording with acute probe(s)'; "
46	'be sure to use the double quote characters to encapsulate the description and a space to separate '
47	'multiple procedures',
48	)
49	parser.add_argument('-w', '--weight', type=float, dest='subject_weight_grams', required=False, default=None)	2✔
50	parser.add_argument('--no-interactive', dest='interactive', action='store_false')	2✔
51	parser.add_argument('--append', dest='append', action='store_true')	2✔
52	parser.add_argument('--stub', type=Path, help='Path to _ibl_experiment.description.yaml stub file.')	2✔
53	parser.add_argument(	2✔
54	'--log-level',
55	default='INFO',
56	help='verbosity of the console logger (default: INFO)',
57	choices=['NOTSET', 'DEBUG', 'INFO', 'WARN', 'ERROR', 'CRITICAL'],
58	)
59	parser.add_argument('--wizard', dest='wizard', action='store_true', help=argparse.SUPPRESS)	2✔
60	return parser	2✔
61
62
63	def _post_parse_arguments(**kwargs):	2✔
64	"""
65	Post-process arguments after parsing.
66
67	This function is used to force the interactive mode to True (as it is a call from a user) and to override the
68	settings file value for the user. This function is split for the purpos of unit-testing.
69
70	Parameters
71	----------
72	kwargs : dict
73	Keyword arguments passed to argparse.ArgumentParser.
74
75	Returns
76	-------
77	kwargs : dict
78	Keyword arguments passed to argparse.ArgumentParser.
79	"""
80	# override the settings file value if the user is specified
81	user = kwargs.pop('user')	2✔
82	if user is not None:	2✔
83	kwargs['iblrig_settings'] = {'ALYX_USER': user}	2✔
84	return kwargs	2✔
85
86
87	def get_task_arguments(parents: Sequence[argparse.ArgumentParser] = None):	2✔
88	"""
89	Parse input to run the tasks. All the variables are fed to the Session instance
90	task.py -s subject_name -p projects_name -c procedures_name --no-interactive
91	:param extra_args: list of dictionaries of additional argparse arguments to add to the parser
92	For example, to add a new toto and titi arguments, use:
93	get_task_arguments({'--toto', type=str, default='toto'}, {'--titi', action='store_true', default=False})
94	:return:
95	"""
96	parser = get_task_argument_parser(parents=parents)	×
97	kwargs = vars(parser.parse_args())	×
UNCOV 98	return _post_parse_arguments(**kwargs)	×
99
100
101	def _is_datetime(x: str) -> bool:	2✔
102	"""
103	Check if a string is a date in the format YYYY-MM-DD.
104
105	Parameters
106	----------
107	x : str
108	The string to check.
109
110	Returns
111	-------
112	bool or None
113	True if the string matches the date format, False otherwise, or None if there's an exception.
114	"""
UNCOV 115	try:	×
UNCOV 116	datetime.strptime(x, '%Y-%m-%d')	×
UNCOV 117	return True	×
UNCOV 118	except ValueError:	×
UNCOV 119	return False	×
120
121
122	def get_session_path(path: str \| Path) -> Path \| None:	2✔
123	"""Returns the session path from any filepath if the date/number pattern is found."""
UNCOV 124	if path is None:	×
UNCOV 125	return	×
UNCOV 126	if isinstance(path, str):	×
UNCOV 127	path = Path(path)	×
UNCOV 128	sess = None	×
UNCOV 129	for i, p in enumerate(path.parts):	×
UNCOV 130	if p.isdigit() and _is_datetime(path.parts[i - 1]):	×
UNCOV 131	sess = Path().joinpath(*path.parts[: i + 1])	×
132
UNCOV 133	return sess	×
134
135
136	def get_port_events(events: dict, name: str = '') -> list:	2✔
UNCOV 137	out: list = []	×
UNCOV 138	for key, value in events.items():	×
UNCOV 139	if name in key:	×
UNCOV 140	out.extend(value)	×
UNCOV 141	return sorted(out)	×
142
143
144	def truncated_exponential(scale: float = 0.35, min_value: float = 0.2, max_value: float = 0.5) -> float:	2✔
145	"""
146	Generate a truncated exponential random variable within a specified range.
147
148	Parameters
149	----------
150	scale : float, optional
151	Scale of the exponential distribution (inverse of rate parameter). Defaults to 0.35.
152	min_value : float, optional
153	Minimum value for the truncated range. Defaults to 0.2.
154	max_value : float, optional
155	Maximum value for the truncated range. Defaults to 0.5.
156
157	Returns
158	-------
159	float
160	Truncated exponential random variable.
161
162	Notes
163	-----
164	This function generates a random variable from an exponential distribution
165	with the specified `scale`. It then checks if the generated value is within
166	the specified range `[min_value, max_value]`. If it is within the range, it returns
167	the generated value; otherwise, it recursively generates a new value until it falls
168	within the specified range.
169
170	The `scale` should typically be greater than or equal to the `min_value` to avoid
171	potential issues with infinite recursion.
172	"""
173	x = np.random.exponential(scale)	2✔
174	if min_value <= x <= max_value:	2✔
175	return x	2✔
176	else:
177	return truncated_exponential(scale, min_value, max_value)	2✔
178
179
180	def get_biased_probs(n: int, idx: int = -1, p_idx: float = 0.5) -> list[float]:	2✔
181	"""
182	Calculate biased probabilities for all elements of an array such that the
183	`i`th value has probability `p_i` for being drawn relative to the remaining
184	values.
185
186	See: https://github.com/int-brain-lab/iblrig/issues/74
187
188	Parameters
189	----------
190	n : int
191	The length of the array, i.e., the number of probabilities to generate.
192	idx : int, optional
193	The index of the value that has the biased probability. Defaults to -1.
194	p_idx : float, optional
195	The probability of the `idx`-th value relative to the rest. Defaults to 0.5.
196
197	Returns
198	-------
199	List[float]
200	List of biased probabilities.
201
202	Raises
203	------
204	IndexError
205	If `idx` is out of range
206	ValueError
207	If `p_idx` is 0.
208	"""
209	if n == 1:	2✔
UNCOV 210	return [1.0]	×
211	if idx not in range(-n, n):	2✔
212	raise IndexError('`idx` is out of range.')	2✔
213	if p_idx == 0:	2✔
UNCOV 214	raise ValueError('Probability must be larger than 0.')	×
215	z = n - 1 + p_idx	2✔
216	p = [1 / z] * n	2✔
217	p[idx] *= p_idx	2✔
218	return p	2✔
219
220
221	def draw_contrast(	2✔
222	contrast_set: list[float],
223	probability_type: Literal['skew_zero', 'biased', 'uniform'] = 'biased',
224	idx: int = -1,
225	idx_probability: float = 0.5,
226	) -> float:
227	"""
228	Draw a contrast value from a given iterable based to the specified probability type.
229
230	Parameters
231	----------
232	contrast_set : list[float]
233	The set of contrast values from which to draw.
234	probability_type : Literal["skew_zero", "biased", "uniform"], optional
235	The type of probability distribution to use.
236	- "skew_zero" or "biased": Draws with a biased probability distribution based on idx and idx_probability,
237	- "uniform": Draws with a uniform probability distribution.
238	Defaults to "biased".
239	idx : int, optional
240	Index for probability manipulation (with "skew_zero" or "biased"), default: -1.
241	idx_probability : float, optional
242	Probability for the specified index (with "skew_zero" or "biased"), default: 0.5.
243
244	Returns
245	-------
246	float
247	The drawn contrast value.
248
249	Raises
250	------
251	ValueError
252	If an unsupported `probability_type` is provided.
253	"""
254	if probability_type in ['skew_zero', 'biased']:	2✔
255	p = get_biased_probs(n=len(contrast_set), idx=idx, p_idx=idx_probability)	2✔
256	return np.random.choice(contrast_set, p=p)	2✔
257	elif probability_type == 'uniform':	2✔
258	return np.random.choice(contrast_set)	2✔
259	else:
260	raise ValueError("Unsupported probability_type. Use 'skew_zero', 'biased', or 'uniform'.")	2✔
261
262
263	def online_std(new_sample: float, new_count: int, old_mean: float, old_std: float) -> tuple[float, float]:	2✔
264	"""
265	Update the mean and standard deviation of a group of values after a sample update.
266
267	Parameters
268	----------
269	new_sample : float
270	The new sample to be included.
271	new_count : int
272	The new count of samples (including new_sample).
273	old_mean : float
274	The previous mean (N - 1).
275	old_std : float
276	The previous standard deviation (N - 1).
277
278	Returns
279	-------
280	tuple[float, float]
281	Updated mean and standard deviation.
282	"""
283	if new_count == 1:	2✔
284	return new_sample, 0.0	2✔
285	new_mean = (old_mean * (new_count - 1) + new_sample) / new_count	2✔
286	new_std = np.sqrt((old_std*2 (new_count - 1) + (new_sample - old_mean) * (new_sample - new_mean)) / new_count)	2✔
287	return new_mean, new_std	2✔

int-brain-lab / iblrig / 14196118657

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