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

88.73

/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_management
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 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 static_vars(**kwargs) -> Callable[..., Any]:
    """
    Decorator to add static variables to a function.

    This decorator allows you to add static variables to a function by providing
    keyword arguments. Static variables are shared across all calls to the
    decorated function.

    Parameters
    ----------
    **kwargs
        Keyword arguments where the keys are variable names and the values are
        the initial values of the static variables.

    Returns
    -------
    function
        A decorated function with the specified static variables.
    """

    def decorate(func: Callable[..., Any]) -> Callable[..., Any]:
        for k in kwargs:
            setattr(func, k, kwargs[k])
        return func

    return decorate


@static_vars(return_value=None)
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 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,
) -> list[str]:
    """
    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
    -------
    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_EXE.exists():
        raise FileNotFoundError(BONSAI_EXE)
    workflow_file = Path(workflow_file)
    if not workflow_file.exists():
        raise FileNotFoundError(workflow_file)
    create_bonsai_layout_from_template(workflow_file)

    cmd = [str(BONSAI_EXE), 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,
) -> 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.

    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)
    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'] = str(version_management.get_local_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'] = version_management.get_branch()
    git['commit_id'] = version_management.get_commit_hash()
    git['is_dirty'] = version_management.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

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