Upload 3 files

DRLCogNet.zip

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:0ea989215f7b0108400126a5d4fe1bc6b6e887f41c675ddf22d9ce99922e2c8f
+size 779770517

gradio_app.py

@@ -0,0 +1,314 @@
+import logging
+import json
+import os
+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
+
+def load_model_with_questions_and_answers(filename="https://drive.google.com/file/d/1Bd7ILLW8ipgoNHSYqZp97VKVt5tR75OM/view?usp=drive_link"):
+    """
+    Lädt das Modell mit Fragen und Antworten aus einer JSON-Datei.
+
+    Args:
+        filename (str): Der Dateiname der 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
+
+    if not os.path.exists(filename):
+        logging.warning(f"Datei {filename} nicht gefunden. Netzwerk wird initialisiert.")
+        return None, None
+
+    try:
+        with open(filename, "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"])
+
+        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 = simulate_question_answering(category_nodes, matched_question['question'], questions)
+        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.activation_history = []
+        self.decay_rate = 0.0  # Initialisierung der decay_rate
+
+    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))
+
+    def save_state(self):
+        """
+        Speichert den Zustand des Knotens.
+
+        Returns:
+            dict: Der gespeicherte Zustand des Knotens.
+        """
+        return {
+            "label": self.label,
+            "activation": self.activation,
+            "activation_history": self.activation_history,
+            "decay_rate": self.decay_rate,  # Speichern der decay_rate
+            "connections": [{"target": conn.target_node.label, "weight": conn.weight} for conn in self.connections]
+        }
+
+    @staticmethod
+    def load_state(state, nodes_dict):
+        """
+        Lädt den Zustand eines Knotens.
+
+        Args:
+            state (dict): Der gespeicherte Zustand des Knotens.
+            nodes_dict (dict): Ein Dictionary der Knoten.
+
+        Returns:
+            Node: Der geladene Knoten.
+        """
+        node = Node(state["label"])
+        node.activation = state["activation"]
+        node.activation_history = state["activation_history"]
+        node.decay_rate = float(state["decay_rate"])  # Laden der decay_rate
+        for conn_state in state["connections"]:
+            target_node = nodes_dict[conn_state["target"]]
+            connection = Connection(target_node, conn_state["weight"])
+            node.connections.append(connection)
+        return node
+
+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 simulate_question_answering(category_nodes, question, questions):
+    """
+    Simuliert die Beantwortung einer Frage im Netzwerk.
+
+    Args:
+        category_nodes (list): Liste der Kategorie-Knoten.
+        question (str): Die Frage, die beantwortet werden soll.
+        questions (list): Liste aller Fragen.
+
+    Returns:
+        float: Die Aktivierung des Kategorie-Knotens.
+    """
+    category = next((q['category'] for q in questions if q['question'] == question), None)
+    if not category:
+        logging.warning(f"Frage '{question}' nicht gefunden!")
+        return None
+
+    category_node = next((node for node in category_nodes if node.label == category), None)
+    if category_node:
+        propagate_signal(category_node, input_signal=0.9, emotion_weights={}, emotional_state=1.0)
+        activation = category_node.activation
+        if activation is None or activation <= 0:
+            logging.warning(f"Kategorie '{category}' hat eine ungültige Aktivierung: {activation}")
+            return 0.0  # Rückgabe von 0, falls die Aktivierung fehlschlägt
+        logging.info(f"Verarbeite Frage: '{question}' → Kategorie: '{category}' mit Aktivierung {activation:.4f}")
+        return activation  # Entfernte doppelte Logging-Ausgabe
+    else:
+        logging.warning(f"Kategorie '{category}' nicht im Netzwerk gefunden. Die Kategorie wird neu hinzugefügt!")
+        return 0.0
+
+def propagate_signal(node, input_signal, emotion_weights, emotional_state=1.0, context_factors=None):
+    """
+    Propagiert ein Signal durch das Netzwerk.
+
+    Args:
+        node (Node): Der Knoten, an dem das Signal beginnt.
+        input_signal (float): Das Eingangssignal.
+        emotion_weights (dict): Die emotionalen Gewichte.
+        emotional_state (float): Der emotionale Zustand.
+        context_factors (dict): Die kontextuellen Faktoren.
+    """
+    node.activation = add_activation_noise(sigmoid(input_signal * random.uniform(0.8, 1.2)))
+    node.activation_history.append(node.activation)  # Aktivierung speichern
+    node.activation = apply_emotion_weight(node.activation, node.label, emotion_weights, emotional_state)
+    if context_factors:
+        node.activation = apply_contextual_factors(node.activation, node, context_factors)
+    logging.info(f"Signalpropagation für {node.label}: Eingangssignal {input_signal:.4f}")
+    for connection in node.connections:
+        logging.info(f"  → Signal an {connection.target_node.label} mit Gewicht {connection.weight:.4f}")
+        connection.target_node.activation += node.activation * connection.weight
+
+def add_activation_noise(activation, noise_level=0.1):
+    """
+    Fügt Rauschen zur Aktivierung hinzu.
+
+    Args:
+        activation (float): Die Aktivierung.
+        noise_level (float): Das Rausch-Level.
+
+    Returns:
+        float: Die Aktivierung mit Rauschen.
+    """
+    noise = np.random.normal(0, noise_level)
+    return np.clip(activation + noise, 0.0, 1.0)
+
+def sigmoid(x):
+    """
+    Berechnet die Sigmoid-Funktion.
+
+    Args:
+        x (float): Der Eingabewert.
+
+    Returns:
+        float: Der Ausgabewert der Sigmoid-Funktion.
+    """
+    return 1 / (1 + np.exp(-x))
+
+def apply_emotion_weight(activation, category_label, emotion_weights, emotional_state=1.0):
+    """
+    Wendet ein emotionales Gewicht auf die Aktivierung an.
+
+    Args:
+        activation (float): Die Aktivierung.
+        category_label (str): Das Label der Kategorie.
+        emotion_weights (dict): Die emotionalen Gewichte.
+        emotional_state (float): Der emotionale Zustand.
+
+    Returns:
+        float: Die gewichtete Aktivierung.
+    """
+    emotion_factor = emotion_weights.get(category_label, 1.0) * emotional_state
+    return activation * emotion_factor
+
+def apply_contextual_factors(activation, node, context_factors):
+    """
+    Wendet kontextuelle Faktoren auf die Aktivierung an.
+
+    Args:
+        activation (float): Die Aktivierung.
+        node (Node): Der Knoten.
+        context_factors (dict): Die kontextuellen Faktoren.
+
+    Returns:
+        float: Die aktualisierte Aktivierung.
+    """
+    context_factor = context_factors.get(node.label, 1.0)
+    return activation * context_factor * random.uniform(0.9, 1.1)
+
+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("https://drive.google.com/file/d/1Bd7ILLW8ipgoNHSYqZp97VKVt5tR75OM/view?usp=drive_link")
+if category_nodes is None or questions is None:
+    logging.error("Fehler beim Laden des Modells. Stellen Sie sicher, dass die Datei 'model_with_qa.json' vorhanden ist.")
+else:
+    # Erstelle die Gradio-App
+    with gr.Blocks(title="DRL-CogNet Dog finder") as demo: # Hinzugefügter Titel
+        gr.Markdown("# Frage an das Modell")
+        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()

requirements.txt

@@ -0,0 +1,9 @@
+logging
+json
+os
+difflib
+base64
+pillow
+random
+numpy
+gradio