9031936551

Committed 10 May 2024 12:05PM UTC coverage: 48.538% (+1.7%) from 46.79%

Build # 9031936551

Build Type

Pull #643

github

Specific Base 53c3e3

Committed by

web-flow

Commit Message

Merge 3c8214f78 into ec2d8e4fe

Pull Request Pull Request #643: 8.19.0

Run Details

377 of 1073 new or added lines in 38 files covered. (35.14%)

977 existing lines in 19 files now uncovered.

3253 of 6702 relevant lines covered (48.54%)

0.97 hits per line

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

69.51

/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 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=None):
    """
    This function returns the task argument parser with extra optional parameters if provided
    This function is kept separate from parsing for unit tests purposes.
    """
    parser = argparse.ArgumentParser(parents=parents or [])
    parser.add_argument('-s', '--subject', required=True, help='--subject ZFM-05725')
    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')
    return parser


def _post_parse_arguments(**kwargs):
    """
    This is called to post-process the arguments after parsing. It 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 unit-test purposes.
    :param kwargs:
    :return:
    """
    # if the user is specified, then override the settings file value
    user = kwargs.pop('user')
    if user is not None:
        kwargs['iblrig_settings'] = {'ALYX_USER': user}
    return kwargs


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

int-brain-lab / iblrig / 9031936551

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