audiofileinput.py

# Copyright (C) 2018 Google 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.
"""Simple file-based sample for the Google Assistant Service."""

import json
import logging
import os
import os.path
import sys

import click
import google.auth.transport.grpc
import google.auth.transport.requests
import google.oauth2.credentials

from google.assistant.embedded.v1alpha2 import (embedded_assistant_pb2,
                                                embedded_assistant_pb2_grpc)

END_OF_UTTERANCE = embedded_assistant_pb2.AssistResponse.END_OF_UTTERANCE


@click.command()
@click.option(
    '--api-endpoint',
    default='embeddedassistant.googleapis.com',
    metavar='<api endpoint>',
    show_default=True,
    help='Address of Google Assistant API service.')
@click.option(
    '--credentials',
    metavar='<credentials>',
    show_default=True,
    default=os.path.join(
        click.get_app_dir('google-oauthlib-tool'), 'credentials.json'),
    help='Path to read OAuth2 credentials.')
@click.option(
    '--device-model-id',
    required=True,
    metavar='<device model id>',
    help='Unique device model identifier.')
@click.option(
    '--device-id',
    required=True,
    metavar='<device id>',
    help='Unique registered device instance identifier.')
@click.option(
    '--lang',
    show_default=True,
    metavar='<language code>',
    default='en-US',
    help='Language code of the Assistant.')
@click.option(
    '--verbose',
    '-v',
    is_flag=True,
    default=False,
    help='Enable verbose logging.')
@click.option(
    '--input-audio-file',
    '-i',
    required=True,
    metavar='<input file>',
    type=click.File('rb'),
    help='Path to input audio file (format: LINEAR16 16000 Hz).')
@click.option(
    '--output-audio-file',
    '-o',
    required=True,
    metavar='<output file>',
    type=click.File('wb'),
    help='Path to output audio file (format: LINEAR16 16000 Hz).')
@click.option(
    '--block-size',
    default=1024,
    metavar='<block size>',
    show_default=True,
    help='Size of each input stream read in bytes.')
@click.option(
    '--grpc-deadline',
    default=300,
    metavar='<grpc deadline>',
    show_default=True,
    help='gRPC deadline in seconds')
def main(api_endpoint, credentials, device_model_id, device_id, lang, verbose,
    input_audio_file, output_audio_file, block_size, grpc_deadline, *args,
    **kwargs):
    """File based sample for the Google Assistant API.

      Examples:
        $ python -m audiofileinput -i <input file> -o <output file>
      """
    # Setup logging.
    logging.basicConfig(level=logging.DEBUG if verbose else logging.INFO)

    # Load OAuth 2.0 credentials.
    try:
        with open(credentials, 'r') as f:
            credentials = google.oauth2.credentials.Credentials(
                token=None, **json.load(f))
            http_request = google.auth.transport.requests.Request()
            credentials.refresh(http_request)
    except Exception as e:
        logging.error('Error loading credentials: %s', e)
        logging.error('Run google-oauthlib-tool to initialize '
                      'new OAuth 2.0 credentials.')
        sys.exit(-1)

    # Create an authorized gRPC channel.
    grpc_channel = google.auth.transport.grpc.secure_authorized_channel(
        credentials, http_request, api_endpoint)
    logging.info('Connecting to %s', api_endpoint)

    # Create gRPC stubs
    assistant = embedded_assistant_pb2_grpc.EmbeddedAssistantStub(grpc_channel)

    # Generate gRPC requests.
    def gen_assist_requests(input_stream):
        dialog_state_in = embedded_assistant_pb2.DialogStateIn(
            language_code=lang, conversation_state=b'')
        config = embedded_assistant_pb2.AssistConfig(
            audio_in_config=embedded_assistant_pb2.AudioInConfig(
                encoding='LINEAR16',
                sample_rate_hertz=16000,
            ),
            audio_out_config=embedded_assistant_pb2.AudioOutConfig(
                encoding='LINEAR16',
                sample_rate_hertz=16000,
                volume_percentage=100,
            ),
            dialog_state_in=dialog_state_in,
            device_config=embedded_assistant_pb2.DeviceConfig(
                device_id=device_id,
                device_model_id=device_model_id,
            ))
        # Send first AssistRequest message with configuration.
        yield embedded_assistant_pb2.AssistRequest(config=config)
        while True:
            # Read user request from file.
            data = input_stream.read(block_size)
            if not data:
                break
            # Send following AssitRequest message with audio chunks.
            yield embedded_assistant_pb2.AssistRequest(audio_in=data)

    for resp in assistant.Assist(
        gen_assist_requests(input_audio_file), grpc_deadline):
        # Iterate on AssistResponse messages.
        if resp.event_type == END_OF_UTTERANCE:
            logging.info('End of audio request detected')
        if resp.speech_results:
            logging.info('Transcript of user request: "%s".',
                         ' '.join(r.transcript for r in resp.speech_results))
        if len(resp.audio_out.audio_data) > 0:
            # Write assistant response to supplied file.
            output_audio_file.write(resp.audio_out.audio_data)
        if resp.dialog_state_out.supplemental_display_text:
            logging.info('Assistant display text: "%s"',
                         resp.dialog_state_out.supplemental_display_text)
        if resp.device_action.device_request_json:
            device_request = json.loads(resp.device_action.device_request_json)
            logging.info('Device request: %s', device_request)


if __name__ == '__main__':
    main()