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

84.03

/iblrig/tools.py

import asyncio
import logging
import os
import platform
import re
import shutil
import socket
import subprocess
from collections.abc import Callable
from dataclasses import dataclass
from datetime import date
from functools import cache
from pathlib import Path
from typing import Any, TypeVar

from iblrig import __version__ as iblrig_version
from iblrig.constants import BONSAI_EXE, IS_GIT
from iblrig.path_helper import create_bonsai_layout_from_template, load_pydantic_yaml
from iblrig.pydantic_definitions import HardwareSettings, RigSettings
from iblrig.version_management import get_branch, get_commit_hash, is_dirty
from iblutil.util import get_mac

log = logging.getLogger(__name__)


def ask_user(prompt: str, default: bool = False) -> bool:
    """
    Prompt the user for a yes/no response.

    This function displays a prompt to the user and expects a yes or no response.
    The response is not case-sensitive. If the user presses Enter without
    typing anything, the function interprets it as the default response.

    Parameters
    ----------
    prompt : str
        The prompt message to display to the user.
    default : bool, optional
        The default response when the user presses Enter without typing
        anything. If True, the default response is 'yes' (Y/y or Enter).
        If False, the default response is 'no' (N/n or Enter).

    Returns
    -------
    bool
        True if the user responds with 'yes'
        False if the user responds with 'no'
    """
    while True:
        user_input = input(f'{prompt} [Y/n] ' if default else f'{prompt} [y/N] ').strip().lower()
        if not user_input:
            return default
        elif user_input in ['y', 'yes']:
            return True
        elif user_input in ['n', 'no']:
            return False


def get_anydesk_id(format_id: bool = True, silent: bool = False) -> str | None:
    """
    Retrieve the AnyDesk ID of the current machine.

    Parameters
    ----------
    format_id : bool, optional
        If True (default), format the ID in blocks separated by spaces.
        If False, return the ID as one continuous block.
    silent : bool, optional
        If True, suppresses exceptions and logs them instead.
        If False (default), raises exceptions.

    Returns
    -------
    str or None
        The AnyDesk ID as a formatted string (e.g., '123 456 789') if successful,
        or None on failure.

    Raises
    ------
    FileNotFoundError
        If the AnyDesk executable is not found.
    subprocess.CalledProcessError
        If an error occurs while executing the AnyDesk command.
    StopIteration
        If the subprocess output is empty.
    UnicodeDecodeError
        If there is an issue decoding the subprocess output.

    Notes
    -----
    The function attempts to find the AnyDesk executable and retrieve the ID using the command line.
    On success, the AnyDesk ID is returned as a formatted string. If silent is True, exceptions are logged,
    and None is returned on failure. If silent is False, exceptions are raised on failure.
    """
    anydesk_id = None
    try:
        if cmd := shutil.which('anydesk'):
            pass
        elif os.name == 'nt':
            cmd = str(Path(os.environ['PROGRAMFILES(X86)'], 'AnyDesk', 'anydesk.exe'))
        if cmd is None or not Path(cmd).exists():
            raise FileNotFoundError('AnyDesk executable not found')

        proc = subprocess.Popen([cmd, '--get-id'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        if proc.stdout and re.match(r'^\d{10}$', id_string := next(proc.stdout).decode()):
            anydesk_id = f'{int(id_string):,}'.replace(',', ' ' if format_id else '')
    except (FileNotFoundError, subprocess.CalledProcessError, StopIteration, UnicodeDecodeError) as e:
        if silent:
            log.debug(e, exc_info=True)
        else:
            raise e
    return anydesk_id


def internet_available(host: str = '8.8.8.8', port: int = 53, timeout: int = 3, force_update: bool = False) -> bool:
    """
    Check if the internet connection is available.

    This function checks if an internet connection is available by attempting to
    establish a connection to a specified host and port. It will use a cached
    result if the latter is available and `force_update` is set to False.

    Parameters
    ----------
    host : str, optional
        The IP address or domain name of the host to check the connection to.
        Default is "8.8.8.8" (Google's DNS server).
    port : int, optional
        The port to use for the connection check. Default is 53 (DNS port).
    timeout : int, optional
        The maximum time (in seconds) to wait for the connection attempt.
        Default is 3 seconds.
    force_update : bool, optional
        If True, force an update and recheck the internet connection even if
        the result is cached. Default is False.

    Returns
    -------
    bool
        True if an internet connection is available, False otherwise.
    """
    if not force_update and hasattr(internet_available, 'return_value'):
        return internet_available.return_value
    try:
        socket.setdefaulttimeout(timeout)
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.connect((host, port))
        internet_available.return_value = True
    except OSError:
        internet_available.return_value = False
    return internet_available.return_value


def alyx_reachable() -> bool:
    """
    Check if Alyx can be connected to.

    Returns
    -------
    bool
        True if Alyx can be connected to, False otherwise.
    """
    settings: RigSettings = load_pydantic_yaml(RigSettings)
    if settings.ALYX_URL is not None:
        return internet_available(host=settings.ALYX_URL.host, port=443, timeout=1, force_update=True)
    return False


def _build_bonsai_cmd(
    workflow_file: str | Path,
    parameters: dict[str, Any] | None = None,
    start: bool = True,
    debug: bool = False,
    bootstrap: bool = True,
    editor: bool = True,
    bonsai_executable: Path = BONSAI_EXE,
) -> subprocess.Popen[bytes] | subprocess.Popen[str | bytes | Any] | subprocess.CompletedProcess:
    """
    Execute a Bonsai workflow within a subprocess call.

    Parameters
    ----------
    workflow_file : str | Path
        Path to the Bonsai workflow file.
    parameters : dict[str, str], optional
        Parameters to be passed to Bonsai workflow.
    start : bool, optional
        Start execution of the workflow within Bonsai (default is True).
    debug : bool, optional
        Enable debugging mode if True (default is False).
        Only applies if editor is True.
    bootstrap : bool, optional
        Enable Bonsai bootstrapping if True (default is True).
    editor : bool, optional
        Enable Bonsai editor if True (default is True).
    bonsai_executable : Path
        Path to bonsai executable. Defaults to iblrig.constants.BONSAI_EXE.

    Returns
    -------
    list of str
        The Bonsai command to pass to subprocess.

    Raises
    ------
    FileNotFoundError
        If the Bonsai executable does not exist.
        If the specified workflow file does not exist.
    """
    if not bonsai_executable.exists():
        raise FileNotFoundError(bonsai_executable)
    workflow_file = Path(workflow_file)
    if not workflow_file.exists():
        raise FileNotFoundError(workflow_file)
    create_bonsai_layout_from_template(workflow_file)

    cmd = [str(bonsai_executable), str(workflow_file)]
    if start:
        cmd.append('--start' if debug else '--start-no-debug')
    if not editor:
        cmd.append('--no-editor')
    if not bootstrap:
        cmd.append('--no-boot')
    if parameters is not None:
        for key, value in parameters.items():
            cmd.append(f'-p:{key}={str(value)}')
    return cmd


def call_bonsai(
    workflow_file: str | Path,
    parameters: dict[str, Any] | None = None,
    start: bool = True,
    debug: bool = False,
    bootstrap: bool = True,
    editor: bool = True,
    wait: bool = True,
    check: bool = False,
    bonsai_executable: Path = BONSAI_EXE,
) -> subprocess.Popen[bytes] | subprocess.Popen[str | bytes | Any] | subprocess.CompletedProcess:
    """
    Execute a Bonsai workflow within a subprocess call.

    Parameters
    ----------
    workflow_file : str | Path
        Path to the Bonsai workflow file.
    parameters : dict[str, str], optional
        Parameters to be passed to Bonsai workflow.
    start : bool, optional
        Start execution of the workflow within Bonsai (default is True).
    debug : bool, optional
        Enable debugging mode if True (default is False).
        Only applies if editor is True.
    bootstrap : bool, optional
        Enable Bonsai bootstrapping if True (default is True).
    editor : bool, optional
        Enable Bonsai editor if True (default is True).
    wait : bool, optional
        Wait for Bonsai process to finish (default is True).
    check : bool, optional
        Raise CalledProcessError if Bonsai process exits with non-zero exit code (default is False).
        Only applies if wait is True.
    bonsai_executable : Path
        Path to bonsai executable. Defaults to iblrig.constants.BONSAI_EXE.

    Returns
    -------
    Popen[bytes] | Popen[str | bytes | Any] | CompletedProcess
        Pointer to the Bonsai subprocess if wait is False, otherwise subprocess.CompletedProcess.

    Raises
    ------
    FileNotFoundError
        If the Bonsai executable does not exist.
        If the specified workflow file does not exist.

    """
    cmd = _build_bonsai_cmd(workflow_file, parameters, start, debug, bootstrap, editor, bonsai_executable)
    cwd = Path(workflow_file).parent
    log.info(f'Starting Bonsai workflow `{workflow_file.name}`')
    log.debug(' '.join(map(str, cmd)))
    if wait:
        return subprocess.run(args=cmd, cwd=cwd, check=check)
    else:
        return subprocess.Popen(args=cmd, cwd=cwd)


async def call_bonsai_async(
    workflow_file: str | Path,
    parameters: dict[str, Any] | None = None,
    start: bool = True,
    debug: bool = False,
    bootstrap: bool = True,
    editor: bool = True,
) -> asyncio.subprocess.Process:
    """
    Asynchronously execute a Bonsai workflow within a subprocess call.

    Parameters
    ----------
    workflow_file : str | Path
        Path to the Bonsai workflow file.
    parameters : dict[str, str], optional
        Parameters to be passed to Bonsai workflow.
    start : bool, optional
        Start execution of the workflow within Bonsai (default is True).
    debug : bool, optional
        Enable debugging mode if True (default is False).
        Only applies if editor is True.
    bootstrap : bool, optional
        Enable Bonsai bootstrapping if True (default is True).
    editor : bool, optional
        Enable Bonsai editor if True (default is True).

    Returns
    -------
    asyncio.subprocess.Process
        Pointer to the Bonsai subprocess if wait is False, otherwise subprocess.CompletedProcess.

    Raises
    ------
    FileNotFoundError
        If the Bonsai executable does not exist.
        If the specified workflow file does not exist.

    """
    program, *cmd = _build_bonsai_cmd(workflow_file, parameters, start, debug, bootstrap, editor)
    log.info(f'Starting Bonsai workflow `{workflow_file.name}`')
    log.debug(' '.join(map(str, cmd)))
    working_dir = Path(workflow_file).parent
    return await asyncio.create_subprocess_exec(
        program, *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=working_dir
    )


T = TypeVar('T', bound=object)


def get_inheritors(cls: T) -> set[T]:
    """Obtain a set of all direct inheritors of a class."""
    subclasses = set(cls.__subclasses__())
    for child in subclasses:
        subclasses = subclasses.union(get_inheritors(child))
    return subclasses


@dataclass
class ANSI:
    """ANSI Codes for formatting text on the CLI."""

    WHITE = '\033[37m'
    PURPLE = '\033[95m'
    CYAN = '\033[96m'
    DARKCYAN = '\033[36m'
    BLUE = '\033[94m'
    GREEN = '\033[92m'
    YELLOW = '\033[93m'
    RED = '\033[91m'
    BOLD = '\033[1m'
    DIM = '\033[2m'
    UNDERLINE = '\033[4m'
    END = '\033[0m'


cached_check_output = cache(subprocess.check_output)


def get_lab_location_dict(hardware_settings: HardwareSettings, iblrig_settings: RigSettings) -> dict[str, Any]:
    lab_location = dict()
    lab_location['rig_name'] = hardware_settings.RIG_NAME
    lab_location['iblrig_version'] = iblrig_version
    lab_location['last_seen'] = date.today().isoformat()

    machine = dict()
    machine['platform'] = platform.platform()
    machine['hostname'] = socket.gethostname()
    machine['fqdn'] = socket.getfqdn()
    machine['ip'] = socket.gethostbyname(machine['hostname'])
    machine['mac'] = get_mac()
    machine['anydesk'] = get_anydesk_id(format_id=False, silent=True)
    lab_location['machine'] = machine

    git = dict()
    git['is_git'] = IS_GIT
    git['branch'] = get_branch()
    git['commit_id'] = get_commit_hash()
    git['is_dirty'] = is_dirty()
    lab_location['git'] = git

    # TODO: add hardware/firmware versions of bpod, soundcard, rotary encoder, frame2ttl, ambient module, etc
    # TODO: add validation errors/warnings
    return lab_location


def get_number(
    prompt: str = 'Enter number: ',
    numeric_type: type(int) | type(float) = int,
    validation: Callable[[int | float], bool] = lambda _: True,
) -> int | float:
    """
    Prompt the user for a number input of a specified numeric type and validate it.

    Parameters
    ----------
    prompt : str, optional
        The message displayed to the user when asking for input.
        Defaults to 'Enter number: '.
    numeric_type : type, optional
        The type of the number to be returned. Can be either `int` or `float`.
        Defaults to `int`.
    validation : callable, optional
        A function that takes a number as input and returns a boolean
        indicating whether the number is valid. Defaults to a function
        that always returns True.

    Returns
    -------
    int or float
        The validated number input by the user, converted to the specified type.

    Notes
    -----
    The function will continue to prompt the user until a valid number
    is entered that passes the validation function.
    """
    value = None
    while not isinstance(value, numeric_type) or validation(value) is False:
        try:
            value = numeric_type(input(prompt).strip())
        except ValueError:
            value = None
    return value

1	import asyncio	2✔
2	import logging	2✔
3	import os	2✔
4	import platform	2✔
5	import re	2✔
6	import shutil	2✔
7	import socket	2✔
8	import subprocess	2✔
9	from collections.abc import Callable	2✔
10	from dataclasses import dataclass	2✔
11	from datetime import date	2✔
12	from functools import cache	2✔
13	from pathlib import Path	2✔
14	from typing import Any, TypeVar	2✔
15
16	from iblrig import __version__ as iblrig_version	2✔
17	from iblrig.constants import BONSAI_EXE, IS_GIT	2✔
18	from iblrig.path_helper import create_bonsai_layout_from_template, load_pydantic_yaml	2✔
19	from iblrig.pydantic_definitions import HardwareSettings, RigSettings	2✔
20	from iblrig.version_management import get_branch, get_commit_hash, is_dirty	2✔
21	from iblutil.util import get_mac	2✔
22
23	log = logging.getLogger(__name__)	2✔
24
25
26	def ask_user(prompt: str, default: bool = False) -> bool:	2✔
27	"""
28	Prompt the user for a yes/no response.
29
30	This function displays a prompt to the user and expects a yes or no response.
31	The response is not case-sensitive. If the user presses Enter without
32	typing anything, the function interprets it as the default response.
33
34	Parameters
35	----------
36	prompt : str
37	The prompt message to display to the user.
38	default : bool, optional
39	The default response when the user presses Enter without typing
40	anything. If True, the default response is 'yes' (Y/y or Enter).
41	If False, the default response is 'no' (N/n or Enter).
42
43	Returns
44	-------
45	bool
46	True if the user responds with 'yes'
47	False if the user responds with 'no'
48	"""
49	while True:	2✔
50	user_input = input(f'{prompt} [Y/n] ' if default else f'{prompt} [y/N] ').strip().lower()	2✔
51	if not user_input:	2✔
52	return default	2✔
53	elif user_input in ['y', 'yes']:	2✔
54	return True	2✔
55	elif user_input in ['n', 'no']:	2✔
56	return False	2✔
57
58
59	def get_anydesk_id(format_id: bool = True, silent: bool = False) -> str \| None:	2✔
60	"""
61	Retrieve the AnyDesk ID of the current machine.
62
63	Parameters
64	----------
65	format_id : bool, optional
66	If True (default), format the ID in blocks separated by spaces.
67	If False, return the ID as one continuous block.
68	silent : bool, optional
69	If True, suppresses exceptions and logs them instead.
70	If False (default), raises exceptions.
71
72	Returns
73	-------
74	str or None
75	The AnyDesk ID as a formatted string (e.g., '123 456 789') if successful,
76	or None on failure.
77
78	Raises
79	------
80	FileNotFoundError
81	If the AnyDesk executable is not found.
82	subprocess.CalledProcessError
83	If an error occurs while executing the AnyDesk command.
84	StopIteration
85	If the subprocess output is empty.
86	UnicodeDecodeError
87	If there is an issue decoding the subprocess output.
88
89	Notes
90	-----
91	The function attempts to find the AnyDesk executable and retrieve the ID using the command line.
92	On success, the AnyDesk ID is returned as a formatted string. If silent is True, exceptions are logged,
93	and None is returned on failure. If silent is False, exceptions are raised on failure.
94	"""
95	anydesk_id = None	2✔
96	try:	2✔
97	if cmd := shutil.which('anydesk'):	2✔
UNCOV 98	pass	×
99	elif os.name == 'nt':	2✔
100	cmd = str(Path(os.environ['PROGRAMFILES(X86)'], 'AnyDesk', 'anydesk.exe'))	1✔
101	if cmd is None or not Path(cmd).exists():	2✔
102	raise FileNotFoundError('AnyDesk executable not found')	2✔
103
UNCOV 104	proc = subprocess.Popen([cmd, '--get-id'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)	×
UNCOV 105	if proc.stdout and re.match(r'^\d{10}$', id_string := next(proc.stdout).decode()):	×
UNCOV 106	anydesk_id = f'{int(id_string):,}'.replace(',', ' ' if format_id else '')	×
107	except (FileNotFoundError, subprocess.CalledProcessError, StopIteration, UnicodeDecodeError) as e:	2✔
108	if silent:	2✔
109	log.debug(e, exc_info=True)	2✔
110	else:
UNCOV 111	raise e	×
112	return anydesk_id	2✔
113
114
115	def internet_available(host: str = '8.8.8.8', port: int = 53, timeout: int = 3, force_update: bool = False) -> bool:	2✔
116	"""
117	Check if the internet connection is available.
118
119	This function checks if an internet connection is available by attempting to
120	establish a connection to a specified host and port. It will use a cached
121	result if the latter is available and `force_update` is set to False.
122
123	Parameters
124	----------
125	host : str, optional
126	The IP address or domain name of the host to check the connection to.
127	Default is "8.8.8.8" (Google's DNS server).
128	port : int, optional
129	The port to use for the connection check. Default is 53 (DNS port).
130	timeout : int, optional
131	The maximum time (in seconds) to wait for the connection attempt.
132	Default is 3 seconds.
133	force_update : bool, optional
134	If True, force an update and recheck the internet connection even if
135	the result is cached. Default is False.
136
137	Returns
138	-------
139	bool
140	True if an internet connection is available, False otherwise.
141	"""
142	if not force_update and hasattr(internet_available, 'return_value'):	2✔
143	return internet_available.return_value	2✔
144	try:	2✔
145	socket.setdefaulttimeout(timeout)	2✔
146	with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:	2✔
147	s.connect((host, port))	2✔
148	internet_available.return_value = True	2✔
149	except OSError:	2✔
150	internet_available.return_value = False	2✔
151	return internet_available.return_value	2✔
152
153
154	def alyx_reachable() -> bool:	2✔
155	"""
156	Check if Alyx can be connected to.
157
158	Returns
159	-------
160	bool
161	True if Alyx can be connected to, False otherwise.
162	"""
UNCOV 163	settings: RigSettings = load_pydantic_yaml(RigSettings)	×
UNCOV 164	if settings.ALYX_URL is not None:	×
UNCOV 165	return internet_available(host=settings.ALYX_URL.host, port=443, timeout=1, force_update=True)	×
UNCOV 166	return False	×
167
168
169	def _build_bonsai_cmd(	2✔
170	workflow_file: str \| Path,
171	parameters: dict[str, Any] \| None = None,
172	start: bool = True,
173	debug: bool = False,
174	bootstrap: bool = True,
175	editor: bool = True,
176	bonsai_executable: Path = BONSAI_EXE,
177	) -> subprocess.Popen[bytes] \| subprocess.Popen[str \| bytes \| Any] \| subprocess.CompletedProcess:
178	"""
179	Execute a Bonsai workflow within a subprocess call.
180
181	Parameters
182	----------
183	workflow_file : str \| Path
184	Path to the Bonsai workflow file.
185	parameters : dict[str, str], optional
186	Parameters to be passed to Bonsai workflow.
187	start : bool, optional
188	Start execution of the workflow within Bonsai (default is True).
189	debug : bool, optional
190	Enable debugging mode if True (default is False).
191	Only applies if editor is True.
192	bootstrap : bool, optional
193	Enable Bonsai bootstrapping if True (default is True).
194	editor : bool, optional
195	Enable Bonsai editor if True (default is True).
196	bonsai_executable : Path
197	Path to bonsai executable. Defaults to iblrig.constants.BONSAI_EXE.
198
199	Returns
200	-------
201	list of str
202	The Bonsai command to pass to subprocess.
203
204	Raises
205	------
206	FileNotFoundError
207	If the Bonsai executable does not exist.
208	If the specified workflow file does not exist.
209	"""
210	if not bonsai_executable.exists():	2✔
211	raise FileNotFoundError(bonsai_executable)	2✔
212	workflow_file = Path(workflow_file)	2✔
213	if not workflow_file.exists():	2✔
UNCOV 214	raise FileNotFoundError(workflow_file)	×
215	create_bonsai_layout_from_template(workflow_file)	2✔
216
217	cmd = [str(bonsai_executable), str(workflow_file)]	2✔
218	if start:	2✔
219	cmd.append('--start' if debug else '--start-no-debug')	2✔
220	if not editor:	2✔
221	cmd.append('--no-editor')	2✔
222	if not bootstrap:	2✔
223	cmd.append('--no-boot')	2✔
224	if parameters is not None:	2✔
225	for key, value in parameters.items():	2✔
226	cmd.append(f'-p:{key}={str(value)}')	2✔
227	return cmd	2✔
228
229
230	def call_bonsai(	2✔
231	workflow_file: str \| Path,
232	parameters: dict[str, Any] \| None = None,
233	start: bool = True,
234	debug: bool = False,
235	bootstrap: bool = True,
236	editor: bool = True,
237	wait: bool = True,
238	check: bool = False,
239	bonsai_executable: Path = BONSAI_EXE,
240	) -> subprocess.Popen[bytes] \| subprocess.Popen[str \| bytes \| Any] \| subprocess.CompletedProcess:
241	"""
242	Execute a Bonsai workflow within a subprocess call.
243
244	Parameters
245	----------
246	workflow_file : str \| Path
247	Path to the Bonsai workflow file.
248	parameters : dict[str, str], optional
249	Parameters to be passed to Bonsai workflow.
250	start : bool, optional
251	Start execution of the workflow within Bonsai (default is True).
252	debug : bool, optional
253	Enable debugging mode if True (default is False).
254	Only applies if editor is True.
255	bootstrap : bool, optional
256	Enable Bonsai bootstrapping if True (default is True).
257	editor : bool, optional
258	Enable Bonsai editor if True (default is True).
259	wait : bool, optional
260	Wait for Bonsai process to finish (default is True).
261	check : bool, optional
262	Raise CalledProcessError if Bonsai process exits with non-zero exit code (default is False).
263	Only applies if wait is True.
264	bonsai_executable : Path
265	Path to bonsai executable. Defaults to iblrig.constants.BONSAI_EXE.
266
267	Returns
268	-------
269	Popen[bytes] \| Popen[str \| bytes \| Any] \| CompletedProcess
270	Pointer to the Bonsai subprocess if wait is False, otherwise subprocess.CompletedProcess.
271
272	Raises
273	------
274	FileNotFoundError
275	If the Bonsai executable does not exist.
276	If the specified workflow file does not exist.
277
278	"""
279	cmd = _build_bonsai_cmd(workflow_file, parameters, start, debug, bootstrap, editor, bonsai_executable)	2✔
280	cwd = Path(workflow_file).parent	2✔
281	log.info(f'Starting Bonsai workflow `{workflow_file.name}`')	2✔
282	log.debug(' '.join(map(str, cmd)))	2✔
283	if wait:	2✔
284	return subprocess.run(args=cmd, cwd=cwd, check=check)	2✔
285	else:
UNCOV 286	return subprocess.Popen(args=cmd, cwd=cwd)	×
287
288
289	async def call_bonsai_async(	2✔
290	workflow_file: str \| Path,
291	parameters: dict[str, Any] \| None = None,
292	start: bool = True,
293	debug: bool = False,
294	bootstrap: bool = True,
295	editor: bool = True,
296	) -> asyncio.subprocess.Process:
297	"""
298	Asynchronously execute a Bonsai workflow within a subprocess call.
299
300	Parameters
301	----------
302	workflow_file : str \| Path
303	Path to the Bonsai workflow file.
304	parameters : dict[str, str], optional
305	Parameters to be passed to Bonsai workflow.
306	start : bool, optional
307	Start execution of the workflow within Bonsai (default is True).
308	debug : bool, optional
309	Enable debugging mode if True (default is False).
310	Only applies if editor is True.
311	bootstrap : bool, optional
312	Enable Bonsai bootstrapping if True (default is True).
313	editor : bool, optional
314	Enable Bonsai editor if True (default is True).
315
316	Returns
317	-------
318	asyncio.subprocess.Process
319	Pointer to the Bonsai subprocess if wait is False, otherwise subprocess.CompletedProcess.
320
321	Raises
322	------
323	FileNotFoundError
324	If the Bonsai executable does not exist.
325	If the specified workflow file does not exist.
326
327	"""
UNCOV 328	program, *cmd = _build_bonsai_cmd(workflow_file, parameters, start, debug, bootstrap, editor)	×
UNCOV 329	log.info(f'Starting Bonsai workflow `{workflow_file.name}`')	×
UNCOV 330	log.debug(' '.join(map(str, cmd)))	×
UNCOV 331	working_dir = Path(workflow_file).parent	×
UNCOV 332	return await asyncio.create_subprocess_exec(	×
333	program, *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=working_dir
334	)
335
336
337	T = TypeVar('T', bound=object)	2✔
338
339
340	def get_inheritors(cls: T) -> set[T]:	2✔
341	"""Obtain a set of all direct inheritors of a class."""
342	subclasses = set(cls.__subclasses__())	2✔
343	for child in subclasses:	2✔
344	subclasses = subclasses.union(get_inheritors(child))	2✔
345	return subclasses	2✔
346
347
348	@dataclass	2✔
349	class ANSI:	2✔
350	"""ANSI Codes for formatting text on the CLI."""
351
352	WHITE = '\033[37m'	2✔
353	PURPLE = '\033[95m'	2✔
354	CYAN = '\033[96m'	2✔
355	DARKCYAN = '\033[36m'	2✔
356	BLUE = '\033[94m'	2✔
357	GREEN = '\033[92m'	2✔
358	YELLOW = '\033[93m'	2✔
359	RED = '\033[91m'	2✔
360	BOLD = '\033[1m'	2✔
361	DIM = '\033[2m'	2✔
362	UNDERLINE = '\033[4m'	2✔
363	END = '\033[0m'	2✔
364
365
366	cached_check_output = cache(subprocess.check_output)	2✔
367
368
369	def get_lab_location_dict(hardware_settings: HardwareSettings, iblrig_settings: RigSettings) -> dict[str, Any]:	2✔
370	lab_location = dict()	2✔
371	lab_location['rig_name'] = hardware_settings.RIG_NAME	2✔
372	lab_location['iblrig_version'] = iblrig_version	2✔
373	lab_location['last_seen'] = date.today().isoformat()	2✔
374
375	machine = dict()	2✔
376	machine['platform'] = platform.platform()	2✔
377	machine['hostname'] = socket.gethostname()	2✔
378	machine['fqdn'] = socket.getfqdn()	2✔
379	machine['ip'] = socket.gethostbyname(machine['hostname'])	2✔
380	machine['mac'] = get_mac()	2✔
381	machine['anydesk'] = get_anydesk_id(format_id=False, silent=True)	2✔
382	lab_location['machine'] = machine	2✔
383
384	git = dict()	2✔
385	git['is_git'] = IS_GIT	2✔
386	git['branch'] = get_branch()	2✔
387	git['commit_id'] = get_commit_hash()	2✔
388	git['is_dirty'] = is_dirty()	2✔
389	lab_location['git'] = git	2✔
390
391	# TODO: add hardware/firmware versions of bpod, soundcard, rotary encoder, frame2ttl, ambient module, etc
392	# TODO: add validation errors/warnings
393	return lab_location	2✔
394
395
396	def get_number(	2✔
397	prompt: str = 'Enter number: ',
398	numeric_type: type(int) \| type(float) = int,
399	validation: Callable[[int \| float], bool] = lambda _: True,
400	) -> int \| float:
401	"""
402	Prompt the user for a number input of a specified numeric type and validate it.
403
404	Parameters
405	----------
406	prompt : str, optional
407	The message displayed to the user when asking for input.
408	Defaults to 'Enter number: '.
409	numeric_type : type, optional
410	The type of the number to be returned. Can be either `int` or `float`.
411	Defaults to `int`.
412	validation : callable, optional
413	A function that takes a number as input and returns a boolean
414	indicating whether the number is valid. Defaults to a function
415	that always returns True.
416
417	Returns
418	-------
419	int or float
420	The validated number input by the user, converted to the specified type.
421
422	Notes
423	-----
424	The function will continue to prompt the user until a valid number
425	is entered that passes the validation function.
426	"""
UNCOV 427	value = None	×
UNCOV 428	while not isinstance(value, numeric_type) or validation(value) is False:	×
UNCOV 429	try:	×
UNCOV 430	value = numeric_type(input(prompt).strip())	×
UNCOV 431	except ValueError:	×
UNCOV 432	value = None	×
UNCOV 433	return value	×

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