network_runner.py

# Copyright 2019 Mycroft AI Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Pieces that convert audio to predictions
"""
importnumpyasnp
fromabcimportabstractmethod, ABCMeta
fromimportlibimportimport_module
fromos.pathimportsplitext
fromtypingimport*
fromtypingimportBinaryIO

fromprecise.threshold_decoderimportThresholdDecoder
fromprecise.modelimportload_precise_model
fromprecise.paramsimportinject_params, pr
fromprecise.utilimportbuffer_to_audio
fromprecise.vectorizationimportvectorize_raw, add_deltas


classRunner(metaclass=ABCMeta):
"""
 Classes that execute trained models on vectorized audio
 and produce prediction values
 """
@abstractmethod
defpredict(self, inputs: np.ndarray) ->np.ndarray:
pass

@abstractmethod
defrun(self, inp: np.ndarray) ->float:
pass


classTensorFlowRunner(Runner):
"""Executes a frozen Tensorflow model created from precise-convert"""
def__init__(self, model_name: str):
ifmodel_name.endswith('.net'):
print('Warning: ', model_name, 'looks like a Keras model.')
self.tf=import_module('tensorflow')
self.graph=self.load_graph(model_name)

self.inp_var=self.graph.get_operation_by_name('import/net_input').outputs[0]
self.out_var=self.graph.get_operation_by_name('import/net_output').outputs[0]

self.sess=self.tf.Session(graph=self.graph)

defload_graph(self, model_file: str) ->'tf.Graph':
graph=self.tf.Graph()
graph_def=self.tf.GraphDef()

withopen(model_file, "rb") asf:
graph_def.ParseFromString(f.read())
withgraph.as_default():
self.tf.import_graph_def(graph_def)

returngraph

defpredict(self, inputs: np.ndarray) ->np.ndarray:
"""Run on multiple inputs"""
returnself.sess.run(self.out_var, {self.inp_var: inputs})

defrun(self, inp: np.ndarray) ->float:
returnself.predict(inp[np.newaxis])[0][0]


classKerasRunner(Runner):
""" Executes a regular Keras model created from precise-train"""
def__init__(self, model_name: str):
importtensorflowastf
# ISSUE 88 - Following 3 lines added to resolve issue 88 - JM 2020-02-04 per liny90626
fromtensorflow.python.keras.backendimportset_session# ISSUE 88
self.sess=tf.Session() # ISSUE 88
set_session(self.sess) # ISSUE 88
self.model=load_precise_model(model_name)
self.graph=tf.get_default_graph()

defpredict(self, inputs: np.ndarray):
fromtensorflow.python.keras.backendimportset_session# ISSUE 88
withself.graph.as_default():
set_session(self.sess) # ISSUE 88
returnself.model.predict(inputs)

defrun(self, inp: np.ndarray) ->float:
returnself.predict(inp[np.newaxis])[0][0]


classListener:
"""Listener that preprocesses audio into MFCC vectors and executes neural networks"""

def__init__(self, model_name: str, chunk_size: int=-1, runner_cls: type=None):
self.window_audio=np.array([])
self.pr=inject_params(model_name)
self.mfccs=np.zeros((self.pr.n_features, self.pr.n_mfcc))
self.chunk_size=chunk_size
runner_cls=runner_clsorself.find_runner(model_name)
self.runner=runner_cls(model_name)
self.threshold_decoder=ThresholdDecoder(self.pr.threshold_config, pr.threshold_center)

@staticmethod
deffind_runner(model_name: str) ->Type[Runner]:
runners= {
'.net': KerasRunner,
'.pb': TensorFlowRunner
 }
ext=splitext(model_name)[-1]
ifextnotinrunners:
raiseValueError('File extension of '+model_name+' must be: '+str(list(runners)))
returnrunners[ext]

defclear(self):
self.window_audio=np.array([])
self.mfccs=np.zeros((self.pr.n_features, self.pr.n_mfcc))

defupdate_vectors(self, stream: Union[BinaryIO, np.ndarray, bytes]) ->np.ndarray:
ifisinstance(stream, np.ndarray):
buffer_audio=stream
else:
ifisinstance(stream, (bytes, bytearray)):
chunk=stream
else:
chunk=stream.read(self.chunk_size)
iflen(chunk) ==0:
raiseEOFError
buffer_audio=buffer_to_audio(chunk)

self.window_audio=np.concatenate((self.window_audio, buffer_audio))

iflen(self.window_audio) >=self.pr.window_samples:
new_features=vectorize_raw(self.window_audio)
self.window_audio=self.window_audio[len(new_features) *self.pr.hop_samples:]
iflen(new_features) >len(self.mfccs):
new_features=new_features[-len(self.mfccs):]
self.mfccs=np.concatenate((self.mfccs[len(new_features):], new_features))

returnself.mfccs

defupdate(self, stream: Union[BinaryIO, np.ndarray, bytes]) ->float:
mfccs=self.update_vectors(stream)
ifself.pr.use_delta:
mfccs=add_deltas(mfccs)
raw_output=self.runner.run(mfccs)
returnself.threshold_decoder.decode(raw_output)