Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def should_convert_wav(profile: Dict[str, Any], wav_io: BinaryIO) -> bool:
expected_rate = int(pydash.get(profile, "audio.format.sample-rate-hertz", 16000))
expected_width = int(pydash.get(profile, "audio.format.sample-width-bits", 16)) // 8
expected_channels = int(pydash.get(profile, "audio.format.channel-count", 1))
with wave.open(wav_io, "rb") as wav_file:
rate, width, channels = (
wav_file.getframerate(),
wav_file.getsampwidth(),
wav_file.getnchannels(),
)
return (
(rate != expected_rate)
or (width != expected_width)
or (channels != expected_channels)
)
def get_pronounce_espeak(
args: argparse.Namespace, core: Voice2JsonCore
) -> typing.Callable[
[str, typing.Iterable[str]], typing.Coroutine[typing.Any, typing.Any, bytes]
]:
"""Get pronounce method for eSpeak."""
# Use eSpeak
espeak_voice = pydash.get(core.profile, "text-to-speech.espeak.voice")
espeak_map_path = core.ppath(
"text-to-speech.espeak.phoneme-map", "espeak_phonemes.txt"
)
assert (
espeak_map_path and espeak_map_path.exists()
), f"Missing eSpeak phoneme map at {espeak_map_path}"
espeak_phoneme_map: typing.Dict[str, str] = {}
with open(espeak_map_path, "r") as map_file:
for line in map_file:
line = line.strip()
if line:
parts = line.split(maxsplit=1)
espeak_phoneme_map[parts[0]] = parts[1]
engine_path = pydash.get(core.profile, "wake-word.precise.engine-executable")
if not engine_path:
engine_path = shutil.which("precise-engine")
model_path: typing.Optional[Path] = None
if args.model:
model_path = Path(args.model)
else:
model_path = core.ppath(
"wake-word.precise.model-file", "precise/hey-mycroft-2.pb"
)
assert model_path, "No model path"
sensitivity = float(pydash.get(core.profile, "wake-word.sensitivity", 0.5))
trigger_level = int(pydash.get(core.profile, "wake-word.precise.trigger-level", 3))
# Load Precise engine
assert engine_path and model_path, "Missing engine or model path"
engine_path = Path(engine_path)
model_path = Path(model_path)
assert engine_path.exists(), f"Engine does not exist at {engine_path}"
assert model_path.exists(), f"Model does not exist at {model_path}"
_LOGGER.debug(
"Loading Precise (engine=%s, model=%s, chunk_size=%s)",
engine_path,
model_path,
args.chunk_size,
)
def get_by_diff_key(dic: dict, diff_key: str) -> Any:
return py_.get(
dic,
diff_key.replace("root", "")
.replace("><", ".")
.replace(">", "")
.replace("<", "")
.replace("'", ""),
)
def _get_stats_metadata(cfg):
""" Build metadata.stats subtree for the product definition
"""
period = pydash.get(cfg, 'date_ranges.stats_duration', '*')
step = pydash.get(cfg, 'date_ranges.step_size', '*')
return dict(period=period, step=step)
async def wake(args: argparse.Namespace, core: Voice2JsonCore) -> None:
"""Wait for wake word in audio stream."""
# Load settings
engine_path = pydash.get(core.profile, "wake-word.precise.engine-executable")
if not engine_path:
engine_path = shutil.which("precise-engine")
model_path: typing.Optional[Path] = None
if args.model:
model_path = Path(args.model)
else:
model_path = core.ppath(
"wake-word.precise.model-file", "precise/hey-mycroft-2.pb"
)
assert model_path, "No model path"
sensitivity = float(pydash.get(core.profile, "wake-word.sensitivity", 0.5))
trigger_level = int(pydash.get(core.profile, "wake-word.precise.trigger-level", 3))
tokens = to_path_tokens(path)
if not pyd.is_list(tokens): # pragma: no cover
tokens = [tokens]
last_key = pyd.last(tokens)
if isinstance(last_key, PathToken):
last_key = last_key.key
target = obj
for idx, token in enumerate(pyd.initial(tokens)):
if isinstance(token, PathToken):
key = token.key
default_factory = pyd.get(tokens,
[idx + 1, 'default_factory'],
default=default_type)
else:
key = token
default_factory = default_type
obj_val = base_get(target, key, default=None)
path_obj = None
if call_customizer:
path_obj = call_customizer(obj_val, key, target)
if path_obj is None:
path_obj = default_factory()
base_set(target, key, path_obj, allow_override=False)
configobj.ConfigObj(data).write(output)
print output.getvalue(),
else:
print data
elif args.command == 'edit':
for action_i in ('append', 'prepend', 'set', 'remove', 'remove_key'):
if getattr(args, action_i):
action = action_i
break
if action in ('append', 'prepend', 'set', 'remove'):
# Unpack key and new value.
key, new_value = getattr(args, action)
# Look up existing value.
config_value = pydash.get(config.data, key)
if action == 'set':
# Set a key to a string value.
# Create dictionary structure containing only the specified key
# and value.
nested_value = pydash.set_({}, key, new_value)
# Merge nested value into existing configuration structure.
pydash.merge(config.data, nested_value)
else:
# Action is a list action.
if config_value is None:
# Create dictionary structure containing only empty list for
# specified key.
config_value = []
def check_body_spec(spec):
'''Base method to check body spec for multi-agent multi-env'''
ae_product = ps.get(spec, 'body.product')
body_num = ps.get(spec, 'body.num')
if ae_product == 'outer':
pass
elif ae_product == 'inner':
agent_num = len(spec['agent'])
env_num = len(spec['env'])
assert agent_num == env_num, 'Agent and Env spec length must be equal for body `inner` product. Given {agent_num}, {env_num}'
else: # custom
assert ps.is_list(body_num)