app.py

from groq import Groq
import base64
import os
from dotenv import load_dotenv
import gradio as gr
import mysql.connector
import cv2
#pip3 install cv-3
import datetime
from sumy.parsers.plaintext import PlaintextParser
from sumy.nlp.tokenizers import Tokenizer
from sumy.summarizers.lsa import LsaSummarizer

load_dotenv(verbose=True)

conn = mysql.connector.connect(
        host=os.environ.get("HOST"),
        user=os.environ.get("USER_NAME"),
        password=os.environ.get("PASSWORD"),
        port=os.environ.get("PORT"),
        database=os.environ.get("DB"),
        ssl_disabled=True,
        connection_timeout=60,
        use_pure=True
    )

cursor = conn.cursor(dictionary=True)

'''def capture_image(image):
    return image  # 撮影した画像をそのまま返す

# Function to encode the image
def encode_image(image_path):
    with open(image_path, "rb") as image_file:
        return base64.b64encode(image_file.read()).decode('utf-8')'''

def encode_image(image_array, file_name):
    # 色空間をBGRからRGBに変換
    image_array = cv2.cvtColor(image_array, cv2.COLOR_BGR2RGB)
    
    cv2.imwrite(file_name, image_array, [int(cv2.IMWRITE_JPEG_QUALITY), 90])
    #cv2.imwrite(file_name, image_array)
    with open(file_name, "rb") as image_file:
        return base64.b64encode(image_file.read()).decode('utf-8')
    #return file_name

# Function to process and identify expressions
def identify_expression(image_path,userid):
    base64_image = encode_image(image_path,userid+".jpg")
    client = Groq(api_key=os.environ.get("GROQ_API_KEY"))

    chat_completion = client.chat.completions.create(
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": "どんな表情ですか？ その顔表情から推測できる心境を詳しく説明してください。尚、メンタルカウンセリングが必要かを教えてください。その場合、どのようにカウンセリングを受ければ良いかも教えてください。"},
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{base64_image}",
                        },
                    },
                ],
            }
        ],
        model="meta-llama/llama-4-scout-17b-16e-instruct",
    )

    # USER Emotion Register
    client2 = Groq(api_key=os.environ.get("GROQ_API_KEY2"))
    # Set the system prompt
    system_prompt = {
        "role": "system",
        "content": "You are a helpful assistant, answer questions concisely."
    }

    # Set the user prompt
    user_input = f'{chat_completion.choices[0].message.content}をカウンセリングの情報を除いて要約してください。'
    user_prompt = {
        "role": "user", "content": user_input
    }

    # Initialize the chat history
    chat_history = [system_prompt, user_prompt]

    response = client2.chat.completions.create(
        model="llama-3.3-70b-versatile",
        messages=chat_history,
        max_tokens=1024,
        temperature=0)

    summary = response.choices[0].message.content
    
    # Update user emotion 
    dt_now = datetime.datetime.now()
    today = dt_now.strftime('%Y/%m/%d')
    update_emotion_query = f"UPDATE emotion_data SET emotion_logs = CONCAT(emotion_logs,CHAR(13),CHAR(13),'"+today+"',CHAR(13),'"+summary+"') WHERE userid = '"+userid+"'"
    
    cursor.execute(update_emotion_query)
    conn.commit()

    return chat_completion.choices[0].message.content

def close_mysql():
    cursor.close()  # カーソルを閉じる
    conn.close()    # 接続を閉じる
    return "DB接続が閉じられました!"

def on_userid_change(user_id):
    if (user_id == ''):
        gr.Info('ユーザーIDが入力されていません。')
    else:
        with open("uid.dat", "wb") as f:
            f.write(user_id.encode())

def uid_changed(new_userid):
    return gr.update(visible=True)

def trend_analyze(userid):
    if (userid == ''):
        gr.Info('ユーザーIDが入力されていません。')
    else:
        select_emotion_query = f"select emotion_logs from emotion_data WHERE userid = '"+userid+"'"
        cursor.execute(select_emotion_query)
        rows = cursor.fetchall()

        data = ""
        for row in rows:
            data = row

        parser = PlaintextParser.from_string(data, Tokenizer("japanese")) 
        summarizer = LsaSummarizer()
        summary = summarizer(parser.document, 4090)
        smry = "\n".join(str(sentence) for sentence in summary)

        client2 = Groq(api_key=os.environ.get("GROQ_API_KEY2"))
        # Set the system prompt
        system_prompt = {
            "role": "system",
            "content": "You are a helpful assistant, answer questions concisely."
        }

        # Set the user prompt
        user_input = f'{smry}からこの人の表情や感情の傾向を全体的に分析してください。必ず、呼称は老人、青年などは除いて、人を使ってください。'
        user_prompt = {
            "role": "user", "content": user_input
        }

        # Initialize the chat history
        chat_history = [system_prompt, user_prompt]

        response = client2.chat.completions.create(
            model="llama-3.3-70b-versatile",
            messages=chat_history,
            max_tokens=1024,
            temperature=0)

        analysis = response.choices[0].message.content

        return analysis

if os.path.exists("uid.dat"):
    with open("uid.dat", "rb") as f:
        taken_userid = f.read().decode()

# Gradio Interface
with gr.Blocks(css="footer {visibility: hidden;} #custom_button {width: 400px; margin: 0 auto; background-color: #E0E7FF;}", theme=gr.themes.Soft(), title="EMOTION DETECTION") as emotion:
    gr.Markdown("# Facial Expression Detector")
    
    with gr.Row():
        with gr.Row():
            image = gr.Image(sources=["webcam"], streaming=False, label="撮影する画像")
            userid = gr.Textbox(label="ユーザーID", value=taken_userid)
            reg_button = gr.Button("登録",visible=False,elem_id="custom_button")

    userid.change(fn=uid_changed, inputs=[userid], outputs=[reg_button])

    submit_button = gr.Button("実行",elem_id="custom_button")
    output_text = gr.Textbox(label="結果")
    
    submit_button.click(fn=identify_expression, inputs=[image, userid], outputs=output_text)
    reg_button.click(fn=on_userid_change, inputs=[userid], outputs=None)

    with gr.Row():
            trend_input = gr.Textbox(label="傾向分析ID", value=taken_userid)
            trend_output = gr.Textbox(label="傾向分析結果")

    trend_button = gr.Button("分析",visible=True,elem_id="custom_button")
    trend_button.click(fn=trend_analyze, inputs=[trend_input], outputs=[trend_output])

emotion.launch(favicon_path='favicon.ico')
emotion.close(close_mysql)