Sample Python runtime client

import argparse
import time
import logging
import grpc
import urllib
import urllib.request
import base64
import json
import threading

from google.protobuf import text_format
from google.protobuf import json_format

from google.protobuf.json_format import MessageToJson

from nuance.nlu.v1.runtime_pb2 import *
from nuance.nlu.v1.runtime_pb2_grpc import *
from nuance.nlu.v1.result_pb2 import *

oauth_token_expiry_threshhold_seconds = 30
oauth_token_expiry_seconds = 0
oauth_token = None

args = None


def parse_args():
    global args
    parser = argparse.ArgumentParser(
        prog="nlu_client.py",
        usage="%(prog)s [-options]",
        add_help=False,
        formatter_class=lambda prog: argparse.HelpFormatter(
            prog, max_help_position=45, width=100)
    )

    options = parser.add_argument_group("options")
    options.add_argument("-h", "--help", action="help",
                         help="Show this help message and exit")
    options.add_argument("--token", nargs="?", help=argparse.SUPPRESS)
    options.add_argument("--oauthURL", metavar="oauthUrl", nargs="?",
                         help="OAuth 2.0 URL")
    options.add_argument("--clientID", metavar="clientID", nargs="?",
                         help="OAuth 2.0 Client ID")
    options.add_argument("--clientSecret", metavar="clientSecret", nargs="?",
                         help="OAuth 2.0 Client Secret")
    options.add_argument("--oauthScope", metavar="oauthScope", nargs="?",
                         help="OAuth 2.0 Scope, default=nlu", default='nlu')
    options.add_argument("-s", "--serverUrl", metavar="serverUrl", nargs="?",
                         help="NLU server URL, default=localhost:8080", default='localhost:8080')
    options.add_argument('--modelUrn', metavar="modelUrn", nargs="?",
                         help="NLU Model URN")
    options.add_argument('--wordsetUrn', metavar="wordsetUrn", nargs="?",
                         help="compiled wordset URN")
    options.add_argument("--textInput", metavar="file", nargs="?",
                         help="Text to perform interpretation on")
    args = parser.parse_args()


def get_oauth2_token():
    global oauth_token
    global oauth_token_expiry_seconds
    global oauth_token_expiry_threshhold_seconds

    if args.oauthURL is None:
        return None
    
    current_time = time.monotonic()

    try:
        if oauth_token and oauth_token_expiry_seconds - oauth_token_expiry_threshhold_seconds > current_time:
            log.debug('OAuth token is still valid')
            return oauth_token

        log.info("Obtaining auth token (Client ID: {}, URL: {})".format(args.clientID, args.oauthURL))

        encoded_credentials = base64.standard_b64encode("{}:{}".format(args.clientID, args.clientSecret).encode()).decode('utf-8')
        headers = { 'Authorization' : "Basic {}".format(encoded_credentials)  }

        data = {
            'grant_type': 'client_credentials',
            'scope': args.oauthScope,
        }

        request = urllib.request.Request(url=args.oauthURL, headers=headers, data=urllib.parse.urlencode(data).encode(), method='POST')
        with urllib.request.urlopen(request) as response:
            response = response.read().decode('utf-8')
            json_response = json.loads(response)

            oauth_token = json_response["access_token"]
            oauth_token_expiry_seconds = time.monotonic() + json_response["expires_in"]
        
            log.debug("Token TTL: %d" % json_response["expires_in"])
            return json_response["access_token"]
    except urllib.error.HTTPError as err:
        raise Exception("Failed to obtain authentication token. Status: {}, Error: {}".format(err.code, err.read().decode()))


def create_channel(args):    
    call_credentials = None
    channel = None

    if args.token:
        log.debug('Adding CallCredentials using token parameter')
        call_credentials = grpc.access_token_call_credentials(args.token)
    else:
        current_oauth_token = get_oauth2_token()
        if current_oauth_token:
            log.debug('Adding CallCredentials from OAuth endpoint')
            call_credentials = grpc.access_token_call_credentials(current_oauth_token)

    log.debug("Creating secure gRPC channel")
    channel_credentials = grpc.ssl_channel_credentials()
    if call_credentials is not None:
        channel_credentials = grpc.composite_channel_credentials(channel_credentials, call_credentials)
    channel = grpc.secure_channel(args.serverUrl, credentials=channel_credentials)

    return channel


def construct_interpret_request(args):
    # Single intent, plain text logging
    params = InterpretationParameters(interpretation_result_type=EnumInterpretationResultType.SINGLE_INTENT, interpretation_input_logging_mode=EnumInterpretationInputLoggingMode.PLAINTEXT)
    # Reference the model via the app config
    model = ResourceReference(type=EnumResourceType.SEMANTIC_MODEL, uri=args.modelUrn)
    # Describe the text to perform interpretation on
    input = InterpretationInput(text=args.textInput)
    # Reference compiled wordset if included as an input
    if args.wordsetUrn:
        wordset_reference = ResourceReference(type=EnumResourceType.COMPILED_WORDSET, uri = args.wordsetUrn)
        resource = InterpretationResource(external_reference = wordset_reference)
        resources = [resource]
        # Build the request
        interpret_req = InterpretRequest(parameters=params, model=model,        resources=resources, input=input)
    else:
        # Build the request
        interpret_req = InterpretRequest(parameters=params, model=model, input=input)
    return interpret_req

def main():
    parse_args()
    log_level = logging.DEBUG
    global log
    log = logging.getLogger('')
    logging.basicConfig(
        format='%(asctime)s %(levelname)-5s: %(message)s', level=log_level)
    
    if args.oauthURL:
        if args.clientID is None:
            log.error("OAuth 2.0 URL was supplied but client ID is missing")
            return
        elif args.clientSecret is None:
            log.error("OAuth 2.0 URL was supplied but client secret is missing")
            return

    get_oauth2_token()

    with create_channel(args) as channel:
        stub = RuntimeStub(channel)
        response = stub.Interpret(construct_interpret_request(args))
        print(MessageToJson(response))
    print("Done")

if __name__ == '__main__':
  main()