| # ============================================================ | |
| # AQARIONZ β POLYGLOT RESEARCH, SIMULATION & TRAINING ENGINE | |
| # ============================================================ | |
| # Fully Integrated Domains: | |
| # - Audio FFT analysis | |
| # - Laser audio modulation (simulated) | |
| # - Solar / photodiode reception (simulated) | |
| # - Cymatic visualization | |
| # - Experiment logging | |
| # - Memory system | |
| # - Knowledge graph (nodes + relations) | |
| # - Creative output tracking | |
| # - Full JSON export for training pipelines | |
| # ============================================================ | |
| import os | |
| import json | |
| import uuid | |
| import sqlite3 | |
| import datetime | |
| from typing import Dict, Any | |
| import numpy as np | |
| import gradio as gr | |
| import matplotlib.pyplot as plt | |
| # ============================================================ | |
| # CONFIG | |
| # ============================================================ | |
| DB_PATH = "aqarionz_system.db" | |
| EXPORT_DIR = "exports" | |
| os.makedirs(EXPORT_DIR, exist_ok=True) | |
| # ============================================================ | |
| # DATABASE LAYER | |
| # ============================================================ | |
| class AqarionzDB: | |
| def __init__(self, path=DB_PATH): | |
| self.conn = sqlite3.connect(path, check_same_thread=False) | |
| self.conn.row_factory = sqlite3.Row | |
| self._init_schema() | |
| def _init_schema(self): | |
| c = self.conn.cursor() | |
| c.execute(""" | |
| CREATE TABLE IF NOT EXISTS experiments ( | |
| id TEXT PRIMARY KEY, | |
| timestamp TEXT, | |
| description TEXT, | |
| parameters TEXT, | |
| results TEXT | |
| ) | |
| """) | |
| c.execute(""" | |
| CREATE TABLE IF NOT EXISTS memories ( | |
| id TEXT PRIMARY KEY, | |
| timestamp TEXT, | |
| content TEXT, | |
| emotion TEXT, | |
| tags TEXT, | |
| source TEXT | |
| ) | |
| """) | |
| c.execute(""" | |
| CREATE TABLE IF NOT EXISTS creative_outputs ( | |
| id TEXT PRIMARY KEY, | |
| timestamp TEXT, | |
| title TEXT, | |
| content TEXT, | |
| tags TEXT | |
| ) | |
| """) | |
| c.execute(""" | |
| CREATE TABLE IF NOT EXISTS knowledge_nodes ( | |
| id TEXT PRIMARY KEY, | |
| label TEXT, | |
| description TEXT, | |
| tags TEXT, | |
| created_at TEXT | |
| ) | |
| """) | |
| c.execute(""" | |
| CREATE TABLE IF NOT EXISTS relationships ( | |
| id TEXT PRIMARY KEY, | |
| from_id TEXT, | |
| to_id TEXT, | |
| relation TEXT, | |
| weight REAL, | |
| created_at TEXT | |
| ) | |
| """) | |
| self.conn.commit() | |
| def _now(self): | |
| return datetime.datetime.utcnow().isoformat() | |
| # ---------------- INSERT ---------------- | |
| def log_experiment(self, desc, params, results): | |
| eid = str(uuid.uuid4()) | |
| self.conn.execute( | |
| "INSERT INTO experiments VALUES (?, ?, ?, ?, ?)", | |
| (eid, self._now(), desc, json.dumps(params), json.dumps(results)) | |
| ) | |
| self.conn.commit() | |
| return eid | |
| def add_memory(self, content, emotion, tags, source="user"): | |
| mid = str(uuid.uuid4()) | |
| self.conn.execute( | |
| "INSERT INTO memories VALUES (?, ?, ?, ?, ?, ?)", | |
| (mid, self._now(), content, emotion, tags, source) | |
| ) | |
| self.conn.commit() | |
| return mid | |
| def add_creative(self, title, content, tags): | |
| cid = str(uuid.uuid4()) | |
| self.conn.execute( | |
| "INSERT INTO creative_outputs VALUES (?, ?, ?, ?, ?)", | |
| (cid, self._now(), title, content, tags) | |
| ) | |
| self.conn.commit() | |
| return cid | |
| def add_node(self, label, description, tags): | |
| nid = str(uuid.uuid4()) | |
| self.conn.execute( | |
| "INSERT INTO knowledge_nodes VALUES (?, ?, ?, ?, ?)", | |
| (nid, label, description, tags, self._now()) | |
| ) | |
| self.conn.commit() | |
| return nid | |
| def add_relationship(self, from_id, to_id, relation, weight): | |
| rid = str(uuid.uuid4()) | |
| self.conn.execute( | |
| "INSERT INTO relationships VALUES (?, ?, ?, ?, ?, ?)", | |
| (rid, from_id, to_id, relation, weight, self._now()) | |
| ) | |
| self.conn.commit() | |
| return rid | |
| # ---------------- EXPORT ---------------- | |
| def export_all(self): | |
| data = {} | |
| for table in [ | |
| "experiments", | |
| "memories", | |
| "creative_outputs", | |
| "knowledge_nodes", | |
| "relationships" | |
| ]: | |
| rows = self.conn.execute(f"SELECT * FROM {table}").fetchall() | |
| data[table] = [dict(r) for r in rows] | |
| path = os.path.join(EXPORT_DIR, f"aqarionz_export_{uuid.uuid4()}.json") | |
| with open(path, "w", encoding="utf-8") as f: | |
| json.dump(data, f, indent=2) | |
| return path | |
| db = AqarionzDB() | |
| # ============================================================ | |
| # SIGNAL PROCESSING | |
| # ============================================================ | |
| def compute_fft(audio, sr): | |
| freqs = np.fft.rfftfreq(len(audio), 1 / sr) | |
| mags = np.abs(np.fft.rfft(audio)) | |
| return freqs, mags | |
| def modulate_laser(audio, depth=0.5): | |
| norm = audio / (np.max(np.abs(audio)) + 1e-9) | |
| return norm * depth | |
| def simulate_solar(laser_signal): | |
| noise = np.random.normal(0, 0.02, size=laser_signal.shape) | |
| return laser_signal + noise | |
| def cymatic_plot(freqs, mags): | |
| fig, ax = plt.subplots(figsize=(4, 4)) | |
| ax.pcolormesh( | |
| freqs, | |
| np.arange(len(mags)), | |
| np.tile(mags, (len(mags), 1)), | |
| shading="auto" | |
| ) | |
| ax.set_title("Cymatic Pattern (Synthetic)") | |
| ax.set_xlabel("Frequency (Hz)") | |
| ax.set_ylabel("Mode Index") | |
| plt.tight_layout() | |
| return fig | |
| # ============================================================ | |
| # INFERENCE HELPERS | |
| # ============================================================ | |
| def infer_emotion(text): | |
| t = text.lower() | |
| if any(w in t for w in ["love", "hope", "joy", "peace"]): | |
| return "positive" | |
| if any(w in t for w in ["fear", "anger", "sad", "loss"]): | |
| return "negative" | |
| if any(w in t for w in ["why", "how", "what"]): | |
| return "curious" | |
| return "neutral" | |
| def auto_tags(text): | |
| keys = [] | |
| for k in ["system", "signal", "memory", "knowledge", "future", "graph", "audio"]: | |
| if k in text.lower(): | |
| keys.append(k) | |
| return ",".join(keys) | |
| # ============================================================ | |
| # UI ACTIONS | |
| # ============================================================ | |
| def audio_fft_ui(file): | |
| if file is None: | |
| return None, "No audio provided" | |
| sr, data = file | |
| data = data.astype(np.float32) | |
| freqs, mags = compute_fft(data, sr) | |
| peak = float(freqs[np.argmax(mags)]) | |
| eid = db.log_experiment( | |
| "Audio FFT Analysis", | |
| {"sample_rate": sr, "length": len(data)}, | |
| {"peak_frequency": peak} | |
| ) | |
| fig = cymatic_plot(freqs, mags) | |
| return fig, f"Experiment {eid}\nPeak Frequency: {peak:.2f} Hz" | |
| def laser_sim_ui(file): | |
| if file is None: | |
| return "No audio provided" | |
| sr, audio = file | |
| audio = audio.astype(np.float32) | |
| laser = modulate_laser(audio) | |
| solar = simulate_solar(laser) | |
| f_tx, m_tx = compute_fft(laser, sr) | |
| f_rx, m_rx = compute_fft(solar, sr) | |
| result = { | |
| "tx_peak": float(f_tx[np.argmax(m_tx)]), | |
| "rx_peak": float(f_rx[np.argmax(m_rx)]) | |
| } | |
| eid = db.log_experiment( | |
| "Laser β Solar Simulation", | |
| {"sample_rate": sr}, | |
| result | |
| ) | |
| return f"Experiment {eid}\nTX Peak: {result['tx_peak']:.1f} Hz\nRX Peak: {result['rx_peak']:.1f} Hz" | |
| def add_memory_ui(text): | |
| emotion = infer_emotion(text) | |
| tags = auto_tags(text) | |
| mid = db.add_memory(text, emotion, tags) | |
| return f"Memory stored\nID: {mid}\nEmotion: {emotion}\nTags: {tags}" | |
| def add_creative_ui(title, content): | |
| tags = auto_tags(content) | |
| cid = db.add_creative(title, content, tags) | |
| return f"Creative output saved\nID: {cid}" | |
| def add_node_ui(label, desc): | |
| tags = auto_tags(desc) | |
| nid = db.add_node(label, desc, tags) | |
| return f"Knowledge node created\nID: {nid}" | |
| def link_nodes_ui(from_id, to_id, relation, weight): | |
| rid = db.add_relationship(from_id, to_id, relation, weight) | |
| return f"Relationship created\nID: {rid}" | |
| def export_ui(): | |
| path = db.export_all() | |
| return f"Export complete:\n{path}" | |
| # ============================================================ | |
| # GRADIO UI | |
| # ============================================================ | |
| with gr.Blocks(title="AQARIONZ POLYGLOT ENGINE") as app: | |
| gr.Markdown(""" | |
| # π AQARIONZ POLYGLOT RESEARCH & TRAINING ENGINE | |
| Signal β’ Simulation β’ Memory β’ Knowledge β’ Export | |
| """) | |
| with gr.Tabs(): | |
| with gr.Tab("π Audio FFT"): | |
| audio_in = gr.Audio(type="numpy") | |
| plot = gr.Plot() | |
| txt = gr.Textbox() | |
| gr.Button("Run FFT").click(audio_fft_ui, audio_in, [plot, txt]) | |
| with gr.Tab("π¦ Laser / Solar"): | |
| laser_audio = gr.Audio(type="numpy") | |
| out = gr.Textbox() | |
| gr.Button("Simulate").click(laser_sim_ui, laser_audio, out) | |
| with gr.Tab("π Memory"): | |
| mem = gr.Textbox(lines=6) | |
| mem_out = gr.Textbox(lines=6) | |
| gr.Button("Store").click(add_memory_ui, mem, mem_out) | |
| with gr.Tab("β¨ Creative"): | |
| title = gr.Textbox() | |
| content = gr.Textbox(lines=6) | |
| cre_out = gr.Textbox(lines=6) | |
| gr.Button("Save").click(add_creative_ui, [title, content], cre_out) | |
| with gr.Tab("π§ Knowledge"): | |
| label = gr.Textbox() | |
| desc = gr.Textbox(lines=6) | |
| node_out = gr.Textbox() | |
| gr.Button("Create Node").click(add_node_ui, [label, desc], node_out) | |
| with gr.Tab("π Relations"): | |
| f = gr.Textbox(label="From ID") | |
| t = gr.Textbox(label="To ID") | |
| r = gr.Textbox(label="Relation") | |
| w = gr.Slider(0, 10, value=1) | |
| rel_out = gr.Textbox() | |
| gr.Button("Link").click(link_nodes_ui, [f, t, r, w], rel_out) | |
| with gr.Tab("π¦ Export"): | |
| exp_out = gr.Textbox(lines=6) | |
| gr.Button("Export All").click(export_ui, None, exp_out) | |
| gr.Markdown(""" | |
| --- | |
| **State:** Unified | |
| **Mode:** Simulation + Research + Training | |
| **Output:** JSON / DB / ML-ready | |
| """) | |
| if __name__ == "__main__": | |
| app.launch() |