diff --git a/aidge_core/unit_tests/test_export.py b/aidge_core/unit_tests/test_export.py index e7ba8a396c418a8dc5b05ec4fdf046667ec4ac13..9fb16128eebed9102cdf0e46e359a832bf6ac140 100644 --- a/aidge_core/unit_tests/test_export.py +++ b/aidge_core/unit_tests/test_export.py @@ -8,15 +8,14 @@ http://www.eclipse.org/legal/epl-2.0. SPDX-License-Identifier: EPL-2.0 """ -import unittest import aidge_core -import pathlib +from aidge_core.utils import run_command +import unittest import os -import sys -import subprocess +import pathlib import shutil -import queue -import threading +import subprocess +import sys def initFiller(model): @@ -44,60 +43,6 @@ def initFiller(model): pass -def enqueue_output(stream, queue): - for line in iter(stream.readline, ""): - queue.put(line) - stream.close() - - -def run_command(command : list[str], cwd : pathlib.Path = None): - """ - This function has the job to run a command and return stdout and stderr that are not shown - by subprocess.check_call / call. - If the subprocess returns smthg else than 0, it will raise an error. - Arg: - command : written with the same syntax as subprocess.call - cwd : path from where the command must be called - """ - process = subprocess.Popen( - command, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True - ) - - stdout_queue = queue.Queue() - stderr_queue = queue.Queue() - - stdout_thread = threading.Thread( - target=enqueue_output, args=(process.stdout, stdout_queue) - ) - stderr_thread = threading.Thread( - target=enqueue_output, args=(process.stderr, stderr_queue) - ) - stdout_thread.start() - stderr_thread.start() - - while ( - stdout_thread.is_alive() - or stderr_thread.is_alive() - or not stdout_queue.empty() - or not stderr_queue.empty() - ): - try: - stdout_line = stdout_queue.get_nowait() - yield stdout_line - except queue.Empty: - pass - - try: - stderr_line = stderr_queue.get_nowait() - yield stderr_line - except queue.Empty: - pass - - return_code = process.wait() - if return_code != 0: - raise subprocess.CalledProcessError(return_code, command) - - def clean_dir(dir: pathlib.Path) -> None: if not dir.is_dir(): print(f"Error : directory {dir} doesn't exist. Exiting clean_dir().") diff --git a/aidge_core/utils.py b/aidge_core/utils.py index d82d524b7e886ed396507376a5934a748a89e44c..48b96a6aab3a1ec827cf48e5fc3dbb0351bfa0e7 100644 --- a/aidge_core/utils.py +++ b/aidge_core/utils.py @@ -1,3 +1,9 @@ +import queue +import threading +import subprocess +import pathlib + + def template_docstring(template_keyword, text_to_replace): """Method to template docstring @@ -6,11 +12,86 @@ def template_docstring(template_keyword, text_to_replace): :param text_to_replace: Text to replace your template with. :type text_to_replace: str """ + def dec(func): - if "{"+template_keyword+"}" not in func.__doc__: + if "{" + template_keyword + "}" not in func.__doc__: raise RuntimeError( - f"The function {function.__name__} docstring does not contain the template keyword: {template_keyword}.") + f"The function {func.__name__} docstring does not contain the template keyword: {template_keyword}." + ) func.__doc__ = func.__doc__.replace( - "{"+template_keyword+"}", text_to_replace) + "{" + template_keyword + "}", text_to_replace + ) return func + return dec + + +def enqueue_output(stream, queue): + for line in iter(stream.readline, ""): + queue.put(line) + stream.close() + + +def run_command(command: list[str], cwd: pathlib.Path = None): + """ + This function has the job to run a command and return stdout and stderr that are not shown + by subprocess.check_call / call. + If the subprocess returns smthg else than 0, it will raise an error. + Arg: + command : written with the same syntax as subprocess.call + cwd : path from where the command must be called + + Call example: + ```python + try: + for std_line in run_command( + [ + "cmake", + str(self.EXPORT_PATH.absolute()), + "-DPYBIND=1", + f"-DCMAKE_INSTALL_PREFIX:PATH={install_path}", + ], + cwd=str(self.BUILD_DIR), + ): + print(std_line, end="") + except subprocess.CalledProcessError as e: + print(f"An error occurred: {e}\nFailed to configure export.") + ``` + """ + process = subprocess.Popen( + command, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True + ) + + stdout_queue = queue.Queue() + stderr_queue = queue.Queue() + + stdout_thread = threading.Thread( + target=enqueue_output, args=(process.stdout, stdout_queue) + ) + stderr_thread = threading.Thread( + target=enqueue_output, args=(process.stderr, stderr_queue) + ) + stdout_thread.start() + stderr_thread.start() + + while ( + stdout_thread.is_alive() + or stderr_thread.is_alive() + or not stdout_queue.empty() + or not stderr_queue.empty() + ): + try: + stdout_line = stdout_queue.get_nowait() + yield stdout_line + except queue.Empty: + pass + + try: + stderr_line = stderr_queue.get_nowait() + yield stderr_line + except queue.Empty: + pass + + return_code = process.wait() + if return_code != 0: + raise subprocess.CalledProcessError(return_code, command)