10568073180

Committed 26 Aug 2024 10:13PM UTC coverage: 47.538% (+0.7%) from 46.79%

Build # 10568073180

Build Type

Pull #711

github

Specific Base eeff82

Committed by

web-flow

Commit Message

Merge 599c9edfb into ad41db25f

Pull Request Pull Request #711: 8.23.2

Run Details

121 of 135 new or added lines in 8 files covered. (89.63%)

1025 existing lines in 22 files now uncovered.

4084 of 8591 relevant lines covered (47.54%)

0.95 hits per line

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

69.88

/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 k in events:
        if name in k:
            out.extend(events[k])
    out = sorted(out)

    return 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 k in events:	×
UNCOV 139	if name in k:	×
UNCOV 140	out.extend(events[k])	×
UNCOV 141	out = sorted(out)	×
142
143	return out	×
144
145
146	def truncated_exponential(scale: float = 0.35, min_value: float = 0.2, max_value: float = 0.5) -> float:	2✔
147	"""
148	Generate a truncated exponential random variable within a specified range.
149
150	Parameters
151	----------
152	scale : float, optional
153	Scale of the exponential distribution (inverse of rate parameter). Defaults to 0.35.
154	min_value : float, optional
155	Minimum value for the truncated range. Defaults to 0.2.
156	max_value : float, optional
157	Maximum value for the truncated range. Defaults to 0.5.
158
159	Returns
160	-------
161	float
162	Truncated exponential random variable.
163
164	Notes
165	-----
166	This function generates a random variable from an exponential distribution
167	with the specified `scale`. It then checks if the generated value is within
168	the specified range `[min_value, max_value]`. If it is within the range, it returns
169	the generated value; otherwise, it recursively generates a new value until it falls
170	within the specified range.
171
172	The `scale` should typically be greater than or equal to the `min_value` to avoid
173	potential issues with infinite recursion.
174	"""
175	x = np.random.exponential(scale)	2✔
176	if min_value <= x <= max_value:	2✔
177	return x	2✔
178	else:
179	return truncated_exponential(scale, min_value, max_value)	2✔
180
181
182	def get_biased_probs(n: int, idx: int = -1, p_idx: float = 0.5) -> list[float]:	2✔
183	"""
184	Calculate biased probabilities for all elements of an array such that the
185	`i`th value has probability `p_i` for being drawn relative to the remaining
186	values.
187
188	See: https://github.com/int-brain-lab/iblrig/issues/74
189
190	Parameters
191	----------
192	n : int
193	The length of the array, i.e., the number of probabilities to generate.
194	idx : int, optional
195	The index of the value that has the biased probability. Defaults to -1.
196	p_idx : float, optional
197	The probability of the `idx`-th value relative to the rest. Defaults to 0.5.
198
199	Returns
200	-------
201	List[float]
202	List of biased probabilities.
203
204	Raises
205	------
206	IndexError
207	If `idx` is out of range
208	ValueError
209	If `p_idx` is 0.
210	"""
211	if n == 1:	2✔
UNCOV 212	return [1.0]	×
213	if idx not in range(-n, n):	2✔
214	raise IndexError('`idx` is out of range.')	2✔
215	if p_idx == 0:	2✔
UNCOV 216	raise ValueError('Probability must be larger than 0.')	×
217	z = n - 1 + p_idx	2✔
218	p = [1 / z] * n	2✔
219	p[idx] *= p_idx	2✔
220	return p	2✔
221
222
223	def draw_contrast(	2✔
224	contrast_set: list[float],
225	probability_type: Literal['skew_zero', 'biased', 'uniform'] = 'biased',
226	idx: int = -1,
227	idx_probability: float = 0.5,
228	) -> float:
229	"""
230	Draw a contrast value from a given iterable based to the specified probability type.
231
232	Parameters
233	----------
234	contrast_set : list[float]
235	The set of contrast values from which to draw.
236	probability_type : Literal["skew_zero", "biased", "uniform"], optional
237	The type of probability distribution to use.
238	- "skew_zero" or "biased": Draws with a biased probability distribution based on idx and idx_probability,
239	- "uniform": Draws with a uniform probability distribution.
240	Defaults to "biased".
241	idx : int, optional
242	Index for probability manipulation (with "skew_zero" or "biased"), default: -1.
243	idx_probability : float, optional
244	Probability for the specified index (with "skew_zero" or "biased"), default: 0.5.
245
246	Returns
247	-------
248	float
249	The drawn contrast value.
250
251	Raises
252	------
253	ValueError
254	If an unsupported `probability_type` is provided.
255	"""
256	if probability_type in ['skew_zero', 'biased']:	2✔
257	p = get_biased_probs(n=len(contrast_set), idx=idx, p_idx=idx_probability)	2✔
258	return np.random.choice(contrast_set, p=p)	2✔
259	elif probability_type == 'uniform':	2✔
260	return np.random.choice(contrast_set)	2✔
261	else:
262	raise ValueError("Unsupported probability_type. Use 'skew_zero', 'biased', or 'uniform'.")	2✔
263
264
265	def online_std(new_sample: float, new_count: int, old_mean: float, old_std: float) -> tuple[float, float]:	2✔
266	"""
267	Update the mean and standard deviation of a group of values after a sample update.
268
269	Parameters
270	----------
271	new_sample : float
272	The new sample to be included.
273	new_count : int
274	The new count of samples (including new_sample).
275	old_mean : float
276	The previous mean (N - 1).
277	old_std : float
278	The previous standard deviation (N - 1).
279
280	Returns
281	-------
282	tuple[float, float]
283	Updated mean and standard deviation.
284	"""
285	if new_count == 1:	2✔
286	return new_sample, 0.0	2✔
287	new_mean = (old_mean * (new_count - 1) + new_sample) / new_count	2✔
288	new_std = np.sqrt((old_std*2 (new_count - 1) + (new_sample - old_mean) * (new_sample - new_mean)) / new_count)	2✔
289	return new_mean, new_std	2✔

int-brain-lab / iblrig / 10568073180

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