From 806ba52117e35d927bf2d80ce894141a9eeca73a Mon Sep 17 00:00:00 2001 From: Gautzilla <72027971+Gautzilla@users.noreply.github.com> Date: Mon, 31 Aug 2026 12:02:20 +0200 Subject: [PATCH 1/7] add Project.get_json_paths method --- src/osekit/public/project.py | 53 +++++++++++++++++++++++++++++++----- 1 file changed, 46 insertions(+), 7 deletions(-) diff --git a/src/osekit/public/project.py b/src/osekit/public/project.py index ba8e64d1..6e9f3281 100644 --- a/src/osekit/public/project.py +++ b/src/osekit/public/project.py @@ -30,7 +30,7 @@ get_umask, locked, ) -from osekit.utils.path import move_tree, ensure_within_base +from osekit.utils.path import ensure_within_base, move_tree if TYPE_CHECKING: from collections.abc import Iterable @@ -522,7 +522,7 @@ def _get_audio_dataset_subpath( def export( self, output_type: OutputType, - ads: AudioDataset | None = None, + ads: AudioDataset, sds: SpectroDataset | LTASDataset | None = None, subtype: str | None = None, spectrum_folder_name: str = "spectrum", @@ -608,12 +608,11 @@ def export( nb_batches=nb_jobs, ) - ads_json = ( - ads.folder / f"{ads.name}.json" - if OutputType.AUDIO in output_type - else "None" + ads_json, sds_json = Project.get_json_paths( + audio_dataset=ads, + spectro_dataset=sds, + output_type=output_type, ) - sds_json = sds.folder / f"{sds.name}.json" if sds is not None else "None" for index, (start, stop) in enumerate(batch_indexes): self.job_builder.create_job( @@ -641,6 +640,46 @@ def export( ) self.job_builder.submit_pbs() + @staticmethod + def get_json_paths( + audio_dataset: AudioDataset, + spectro_dataset: SpectroDataset | None, + output_type: OutputType, + ) -> tuple[Path | str, Path | str]: + """Return the paths of the audio and spectro output JSON files. + + Parameters + ---------- + audio_dataset: AudioDataset + The ``AudioDataset`` the transform is based on. + spectro_dataset: SpectroDataset | None + The ``SpectroDataset`` that is output by the transform. + ``None`` if the transform is audio-only. + output_type: OutputType + The ``OutputType`` of the transform. + + Returns + ------- + tuple[Path | str, Path | str]: + Paths of the audio and spectro output JSON files, respectively. + If there is no output dataset for the given ``OutputType``, + the corresponding path in the tuple is replaced with "None". + + """ + ads_json = ( + audio_dataset.folder / f"{audio_dataset.name}.json" + if OutputType.AUDIO in output_type + else "None" + ) + + sds_json = ( + spectro_dataset.folder / f"{spectro_dataset.name}.json" + if spectro_dataset is not None + else "None" + ) + + return ads_json, sds_json + def _add_spectro_dataset( self, sds: SpectroDataset | LTASDataset, From 50c8440d55169524189f78d5e1e262f850a15041 Mon Sep 17 00:00:00 2001 From: Gautzilla <72027971+Gautzilla@users.noreply.github.com> Date: Mon, 31 Aug 2026 12:13:02 +0200 Subject: [PATCH 2/7] add initial test for get_json_paths --- tests/test_public_api.py | 40 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/tests/test_public_api.py b/tests/test_public_api.py index 842df70a..3e4966ae 100644 --- a/tests/test_public_api.py +++ b/tests/test_public_api.py @@ -1840,3 +1840,43 @@ def test_run_transform_with_same_name_in_different_process( transform.output_type = OutputType.SPECTROGRAM with pytest.raises(FileExistsError, match="already exists"): project.run(transform=transform) + + +@pytest.mark.parametrize( + ("ads_folder_and_name", "sds_folder_and_name", "output_type", "expected"), + [ + pytest.param( + None, + None, + OutputType.SPECTROGRAM, + ("None", "None"), + id="both_to_none", + ), + ], +) +def test_get_json_paths( + monkeypatch: pytest.MonkeyPatch, + ads_folder_and_name: tuple[Path, str] | None, + sds_folder_and_name: tuple[Path, str] | None, + output_type: OutputType, + expected: tuple[Path | str, Path | str], +) -> None: + class DummyDataset: + def __init__(self, folder: Path, name: str) -> None: + self.folder = folder + self.name = name + + monkeypatch.setattr("osekit.public.project.AudioDataset", DummyDataset) + monkeypatch.setattr("osekit.public.project.SpectroDataset", DummyDataset) + + ads = AudioDataset(*ads_folder_and_name) if ads_folder_and_name else None + sds = AudioDataset(*sds_folder_and_name) if ads_folder_and_name else None + + assert ( + Project.get_json_paths( + audio_dataset=ads, + spectro_dataset=sds, + output_type=output_type, + ) + == expected + ) From 5039f190d5264aae779bf33f94869aae024e6fb1 Mon Sep 17 00:00:00 2001 From: Gautzilla <72027971+Gautzilla@users.noreply.github.com> Date: Mon, 31 Aug 2026 13:06:14 +0200 Subject: [PATCH 3/7] add tests for get_json_paths --- tests/test_public_api.py | 31 ++++++++++++++++++++++++++----- 1 file changed, 26 insertions(+), 5 deletions(-) diff --git a/tests/test_public_api.py b/tests/test_public_api.py index 3e4966ae..478dab16 100644 --- a/tests/test_public_api.py +++ b/tests/test_public_api.py @@ -1846,17 +1846,38 @@ def test_run_transform_with_same_name_in_different_process( ("ads_folder_and_name", "sds_folder_and_name", "output_type", "expected"), [ pytest.param( - None, + (Path("cool"), "cool_ads"), None, OutputType.SPECTROGRAM, ("None", "None"), - id="both_to_none", + id="no_audio_output_is_none", + ), + pytest.param( + (Path("cool"), "cool_ads"), + None, + OutputType.AUDIO, + (Path(r"cool/cool_ads.json"), "None"), + id="audio_json_only", + ), + pytest.param( + (Path("cool"), "cool_ads"), + (Path("fun"), "fun_sds"), + OutputType.SPECTROGRAM, + ("None", Path(r"fun/fun_sds.json")), + id="spectro_json_only", + ), + pytest.param( + (Path("cool"), "cool_ads"), + (Path("fun"), "fun_sds"), + OutputType.AUDIO | OutputType.SPECTROGRAM, + (Path(r"cool/cool_ads.json"), Path(r"fun/fun_sds.json")), + id="both_ads_and_sds_jsons", ), ], ) def test_get_json_paths( monkeypatch: pytest.MonkeyPatch, - ads_folder_and_name: tuple[Path, str] | None, + ads_folder_and_name: tuple[Path, str], sds_folder_and_name: tuple[Path, str] | None, output_type: OutputType, expected: tuple[Path | str, Path | str], @@ -1869,8 +1890,8 @@ def __init__(self, folder: Path, name: str) -> None: monkeypatch.setattr("osekit.public.project.AudioDataset", DummyDataset) monkeypatch.setattr("osekit.public.project.SpectroDataset", DummyDataset) - ads = AudioDataset(*ads_folder_and_name) if ads_folder_and_name else None - sds = AudioDataset(*sds_folder_and_name) if ads_folder_and_name else None + ads = DummyDataset(*ads_folder_and_name) if ads_folder_and_name else None + sds = DummyDataset(*sds_folder_and_name) if sds_folder_and_name else None assert ( Project.get_json_paths( From d53baaab6a027b759d532ba2b9f14047d0a86f43 Mon Sep 17 00:00:00 2001 From: Gautzilla <72027971+Gautzilla@users.noreply.github.com> Date: Mon, 31 Aug 2026 13:30:43 +0200 Subject: [PATCH 4/7] add Project.create_jobs() method --- src/osekit/public/project.py | 93 +++++++++++++++++++++++++----------- 1 file changed, 64 insertions(+), 29 deletions(-) diff --git a/src/osekit/public/project.py b/src/osekit/public/project.py index 6e9f3281..ba79acbf 100644 --- a/src/osekit/public/project.py +++ b/src/osekit/public/project.py @@ -603,41 +603,37 @@ def export( ) return - batch_indexes = file_indexes_per_batch( - total_nb_files=len(ads.data), - nb_batches=nb_jobs, - ) - ads_json, sds_json = Project.get_json_paths( audio_dataset=ads, spectro_dataset=sds, output_type=output_type, ) - for index, (start, stop) in enumerate(batch_indexes): - self.job_builder.create_job( - script_path=Path(export_transform.__file__), - script_args={ - "output-type": output_type.value, - "ads-json": ads_json, - "sds-json": sds_json, - "subtype": subtype, - "spectrum-folder-path": spectrum_folder_path, - "spectrogram-folder-path": spectrogram_folder_path, - "welch-folder-path": welch_folder_path, - "first": start, - "last": stop, - "downsampling-quality": resample_quality_settings["downsample"], - "upsampling-quality": resample_quality_settings["upsample"], - "umask": get_umask(), - "multiprocessing": config.multiprocessing["is_active"], - "nb-processes": config.multiprocessing["nb_processes"], - "use-logging-setup": True, - "dataset-json-path": self.folder / "project.json", - }, - name=name + (f"_{index}" if len(batch_indexes) > 1 else ""), - output_folder=self.folder / self.SUBFOLDERS["log"], - ) + script_args = { + "output-type": output_type.value, + "ads-json": ads_json, + "sds-json": sds_json, + "subtype": subtype, + "spectrum-folder-path": spectrum_folder_path, + "spectrogram-folder-path": spectrogram_folder_path, + "welch-folder-path": welch_folder_path, + "downsampling-quality": resample_quality_settings["downsample"], + "upsampling-quality": resample_quality_settings["upsample"], + "umask": get_umask(), + "multiprocessing": config.multiprocessing["is_active"], + "nb-processes": config.multiprocessing["nb_processes"], + "use-logging-setup": True, + "dataset-json-path": self.folder / "project.json", + } + + self.create_jobs( + audio_dataset=ads, + script_path=Path(export_transform.__file__), + script_args=script_args, + job_name=name, + nb_jobs=nb_jobs, + ) + self.job_builder.submit_pbs() @staticmethod @@ -680,6 +676,45 @@ def get_json_paths( return ads_json, sds_json + def create_jobs( + self, + audio_dataset: AudioDataset, + script_path: Path, + script_args: dict, + job_name: str, + nb_jobs: int = 1, + ) -> None: + """Create the jobs corresponding to each batch. + + Parameters + ---------- + audio_dataset: AudioDataset + The ``AudioDataset`` the transform is based on. + script_path: Path + Path to the export script. + script_args: dict + Arguments passed to the export script. + job_name: str + Name of the job. + If there are multiple batches, each batch will be suffixed + with "_{index}". + nb_jobs: int + Number of batches used to run the transform. + Each batch will run in a separate job. + + """ + batch_indexes = file_indexes_per_batch( + total_nb_files=len(audio_dataset.data), + nb_batches=nb_jobs, + ) + for index, (start, stop) in enumerate(batch_indexes): + self.job_builder.create_job( + script_path=script_path, + script_args=script_args | {"first": start, "last": stop}, + name=job_name + (f"_{index}" if len(batch_indexes) > 1 else ""), + output_folder=self.folder / self.SUBFOLDERS["log"], + ) + def _add_spectro_dataset( self, sds: SpectroDataset | LTASDataset, From 3c1a5f2e13a7e3508541bcceeb66085639856608 Mon Sep 17 00:00:00 2001 From: Gautzilla <72027971+Gautzilla@users.noreply.github.com> Date: Mon, 31 Aug 2026 13:41:56 +0200 Subject: [PATCH 5/7] =?UTF-8?q?move=20create=5Fjobs(=C3=83=C3=83)=20to=20J?= =?UTF-8?q?obBuilder=20class?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/osekit/public/project.py | 43 ++------------------------------- src/osekit/utils/job.py | 47 +++++++++++++++++++++++++++++++++++- 2 files changed, 48 insertions(+), 42 deletions(-) diff --git a/src/osekit/public/project.py b/src/osekit/public/project.py index ba79acbf..3e14bdac 100644 --- a/src/osekit/public/project.py +++ b/src/osekit/public/project.py @@ -26,7 +26,6 @@ from osekit.core.spectro_dataset import SpectroDataset from osekit.public.transform import OutputType, Transform from osekit.utils.core import ( - file_indexes_per_batch, get_umask, locked, ) @@ -626,11 +625,12 @@ def export( "dataset-json-path": self.folder / "project.json", } - self.create_jobs( + self.job_builder.create_jobs( audio_dataset=ads, script_path=Path(export_transform.__file__), script_args=script_args, job_name=name, + output_folder=self.folder / self.SUBFOLDERS["log"], nb_jobs=nb_jobs, ) @@ -676,45 +676,6 @@ def get_json_paths( return ads_json, sds_json - def create_jobs( - self, - audio_dataset: AudioDataset, - script_path: Path, - script_args: dict, - job_name: str, - nb_jobs: int = 1, - ) -> None: - """Create the jobs corresponding to each batch. - - Parameters - ---------- - audio_dataset: AudioDataset - The ``AudioDataset`` the transform is based on. - script_path: Path - Path to the export script. - script_args: dict - Arguments passed to the export script. - job_name: str - Name of the job. - If there are multiple batches, each batch will be suffixed - with "_{index}". - nb_jobs: int - Number of batches used to run the transform. - Each batch will run in a separate job. - - """ - batch_indexes = file_indexes_per_batch( - total_nb_files=len(audio_dataset.data), - nb_batches=nb_jobs, - ) - for index, (start, stop) in enumerate(batch_indexes): - self.job_builder.create_job( - script_path=script_path, - script_args=script_args | {"first": start, "last": stop}, - name=job_name + (f"_{index}" if len(batch_indexes) > 1 else ""), - output_folder=self.folder / self.SUBFOLDERS["log"], - ) - def _add_spectro_dataset( self, sds: SpectroDataset | LTASDataset, diff --git a/src/osekit/utils/job.py b/src/osekit/utils/job.py index edb78931..6d47406b 100644 --- a/src/osekit/utils/job.py +++ b/src/osekit/utils/job.py @@ -14,6 +14,9 @@ from pandas import Timedelta +from osekit.core.audio_dataset import AudioDataset +from osekit.utils.core import file_indexes_per_batch + if TYPE_CHECKING: from pathlib import Path @@ -97,7 +100,7 @@ def __init__( """ config = JobConfig() if config is None else config self.script_path = script_path - self.script_args = script_args if script_args else {} + self.script_args = script_args or {} self.nb_nodes = config.nb_nodes self.ncpus = config.ncpus self.ngpus = config.ngpus @@ -535,6 +538,48 @@ def __init__(self, config: JobConfig = JobConfig) -> None: self.config = config self.jobs = [] + def create_jobs( + self, + audio_dataset: AudioDataset, + script_path: Path, + script_args: dict, + output_folder: Path, + job_name: str = "osekit_transform", + nb_jobs: int = 1, + ) -> None: + """Create the jobs corresponding to each batch. + + Parameters + ---------- + audio_dataset: AudioDataset + The ``AudioDataset`` the transform is based on. + script_path: Path + Path to the export script. + script_args: dict + Arguments passed to the export script. + job_name: str + Name of the job. + If there are multiple batches, each batch will be suffixed + with "_{index}". + output_folder: Path + Folder in which the job output log files are saved. + nb_jobs: int + Number of batches used to run the transform. + Each batch will run in a separate job. + + """ + batch_indexes = file_indexes_per_batch( + total_nb_files=len(audio_dataset.data), + nb_batches=nb_jobs, + ) + for index, (start, stop) in enumerate(batch_indexes): + self.create_job( + script_path=script_path, + script_args=script_args | {"first": start, "last": stop}, + name=job_name + (f"_{index}" if len(batch_indexes) > 1 else ""), + output_folder=output_folder, + ) + def create_job( self, script_path: Path, From 566e6a62b2856f7cb3327e7fd4349b2e0f19d393 Mon Sep 17 00:00:00 2001 From: Gautzilla <72027971+Gautzilla@users.noreply.github.com> Date: Mon, 31 Aug 2026 14:39:58 +0200 Subject: [PATCH 6/7] change create_jobs() ads parameter to nb_tasks --- src/osekit/public/project.py | 2 +- src/osekit/utils/job.py | 9 ++++----- 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/src/osekit/public/project.py b/src/osekit/public/project.py index 3e14bdac..ff9fff0c 100644 --- a/src/osekit/public/project.py +++ b/src/osekit/public/project.py @@ -626,7 +626,7 @@ def export( } self.job_builder.create_jobs( - audio_dataset=ads, + nb_tasks=len(ads.data), script_path=Path(export_transform.__file__), script_args=script_args, job_name=name, diff --git a/src/osekit/utils/job.py b/src/osekit/utils/job.py index 6d47406b..63494ae8 100644 --- a/src/osekit/utils/job.py +++ b/src/osekit/utils/job.py @@ -14,7 +14,6 @@ from pandas import Timedelta -from osekit.core.audio_dataset import AudioDataset from osekit.utils.core import file_indexes_per_batch if TYPE_CHECKING: @@ -540,7 +539,7 @@ def __init__(self, config: JobConfig = JobConfig) -> None: def create_jobs( self, - audio_dataset: AudioDataset, + nb_tasks: int, script_path: Path, script_args: dict, output_folder: Path, @@ -551,8 +550,8 @@ def create_jobs( Parameters ---------- - audio_dataset: AudioDataset - The ``AudioDataset`` the transform is based on. + nb_tasks: + The number of tasks that are distributed across ``nb_jobs`` jobs. script_path: Path Path to the export script. script_args: dict @@ -569,7 +568,7 @@ def create_jobs( """ batch_indexes = file_indexes_per_batch( - total_nb_files=len(audio_dataset.data), + total_nb_files=nb_tasks, nb_batches=nb_jobs, ) for index, (start, stop) in enumerate(batch_indexes): From 73d08de0f0a86741c7994fdd521f22b3a164faea Mon Sep 17 00:00:00 2001 From: Gautzilla <72027971+Gautzilla@users.noreply.github.com> Date: Mon, 31 Aug 2026 15:28:33 +0200 Subject: [PATCH 7/7] =?UTF-8?q?add=20tests=20for=20JobBuilder.create=C3=83?= =?UTF-8?q?=5Fjobs()?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_job.py | 94 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 94 insertions(+) diff --git a/tests/test_job.py b/tests/test_job.py index 76633663..f3b4c842 100644 --- a/tests/test_job.py +++ b/tests/test_job.py @@ -703,3 +703,97 @@ def test_build_dependency_string_with_different_types( def test_job_walltime(walltime: str | Timedelta) -> None: job = Job(Path(), config=JobConfig(walltime=walltime)) assert Timedelta(job.walltime_str) == Timedelta(walltime) + + +@pytest.mark.parametrize( + ( + "nb_tasks", + "script_path", + "script_args", + "output_folder", + "job_name", + "nb_jobs", + "expected_task_indexes", + ), + [ + pytest.param( + 10, + Path("path/to/script.py"), + {"int_arg": 1, "str_arg": "cool"}, + Path("path/to/output"), + "cool_name", + 1, + [(0, 10)], + id="one_job_covers_all_tasks", + ), + pytest.param( + 10, + Path("path/to/script.py"), + {"int_arg": 1, "str_arg": "cool"}, + Path("path/to/output"), + "cool_name", + 5, + [(0, 2), (2, 4), (4, 6), (6, 8), (8, 10)], + id="tasks_are_equally_distributed", + ), + ], +) +def test_create_jobs( # noqa: PLR0917 + monkeypatch: pytest.MonkeyPatch, + nb_tasks: int, + script_path: Path, + script_args: dict, + output_folder: Path, + job_name: str, + nb_jobs: int, + expected_task_indexes: list[tuple[int, int]], +) -> None: + created_jobs = {} + + def patch_create_job(self: JobBuilder, **kwargs: str) -> None: + job_name = kwargs.pop("name") + created_jobs[job_name] = kwargs + + monkeypatch.setattr(JobBuilder, "create_job", patch_create_job) + + JobBuilder().create_jobs( + nb_tasks=nb_tasks, + script_path=script_path, + script_args=script_args, + output_folder=output_folder, + job_name=job_name, + nb_jobs=nb_jobs, + ) + + # Correct number of jobs + assert len(created_jobs) == nb_jobs + + # Correct distribution across jobs + for job in created_jobs.values(): + assert ( + job["script_args"]["first"], + job["script_args"]["last"], + ) in expected_task_indexes + + # Script path + assert all(job["script_path"] == script_path for job in created_jobs.values()) + + # Script args + for job in created_jobs.values(): + for arg in script_args: + assert arg in job["script_args"] + + # Output folder + assert all(job["output_folder"] == output_folder for job in created_jobs.values()) + + # Job names + if nb_jobs == 1: + assert np.array_equal(list(created_jobs.keys()), [job_name]) + else: + for idx, job in enumerate( + sorted( + created_jobs.items(), + key=lambda kvp: kvp[1]["script_args"]["first"], + ), + ): + assert job[0] == f"{job_name}_{idx}"