Commit 356ab5a5 authored by Anika Treffehn's avatar Anika Treffehn
Browse files

processing 2 now works for ISM

parent 03065a16
Loading
Loading
Loading
Loading
Loading
+0 −18
Original line number Diff line number Diff line
@@ -39,7 +39,6 @@ import numpy as np
import scipy.io.wavfile as wav

from .audioarray import trim, window
from ivas_processing_scripts.audiotools import audio

logger = logging.getLogger("__main__")
logger.setLevel(logging.DEBUG)
@@ -433,20 +432,3 @@ def parse_wave_header(
        "ext_param": ext_param,
    }

def remove_preamble(cfg):
    # get number of channels from output format
    num_channels = audio.fromtype(cfg.postprocessing["fmt"]).num_channels
    for odir in cfg.out_dirs:
        for item in cfg.items_list:
            path_input = odir / item.name
            # read file
            x, fs = read(path_input, nchannels=num_channels, fs=cfg.postprocessing["fs"])

            # remove preamble
            x = trim(x, fs, (cfg.pre2.preamble, 0))

            # write file
            write(path_input, x, fs)

    return
+38 −25
Original line number Diff line number Diff line
@@ -188,7 +188,7 @@ def write_ISM_metadata_in_file(
        List of acutally used file names
    """

    if len(metadata) != len(file_name) and not automatic_naming:
    if not automatic_naming and len(metadata) != len(file_name):
        raise ValueError("Number of metadata objects and file names has to match")
    number_objects = len(metadata)

@@ -302,7 +302,7 @@ def concat_meta_from_file(
    input_fmt: str,
    silence_pre: Optional[int] = 0,
    silence_post: Optional[int] = 0,
    preamble: Optional[int] = None,  # TODO: remove preamble here
    preamble: Optional[int] = None,
) -> None:
    """
    Concatenate ISM metadata from files
@@ -362,7 +362,7 @@ def concat_meta_from_file(
        # pad
        trim_meta(
            audio_item, (-silence_pre, -silence_post)
        )  # use negative value since we wante to pad, not trim
        )  # use negative value since we want to pad, not trim

        # concatenate
        for idx, obj_pos in enumerate(audio_item.object_pos):
@@ -374,26 +374,7 @@ def concat_meta_from_file(

    # add preamble
    if preamble:
        preamble_frames = preamble / IVAS_FRAME_LEN_MS
        if not preamble_frames.is_integer():
            raise ValueError(
                f"ISM metadata padding and trimming only possible if pad/trim length is multiple of frame length. "
                f"Frame length: {IVAS_FRAME_LEN_MS}ms"
            )
        for obj_idx in range(len(concat_meta_all_obj)):
            if (
                concat_meta_all_obj is not None
                and concat_meta_all_obj[obj_idx] is not None
            ):
                concat_meta_all_obj[obj_idx] = trim(
                    concat_meta_all_obj[obj_idx],
                    limits=(-int(preamble_frames), 0),
                    samples=True,
                )

                # add radius 1
                concat_meta_all_obj[obj_idx][: int(preamble_frames), 2] = 1
        pass
        concat_meta_all_obj = add_remove_preamble(concat_meta_all_obj, preamble)

    write_ISM_metadata_in_file(concat_meta_all_obj, out_file)

@@ -529,7 +510,7 @@ def check_ISM_metadata(


def metadata_search(
    in_meta: Union[str, Path],
    in_meta_path: Union[str, Path],
    item_names: list[Union[str, Path]],
    num_objects: int,
) -> list[list[Union[Path, str]]]:
@@ -542,7 +523,7 @@ def metadata_search(
    for item in item_names:
        list_item = []
        for obj_idx in range(num_objects):
            file_name_meta = in_meta / Path(item.stem).with_suffix(
            file_name_meta = in_meta_path / Path(item.stem).with_suffix(
                f"{item.suffix}.{obj_idx}.csv"
            )
            # check if file exists and add to list
@@ -556,3 +537,35 @@ def metadata_search(
            list_meta.append(list_item)

    return list_meta


def add_remove_preamble(
    metadata,
    preamble,
    add: Optional[bool] = True,
):
    preamble_frames = preamble / IVAS_FRAME_LEN_MS
    if not preamble_frames.is_integer():
        raise ValueError(
            f"Application of preamble for ISM metadata is only possible if preamble length is multiple of frame length. "
            f"Frame length: {IVAS_FRAME_LEN_MS}ms"
        )
    for obj_idx in range(len(metadata)):
        if metadata is not None and metadata[obj_idx] is not None:
            if add:
                metadata[obj_idx] = trim(
                    metadata[obj_idx],
                    limits=(-int(preamble_frames), 0),
                    samples=True,
                )

                # add radius 1
                metadata[obj_idx][: int(preamble_frames), 2] = 1
            else:
                metadata[obj_idx] = trim(
                    metadata[obj_idx],
                    limits=(int(preamble_frames), 0),
                    samples=True,
                )

    return metadata
+11 −2
Original line number Diff line number Diff line
@@ -37,9 +37,11 @@ import numpy as np
from ivas_processing_scripts.processing.processing import Processing
from ivas_processing_scripts.audiotools.audiofile import read, write
from ivas_processing_scripts.audiotools.audioarray import trim
from ivas_processing_scripts.audiotools import audio
from ivas_processing_scripts.audiotools.metadata import add_remove_preamble, write_ISM_metadata_in_file


class Preprocessing2(Processing):  # TODO
class Preprocessing2(Processing):
    def __init__(self, attrs: dict):
        super().__init__(attrs)
        self.name = "pre_2"
@@ -49,10 +51,17 @@ class Preprocessing2(Processing): # TODO
        logger.debug(f"Preprocessing2 {in_file.absolute()} -> {out_file.absolute()}")

        # load in file
        x, fs = read(in_file, self.in_fs, self.in_fmt)
        number_channels = audio.fromtype(self.in_fmt).num_channels
        x, fs = read(in_file, fs=self.in_fs, nchannels=number_channels)

        # add preamble
        if self.preamble:
            if self.in_fmt.startswith("ISM"):
                metadata = []
                for meta in in_meta:
                    metadata.append(np.genfromtxt(meta, delimiter=","))
                metadata = add_remove_preamble(metadata, self.preamble)
                write_ISM_metadata_in_file(metadata, [out_file], True)
            x = trim(x, fs, (-self.preamble, 0), self.pad_noise_preamble)

        # add background noise
+51 −7
Original line number Diff line number Diff line
@@ -37,12 +37,16 @@ from pathlib import Path
from shutil import copyfile
from typing import Iterable, Union
from warnings import warn
import numpy as np

from ivas_processing_scripts.audiotools import audio
from ivas_processing_scripts.audiotools.audiofile import concat, split, remove_preamble
from ivas_processing_scripts.audiotools.audiofile import concat, split, read, trim, write
from ivas_processing_scripts.audiotools.metadata import (
    concat_meta_from_file,
    split_meta_in_file,
    metadata_search,
    add_remove_preamble,
    write_ISM_metadata_in_file,
)
from ivas_processing_scripts.constants import LOGGER_DATEFMT, LOGGER_FORMAT
from ivas_processing_scripts.processing.config import TestConfig
@@ -95,11 +99,10 @@ def concat_setup(cfg: TestConfig, chain, logger: logging.Logger):
        cfg.concat_meta = []
        for obj_idx in range(len(cfg.metadata_path[0])):
            cfg.concat_meta.append(
                cfg.output_path.joinpath(
                cfg.tmp_dirs[0].joinpath(
                    f"{cfg.input_path.name}_concatenated.wav.{obj_idx}.csv"
                )
            )
        # TODO: check this
        concat_meta_from_file(
            cfg.items_list,
            cfg.metadata_path,
@@ -111,7 +114,7 @@ def concat_setup(cfg: TestConfig, chain, logger: logging.Logger):
        cfg.metadata_path = [cfg.concat_meta]

    # concatenate audio
    cfg.concat_file = cfg.output_path.joinpath(
    cfg.concat_file = cfg.tmp_dirs[0].joinpath(
        f"{cfg.input_path.name}_concatenated.wav"
    )

@@ -269,7 +272,7 @@ def preprocess_2(cfg, logger):
def reverse_process_2(cfg, logger):

    # remove preamble
    if cfg.pre2.preamble:
    if cfg.pre2.preamble:  # TODO: remove preamble for ISM
        remove_preamble(cfg)

    # reverse concatenation
@@ -277,9 +280,16 @@ def reverse_process_2(cfg, logger):
        # write out the splits, optionally remove file
        out_paths_splits, out_meta_splits = concat_teardown(cfg, logger)
    else:
        out_path_splits = 0  # TODO: list of files
        # if no concatenation read files from folder
        out_paths_splits = []
        for out_dir in cfg.out_dirs:
            list_audio_dir = list_audio(out_dir, absolute=True)
            out_paths_splits.append(list_audio_dir)
        if cfg.postprocessing["fmt"].startswith("ISM"):
            out_meta_splits = 0  # TODO: list of files
            out_meta_splits = []
            for i, condition in enumerate(out_paths_splits):
                meta_condition = metadata_search(cfg.out_dirs[i], condition, num_objects=int(cfg.postprocessing["fmt"][-1]))
                out_meta_splits.append(meta_condition)
        else:
            out_meta_splits = None

@@ -369,3 +379,37 @@ def process_item(
    if processing_paths_meta[-1]:
        for idx, ppm in enumerate(processing_paths_meta[-1]):
            copyfile(ppm, out_meta[idx])


def remove_preamble(cfg):
    # get number of channels from output format
    num_channels = audio.fromtype(cfg.postprocessing["fmt"]).num_channels
    for odir in cfg.out_dirs:
        for item in cfg.items_list:
            path_input = odir / item.name

            # remove preamble for ISM metadata
            if cfg.postprocessing["fmt"].startswith("ISM"):

                # search for metadata
                meta_item = metadata_search(odir, [Path(item.name)], num_objects=num_channels)
                metadata_array = []
                for meta_i in meta_item:
                    metadata_array.append(np.genfromtxt(meta_i, delimiter=","))

                # remove preamble
                metadata_array = add_remove_preamble(metadata_array, cfg.pre2.preamble, add=False)

                # write csv files
                write_ISM_metadata_in_file(metadata_array, [path_input], automatic_naming=True)

            # read file
            x, fs = read(path_input, nchannels=num_channels, fs=cfg.postprocessing["fs"])

            # remove preamble
            x = trim(x, fs, (cfg.pre2.preamble, 0))

            # write file
            write(path_input, x, fs)

    return