import spaces import gradio as gr import pandas as pd import numpy as np import os import re import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt import seaborn as sns import networkx as nx from datetime import datetime from huggingface_hub import HfApi, hf_hub_download from sentence_transformers import SentenceTransformer from sklearn.metrics.pairwise import cosine_similarity from sklearn.manifold import TSNE # ========================================== # ИНИЦИАЛИЗАЦИЯ # ========================================== model = SentenceTransformer('all-MiniLM-L6-v2') def split_into_sentences(text): sentences = re.split(r'(?<=[.!?])\s+', text.strip()) return [s for s in sentences if len(s) > 10] # ========================================== # БАЗА ДАННЫХ # ========================================== DATASET_REPO_ID = "Masterogon/dream-database" DB_FILE = "dream_database.csv" HF_TOKEN = os.environ.get("HF_TOKEN") api = HfApi() def init_db(): if HF_TOKEN: try: print("Downloading database from Hugging Face Hub...") file_path = hf_hub_download(repo_id=DATASET_REPO_ID, filename=DB_FILE, repo_type="dataset", token=HF_TOKEN) import shutil shutil.copy(file_path, DB_FILE) print("Database loaded.") return except Exception as e: print("Could not download DB:", e) if not os.path.exists(DB_FILE): df = pd.DataFrame(columns=["Timestamp", "Alias", "ASC_Type", "Emotion", "Intensity", "Narrative"]) df.to_csv(DB_FILE, index=False) print("Created fresh local database.") init_db() # ========================================== # ЗАПИСЬ НОВОГО ОТЧЁТА # ========================================== def process_entry(alias, asc_type, emotion, intensity, narrative): if not narrative.strip(): return "Error: Please describe your experience.", pd.read_csv(DB_FILE).tail(5) timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") new_data = pd.DataFrame([[timestamp, alias, asc_type, emotion, intensity, narrative]], columns=["Timestamp", "Alias", "ASC_Type", "Emotion", "Intensity", "Narrative"]) new_data.to_csv(DB_FILE, mode='a', header=False, index=False) if HF_TOKEN: try: api.upload_file( path_or_fileobj=DB_FILE, path_in_repo=DB_FILE, repo_id=DATASET_REPO_ID, repo_type="dataset", token=HF_TOKEN, commit_message=f"Added new report by {alias}" ) backup_status = "Successfully synced to cloud." except Exception as e: backup_status = f"Warning: Cloud sync failed ({e})" else: backup_status = "Warning: No HF_TOKEN. Data is local only." return f"Success! Entry added. {backup_status}", pd.read_csv(DB_FILE).tail(10) # ========================================== # MACRO ANALYSIS (Tab 2) — фокус на структурах # ========================================== @spaces.GPU def macro_analysis(): df = pd.read_csv(DB_FILE).dropna(subset=['Narrative']) if len(df) < 3: return None, None, "Недостаточно данных (минимум 3 отчёта)" texts = df['Narrative'].tolist() embeddings = model.encode(texts) sim_matrix = cosine_similarity(embeddings) # Heatmap fig_heat, ax_heat = plt.subplots(figsize=(9, 7)) labels = [f"R{i+1}" for i in range(len(texts))] sns.heatmap(sim_matrix, xticklabels=labels, yticklabels=labels, annot=True, cmap="YlOrRd", fmt=".2f", ax=ax_heat) ax_heat.set_title("Similarity Matrix of Reports (Shared Information Structures)") plt.tight_layout() # Network Graph fig_graph, ax_graph = plt.subplots(figsize=(9, 7)) G = nx.Graph() threshold = 0.45 for i in range(len(texts)): G.add_node(f"R{i+1}", size=1000) for i in range(len(texts)): for j in range(i + 1, len(texts)): if sim_matrix[i, j] > threshold: G.add_edge(f"R{i+1}", f"R{j+1}", weight=sim_matrix[i, j]) pos = nx.spring_layout(G, seed=42) nx.draw_networkx_nodes(G, pos, node_color="lightblue", node_size=1400, ax=ax_graph) nx.draw_networkx_labels(G, pos, font_size=10, font_weight="bold", ax=ax_graph) edges = G.edges() weights = [G[u][v]['weight'] * 7 for u, v in edges] nx.draw_networkx_edges(G, pos, width=weights, edge_color='darkred', alpha=0.7, ax=ax_graph) ax_graph.set_title(f"Network of Shared Semantic Structures (Threshold > {threshold})") ax_graph.axis('off') summary = f"Анализ {len(df)} отчётов. Выявлено {len(G.edges())} сильных связей между информационными структурами." return fig_heat, fig_graph, summary # ========================================== # MICRO ANALYSIS (Tab 3) — sentence-level archetypes # ========================================== @spaces.GPU def micro_analysis(): df = pd.read_csv(DB_FILE).dropna(subset=['Narrative']) if len(df) < 2: return None, "Недостаточно данных" all_sentences = [] parent_report = [] full_preview = [] for idx, row in df.iterrows(): sents = split_into_sentences(row['Narrative']) all_sentences.extend(sents) report_id = f"R{idx+1}" parent_report.extend([report_id] * len(sents)) preview = row['Narrative'][:400] + "..." if len(row['Narrative']) > 400 else row['Narrative'] full_preview.extend([preview] * len(sents)) if len(all_sentences) < 5: return None, "Недостаточно предложений" sent_embeddings = model.encode(all_sentences) perplexity = min(30, len(all_sentences) - 1) tsne = TSNE(n_components=2, random_state=42, perplexity=perplexity) vecs_2d = tsne.fit_transform(sent_embeddings) fig_tsne, ax_tsne = plt.subplots(figsize=(10, 8)) sns.scatterplot(x=vecs_2d[:,0], y=vecs_2d[:,1], hue=parent_report, palette="tab10", s=90, alpha=0.85, ax=ax_tsne) ax_tsne.set_title("Clustering of Individual Semantic Scenes / Fragments") plt.legend(title="Report ID", bbox_to_anchor=(1.05, 1)) plt.tight_layout() # Центральные архетипы sim_matrix = cosine_similarity(sent_embeddings) mean_sims = sim_matrix.mean(axis=1) top_indices = mean_sims.argsort()[-8:][::-1] central_text = "### 🔬 Most Central / Reproducible Semantic Fragments\n\n" for idx in top_indices: central_text += f"**{parent_report[idx]}** — Centrality: {mean_sims[idx]:.3f}\n" central_text += f"\"{all_sentences[idx]}\"\n\n" central_text += f"*Preview of full report:* {full_preview[idx]}\n---\n" return fig_tsne, central_text # ========================================== # HYBRID PATTERN DISCOVERY (Tab 4) # ========================================== @spaces.GPU def hybrid_pattern_discovery(mode, custom_motifs_text): df = pd.read_csv(DB_FILE).dropna(subset=['Narrative']) if len(df) < 3: return "Недостаточно данных (минимум 3 отчёта)" all_sentences = [] parent_report = [] full_preview = [] for idx, row in df.iterrows(): sents = split_into_sentences(row['Narrative']) all_sentences.extend(sents) report_id = f"R{idx+1}" parent_report.extend([report_id] * len(sents)) preview = row['Narrative'][:350] + "..." if len(row['Narrative']) > 350 else row['Narrative'] full_preview.extend([preview] * len(sents)) if len(all_sentences) < 5: return "Недостаточно фрагментов для анализа" if mode == "Blind Extraction (Unsupervised)": sent_embeddings = model.encode(all_sentences) num_clusters = max(3, min(12, len(all_sentences) // 8)) from sklearn.cluster import KMeans kmeans = KMeans(n_clusters=num_clusters, random_state=42, n_init=10) labels = kmeans.fit_predict(sent_embeddings) results = f"### 🧬 Blind Discovery of Reproducible Information Structures\n" results += f"Извлечено {num_clusters} кластеров из {len(all_sentences)} фрагментов.\n\n" for i in range(num_clusters): cluster_idx = np.where(labels == i)[0] if len(cluster_idx) < 2: continue cluster_emb = sent_embeddings[cluster_idx] centroid = kmeans.cluster_centers_[i] distances = cosine_similarity([centroid], cluster_emb)[0] central_local_idx = np.argmax(distances) global_idx = cluster_idx[central_local_idx] reports_in_cluster = set([parent_report[k] for k in cluster_idx]) results += f"#### Archetype Cluster {i+1} — Found across {len(reports_in_cluster)} reports\n" results += f"**Core Signal**: \"{all_sentences[global_idx]}\"\n" results += f"**Strength**: {len(cluster_idx)} fragments | Cross-report validation: {len(reports_in_cluster)}\n\n" count = 0 for k in cluster_idx: if k != global_idx and count < 4: results += f"- {parent_report[k]}: \"{all_sentences[k][:140]}...\"\n" count += 1 results += f"*Full report previews available in data tab.*\n---\n" return results else: # Targeted Search seed_motifs = [m.strip() for m in re.split(r'[,|\n]', custom_motifs_text) if m.strip()] if not seed_motifs: return "Введите хотя бы один мотив для поиска." motif_embeddings = model.encode(seed_motifs) results = "### 🎯 Targeted Search: Testing Specific Hypotheses\n\n" for m_idx, motif in enumerate(seed_motifs): motif_emb = motif_embeddings[m_idx] reports_with_motif = 0 best_matches = [] for idx, row in df.iterrows(): sents = split_into_sentences(row['Narrative']) if not sents: continue sent_embs = model.encode(sents) sims = cosine_similarity([motif_emb], sent_embs)[0] max_sim = np.max(sims) if max_sim > 0.42: reports_with_motif += 1 best_idx = np.argmax(sims) best_matches.append(f"**{row['Alias']} (R{idx+1})**: \"{sents[best_idx][:180]}...\" (sim: {max_sim:.3f})") results += f"#### Target Motif: '{motif}'\n" results += f"**Detected in {reports_with_motif} reports**\n" for match in best_matches[:5]: results += f"- {match}\n" results += "---\n" return results # ========================================== # GRADIO INTERFACE # ========================================== with gr.Blocks(theme=gr.themes.Monochrome()) as app: gr.Markdown("# 🌌 DreamCode — Detector of Reproducible Informational Signals") with gr.Tabs(): # TAB 1: Data Ingestion with gr.TabItem("1. Data Ingestion"): with gr.Row(): with gr.Column(): alias = gr.Textbox(label="Participant ID / Alias (optional)") asc_type = gr.Dropdown(choices=["Ordinary Dream", "Lucid Dream (LD)", "OBE", "NDE", "Other"], label="Altered State") emotion = gr.Radio(choices=["Positive", "Neutral", "Negative"], label="Dominant Emotion") intensity = gr.Slider(1, 3, value=2, step=1, label="Intensity") narrative = gr.Textbox(label="Full Narrative / Description", lines=8) submit_btn = gr.Button("Submit Experience", variant="primary") with gr.Column(): status_output = gr.Textbox(label="Status") data_preview = gr.Dataframe(label="Recent Entries", headers=["Timestamp", "Alias", "ASC_Type", "Emotion", "Intensity", "Narrative"]) submit_btn.click(fn=process_entry, inputs=[alias, asc_type, emotion, intensity, narrative], outputs=[status_output, data_preview]) # TAB 2: Macro Analysis with gr.TabItem("2. Document-Level Structures"): gr.Markdown("Similarity between full reports — looking for shared informational structures.") analyze_macro_btn = gr.Button("Run Macro Analysis", variant="primary") with gr.Row(): heat_plot = gr.Plot(label="Similarity Heatmap") network_plot = gr.Plot(label="Semantic Network") macro_summary = gr.Textbox(label="Summary") analyze_macro_btn.click(fn=macro_analysis, inputs=[], outputs=[heat_plot, network_plot, macro_summary]) # TAB 3: Sentence-Level Analysis with gr.TabItem("3. Scene & Fragment Clustering"): gr.Markdown("Breaks narratives into sentences/scenes and finds central reproducible fragments.") analyze_micro_btn = gr.Button("Run Sentence Analysis", variant="primary") with gr.Row(): tsne_plot = gr.Plot(label="t-SNE of Semantic Fragments") central_text = gr.Markdown(label="Central Archetypal Fragments") analyze_micro_btn.click(fn=micro_analysis, inputs=[], outputs=[tsne_plot, central_text]) # TAB 4: AI Pattern Discovery with gr.TabItem("4. AI Pattern Discovery"): gr.Markdown("### Main Engine: Extracting Reproducible Signals") search_mode = gr.Radio( choices=["Blind Extraction (Unsupervised)", "Targeted Search (Zero-Shot)"], value="Blind Extraction (Unsupervised)", label="Analysis Mode" ) motif_input = gr.Textbox( label="Custom motifs / fragments (comma or new line separated)", lines=3, value="Huge red planet, Approaching celestial body, Global catastrophe feeling", visible=False ) def toggle_input(mode): return gr.update(visible=(mode == "Targeted Search (Zero-Shot)")) search_mode.change(fn=toggle_input, inputs=[search_mode], outputs=[motif_input]) analyze_btn = gr.Button("Run Pattern Discovery Engine", variant="primary") output_md = gr.Markdown() analyze_btn.click(fn=hybrid_pattern_discovery, inputs=[search_mode, motif_input], outputs=[output_md]) app.launch()