|
|
|
|
| from __future__ import annotations
|
|
|
| import json
|
| import subprocess
|
| import sys
|
| from pathlib import Path
|
|
|
| RUNTIME_NAME = "sle_spark.py"
|
|
|
| def _sequence(value):
|
| if value is None:
|
| return ()
|
| if isinstance(value, str):
|
| return (value,)
|
| return tuple(value)
|
|
|
|
|
| def _append_optional(args, payload, key, flag):
|
| value = payload.get(key)
|
| if value is not None:
|
| args.extend((flag, str(value)))
|
|
|
|
|
| def _run_artifact_args(payload):
|
| mode = str(payload.get('mode', 'response'))
|
| args = ['run-artifact', '--artifact', 'model.cra', '--mode', mode]
|
| if mode == 'acoustic':
|
| args.extend(('--audio-samples', json.dumps(payload.get('audio_samples', []))))
|
| else:
|
| args.extend(('--text', str(payload.get('text', ''))))
|
| for tool in _sequence(payload.get('host_tools', payload.get('host_tool'))):
|
| args.extend(('--host-tool', str(tool)))
|
| if payload.get('avoid_replay'):
|
| args.append('--avoid-replay')
|
| for text in _sequence(payload.get('avoid_texts')):
|
| args.extend(('--avoid-text', str(text)))
|
| if payload.get('compose_speech'):
|
| args.append('--compose-speech')
|
| if payload.get('prefer_composed'):
|
| args.append('--prefer-composed')
|
| if payload.get('canonicalize_stream'):
|
| args.append('--canonicalize-stream')
|
| if payload.get('recover_on_failure'):
|
| args.append('--recover-on-failure')
|
| _append_optional(args, payload, 'audio_symbol_score_floor', '--audio-symbol-score-floor')
|
| _append_optional(args, payload, 'diverse_speech_limit', '--diverse-speech-limit')
|
| _append_optional(args, payload, 'turns', '--turns')
|
| _append_optional(args, payload, 'punctuation_floor', '--punctuation-floor')
|
| _append_optional(args, payload, 'output_artifact', '--output-artifact')
|
| _append_optional(args, payload, 'dt', '--dt')
|
| _append_optional(args, payload, 'max_speech_steps', '--max-speech-steps')
|
| return args
|
|
|
|
|
| def _runtime_command(args):
|
| runtime = Path(__file__).resolve().with_name(RUNTIME_NAME)
|
| if runtime.is_file():
|
| if runtime.suffix.lower() == '.py':
|
| return [sys.executable, str(runtime), *args]
|
| return [str(runtime), *args]
|
| return [sys.executable, '-m', 'sle.cli', *args]
|
|
|
|
|
| def main() -> int:
|
| payload = json.loads(sys.stdin.read() or '{}')
|
| completed = subprocess.run(
|
| _runtime_command(_run_artifact_args(payload)),
|
| check=False,
|
| text=True,
|
| capture_output=True,
|
| )
|
| sys.stdout.write(completed.stdout)
|
| if completed.stderr:
|
| sys.stderr.write(completed.stderr)
|
| return int(completed.returncode)
|
|
|
|
|
| if __name__ == '__main__':
|
| raise SystemExit(main())
|
|
|
