genesyscloud/genesyscloud-audiohook/audiohook_blueprint.py

# Copyright 2024 Google LLC
#
# 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.

""" Module for receiving audio streaming from Audiohook Monitor and
call Agent Assist backend
"""
import json
import logging
from dataclasses import dataclass, field
from threading import Thread

import numpy as np
from flask import Blueprint
from flask_sock import Sock
from google.api_core.exceptions import NotFound
from google.cloud import dialogflow_v2beta1 as dialogflow
from simple_websocket import Server

from audio_stream import Stream
from audiohook import DEFAULT_CONVERSATION_ID, AudioHook
from audiohook_config import config
from dialogflow_api import (DialogflowAPI, await_redis, create_conversation_name,
                            find_participant_by_role, location_id, project)

audiohook_bp = Blueprint("audiohook", __name__)
sock = Sock(audiohook_bp)


logging.getLogger()
logging.basicConfig(
    format='%(levelname)-8s [%(filename)s:%(lineno)d in '
           'function %(funcName)s] %(message)s',
    datefmt='%Y-%m-%d:%H:%M:%S',
    level=config.log_level.upper()
)


@dataclass
class OpenConversationState:
    """ Memorize the state after open message that is not a connection prob
    """
    agent_thread: Thread
    user_thread: Thread
    conversation_name: str
    is_opened: bool


def process_open_conversation_message(
        conversation_id: str,
        dialogflow_api: DialogflowAPI,
        agent_stream: Stream,
        customer_stream: Stream,
        ws: Server,
        audiohook: AudioHook
) -> OpenConversationState:
    """Process "open" message get from Audiohook Monitor, and establish a state
    object for conversation_name, agent_thread, user_thread, and is_opened bool
    """

    conversation_profile = dialogflow_api.get_conversation_profile(
        conversation_profile_name=config.conversation_profile_name)
    agent_audio_config = agent_stream.define_audio_config(conversation_profile)
    customer_audio_config = customer_stream.define_audio_config(
        conversation_profile)
    normalized_conversation_id = 'a' + conversation_id
    conversation_name = create_conversation_name(
        normalized_conversation_id, location_id, project)
    try:
        dialogflow_api.get_conversation(
            conversation_name)
    except NotFound as e:
        logging.warning("Error getting the conversation : %s", e)
        dialogflow_api.create_conversation(
            conversation_profile, normalized_conversation_id)

    try:
        participants_list = dialogflow_api.list_participant(
            conversation_name)
        participant_agent = find_participant_by_role(
            dialogflow.Participant.Role.HUMAN_AGENT, participants_list)
        participant_user = find_participant_by_role(
            dialogflow.Participant.Role.END_USER, participants_list)
        if not participant_agent:
            participant_agent = dialogflow_api.create_participant(
                conversation_name=conversation_name, role="HUMAN_AGENT")
        if not participant_user:
            participant_user = dialogflow_api.create_participant(
                conversation_name=conversation_name, role="END_USER")
    except NotFound as e:
        logging.error("Participants not found %s: ", e)

    agent_thread = Thread(
        target=dialogflow_api.maintained_streaming_analyze_content, args=(
            agent_stream, participant_agent, agent_audio_config))
    user_thread = Thread(
        target=dialogflow_api.maintained_streaming_analyze_content, args=(
            customer_stream, participant_user, customer_audio_config))
    ws.send(json.dumps(audiohook.create_opened_message()))
    return OpenConversationState(
        agent_thread,
        user_thread,
        conversation_name,
        True)


def process_ongoing_conversation_messages(
        message: json,
        dialogflow_api: DialogflowAPI,
        audiohook: AudioHook,
        agent_stream: Stream,
        customer_stream: Stream,
        open_conversation_state: OpenConversationState,
        ws: Server) -> bool:
    """Process string messages that are not "open" and "ping" from Audiohook client through websocket.

    Note:
        Audiohook client passes Null UUIDs (00000000-0000-0000-0000-000000000000)
        as conversationId and participant.id parameters to identify connection probes.
        https://developer.genesys.cloud/devapps/audiohook/protocol-reference#openparameters

    Reference: https://developer.genesys.cloud/devapps/audiohook/protocol-reference

    Return:
        True, if there is a connection close message. So the outer loop for
        receiving audio can be completed
        False, for other messages to continue process messages from websocket
    """
    message_type = message.get("type")
    match message_type:
        case "resumed":
            # The first paused message after open message sets the
            # closed to True, now after resume, need to flip the bit
            customer_stream.closed = False
            agent_stream.closed = False
            open_conversation_state.agent_thread.start()
            open_conversation_state.user_thread.start()
        case "paused":
            customer_stream.closed = True
            agent_stream.closed = True
            logging.debug("Audio stream is paused")
        case "close":
            # This "close" is for ending a real conversation
            agent_stream.closed = True
            customer_stream.closed = True
            agent_stream.terminate = True
            customer_stream.terminate = True
            ws.send(json.dumps(audiohook.create_close_message()))
            try:
                dialogflow_api.complete_conversation(
                    open_conversation_state.conversation_name)
            except Exception as e:
                logging.error("Error completing conversation %s", e)
            # wait for the two thread to finish then terminate
            logging.debug("Stop streaming threads for customers and agents")
            return True
        case "discarded":
            start_time = message.get("START_TIME")
            duration = message.get("DURATION")
            logging.info(
                "Currently the audio stream has been paused from %s for about %s second",
                start_time,
                duration)
    return False


def wait_for_redis_resume(open_conversation_state: OpenConversationState,
                          audiohook: AudioHook, ws: Server):
    await_redis(
            open_conversation_state.conversation_name)
    # Always send the resume after awaiting the redis, don't stop the audio streaming
    # event if redis client is not set
    ws.send(json.dumps(audiohook.create_resume_message()))


@sock.route('/connect')
def audiohook_connect(ws: Server):
    """Genesys Cloud Audiohook connector

    Args:
        ws (Server): Websocket server for exchange messages
    """
    agent_stream = Stream(
        config.rate, chunk_size=config.chunk_size)
    customer_stream = Stream(
        config.rate, chunk_size=config.chunk_size)

    dialogflow_api = DialogflowAPI()
    audiohook = AudioHook()
    logging.info(
        "Audiohook client connected with the interceptor server")
    open_conversation_state = None
    while True:
        data = ws.receive()
        if isinstance(data, str):
            try:
                json_message = json.loads(data)
            except ValueError as e:
                logging.warning(
                    "Not a valid JSON message %s, error details %s ", data, e)
                continue
            message_type = json_message.get("type")
            logging.info(
                "Handle %s message %s", message_type, json_message)
            conversation_id = json_message.get("parameters", {}).get(
                "conversationId", DEFAULT_CONVERSATION_ID)
            audiohook.set_session_id(json_message.get("id", 0))
            audiohook.set_client_sequence(json_message.get("seq"))
            if message_type == "open":
                if conversation_id == DEFAULT_CONVERSATION_ID:
                    logging.debug(
                        "Connection Probe, not creating Dialogflow Conversation")
                    ws.send(json.dumps(audiohook.create_opened_message()))
                elif conversation_id != DEFAULT_CONVERSATION_ID and open_conversation_state is None:
                    # Get the first "open" message for real conversation
                    # open_state contains the agent and user thread for
                    # calling streaming_analyze_content
                    # a bool flag indicating if conversation, participants have been initialized
                    # and the conversation_name for the dialogflow.Conversation object
                    open_conversation_state = process_open_conversation_message(
                        conversation_id,
                        dialogflow_api,
                        agent_stream,
                        customer_stream,
                        ws,
                        audiohook,
                    )
                    logging.debug(
                        "open conversation message %s ", open_conversation_state)
                    # Check if the redis client has join_room called from the
                    # agent assist backend. Before setting conversation_name in the redis client,
                    # we should not publish any messages to the redis client
                    # otherwise e UI modules will not receive pub/subs until redis connects the conversation
                    await_redis_thread = Thread(target=wait_for_redis_resume, args=(
                        open_conversation_state, audiohook, ws))
                    await_redis_thread.start()
            elif message_type == "ping":
                ws.send(json.dumps(audiohook.create_pong_message()))
            elif message_type == "close" and open_conversation_state is None:
                # This "close" is for a connection prob, we don't need to call dialogflow
                # to complete conversation and terminate the stream in this case
                ws.send(json.dumps(audiohook.create_close_message()))
                break
            elif open_conversation_state is not None:
                # Close websocket connection when receive a "close message"
                if (process_ongoing_conversation_messages(json_message,
                                                          dialogflow_api,
                                                          audiohook,
                                                          agent_stream,
                                                          customer_stream,
                                                          open_conversation_state, ws,
                                                          )):
                    logging.info(
                        "Disconnecting Audiohook with the server")
                    break
        else:
            # audio is a 2-channel interleaved 8-bit PCMU audio stream
            # which is separated into single streams
            # using numpy
            # stream the audio to pub/sub
            array = np.frombuffer(data, dtype=np.int8)
            reshaped = array.reshape(
                (int(len(array) / 2), 2))
            # append audio to customer audio buffer
            customer_stream.fill_buffer(reshaped[:, 0].tobytes())
            # append audio to agent audio buffer
            agent_stream.fill_buffer(reshaped[:, 1].tobytes())