app.py

import logging
import json
import os
import zipfile
from difflib import get_close_matches
import base64
from io import BytesIO
from PIL import Image
import random
import numpy as np
import gradio as gr

# Konfiguration des Loggers
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# Globale Variablen
category_nodes = []
questions = []
initialized = False

# Dateipfade
ZIP_FILE = "DRLCogNet.zip"
MODEL_FILE = "model.DRLCogNet"  # Datei ist direkt in der ZIP

def extract_model():
    """
    Entpackt die ZIP-Datei in das aktuelle Verzeichnis, wenn die Modelldatei nicht existiert.
    """
    if not os.path.exists(MODEL_FILE):
        logging.info("Entpacke Modelldatei...")
        with zipfile.ZipFile(ZIP_FILE, 'r') as zip_ref:
            zip_ref.extractall(".")  # Entpackt alle Dateien ins aktuelle Verzeichnis
        logging.info("Modell erfolgreich entpackt.")
        
        # Überprüfe, ob die Dateien entpackt wurden
        extracted_files = os.listdir(".")
        logging.info(f"Entpackte Dateien: {extracted_files}")

        # Prüfe, ob die Modelldatei tatsächlich vorhanden ist
        if MODEL_FILE in extracted_files:
            logging.info(f"Die Datei {MODEL_FILE} wurde erfolgreich extrahiert.")
        else:
            logging.error(f"Die Datei {MODEL_FILE} wurde NICHT gefunden.")

    else:
        logging.info("Modell ist bereits entpackt.")


def load_model_with_questions_and_answers():
    """
    Lädt das Modell mit Fragen und Antworten aus einer JSON-Datei.

    Returns:
        tuple: Die Liste der Kategorie-Knoten und die Liste der Fragen.
    """
    global initialized
    if initialized:
        logging.info("Modell bereits initialisiert.")
        return None, None

    extract_model()

    if not os.path.exists(MODEL_FILE):
        logging.error(f"Modelldatei {MODEL_FILE} nicht gefunden. Überprüfe die ZIP-Datei.")
        return None, None

    try:
        with open(MODEL_FILE, "r", encoding="utf-8") as file:
            model_data = json.load(file)

        nodes_dict = {node_data["label"]: Node(node_data["label"]) for node_data in model_data["nodes"]}

        for node_data in model_data["nodes"]:
            node = nodes_dict[node_data["label"]]
            node.activation = node_data.get("activation", 0.0)
            node.decay_rate = float(node_data.get("decay_rate", 0.0))  # Laden der decay_rate
            for conn_state in node_data["connections"]:
                target_node = nodes_dict.get(conn_state["target"])
                if target_node:
                    node.add_connection(target_node, conn_state["weight"])

        global questions
        questions = model_data.get("questions", [])
        logging.info(f"Modell geladen mit {len(nodes_dict)} Knoten und {len(questions)} Fragen")
        initialized = True
        return list(nodes_dict.values()), questions

    except json.JSONDecodeError as e:
        logging.error(f"Fehler beim Parsen der JSON-Datei: {e}")
        return None, None

def find_similar_question(questions, query):
    """
    Findet die ähnlichste Frage basierend auf einfachen Ähnlichkeitsmetriken.

    Args:
        questions (list): Liste aller Fragen.
        query (str): Die Abfrage, nach der gesucht werden soll.

    Returns:
        dict: Die ähnlichste Frage.
    """
    question_texts = [q['question'] for q in questions]
    closest_matches = get_close_matches(query, question_texts, n=1, cutoff=0.3)

    if closest_matches:
        matched_question = next((q for q in questions if q['question'] == closest_matches[0]), None)
        return matched_question
    else:
        return {"question": "Keine passende Frage gefunden", "category": "Unbekannt"}

def find_best_answer(category_nodes, questions, query):
    """
    Findet die beste Antwort auf eine Abfrage.

    Args:
        category_nodes (list): Liste der Kategorie-Knoten.
        questions (list): Liste der Fragen.
        query (str): Die Abfrage.

    Returns:
        str: Die beste Antwort.
        float: Die Aktivierung des Kategorie-Knotens.
    """
    matched_question = find_similar_question(questions, query)
    if matched_question:
        logging.info(f"Gefundene Frage: {matched_question['question']} -> Kategorie: {matched_question['category']}")
        answer = matched_question.get("answer", "Keine Antwort verfügbar")
        logging.info(f"Antwort: {answer}")
        activation = 0.8  # Beispielhafte Aktivierung
        return answer, activation
    else:
        logging.warning("Keine passende Frage gefunden.")
        return None, None

class Node:
    """
    Ein Knoten im Netzwerk.
    """
    def __init__(self, label):
        self.label = label
        self.connections = []
        self.activation = 0.0
        self.decay_rate = 0.0

    def add_connection(self, target_node, weight=None):
        """
        Fügt eine Verbindung zu einem Zielknoten hinzu.

        Args:
            target_node (Node): Der Zielknoten.
            weight (float): Das Gewicht der Verbindung.
        """
        self.connections.append(Connection(target_node, weight))

class Connection:
    """
    Eine Verbindung zwischen zwei Knoten im Netzwerk.
    """
    def __init__(self, target_node, weight=None):
        self.target_node = target_node
        self.weight = weight if weight is not None else random.uniform(0.1, 1.0)

def get_answer(query):
    answer, activation = find_best_answer(category_nodes, questions, query)
    if answer:
        # Dekodieren der Base64-Bilddaten
        try:
            image_data = base64.b64decode(answer)
            image = Image.open(BytesIO(image_data))
            return image, activation, f"{activation:.2f}"
        except Exception as e:
            logging.error(f"Fehler beim Dekodieren des Bildes: {e}")
            return None, None, "0.00"
    else:
        return None, None, "0.00"

# Lade das Modell und die Fragen
category_nodes, questions = load_model_with_questions_and_answers()

if category_nodes is None or questions is None:
    logging.error("Fehler beim Laden des Modells. Stelle sicher, dass die ZIP-Datei vorhanden ist.")
else:
    # Erstelle die Gradio-App
    with gr.Blocks(title="DRL-CogNet Dog Finder: wrote a number behind the dog name") as demo:
        gr.Markdown("# Frage an das Modell (wrote a numer behind the dog name: ein golden retriever 2")
        with gr.Row():
            with gr.Column():
                question_input = gr.Textbox(label="Frage")
                submit_button = gr.Button("Antwort abrufen")
            with gr.Column():
                image_output = gr.Image(label="Antwortbild")
                activation_output = gr.Number(label="Aktivierung")
                weight_output = gr.Text(label="Gewichtung")

        submit_button.click(fn=get_answer, inputs=question_input, outputs=[image_output, activation_output, weight_output])

    demo.launch()