Commit 1b23b6de authored by Fabian Müller's avatar Fabian Müller
Browse files

Add collection scripts for P.800 characterization processing

parent 7150b926
Loading
Loading
Loading
Loading
Loading
+36 −0
Original line number Diff line number Diff line
# Characterization Collection Scripts

These scripts collect the final listening test material from the `proc_output` directories every test generates into
a `proc_final` directory next to it.


## P.800 Quickstart

First, install Python 3 on your computer.

To collect the categorized samples and the preliminaries, run the following command from the repository root in a
terminal:

```shell
# on Windows
python collection-scripts\characterization\p800_categorized_samples.py
python collection-scripts\characterization\p800_preliminaries.py

# on Linux/macOS/...
python collection-scripts/characterization/p800_categorized_samples.py
python collection-scripts/characterization/p800_preliminaries.py
```

Some commands may support additional parameters or options. You can run them with `-h` or `--help` to print the help
text. See below for an example (which may be out of date).

```shell
> python collection-scripts/characterization/p800_categorized_samples.py --help
usage: p800_categorized_samples.py [-h] [-v]

Copy samples into proc_final/cXX directories for P.800 experiments.

options:
  -h, --help     show this help message and exit
  -v, --verbose
```
+115 −0
Original line number Diff line number Diff line
import argparse
import glob
import re
import shutil
from pathlib import Path


# table 5 from IVAS-8b
TABLE_5 = {
    "s01": {
        "cat1": -16,
        "cat2": -36,
        "cat3": -36,
        "cat4": -26,
        "cat5": -26,
        "cat6": -16,
    },
    "s02": {
        "cat1": -16,
        "cat2": -16,
        "cat3": -36,
        "cat4": -36,
        "cat5": -26,
        "cat6": -26,
    },
    "s03": {
        "cat1": -26,
        "cat2": -16,
        "cat3": -16,
        "cat4": -36,
        "cat5": -36,
        "cat6": -26,
    },
    "s04": {
        "cat1": -26,
        "cat2": -26,
        "cat3": -16,
        "cat4": -16,
        "cat5": -36,
        "cat6": -36,
    },
    "s05": {
        "cat1": -36,
        "cat2": -26,
        "cat3": -26,
        "cat4": -16,
        "cat5": -16,
        "cat6": -36,
    },
    "s06": {
        "cat1": -36,
        "cat2": -36,
        "cat3": -26,
        "cat4": -26,
        "cat5": -16,
        "cat6": -16,
    },
}


def repo_root_dir() -> Path:
    this_dir = Path(__file__).parent
    return this_dir.parent.parent


def copy_final_items(verbose: bool = False):
    files_copied = 0

    for exp_dir in map(Path, glob.glob(str(repo_root_dir() / "experiments/characterization/P800-*/"))):
        exp_dir = Path(exp_dir)

        final_dir = exp_dir / "proc_final"
        final_dir.mkdir(parents=True, exist_ok=True)

        proc_output = exp_dir / "proc_output"

        for sample, categories_map in TABLE_5.items():
            for category, loudness in categories_map.items():
                item_dir_pattern = proc_output / glob.escape(category) / f"out_{glob.escape(str(loudness))}LKFS" / "c*"
                item_filename_pattern = f"*{glob.escape(sample)}.c*.wav"

                items_to_copy = list(map(Path, glob.glob(str(item_dir_pattern / item_filename_pattern))))

                if not items_to_copy:
                    if verbose:
                        print(f"Could not find processed item for {sample}, {category}, {loudness}LKFS, skipping")
                    continue

                for item_to_copy in items_to_copy:
                    match = re.match(r".*/c(\d+)/[^/]*s\d+\.c\d+\.wav$", str(item_to_copy.as_posix()))
                    assert match, item_to_copy
                    condition = match.group(1)

                    item_target_dir = final_dir / f"c{condition}"
                    item_target_dir.mkdir(parents=True, exist_ok=True)

                    print(f"Copying {item_to_copy.relative_to(Path.cwd())} to {item_target_dir.relative_to(Path.cwd())}")

                    if (item_target_dir / item_to_copy.name).is_file():
                        print(f"WARNING: item {item_to_copy.name} already exists in target dir, overwriting")

                    shutil.copy(item_to_copy, item_target_dir)

                    files_copied += 1

    print(f"Done. {files_copied} files copied.")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Copy samples into proc_final/cXX directories for P.800 experiments.")
    parser.add_argument("-v", "--verbose", action="store_true")

    parsed = parser.parse_args()

    copy_final_items(verbose=parsed.verbose)
+80 −0
Original line number Diff line number Diff line
import argparse
import glob
import shutil
import warnings
from pathlib import Path

import yaml


def this_dir() -> Path:
    return Path(__file__).parent


def repo_root_dir() -> Path:
    return this_dir().parent.parent


def load_config(yaml_path):
    with open(yaml_path, "r") as f:
        return yaml.safe_load(f)


# TODO: temporary map as of 2025-09-05, waiting for final confirmation
LOUDNESS_MAP = {
    "cat1": -16,
    "cat2": -26,
    "cat3": -36,
    "cat4": -16,
    "cat5": -26,
    "cat6": -36,
}


def copy_preliminaries(root_dir, config_yaml):
    config = load_config(config_yaml)

    for exp, exp_cfg in config.items():
        exp = Path(exp)
        exp_dir = root_dir / exp

        # this is a list to handle multiple proc_outputs in P.800
        proc_out_list = [
            p for p in exp_dir.iterdir() if p.name.startswith("proc_output")
        ]

        out_dir = root_dir / exp / "proc_final/preliminaries/"

        print(f"Processing experiment: {exp}")

        if out_dir.exists():
            warnings.warn(f"Preliminaries directory already exists for {exp}, removing")
            shutil.rmtree(out_dir)

        out_dir.mkdir(parents=True, exist_ok=True)

        for item in exp_cfg["preliminaries"]:
            cat = item["category"]
            cond = item["label"]

            # Find the condition folder in proc_output
            for proc_out in proc_out_list:
                src = proc_out / cat / f"out_{LOUDNESS_MAP[cat]}LKFS" / cond

                for file in glob.glob(str(src / "*s07.c*.wav")):
                    print(f"  Copying {file} -> {out_dir}")

                    if (out_dir / Path(file).name).is_file():
                        raise FileExistsError(f"File {file} already exists in {out_dir}")

                    shutil.copy(file, out_dir)

    print("All done.")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Copy preliminaries into proc_final/preliminaries directory for P.800 experiments.")

    parsed = parser.parse_args()

    copy_preliminaries(repo_root_dir() / "experiments" / "characterization", this_dir() / "p800_preliminaries.yml")
+1760 −0

File added.

Preview size limit exceeded, changes collapsed.