Merge branch 'tests' into main

src/OSmOSE/cluster/audio_reshaper.py

@@ -202,6 +202,9 @@ def reshape(
     while i < len(files):
         audio_data, sample_rate = sf.read(input_dir_path.joinpath(files[i]))
         file_duration = len(audio_data)//sample_rate
+
+        if not merge_files and file_duration < chunk_size:
+            raise ValueError("When not merging files, the file duration must be smaller than the target duration.")
 
         if overwrite and not implicit_output and output_dir_path == input_dir_path and output_dir_path == input_dir_path and i<len(files)-1:
             print(f"Deleting {files[i]}")
@@ -566,7 +569,7 @@ def reshape(
         "--no-merge",
         action="store_false",
         help="Don't try to merge the reshaped files."
-    )
+    ) # When absent = we merge file; when present = we don't merge -> merge_file is False
 
     args = parser.parse_args()
 

src/OSmOSE/job.py

@@ -250,12 +250,15 @@ def build_job_file(
             The path to the created job file.
         """
         set_umask()
-        if preset and preset.lower() not in self.__config.Presets._fields:
-            raise ValueError(
-                f"Unrecognized preset {preset}. Valid presets are: {', '.join(self.__config.Presets._fields)}"
-            )
+        if "Presets" in self.__config._fields:
+            if preset and preset.lower() not in self.__config.Presets._fields:
+                raise ValueError(
+                    f"Unrecognized preset {preset}. Valid presets are: {', '.join(self.__config.Presets._fields)}"
+                )
 
-        job_preset = getattr(self.__config.Presets, preset)
+            job_preset = getattr(self.__config.Presets, preset)
+        else:
+            preset = None
 
         pwd = Path(__file__).parent
 

src/OSmOSE/timestamps.py

@@ -48,13 +48,11 @@ def write_timestamp(
     audio_path: str,
     date_template: str,
     timezone: str = "UTC",
-    offsets: tuple = None,
+    offset: tuple = None,
 ):
     """Read the dates in the filenames of audio files in the `audio_path` folder,
     according to the date template in strftime format or the offsets from the beginning and end of the date.
 
-    If both `date_template` and `offsets` are provided, the latter has priority and `date_template` is ignored.
-
     The result is written in a file named `timestamp.csv` with no header and two columns in this format : [filename],[timestamp].
     The output date is in the template `'%Y-%m-%dT%H:%M:%S.%fZ'.
 
@@ -71,11 +69,6 @@ def write_timestamp(
             a tuple containing the beginning and end offset of the date.
             The first element is the first character of the date, and the second is the last.
     """
-    if offsets:
-        if "-" in offsets:
-            offset = [int(off) for off in offsets.split("-")]
-        else:
-            offset = [int(offsets), 0]
     # TODO: extension-agnostic
     list_audio_file = sorted([file for file in Path(audio_path).glob("*.wav")])
 
@@ -89,15 +82,15 @@ def write_timestamp(
 
     converted = convert_template_to_re(date_template)
     for filename in list_audio_file:
-        if offsets:
-            date_extracted = filename.stem[offset[0] : offset[1] + 1]
-        else:
-            try:
+        try:
+            if offset:
+                date_extracted = re.search(converted, filename.stem[offset[0] : offset[1] + 1])[0]
+            else:
                 date_extracted = re.search(converted, str(filename))[0]
-            except TypeError:
-                raise ValueError(
-                    f"The date template does not match any set of character in the file name {filename}\nMake sure you are not forgetting separator characters, or use the offsets parameter."
-                )
+        except TypeError:
+            raise ValueError(
+                f"The date template does not match any set of character in the file name {filename}\nMake sure you are not forgetting separator characters, or use the offsets parameter."
+            )
 
         date_obj = datetime.datetime.strptime(date_extracted, date_template)
         dates = datetime.datetime.strftime(date_obj, "%Y-%m-%dT%H:%M:%S.%f")
@@ -145,9 +138,17 @@ def write_timestamp(
     )
     args = argparser.parse_args()
 
+    if args.offset and "-" in args.offset:
+        split = args.offset.split("-")
+        offset = (int(split[0]), int(split[1]))
+    elif args.offset:
+        offset = int(args.offset)
+    else:
+        offset = None
+
     write_timestamp(
         audio_path=args.dataset_name,
         date_template=args.date_template,
         timezone=args.timezone,
-        offsets=args.offset,
+        offsets=offset,
     )

tests/test_create_timestamps.py

@@ -1,7 +1,6 @@
 import OSmOSE.timestamps as tm
 import re
-import importlib
-
+import pytest
 
 def test_convert_template_to_re():
     raw_all = "".join(tm.__converter.keys())
@@ -20,8 +19,26 @@ def test_convert_template_to_re():
 
 
 # a monkeypatch
-def test_write_timestamp():
+def test_write_timestamp(tmp_path):
     true_template = "%d%m%y_%H%M%S"
-    true_offsets = (8, 4)
+    bad_template = "%Y%I%S%p"
+    true_offsets = (5, 4)
+    expected_result = []
+    resfile = tmp_path.joinpath("timestamp.csv")
+
+    for i in range(10):
+        filename = f"test_120723_1815{str(3*i).zfill(2)}.wav"
+        open(tmp_path.joinpath(filename), "w").close()
+        expected_result.append(f"{filename},2023-07-12T18:15:{str(3*i).zfill(2)}.000Z,UTC\n")
+
+    tm.write_timestamp(audio_path=tmp_path, date_template = true_template)
+
+    with open(resfile, "r") as f:
+        assert expected_result == f.readlines()
+    resfile.unlink()
+
+    with pytest.raises(ValueError) as excinfo:
+        tm.write_timestamp(audio_path=tmp_path, date_template = bad_template)
+
+    assert "The date template does not match" in str(excinfo.value)
 
-    # TODO

tests/test_job.py

@@ -0,0 +1,41 @@
+import os
+from pathlib import Path
+
+import pytest
+from OSmOSE import Job_builder
+from OSmOSE.utils import read_config, convert
+
+custom_config = convert({
+    "job_scheduler":"Torque",
+    "env_script":"conda activate",
+    "env_name":"osmose",
+    "queue":"normal",
+    "nodes":2,
+    "walltime":"12:00:00",
+    "ncpus":16,
+    "mem":"42G",
+    "outfile":"%j.out",
+    "errfile":"%j.err"
+})
+
+pbshead = """#!/bin/bash
+#PBS -N test_job
+#PBS -q normal
+#PBS -l nodes=2
+#PBS -l ncpus=16
+#PBS -l walltime=12:00:00
+#PBS -l mem=42G"""
+
+def test_build_job_file(output_dir):
+    script_path = "/path/to/script"
+    script_args = "--arg1 value1 --arg2 value2"
+    jobname = "test_job"
+
+    jb = Job_builder()
+    jb._Job_builder__config = custom_config
+    jb.build_job_file(script_path = script_path, script_args=script_args, jobname=jobname, logdir=output_dir)
+
+    with open(jb.prepared_jobs[0]["path"], "r") as f:
+        text = "".join(f.readlines())
+
+    assert pbshead in text

tests/test_reshaper.py

@@ -321,4 +321,4 @@ def test_reshape_max_delta_interval(input_reshape: Path, output_dir: Path, monke
             last_file_behavior="pad",
             verbose=True
         )
-    assert str(e.value) == "Error: Cannot merge non-continuous audio files if force_reshape is false."
+    assert str(e.value) == "Error: Cannot merge non-continuous audio files if force_reshape is false."

tests/test_spectrogram.py

@@ -190,3 +190,6 @@ def test_initialize_2s(input_dataset):
     )
 
     assert np.allclose(full_input, full_output)
+
+def test_process_file_adjust(input_spectrogram, output_dir):
+