diff --git a/SofaRegressionProgram/SofaRegressionProgram.py b/SofaRegressionProgram/SofaRegressionProgram.py index 15e37ac..30339d1 100644 --- a/SofaRegressionProgram/SofaRegressionProgram.py +++ b/SofaRegressionProgram/SofaRegressionProgram.py @@ -13,12 +13,12 @@ import Sofa import SofaRuntime # importing SofaRuntime will add the py3 loader to the scene loaders import tools.RegressionSceneList as RegressionSceneList -from tools import ProgressBarHandler as pbh +import tools.RegressionWorker as RegressionWorker regression_file_extension = ".regression-tests" class RegressionProgram: - def __init__(self, input_folder, filter = None, disable_progress_bar = False, verbose = False): + def __init__(self, input_folder, filter = None, disable_progress_bar = False, verbose = False, nbr_jobs = 1): """Initialize the RegressionProgram Args: @@ -26,18 +26,20 @@ def __init__(self, input_folder, filter = None, disable_progress_bar = False, ve filter (str): Regex pattern to filter scene files (e.g., '^demo.*.scn$'). If None, no filter is applied. Defaults to None. disable_progress_bar (bool, optional): If True, disable progress bars. Defaults to False. verbose (bool, optional): If True, enable verbose output. Defaults to False. + nbr_jobs (int, optional): Number of scenes to write/compare at the same time. 0 means one per logical core. Defaults to 1. """ self.scene_sets = [] # List self.disable_progress_bar = disable_progress_bar self.verbose = verbose self.legacy_mode = False + self.nbr_jobs = RegressionWorker.resolve_nbr_jobs(nbr_jobs) for root, dirs, files in os.walk(input_folder): for file in files: if file.endswith(regression_file_extension): file_path = os.path.join(root, file) - scene_list = RegressionSceneList.RegressionSceneList(file_path, filter, self.disable_progress_bar, verbose) + scene_list = RegressionSceneList.RegressionSceneList(file_path, filter, self.disable_progress_bar, verbose, self.nbr_jobs) scene_list.process_file() self.scene_sets.append(scene_list) @@ -48,30 +50,42 @@ def nbr_error_in_sets(self): nbr_errors = nbr_errors + scene_list.get_nbr_errors() return nbr_errors + def nbr_parsing_error_in_sets(self): + nbr_errors = 0 + for scene_list in self.scene_sets: + nbr_errors = nbr_errors + scene_list.get_nbr_parsing_errors() + return nbr_errors + def log_errors_in_sets(self): for scene_list in self.scene_sets: scene_list.log_scenes_errors() + def run_all_sets(self, mode, description): + """Run every scene of every set in `mode` ("write" or "compare"). + + When several jobs are allowed, the scenes of all the sets are scheduled + in a single pool: a set holding fewer scenes than the number of jobs + would otherwise leave most of the workers idle. + """ + tasks = [] + for scene_list in self.scene_sets: + scene_list.legacy_mode = self.legacy_mode + tasks.extend(scene_list.build_tasks(mode)) + + return RegressionWorker.run_scene_tasks( + tasks, + nbr_jobs=self.nbr_jobs, + on_result=lambda task, result: task["scene_list"].apply_result(task, result), + description=description, + disable_progress_bar=self.disable_progress_bar) + def write_sets_references(self, id_set=0): scene_list = self.scene_sets[id_set] nbr_scenes = scene_list.write_all_references() return nbr_scenes def write_all_sets_references(self): - nbr_sets = len(self.scene_sets) - - pbar_sets = pbh.ProgressBarHandler(total=nbr_sets, disable=self.disable_progress_bar) - pbar_sets.set_description("Write All sets") - - nbr_scenes = 0 - for i in range(0, nbr_sets): - nbr_scenes = nbr_scenes + self.write_sets_references(i) - pbar_sets.update(1) - - if not self.disable_progress_bar: - pbar_sets.close() - - return nbr_scenes + return self.run_all_sets("write", "Write All sets") def compare_sets_references(self, id_set=0): scene_list = self.scene_sets[id_set] @@ -80,18 +94,7 @@ def compare_sets_references(self, id_set=0): return nbr_scenes def compare_all_sets_references(self): - nbr_sets = len(self.scene_sets) - pbar_sets = pbh.ProgressBarHandler(total=nbr_sets, disable=self.disable_progress_bar) - pbar_sets.set_description("Compare All sets") - - nbr_scenes = 0 - for i in range(0, nbr_sets): - nbr_scenes = nbr_scenes + self.compare_sets_references(i) - pbar_sets.update(1) - - pbar_sets.close() - - return nbr_scenes + return self.run_all_sets("compare", "Compare All sets") def replay_references(self, id_scene, id_set=0): scene_list = self.scene_sets[id_set] @@ -122,7 +125,15 @@ def make_parser(): help="A regex filter to select scenes to test (e.g., '^demo.*.scn$')", type=str) - parser.add_argument('--replay', + parser.add_argument('-j', '--jobs', + dest='jobs', + help="Number of scenes to process at the same time (each one still runs in its own\n" + "isolated process, so the results are unchanged). 0 means one job per logical\n" + "core. Default: 1 (sequential).", + type=int, + default=1) + + parser.add_argument('--replay', dest='replay', help=f"Will launch runSofa on the scene number X (input number) in the input the list of the {regression_file_extension} file given as input and display the scene references aside from the simulation", type=int) @@ -163,6 +174,8 @@ def make_parser(): python SofaRegressionProgram.py --input ./scenes python SofaRegressionProgram.py --input ./scenes --filter \"$demo.*.scn\" python SofaRegressionProgram.py --input ./scenes --replay 5 + python SofaRegressionProgram.py --input ./scenes --jobs 8 + python SofaRegressionProgram.py --input ./scenes --write-references -j 0 ''' return parser @@ -175,7 +188,7 @@ def make_parser(): # 2- Process file if args.input is not None: - reg_prog = RegressionProgram(args.input, args.filter, args.progress_bar_is_disabled, args.verbose) + reg_prog = RegressionProgram(args.input, args.filter, args.progress_bar_is_disabled, args.verbose, args.jobs) else: parser.print_help() exit("Error: Argument is required ! Quitting.") @@ -185,7 +198,11 @@ def make_parser(): if args.legacy_mode: print("Legacy regression mode activated.") reg_prog.legacy_mode = True - + + if reg_prog.nbr_jobs > 1: + print(f"Processing up to {reg_prog.nbr_jobs} scenes at the same time.") + + if args.replay is not None: replayId = int(args.replay) reg_prog.replay_references(replayId) @@ -212,14 +229,23 @@ def make_parser(): np.set_printoptions(legacy='1.25') # revert printing floating-point type in numpy (concretely remove np.array when displaying a list of np.float) + nbr_parsing_errors = reg_prog.nbr_parsing_error_in_sets() + print ("### Number of sets Done: " + str(len(reg_prog.scene_sets))) print ("### Number of scenes Done: " + str(nbr_scenes)) + if nbr_parsing_errors > 0: + # Those scenes have not been processed at all: report them as an error + # so that an invalid list file cannot silently reduce the test coverage. + print ("### Number of invalid lines skipped: " + str(nbr_parsing_errors)) if args.write_mode is False: print ("### Number of scenes failed: " + str(reg_prog.nbr_error_in_sets())) reg_prog.log_errors_in_sets() if reg_prog.nbr_error_in_sets() > 0: sys.exit(1) # exit with error(s) + if nbr_parsing_errors > 0: + sys.exit(1) # exit with error(s) + sys.exit(0) # exit without error diff --git a/SofaRegressionProgram/tools/RegressionSceneData.py b/SofaRegressionProgram/tools/RegressionSceneData.py index ec6be5c..e9f8a99 100644 --- a/SofaRegressionProgram/tools/RegressionSceneData.py +++ b/SofaRegressionProgram/tools/RegressionSceneData.py @@ -108,6 +108,16 @@ def log_errors(self): else: helper.writeSuccess(f"{self.file_scene_path} | Number of key frames compared: {self.nbr_tested_frame} | run time: {self.total_run_time/1e9} seconds. ") + def apply_worker_result(self, result): + """Copy the fields reported by an isolated worker process back onto this + object so that log_errors() and error counting behave as if the scene had + been compared in-process.""" + self.regression_failed = bool(result.get("regression_failed", False)) + self.nbr_tested_frame = int(result.get("nbr_tested_frame", 0)) + self.total_run_time = result.get("total_run_time", 0) + self.error_by_dof = result.get("error_by_dof", []) + self.total_error = result.get("total_error", []) + def print_meca_objs(self): helper.writeLog("# Nbr Meca: " + str(len(self.meca_objs))) counter = 0 diff --git a/SofaRegressionProgram/tools/RegressionSceneList.py b/SofaRegressionProgram/tools/RegressionSceneList.py index 830fd5e..b8c2b43 100644 --- a/SofaRegressionProgram/tools/RegressionSceneList.py +++ b/SofaRegressionProgram/tools/RegressionSceneList.py @@ -1,14 +1,14 @@ import os import tools.RegressionSceneData as RegressionSceneData import tools.RegressionHelper as helper -from tools import ProgressBarHandler as pbh +import tools.RegressionWorker as RegressionWorker import re ## This class is responsible for loading a file.regression-tests to gather the list of scene to test with all arguments ## It will provide the API to launch the tests or write refs on all scenes contained in this file class RegressionSceneList: - def __init__(self, file_path, filter, disable_progress_bar = False, verbose = False): + def __init__(self, file_path, filter, disable_progress_bar = False, verbose = False, nbr_jobs = 1): """ /// Path to the file.regression-tests containing the list of scene to tests with all arguments std::string filePath; @@ -18,18 +18,23 @@ def __init__(self, file_path, filter, disable_progress_bar = False, verbose = Fa self.file_dir = os.path.dirname(file_path) self.scenes_data_sets = [] # List self.nbr_errors = 0 + self.nbr_parsing_errors = 0 # number of lines of the list file that could not be used self.ref_dir_path = None self.disable_progress_bar = disable_progress_bar self.verbose = verbose self.legacy_mode = False + self.nbr_jobs = nbr_jobs # number of scenes simulated at the same time def get_nbr_scenes(self): return len(self.scenes_data_sets) - + def get_nbr_errors(self): return self.nbr_errors - + + def get_nbr_parsing_errors(self): + return self.nbr_parsing_errors + def log_scenes_errors(self): for scene in self.scenes_data_sets: scene.log_errors() @@ -37,6 +42,107 @@ def log_scenes_errors(self): def set_legacy_mode(self, legacy_mode): self.legacy_mode = legacy_mode + + def parsing_error(self, line_number, message): + """Report a line of the list file that cannot be used, and count it. + + A malformed line only invalidates the scene it describes: it must never + interrupt the parsing of the file, nor the whole regression run. + """ + self.nbr_parsing_errors = self.nbr_parsing_errors + 1 + helper.writeError(f"{self.file_path}:{line_number}: {message}") + + + def parse_scene_line(self, values, line_number): + """Parse one scene line of the list file. + + Args: + values (list): the whitespace separated fields of the line. + line_number (int): line number in the list file, for error reporting. + + Returns: + RegressionSceneData: the described scene, or None if the line is + invalid. In that case the error has already been reported. + """ + expected_fields = " " + if len(values) > 5: + helper.writeWarning(f"{self.file_path}:{line_number}: expecting at most 5 fields " + f"({expected_fields}), got {len(values)}. Extra fields are ignored.") + + steps = 1000 + epsilon = 0.0001 + meca_in_mapping = False + dump_number_step = 1 + + if len(values) < 2: + helper.writeWarning(f"{self.file_path}:{line_number}: cannot evaluate steps. " + f"Default value {steps} will be used instead.") + else: + try: + steps = int(values[1]) + except ValueError: + self.parsing_error(line_number, f"steps must be an integer, got '{values[1]}'. " + f"Expecting: {expected_fields}. Skipping this scene.") + return None + if steps <= 0: + self.parsing_error(line_number, f"steps must be strictly positive, got {steps}. " + f"Skipping this scene.") + return None + + if len(values) < 3: + helper.writeWarning(f"{self.file_path}:{line_number}: cannot evaluate epsilon. " + f"Default value {epsilon} will be used instead.") + else: + try: + epsilon = float(values[2]) + except ValueError: + self.parsing_error(line_number, f"epsilon must be a number, got '{values[2]}'. " + f"Expecting: {expected_fields}. Skipping this scene.") + return None + if epsilon < 0: + self.parsing_error(line_number, f"epsilon must be positive, got {epsilon}. " + f"Skipping this scene.") + return None + + if len(values) < 4: + helper.writeWarning(f"{self.file_path}:{line_number}: cannot evaluate meca_in_mapping. " + f"Default value {meca_in_mapping} will be used instead.") + elif values[3] not in ('0', '1'): + self.parsing_error(line_number, f"meca_in_mapping must be 0 or 1, got '{values[3]}'. " + f"Expecting: {expected_fields}. Skipping this scene.") + return None + else: + meca_in_mapping = (values[3] == '1') # converting string to Bool always gives True + + if len(values) < 5: + helper.writeWarning(f"{self.file_path}:{line_number}: cannot evaluate dump_number_step. " + f"Default value {dump_number_step} will be used instead.") + else: + try: + dump_number_step = int(values[4]) + except ValueError: + self.parsing_error(line_number, f"dump_number_step must be an integer, got '{values[4]}'. " + f"Expecting: {expected_fields}. Skipping this scene.") + return None + # dump_number_step is used as a divider of the number of steps + if dump_number_step <= 0: + self.parsing_error(line_number, f"dump_number_step must be strictly positive, " + f"got {dump_number_step}. Skipping this scene.") + return None + + full_file_path = os.path.normpath(os.path.join(self.file_dir, values[0])) + if not os.path.isfile(full_file_path): + self.parsing_error(line_number, f"scene file does not exist: {full_file_path}. " + f"Skipping this scene.") + return None + + full_ref_file_path = os.path.normpath(os.path.join(self.ref_dir_path, values[0])) + + return RegressionSceneData.RegressionSceneData(full_file_path, full_ref_file_path, + steps, epsilon, meca_in_mapping, dump_number_step, + self.disable_progress_bar, self.verbose) + + def process_file(self): with open(self.file_path, 'r') as the_file: data = the_file.readlines() @@ -44,7 +150,9 @@ def process_file(self): count = 0 for idx, line in enumerate(data): - if line[0] == "#": + line_number = idx + 1 + + if line.startswith("#"): continue values = line.split() @@ -56,14 +164,17 @@ def process_file(self): if ("REGRESSION_DIR" in os.environ): self.ref_dir_path = values[0].replace("$REGRESSION_DIR", os.environ["REGRESSION_DIR"]) else: - helper.writeError(f"The environment variable $REGRESSION_DIR is required in {self.file_path} but not set. Please set this variable to the root directory of your regression tests to proceed.") + self.parsing_error(line_number, f"the environment variable $REGRESSION_DIR is required but not set. " + f"Please set this variable to the root directory of your regression tests to proceed. " + f"No scene of this file will be processed.") return else: # direct absolute or relative path self.ref_dir_path = os.path.join(self.file_dir, values[0]) self.ref_dir_path = os.path.abspath(self.ref_dir_path) if not os.path.isdir(self.ref_dir_path): - helper.writeError(f'Reference directory mentioned by file \'{self.file_path}\' does not exist: {self.ref_dir_path}') + self.parsing_error(line_number, f"reference directory does not exist: {self.ref_dir_path}. " + f"No scene of this file will be processed.") return if self.verbose: @@ -76,101 +187,99 @@ def process_file(self): helper.writeLog(f'Filtered out {self.filter}: {values[0]}') continue - steps = 1000 - epsilon = 0.0001 - meca_in_mapping = False - dump_number_step = 1 + # An invalid line is reported and skipped: the other scenes of the + # file must still be processed. + scene_data = self.parse_scene_line(values, line_number) + if scene_data is None: + continue - if len(values) < 2: - helper.writeWarning(f"Cannot evaluate steps from line: {line}. Default value {steps} will be used instead.") - else: - steps = int(values[1]) + #scene_data.printInfo() + self.scenes_data_sets.append(scene_data) - if len(values) < 3: - helper.writeWarning( - f"Cannot evaluate epsilon from line: {line}. Default value {epsilon} will be used instead.") - else: - epsilon = float(values[2]) - if len(values) < 4: - helper.writeWarning( - f"Cannot evaluate meca_in_mapping from line: {line}. Default value {meca_in_mapping} will be used instead.") - else: - if values[3] == '1': # converting string to Bool always gives True - meca_in_mapping = True - - if len(values) < 5: - helper.writeWarning( - f"Cannot evaluate dump_number_step from line: {line}. Default value {dump_number_step} will be used instead.") - else: - dump_number_step = int(values[4]) + def build_task(self, id_scene, mode): + """Return the task descriptor handed over to RegressionWorker for one scene. + + Each scene is run in its own process to guarantee a clean SOFA state + (SOFA does not fully reset its global state between load/unload), which + also makes it safe to run several of them at the same time. + """ + return { + "scene_list": self, + "id_scene": id_scene, + "scene_data": self.scenes_data_sets[id_scene], + "mode": mode, + "legacy": self.legacy_mode, + "verbose": self.verbose, + } - full_file_path = os.path.normpath(os.path.join(self.file_dir, values[0])) - full_ref_file_path = os.path.normpath(os.path.join(self.ref_dir_path, values[0])) - scene_data = RegressionSceneData.RegressionSceneData(full_file_path, full_ref_file_path, - steps, epsilon, meca_in_mapping, dump_number_step, - self.disable_progress_bar, self.verbose) + def build_tasks(self, mode): + """Return the task descriptors of every scene of this list.""" + return [self.build_task(i, mode) for i in range(len(self.scenes_data_sets))] - #scene_data.printInfo() - self.scenes_data_sets.append(scene_data) + + def apply_result(self, task, result): + """Collect the outcome reported by a worker process for one scene.""" + scene = self.scenes_data_sets[task["id_scene"]] + + if task["mode"] == "write": + if not result.get("ok", False): + helper.writeError(f"While writing references for {scene.file_scene_path}: {result.get('error')}") + return + + if not result.get("ok", False): + # Hard failure (scene could not be loaded / worker crashed). + self.nbr_errors = self.nbr_errors + 1 + helper.writeError(f"While trying to compare {scene.file_scene_path}: {result.get('error')}") + return + + # Bring the worker's outcome back so log_errors() reports it as usual. + scene.apply_worker_result(result) + if not result.get("result", False): + self.nbr_errors = self.nbr_errors + 1 + + + def _run_tasks(self, mode, description): + tasks = self.build_tasks(mode) + return RegressionWorker.run_scene_tasks( + tasks, + nbr_jobs=self.nbr_jobs, + on_result=self.apply_result, + description=description, + disable_progress_bar=self.disable_progress_bar) def write_references(self, id_scene, print_log = False): + scene = self.scenes_data_sets[id_scene] if self.verbose: - helper.writeLog(f'Writing reference files for {self.scenes_data_sets[id_scene].file_scene_path}.') + helper.writeLog(f'Writing reference files for {scene.file_scene_path}.') - self.scenes_data_sets[id_scene].load_scene() - if print_log is True: - self.scenes_data_sets[id_scene].print_meca_objs() - - self.scenes_data_sets[id_scene].write_references() + task = self.build_task(id_scene, "write") + result = RegressionWorker.run_scene_in_subprocess( + scene, mode="write", + disable_progress_bar=self.disable_progress_bar, verbose=self.verbose) + self.apply_result(task, result) - def write_all_references(self): - nbr_scenes = len(self.scenes_data_sets) - pbar_scenes = pbh.ProgressBarHandler(total=nbr_scenes, disable=self.disable_progress_bar) - pbar_scenes.set_description("Write all scenes from: " + self.file_path) - - for i in range(0, nbr_scenes): - self.write_references(i) - pbar_scenes.update(1) - - pbar_scenes.close() - - return nbr_scenes + def write_all_references(self): + return self._run_tasks("write", "Write all scenes from: " + self.file_path) def compare_references(self, id_scene): + scene = self.scenes_data_sets[id_scene] if self.verbose: - self.scenes_data_sets[id_scene].print_info() + scene.print_info() - try: - self.scenes_data_sets[id_scene].load_scene() - except Exception as e: - self.nbr_errors = self.nbr_errors + 1 - helper.writeError(f"While trying to load: {str(e)}") - else: - if self.legacy_mode: - result = self.scenes_data_sets[id_scene].compare_legacy_references() - else: - result = self.scenes_data_sets[id_scene].compare_references() - - if not result: - self.nbr_errors = self.nbr_errors + 1 - + task = self.build_task(id_scene, "compare") + result = RegressionWorker.run_scene_in_subprocess( + scene, mode="compare", legacy=self.legacy_mode, + disable_progress_bar=self.disable_progress_bar, verbose=self.verbose) + self.apply_result(task, result) - def compare_all_references(self): - nbr_scenes = len(self.scenes_data_sets) - pbar_scenes = pbh.ProgressBarHandler(total=nbr_scenes, disable=self.disable_progress_bar) - pbar_scenes.set_description("Compare all scenes from: " + self.file_path) - - for i in range(0, nbr_scenes): - self.compare_references(i) - pbar_scenes.update(1) - pbar_scenes.close() - return nbr_scenes + def compare_all_references(self): + return self._run_tasks("compare", "Compare all scenes from: " + self.file_path) def replay_references(self, id_scene): diff --git a/SofaRegressionProgram/tools/RegressionWorker.py b/SofaRegressionProgram/tools/RegressionWorker.py new file mode 100644 index 0000000..293bd48 --- /dev/null +++ b/SofaRegressionProgram/tools/RegressionWorker.py @@ -0,0 +1,329 @@ +""" +Per-scene subprocess isolation for the SOFA regression program. + +SOFA does not fully reset its global/static state between two load/unload +cycles inside a single process. As a consequence, the simulation result of a +scene depends on which scenes were simulated before it in the same process. +This makes references order-dependent: a scene simulated during the batch +`--write-references` pass can produce a different result than the same scene +simulated during the batch compare pass, causing regressions to fail right +after regenerating the references without anything having changed. + +To guarantee reproducibility, every scene is simulated in its own freshly +spawned Python process. Each child starts from a clean SOFA state, so the +write pass and the compare pass always see identical conditions. + +This module has two roles: + * Parent side: `run_scene_in_subprocess()` spawns a child for one scene and + marshals the result back through a temporary JSON file. `run_scene_tasks()` + schedules a list of scenes over a pool of such children, so that several + scenes are simulated at the same time. + * Child side: executed as `python RegressionWorker.py ...`, it sets up the + SOFA environment, runs a single scene (write or compare) and writes its + result to the file given by `--result-file`. + +Because every scene already runs in its own process, running several of them +concurrently changes nothing to the results: children never share any SOFA +state. The parent only has to schedule them and collect their outcome. + +Only the standard library is imported at module top-level so that importing +this module in the parent does NOT import SOFA (the parent must never load or +simulate a scene, otherwise the isolation would be defeated). +""" + +import os +import sys +import json +import argparse +import subprocess +import tempfile +from concurrent.futures import ThreadPoolExecutor, as_completed + + +def _safe_remove(path): + try: + os.remove(path) + except OSError: + pass + + +# -------------------------------------------------- +# Parent side: spawn one child process for one scene +# -------------------------------------------------- +def run_scene_in_subprocess(scene_data, mode, legacy=False, + disable_progress_bar=False, verbose=False, + format="JSON", python_exe=None, + capture_output=False): + """Run a single scene (write or compare) in an isolated child process. + + Args: + scene_data: the RegressionSceneData describing the scene to run. + mode (str): "write" to generate references, "compare" to check them. + legacy (bool): use the legacy reference format (compare only). + disable_progress_bar (bool): forwarded to the child. + verbose (bool): forwarded to the child. + format (str): reference file format ("JSON" or "CSV"). + python_exe (str): interpreter to use for the child (defaults to the + current one). + capture_output (bool): if True, the child output is captured and + returned in the "stdout"/"stderr" keys of the result instead of + being interleaved with the output of the other children. Used when + several scenes run concurrently. + + Returns: + dict: the result reported by the child. Always contains an "ok" key. + For compare runs it also contains "result", "regression_failed", + "nbr_tested_frame", "total_run_time", "error_by_dof" and + "total_error". + """ + python_exe = python_exe or sys.executable + worker_path = os.path.abspath(__file__) + + fd, result_path = tempfile.mkstemp(suffix=".json", prefix="regression_result_") + os.close(fd) + + cmd = [ + python_exe, worker_path, + "--mode", mode, + "--scene", str(scene_data.file_scene_path), + "--ref", str(scene_data.file_ref_path), + "--steps", str(scene_data.steps), + "--epsilon", repr(scene_data.epsilon), + "--meca-in-mapping", "1" if scene_data.meca_in_mapping else "0", + "--dump-number-step", str(scene_data.dump_number_step), + "--format", format, + "--result-file", result_path, + ] + if legacy: + cmd.append("--legacy") + if verbose: + cmd.append("--verbose") + if disable_progress_bar: + cmd.append("--disable-progress-bar") + + # When a single scene runs at a time, stdout/stderr are inherited so SOFA + # logs and progress bars behave exactly as before (and the parent's --quiet + # redirection propagates to the child). When several children run at the + # same time their output is captured instead, and replayed as one block by + # the caller, otherwise the logs of all the scenes would be interleaved. + try: + completed = subprocess.run(cmd, capture_output=capture_output, text=capture_output) + except Exception as e: + _safe_remove(result_path) + return {"ok": False, "error": f"Failed to launch worker subprocess: {e}"} + + result = None + try: + with open(result_path, "r") as f: + result = json.load(f) + except Exception: + result = None + _safe_remove(result_path) + + if result is None: + result = {"ok": False, + "error": f"Worker produced no result (exit code {completed.returncode})."} + + if capture_output: + result["stdout"] = completed.stdout + result["stderr"] = completed.stderr + return result + + +# -------------------------------------------------- +# Parent side: schedule several scenes concurrently +# -------------------------------------------------- +def resolve_nbr_jobs(nbr_jobs): + """Turn the user-provided job count into a usable number of workers. + + 0 (or a negative value) means "one job per logical core". + """ + if nbr_jobs is None: + return 1 + nbr_jobs = int(nbr_jobs) + if nbr_jobs <= 0: + return os.cpu_count() or 1 + return nbr_jobs + + +def _echo_captured_output(header, result): + """Print in one block the output captured from a child process.""" + out = result.get("stdout") + err = result.get("stderr") + if not (out or err): + return + + if out: + sys.stdout.write(header + "\n") + sys.stdout.write(out if out.endswith("\n") else out + "\n") + sys.stdout.flush() + if err: + sys.stderr.write(header + "\n") + sys.stderr.write(err if err.endswith("\n") else err + "\n") + sys.stderr.flush() + + +def run_scene_tasks(tasks, nbr_jobs=1, format="JSON", on_result=None, + description=None, disable_progress_bar=False): + """Run a list of scenes, up to `nbr_jobs` of them at the same time. + + Args: + tasks (list): task descriptors. Each one is a dict containing at least + "scene_data" (RegressionSceneData), "mode" ("write" or "compare"), + and optionally "legacy" and "verbose". Any other key is ignored + here and simply handed back to `on_result`, which lets the caller + attach whatever context it needs to identify the task. + nbr_jobs (int): maximum number of scenes simulated concurrently. + format (str): reference file format ("JSON" or "CSV"). + on_result (callable): called as `on_result(task, result)` for every + finished task, always from the calling thread so that the callback + does not need any locking. + description (str): label of the progress bar. + disable_progress_bar (bool): disable the progress bar of this run. + + Returns: + int: the number of tasks that were run. + """ + from tools import ProgressBarHandler as pbh + + nbr_jobs = max(1, resolve_nbr_jobs(nbr_jobs)) + # Never spawn more workers than there is work to do. + nbr_jobs = min(nbr_jobs, len(tasks)) if tasks else 1 + + pbar = pbh.ProgressBarHandler(total=len(tasks), disable=disable_progress_bar) + if description is not None: + pbar.set_description(description) + + def _run(task): + return run_scene_in_subprocess( + task["scene_data"], + mode=task["mode"], + legacy=task.get("legacy", False), + # In parallel the per-step progress bars of the children are + # captured along with their output: they would only produce noise. + disable_progress_bar=disable_progress_bar or nbr_jobs > 1, + verbose=task.get("verbose", False), + format=format, + capture_output=nbr_jobs > 1, + ) + + try: + if nbr_jobs == 1: + for task in tasks: + result = _run(task) + if on_result is not None: + on_result(task, result) + pbar.update(1) + else: + with ThreadPoolExecutor(max_workers=nbr_jobs) as executor: + # The threads only wait on their child process: all the result + # handling happens here, in the calling thread. + futures = {executor.submit(_run, task): task for task in tasks} + try: + for future in as_completed(futures): + task = futures[future] + result = future.result() + _echo_captured_output( + f"--- {task['mode']}: {task['scene_data'].file_scene_path}", result) + if on_result is not None: + on_result(task, result) + pbar.update(1) + except (KeyboardInterrupt, SystemExit): + executor.shutdown(wait=False, cancel_futures=True) + raise + finally: + pbar.close() + + return len(tasks) + + +# -------------------------------------------------- +# Child side: run one scene in a fresh SOFA process +# -------------------------------------------------- +def _make_worker_parser(): + parser = argparse.ArgumentParser(description="Regression per-scene worker (internal)") + parser.add_argument("--mode", choices=["write", "compare"], required=True) + parser.add_argument("--scene", required=True) + parser.add_argument("--ref", required=True) + parser.add_argument("--steps", type=int, required=True) + parser.add_argument("--epsilon", type=float, required=True) + parser.add_argument("--meca-in-mapping", dest="meca_in_mapping", choices=["0", "1"], required=True) + parser.add_argument("--dump-number-step", dest="dump_number_step", type=int, required=True) + parser.add_argument("--format", default="JSON") + parser.add_argument("--result-file", dest="result_file", required=True) + parser.add_argument("--legacy", action="store_true") + parser.add_argument("--verbose", action="store_true") + parser.add_argument("--disable-progress-bar", dest="disable_progress_bar", action="store_true") + return parser + + +def _worker_main(): + args = _make_worker_parser().parse_args() + + result = {"ok": False, "error": None} + try: + # SOFA and the tools package must be imported inside this fresh process. + if "SOFA_ROOT" not in os.environ: + raise RuntimeError("SOFA_ROOT environment variable is not set.") + + sofapython3_path = os.path.join(os.environ["SOFA_ROOT"], "lib", "python3", "site-packages") + if sofapython3_path not in sys.path: + sys.path.append(sofapython3_path) + + # Make the "tools" package importable (program root = parent of this dir). + program_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + if program_root not in sys.path: + sys.path.insert(0, program_root) + + import SofaRuntime # noqa: F401 (registers the py3 scene loader) + import tools.RegressionSceneData as RegressionSceneData + + scene = RegressionSceneData.RegressionSceneData( + file_scene_path=args.scene, + file_ref_path=args.ref, + steps=args.steps, + epsilon=args.epsilon, + meca_in_mapping=(args.meca_in_mapping == "1"), + dump_number_step=args.dump_number_step, + disable_progress_bar=args.disable_progress_bar, + verbose=args.verbose, + ) + + scene.load_scene(args.format) + + if args.mode == "write": + scene.write_references(args.format) + result = {"ok": True, "error": None} + else: # compare + if args.legacy: + passed = scene.compare_legacy_references() + else: + passed = scene.compare_references(args.format) + + result = { + "ok": True, + "result": bool(passed), + "regression_failed": bool(scene.regression_failed), + "nbr_tested_frame": int(scene.nbr_tested_frame), + "total_run_time": int(scene.total_run_time), + "error_by_dof": [float(v) for v in scene.error_by_dof], + "total_error": [float(v) for v in scene.total_error], + "error": None, + } + except Exception as e: + import traceback + result = {"ok": False, "error": str(e), "traceback": traceback.format_exc()} + finally: + try: + with open(args.result_file, "w") as f: + json.dump(result, f) + except Exception: + pass + + # The outcome is communicated through the result file, so always exit 0 + # unless the result could not be written at all. + sys.exit(0) + + +if __name__ == "__main__": + _worker_main()