bobby/open_dbm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix code refactoring in hnr

Browse Source
This commit is contained in:
jordi.hasianta
2022-09-15 21:33:19 +07:00
parent 2498e93e20
commit 4218ea4bef
1 changed files with 49 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

144
opendbm/dbm_lib/dbm_features/raw_features/audio/hnr.py
View File

@@ -1,5 +1,5 @@
""" """
file_name: gne file_name: hnr
project_name: DBM project_name: DBM
created: 2020-20-07 created: 2020-20-07
""" """
@@ -9,7 +9,6 @@ import logging
import os import os
from os.path import join from os.path import join
import more_itertools as mit
import numpy as np import numpy as np
import pandas as pd import pandas as pd
import parselmouth import parselmouth
@@ -19,115 +18,70 @@ from opendbm.dbm_lib.dbm_features.raw_features.util import util as ut
logging.basicConfig(level=logging.INFO) logging.basicConfig(level=logging.INFO)
logger = logging.getLogger() logger = logging.getLogger()
gne_dir = "acoustic/glottal_noise" hnr_dir = "acoustic/harmonic_noise"
ff_dir = "acoustic/pitch" csv_ext = "_hnr.csv"
csv_ext = "_gne.csv" error_txt = "error: length less than 0.064"
def gne_ratio(sound): def hnr_ratio(filepath):
""" """
Using parselmouth library fetching glottal noise excitation ratio Using parselmouth library fetching harmonic noise ratio ratio
Args: Args:
sound: parselmouth object path: (.wav) audio file location
Returns: Returns:
(list) list of gne ratio for each voice frame (list) list of hnr ratio for each voice frame, min,max and mean hnr
""" """
harmonicity_gne = sound.to_harmonicity_gne() sound = parselmouth.Sound(filepath)
gne_all_bands = harmonicity_gne.values harmonicity = sound.to_harmonicity_ac(time_step=0.001)
gne_all_bands = np.where(gne_all_bands == -200, np.NaN, gne_all_bands)
gne = np.nanmax( hnr_all_frames = harmonicity.values # [harmonicity.values != -200] nan it (****)
gne_all_bands hnr_all_frames = np.where(hnr_all_frames == -200, np.NaN, hnr_all_frames)
) # following http://www.fon.hum.uva.nl/rob/NKI_TEVA/TEVA/HTML/NKI_TEVA.pdf return hnr_all_frames.transpose()
return gne
def empty_gne(video_uri, out_loc, fl_name, r_config, error_txt, save=True): def calc_hnr(video_uri, audio_file, out_loc, fl_name, r_config, save=True):
""" """
Preparing empty GNE matrix if something fails Preparing harmonic noise matrix
"""
cols = ["Frames", r_config.aco_gne, r_config.err_reason]
out_val = [[np.nan, np.nan, error_txt]]
df_gne = pd.DataFrame(out_val, columns=cols)
df_gne["dbm_master_url"] = video_uri
if save:
logger.info("Saving Output file {} ".format(out_loc))
ut.save_output(df_gne, out_loc, fl_name, gne_dir, csv_ext)
return df_gne
def segment_gne(com_speech_sort, voiced_yes, voiced_no, gne_all_frames, audio_file):
"""
calculating gne for each voice segment
"""
snd = parselmouth.Sound(audio_file)
pitch = snd.to_pitch(time_step=0.001)
for idx, vs in enumerate(com_speech_sort):
try:
max_gne = np.NaN
if vs in voiced_yes and len(vs) > 1:
start_time = pitch.get_time_from_frame_number(vs[0])
end_time = pitch.get_time_from_frame_number(vs[-1])
snd_start = int(snd.get_frame_number_from_time(start_time))
snd_end = int(snd.get_frame_number_from_time(end_time))
samples = parselmouth.Sound(snd.as_array()[0][snd_start:snd_end])
max_gne = gne_ratio(samples)
except:
pass
gne_all_frames[idx] = max_gne
return gne_all_frames
def calc_gne(video_uri, audio_file, out_loc, fl_name, r_config, save=True, ff_df=None):
"""
Preparing gne matrix
Args: Args:
audio_file: (.wav) parsed audio file audio_file: (.wav) parsed audio file
out_loc: (str) Output directory for csv's out_loc: (str) Output directory for csv's
""" """
dir_path = os.path.join(out_loc, ff_dir)
if os.path.isdir(dir_path) or ff_df is not None:
if ff_df is not None:
voice_seg = ut.process_segment_pitch(ff_df, r_config)
else:
voice_seg = ut.segment_pitch(dir_path, r_config, ff_df=ff_df)
gne_all_frames = [np.NaN] * len(voice_seg[0]) hnr_all_frames = hnr_ratio(audio_file)
gne_segment_frames = segment_gne( df_hnr = pd.DataFrame(hnr_all_frames, columns=[r_config.aco_hnr])
voice_seg[0], voice_seg[1], voice_seg[2], gne_all_frames, audio_file
)
df_gne = pd.DataFrame(gne_segment_frames, columns=[r_config.aco_gne]) df_hnr["Frames"] = df_hnr.index
df_gne[ df_hnr["dbm_master_url"] = video_uri
df_hnr[
r_config.err_reason r_config.err_reason
] = "Pass" # will replace with threshold in future release ] = "Pass" # will replace with threshold in future release
df_gne["Frames"] = df_gne.index if save:
df_gne["dbm_master_url"] = video_uri logger.info("Saving Output file {} ".format(out_loc))
ut.save_output(df_hnr, out_loc, fl_name, hnr_dir, csv_ext)
return df_hnr
def empty_hnr(video_uri, out_loc, fl_name, r_config, save=True):
"""
Preparing empty HNR matrix if something fails
"""
cols = ["Frames", r_config.aco_hnr, r_config.err_reason]
out_val = [[np.nan, np.nan, error_txt]]
df_hnr = pd.DataFrame(out_val, columns=cols)
df_hnr["dbm_master_url"] = video_uri
if save: if save:
logger.info("Processing Output file {} ".format(out_loc)) logger.info("Saving Output file {} ".format(out_loc))
ut.save_output(df_gne, out_loc, fl_name, gne_dir, csv_ext) ut.save_output(df_hnr, out_loc, fl_name, hnr_dir, csv_ext)
return df_gne return df_hnr
else:
error_txt = "error: pitch freq not available"
return empty_gne(video_uri, out_loc, fl_name, r_config, error_txt, save=save)
def run_gne(video_uri, out_dir, r_config, save=True, ff_df=None): def run_hnr(video_uri, out_dir, r_config, save=True):
""" """
Processing all patient's for fetching glottal noise ratio Processing all patient's for fetching harmonic noise ratio
--------------- -------------------
--------------- -------------------
Args: Args:
video_uri: video path; r_config: raw variable config object video_uri: video path; r_config: raw variable config object
out_dir: (str) Output directory for processed output out_dir: (str) Output directory for processed output
@@ -146,19 +100,11 @@ def run_gne(video_uri, out_dir, r_config, save=True, ff_df=None):
"Output file {} size is less than 0.064sec".format(audio_file) "Output file {} size is less than 0.064sec".format(audio_file)
) )
error_txt = "error: length less than 0.064" df = empty_hnr(video_uri, out_loc, fl_name, r_config, save=save)
df = empty_gne(
video_uri, out_loc, fl_name, r_config, error_txt, save=save
)
else: else:
df = calc_gne( df = calc_hnr(
video_uri, video_uri, audio_file, out_loc, fl_name, r_config, save=save
audio_file,
out_loc,
fl_name,
r_config,
save=save,
ff_df=ff_df,
) )
return df return df
except Exception as e: except Exception as e: