app.py

import torch
import numpy as np
import gc
from typing import Dict, Any, Optional, List

from .llm_iface import get_or_load_model, LLM
from .resonance_seismograph import run_cogitation_loop, run_silent_cogitation_seismic
from .concepts import get_concept_vector
from .introspection import generate_introspective_report
from .signal_analysis import analyze_cognitive_signal, get_power_spectrum_for_plotting
from .utils import dbg

def run_seismic_analysis(
    model_id: str,
    prompt_type: str,
    seed: int,
    num_steps: int,
    concept_to_inject: str,
    injection_strength: float,
    progress_callback,
    llm_instance: Optional[LLM] = None,
    injection_vector_cache: Optional[torch.Tensor] = None
) -> Dict[str, Any]:
    """
    Orchestriert eine einzelne seismische Analyse und integriert nun standardmäßig
    die fortgeschrittene Signal-Analyse.
    """
    local_llm_instance = False
    if llm_instance is None:
        progress_callback(0.0, desc=f"Loading model '{model_id}'...")
        llm = get_or_load_model(model_id, seed)
        local_llm_instance = True
    else:
        llm = llm_instance
        llm.set_all_seeds(seed)

    injection_vector = None
    if concept_to_inject and concept_to_inject.strip():
        if injection_vector_cache is not None:
            dbg(f"Using cached injection vector for '{concept_to_inject}'.")
            injection_vector = injection_vector_cache
        else:
            progress_callback(0.2, desc=f"Vectorizing '{concept_to_inject}'...")
            injection_vector = get_concept_vector(llm, concept_to_inject.strip())

    progress_callback(0.3, desc=f"Recording dynamics for '{prompt_type}'...")

    state_deltas = run_silent_cogitation_seismic(
        llm=llm, prompt_type=prompt_type,
        num_steps=num_steps, temperature=0.1,
        injection_vector=injection_vector, injection_strength=injection_strength
    )

    progress_callback(0.9, desc="Analyzing...")
    
    stats = {}
    results = {}
    verdict = "### ⚠️ Analysis Warning\nNo state changes recorded."

    if state_deltas:
        deltas_np = np.array(state_deltas)
        stats = { 
            "mean_delta": float(np.mean(deltas_np)), 
            "std_delta": float(np.std(deltas_np)), 
            "max_delta": float(np.max(deltas_np)), 
            "min_delta": float(np.min(deltas_np)), 
        }
        
        # FINALE KORREKTUR: Führe die Signal-Analyse hier standardmäßig durch
        signal_metrics = analyze_cognitive_signal(deltas_np)
        stats.update(signal_metrics)

        freqs, power = get_power_spectrum_for_plotting(deltas_np)

        verdict = f"### ✅ Seismic Analysis Complete\nRecorded {len(deltas_np)} steps for '{prompt_type}'."
        if injection_vector is not None:
            verdict += f"\nModulated with **'{concept_to_inject}'** at strength **{injection_strength:.2f}**."
        
        results["power_spectrum"] = {"frequencies": freqs.tolist(), "power": power.tolist()}

    results.update({ "verdict": verdict, "stats": stats, "state_deltas": state_deltas })

    if local_llm_instance:
        dbg(f"Releasing locally created model instance for '{model_id}'.")
        del llm, injection_vector
        gc.collect()
        if torch.cuda.is_available(): torch.cuda.empty_cache()

    return results

# Die anderen Orchestrator-Funktionen (run_triangulation_probe, run_causal_surgery_probe, run_act_titration_probe)
# bleiben unverändert, da sie ihre eigene, spezifische Analyse-Logik enthalten.```
[File Ends] `cognitive_mapping_probe/orchestrator_seismograph.py`

[File Begins] `app.py`
```python
import gradio as gr
import pandas as pd
import gc
import torch
import json

from cognitive_mapping_probe.orchestrator_seismograph import run_seismic_analysis
from cognitive_mapping_probe.auto_experiment import run_auto_suite, get_curated_experiments
from cognitive_mapping_probe.prompts import RESONANCE_PROMPTS
from cognitive_mapping_probe.utils import dbg, cleanup_memory

theme = gr.themes.Soft(primary_hue="indigo", secondary_hue="blue").set(body_background_fill="#f0f4f9", block_background_fill="white")

def run_single_analysis_display(*args, progress=gr.Progress(track_tqdm=True)):
    """
    Wrapper für den 'Manual Single Run'-Tab, jetzt mit Frequenz-Analyse.
    """
    try:
        results = run_seismic_analysis(*args, progress_callback=progress)
        stats, deltas = results.get("stats", {}), results.get("state_deltas", [])
        
        # Zeitreihen-Plot
        df_time = pd.DataFrame({"Internal Step": range(len(deltas)), "State Change (Delta)": deltas})
        
        # Frequenz-Plot
        spectrum_data = []
        if "power_spectrum" in results:
            spectrum = results["power_spectrum"]
            for freq, power in zip(spectrum["frequencies"], spectrum["power"]):
                if freq > 0.001:
                    spectrum_data.append({"Frequency": freq, "Power": power})
        df_freq = pd.DataFrame(spectrum_data)

        # Update der Statistik-Anzeige
        stats_md = f"""### Statistical Signature
- **Mean Delta:** {stats.get('mean_delta', 0):.4f}
- **Std Dev Delta:** {stats.get('std_delta', 0):.4f}
- **Dominant Frequency:** {stats.get('dominant_frequency', 0):.4f} Hz
- **Spectral Entropy:** {stats.get('spectral_entropy', 0):.4f}"""
        
        serializable_results = json.dumps(results, indent=2, default=str)
        return f"{results.get('verdict', 'Error')}\n\n{stats_md}", df_time, df_freq, serializable_results
    finally:
        cleanup_memory()


def run_auto_suite_display(model_id, num_steps, seed, experiment_name, progress=gr.Progress(track_tqdm=True)):
    """Wrapper für den 'Automated Suite'-Tab."""
    try:
        summary_df, plot_df, all_results = run_auto_suite(model_id, int(num_steps), int(seed), experiment_name, progress)
        
        dataframe_component = gr.DataFrame(label="Comparative Signature (incl. Signal Metrics)", value=summary_df, wrap=True, row_count=(len(summary_df), "dynamic"))

        # Zeitreihen-Plot
        plot_params_time = {
            "title": "Comparative Cognitive Dynamics (Time Domain)",
            "color_legend_position": "bottom", "show_label": True, "height": 300, "interactive": True
        }
        if experiment_name == "Mechanistic Probe (Attention Entropies)":
            plot_params_time.update({"x": "Step", "y": "Value", "color": "Metric", "color_legend_title": "Metric"})
        else:
            plot_params_time.update({"x": "Step", "y": "Delta", "color": "Experiment", "color_legend_title": "Experiment Runs"})
        
        time_domain_plot = gr.LinePlot(value=plot_df, **plot_params_time)

        # Frequenz-Spektrum-Plot
        spectrum_data = []
        for label, result in all_results.items():
            if "power_spectrum" in result:
                spectrum = result["power_spectrum"]
                for freq, power in zip(spectrum["frequencies"], spectrum["power"]):
                    if freq > 0.001:
                        spectrum_data.append({"Frequency": freq, "Power": power, "Experiment": label})
        
        spectrum_df = pd.DataFrame(spectrum_data)
        
        spectrum_plot_params = {
            "x": "Frequency", "y": "Power", "color": "Experiment",
            "title": "Cognitive Frequency Fingerprint (Frequency Domain)", "height": 300,
            "color_legend_position": "bottom", "show_label": True, "interactive": True,
            "color_legend_title": "Experiment Runs",
        }
        frequency_domain_plot = gr.LinePlot(value=spectrum_df, **spectrum_plot_params)

        serializable_results = json.dumps(all_results, indent=2, default=str)
        return dataframe_component, time_domain_plot, frequency_domain_plot, serializable_results
    finally:
        cleanup_memory()

with gr.Blocks(theme=theme, title="Cognitive Seismograph 2.3") as demo:
    gr.Markdown("# 🧠 Cognitive Seismograph 2.3: Advanced Experiment Suite")

    with gr.Tabs():
        with gr.TabItem("🔬 Manual Single Run"):
            gr.Markdown("Run a single experiment with manual parameters to explore specific hypotheses.")
            with gr.Row(variant='panel'):
                with gr.Column(scale=1):
                    gr.Markdown("### 1. General Parameters")
                    manual_model_id = gr.Textbox(value="google/gemma-3-1b-it", label="Model ID")
                    manual_prompt_type = gr.Radio(choices=list(RESONANCE_PROMPTS.keys()), value="resonance_prompt", label="Prompt Type")
                    manual_seed = gr.Slider(1, 1000, 42, step=1, label="Seed")
                    manual_num_steps = gr.Slider(50, 1000, 300, step=10, label="Number of Internal Steps")
                    
                    gr.Markdown("### 2. Modulation Parameters")
                    manual_concept = gr.Textbox(label="Concept to Inject", placeholder="e.g., 'calmness'")
                    manual_strength = gr.Slider(0.0, 5.0, 1.5, step=0.1, label="Injection Strength")
                    manual_run_btn = gr.Button("Run Single Analysis", variant="primary")
                
                with gr.Column(scale=2):
                    gr.Markdown("### Single Run Results")
                    manual_verdict = gr.Markdown("Analysis results will appear here.")
                    with gr.Row():
                        manual_time_plot = gr.LinePlot(x="Internal Step", y="State Change (Delta)", title="Time Domain")
                        manual_freq_plot = gr.LinePlot(x="Frequency", y="Power", title="Frequency Domain")
                    with gr.Accordion("Raw JSON Output", open=False):
                        manual_raw_json = gr.JSON()

            manual_run_btn.click(
                fn=run_single_analysis_display,
                inputs=[manual_model_id, manual_prompt_type, manual_seed, manual_num_steps, manual_concept, manual_strength],
                outputs=[manual_verdict, manual_time_plot, manual_freq_plot, manual_raw_json]
            )

        with gr.TabItem("🚀 Automated Suite"):
            gr.Markdown("Run a predefined, curated suite of experiments and visualize the results comparatively.")
            with gr.Row(variant='panel'):
                with gr.Column(scale=1):
                    gr.Markdown("### Auto-Experiment Parameters")
                    auto_model_id = gr.Textbox(value="google/gemma-3-12b-it", label="Model ID")
                    auto_num_steps = gr.Slider(50, 1000, 300, step=10, label="Steps per Run")
                    auto_seed = gr.Slider(1, 1000, 42, step=1, label="Seed")
                    auto_experiment_name = gr.Dropdown(
                        choices=list(get_curated_experiments().keys()), 
                        value="Causal Verification & Crisis Dynamics", 
                        label="Curated Experiment Protocol"
                    )
                    auto_run_btn = gr.Button("Run Curated Auto-Experiment", variant="primary")
                
                with gr.Column(scale=2):
                    gr.Markdown("### Suite Results Summary")
                    auto_summary_df = gr.DataFrame(label="Comparative Signature (incl. Signal Metrics)", wrap=True)
                    with gr.Row():
                        auto_time_plot_output = gr.LinePlot()
                        auto_freq_plot_output = gr.LinePlot()

                    with gr.Accordion("Raw JSON for all runs", open=False):
                        auto_raw_json = gr.JSON()
            
            auto_run_btn.click(
                fn=run_auto_suite_display,
                inputs=[auto_model_id, auto_num_steps, auto_seed, auto_experiment_name],
                outputs=[auto_summary_df, auto_time_plot_output, auto_freq_plot_output, auto_raw_json]
            )

if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860, debug=True)