In [1]:
from configparser import ConfigParser
from glob import glob
import json
import os
from pathlib import Path
import shutil
import joblib
import numpy as np
import tensorflow as tf
import tqdm
import vak
In [6]:
VDS_PATH = Path(
'/home/art/Documents/data/birdsong/BirdsongRecognition/vak/Bird6/'
)
In [7]:
train_vds_path = list(VDS_PATH.glob('*train.vds.json'))[0]
In [8]:
train_vds = vak.Dataset.load(json_fname=train_vds_path)
if train_vds.are_spects_loaded() is False:
train_vds = train_vds.load_spects()
X_train = train_vds.spects_list()
X_train = np.concatenate(X_train, axis=1)
Y_train = train_vds.lbl_tb_list()
Y_train = np.concatenate(Y_train)
# transpose so rows are time bins
X_train = X_train.T
n_classes = len(train_vds.labelmap)
print(n_classes)
In [9]:
test_vds_path = list(VDS_PATH.glob('*test.vds.json'))[0]
In [10]:
num_replicates = 4
train_set_durs = [60, 120, 480]
In [11]:
test_vds = vak.Dataset.load(json_fname=test_vds_path)
if test_vds.are_spects_loaded() is False:
test_vds = test_vds.load_spects()
if test_vds.labelmap != train_vds.labelmap:
raise ValueError(
f'labelmap of test set, {test_vds.labelmap}, does not match labelmap of training set, '
f'{train_vds.labelmap}'
)
def unpack_test():
"""helper function because we want to get back test set unmodified every time we go through
main loop below, without copying giant arrays"""
X_test = test_vds.spects_list()
X_test = np.concatenate(X_test, axis=1)
# transpose so rows are time bins
X_test = X_test.T
Y_test = test_vds.lbl_tb_list()
Y_test = np.concatenate(Y_test)
return X_test, Y_test
# just get X_test to make sure it has the right shape
X_test, _ = unpack_test()
if X_train.shape[-1] != X_test.shape[-1]:
raise ValueError(f'Number of frequency bins in training set spectrograms, {X_train.shape[-1]}, '
f'does not equal number in test set spectrograms, {X_test.shape[-1]}.')
freq_bins = X_test.shape[-1] # number of columns
# concatenate labels into one big string
# used for Levenshtein distance + syllable error rate
Y_train_labels = [voc.annot.labels.tolist() for voc in train_vds.voc_list]
Y_train_labels_for_lev = ''.join([chr(lbl) if type(lbl) is int else lbl
for labels in Y_train_labels for lbl in labels])
Y_test_labels = [voc.annot.labels.tolist() for voc in test_vds.voc_list]
Y_test_labels_for_lev = ''.join([chr(lbl) if type(lbl) is int else lbl
for labels in Y_test_labels for lbl in labels])
replicates = range(1, num_replicates + 1)
NETWORKS = vak.network._load()
In [12]:
# concatenate labels into one big string
# used for Levenshtein distance + syllable error rate
Y_train_labels = [voc.annot.labels.tolist() for voc in train_vds.voc_list]
Y_train_labels_for_lev = ''.join([chr(lbl) if type(lbl) is int else lbl
for labels in Y_train_labels for lbl in labels])
Y_test_labels = [voc.annot.labels.tolist() for voc in test_vds.voc_list]
Y_test_labels_for_lev = ''.join([chr(lbl) if type(lbl) is int else lbl
for labels in Y_test_labels for lbl in labels])
In [13]:
config_path = str(
'/home/art/Documents/data/birdsong/BirdsongRecognition/vak/Bird6/'
'learning_curve.190621_123313/'
'config_BirdsongRecognition_bird06_copy.ini'
)
In [14]:
a_config = vak.config.parse_config(config_path)
In [15]:
results_dirname = str(
'/home/art/Documents/data/birdsong/BirdsongRecognition/vak/Bird6/'
'learning_curve.190621_123313'
)
In [16]:
train_set_dur = 60
replicate = 1
In [17]:
training_records_dir = ('records_for_training_set_with_duration_of_'
+ str(train_set_dur) + '_sec_replicate_'
+ str(replicate))
training_records_path = os.path.join(results_dirname,
'train',
training_records_dir)
In [18]:
scaler_name = ('spect_scaler_duration_{}_replicate_{}'
.format(train_set_dur, replicate))
spect_scaler = joblib.load(
os.path.join(training_records_path, scaler_name))
In [21]:
(net_name, net_config) = tuple(a_config.networks.items())[0]
In [22]:
X_test, Y_test = unpack_test()
# Normalize before reshaping to avoid even more convoluted array reshaping.
X_test = spect_scaler.transform(X_test)
In [23]:
# Notice we don't reshape Y_test
(X_test,
_,
num_batches_test) = vak.utils.data.reshape_data_for_batching(
X_test,
net_config.batch_size,
net_config.time_bins,
Y_test)
In [24]:
net_config_dict = net_config._asdict()
net_config_dict['n_syllables'] = n_classes
if 'freq_bins' in net_config_dict:
net_config_dict['freq_bins'] = freq_bins
In [25]:
results_dirname_this_net = os.path.join(training_records_path, net_name)
checkpoint_path = os.path.join(results_dirname_this_net, 'checkpoints')
In [26]:
net = NETWORKS[net_name](**net_config_dict)
# we use latest checkpoint when doing summary for learncurve, assume that's "best trained"
checkpoint_file = tf.train.latest_checkpoint(checkpoint_dir=checkpoint_path)
meta_file = glob(checkpoint_file + '*meta')
if len(meta_file) != 1:
raise ValueError('Incorrect number of meta files for last saved checkpoint.\n'
'For checkpoint {}, found these files:\n'
'{}'
.format(checkpoint_file, meta_file))
else:
meta_file = meta_file[0]
data_file = glob(checkpoint_file + '*data*')
if len(data_file) != 1:
raise ValueError('Incorrect number of data files for last saved checkpoint.\n'
'For checkpoint {}, found these files:\n'
'{}'
.format(checkpoint_file, data_file))
else:
data_file = data_file[0]
with tf.Session(graph=net.graph) as sess:
tf.logging.set_verbosity(tf.logging.ERROR)
net.restore(sess=sess,
meta_file=meta_file,
data_file=data_file)
for b in range(num_batches_test): # "b" is "batch number"
d = {
net.X: X_test[:, b * net_config_dict['time_bins']: (b + 1) * net_config_dict['time_bins'], :],
net.lng: [net_config_dict['time_bins']] * net_config_dict['batch_size']}
if 'Y_pred_test' in locals():
preds = sess.run(net.predict, feed_dict=d)
preds = preds.reshape(net_config_dict['batch_size'], -1)
Y_pred_test = np.concatenate((Y_pred_test, preds),
axis=1)
else:
Y_pred_test = sess.run(net.predict, feed_dict=d)
Y_pred_test = Y_pred_test.reshape(net_config_dict['batch_size'], -1)
# again get rid of zero padding predictions
Y_pred_test = Y_pred_test.ravel()[:Y_test.shape[0], np.newaxis]
test_err = np.sum(Y_pred_test != Y_test) / Y_test.shape[0]
In [27]:
Y_pred_test
Out[27]:
In [28]:
Y_test_lbl_tb_list = test_vds.lbl_tb_list()
In [30]:
Y_test_lens = [arr.shape for arr in Y_test_lbl_tb_list]
How different is output of lbl_tb2segments (used by vak.core.predict) compared to output of lbl_tb2label (used by vak.core.learncurve.test)?
In [34]:
Y_pred_test_seg = vak.utils.labels.lbl_tb2labels(Y_pred_test, train_vds.labelmap)
In [36]:
len(Y_pred_test_seg)
Out[36]:
In [39]:
timebin_dur = set([voc.metaspect.timebin_dur for voc in train_vds.voc_list])
timebin_dur = timebin_dur.pop()
In [44]:
Y_pred_test_lbl, onsets, offsets = vak.utils.labels.lbl_tb2segments(Y_pred_test,
train_vds.labelmap,
timebin_dur)
In [45]:
Y_pred_test_lbl.shape
Out[45]:
In [48]:
Y_pred_test_lbl_str = ''.join(Y_pred_test_lbl.tolist())
In [50]:
vak.metrics.levenshtein(Y_pred_test_seg, Y_pred_test_lbl_str)
Out[50]:
In [ ]: