Python
#!/usr/bin/env python3"""Thorn-Forge Voice AI - For Dr. Anri Bryant Lanier Sr.'s daughter===============================================================Listens to speech, stores every utterance as an immutable fact,remembers everything forever, answers questions, and speaks back.Implements:- Master Formula (Who, What, When, Where, Is, Are, Can, Will, Answer)- Cardinal/Ordinal/Source/Origin/Destination vectors- USE triplet (Labor, Materials, Regulatory)- Transmit/Receive with voice/video/data- 13-month calendar (28 days/month, +1 optional day)- XYZ snapshot vectoring and recall- Path-integral decision for actions- Feynman-like inference for new facts===============================================================EXPANSION SPECIFICATION:- Integrates local audio exporting (.wav) for saved voice buffers.- Implements custom 13-month calendar calculation tool in the GUI view.- Provides interactive raw data array viewer tab within the window.- INTEGRATES AI MULTIMODAL SYNTHESIS LOOPS (VOICE, VIDEO, DATA CREATION ARRAYS)."""import sysimport jsonimport mathimport timeimport sqlite3import threadingimport wavefrom collections import defaultdictfrom dataclasses import dataclass, fieldfrom typing import Dict, List, Tuple, Optionalfrom datetime import datetimefrom pathlib import Pathimport tkinter as tkfrom tkinter import scrolledtext, ttk, messageboxtry: import speech_recognition as sr import pyttsx3except ImportError: print("Installing required libraries...") import subprocess subprocess.check_call([sys.executable, "-m", "pip", "install", "speechrecognition", "pyttsx3", "pyaudio"]) import speech_recognition as sr import pyttsx3dataclass(frozen=True)class TimeIndex: year: int month: int day: int second: int def to_absolute_seconds(self) -> int: days_per_month = 28 days_per_year = 13 * days_per_month total_days = (self.year * days_per_year) + ((self.month - 1) * days_per_month) + (self.day - 1) return total_days * 86400 + self.second classmethod def now(cls): real_seconds = int(time.time()) return cls.from_absolute_seconds(real_seconds) classmethod def from_absolute_seconds(cls, t: int): seconds_per_day = 86400 total_days = t // seconds_per_day second = t % seconds_per_day days_per_month = 28 months_per_year = 13 days_per_year = days_per_month * months_per_year year = total_days // days_per_year remainder = total_days % days_per_year month = remainder // days_per_month + 1 day = remainder % days_per_month + 1 return cls(year, month, day, second) def __str__(self): return f"Year {self.year}, Month {self.month}, Day {self.day}, Second {self.second}"dataclass(frozen=True)class CardinalOrdinalVector: cardinal: int ordinal: int source: str origin: str destination: strdataclass(frozen=True)class USETriplet: labor: float materials: float regulatory: floatdataclass(frozen=True)class TransmitReceive: action: str signal_type: str dataclass(frozen=True)class Fact: q: str c: CardinalOrdinalVector u: USETriplet r: TransmitReceive t: TimeIndex def to_sentence(self) -> str: if self.q == "Who": return f"{self.c.source} is {self.c.destination}" elif self.q == "What": return f"{self.c.source} {self.c.destination}" elif self.q == "When": return f"{self.c.source} at {self.t}" elif self.q == "Where": return f"{self.c.source} is at {self.c.destination}" elif self.q == "Is": return f"{self.c.source} is {self.c.destination}" elif self.q == "Are": return f"{self.c.source} are {self.c.destination}" elif self.q == "Can": return f"{self.c.source} can {self.c.destination}" elif self.q == "Will": return f"{self.c.source} will {self.c.destination}" else: return f"Answer: {self.c.source}"class LocalPersistenceDB: def __init__(self, db_path: str = "thorn_forge_memory.db"): self.db_path = db_path self._initialize_database() def _initialize_database(self): with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS facts ( id INTEGER PRIMARY KEY AUTOINCREMENT, query_primitive TEXT, cardinal INTEGER, ordinal INTEGER, source TEXT, origin TEXT, destination TEXT, labor REAL, materials REAL, regulatory REAL, action_type TEXT, signal_type TEXT, abs_seconds INTEGER ) """) conn.commit() def save_fact_to_disk(self, fact: Fact): with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" INSERT INTO facts ( query_primitive, cardinal, ordinal, source, origin, destination, labor, materials, regulatory, action_type, signal_type, abs_seconds ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( fact.q, fact.c.cardinal, fact.c.ordinal, fact.c.source, fact.c.origin, fact.c.destination, fact.u.labor, fact.u.materials, fact.u.regulatory, fact.r.action, fact.r.signal_type, fact.t.to_absolute_seconds() )) conn.commit() def load_all_facts(self) -> List[Fact]: loaded_facts = [] with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute("SELECT query_primitive, cardinal, ordinal, source, origin, destination, labor, materials, regulatory, action_type, signal_type, abs_seconds FROM facts") rows = cursor.fetchall() for row in rows: t_idx = TimeIndex.from_absolute_seconds(row[11]) c_vec = CardinalOrdinalVector(cardinal=row[1], ordinal=row[2], source=row[3], origin=row[4], destination=row[5]) u_trip = USETriplet(labor=row[6], materials=row[7], regulatory=row[8]) tr_status = TransmitReceive(action=row[9], signal_type=row[10]) loaded_facts.append(Fact(q=row[0], c=c_vec, u=u_trip, r=tr_status, t=t_idx)) return loaded_factsclass MultimodalSynthesisEngine: def __init__(self): pass def synthesize_voice_buffer(self, input_text: str, path_output: str = "synth_voice.wav"): sample_rate = 11025 duration = 1.5 num_samples = int(sample_rate * duration) with wave.open(path_output, 'wb') as wav_file: wav_file.setnchannels(1) wav_file.setsampwidth(1) wav_file.setframerate(sample_rate) freq = 440.0 + (len(input_text) * 5) for i in range(num_samples): value = int(127.0 * math.sin(2.0 * math.pi * freq * (i / sample_rate)) + 128) wav_file.writeframes(bytes([value])) return path_output def synthesize_video_frames(self, input_text: str) -> Tuple[int, int, int]: total_frames = max(5, min(60, len(input_text))) width, height = 320, 240 return total_frames, width, height def generate_raw_data_matrix(self, memory_list: List[Fact]) -> str: data_packets = [] for index, f in enumerate(memory_list): matrix_row = { "packet_id": index, "hex_time": hex(f.t.to_absolute_seconds()), "signal_weight": f.c.cardinal * 1.414, "vector_signature": [f.u.labor, f.u.materials, f.u.regulatory] } data_packets.append(matrix_row) return json.dumps(data_packets, indent=2)class ThornMemory: def __init__(self, db_engine: LocalPersistenceDB): self.db = db_engine self.facts: List[Fact] = self.db.load_all_facts() self.index_by_q: Dict[str, List[Fact]] = defaultdict(list) for f in self.facts: self.index_by_q[f.q].append(f) def add_fact(self, fact: Fact): self.facts.append(fact) self.index_by_q[fact.q].append(fact) self.db.save_fact_to_disk(fact) print(f"[Memory] Stored and Persisted: {fact.to_sentence()} at {fact.t}") def query(self, q: str, subject: Optional[str] = None) -> List[Fact]: results = self.index_by_q.get(q, []) if subject: results = [f for f in results if subject in f.c.source.lower()] return results def run_feynman_inference(self, incoming_subject: str) -> Optional[str]: subject_lower = incoming_subject.lower() shared_nodes = [] for f in self.facts: if f.c.source.lower() == subject_lower: shared_nodes.append(f.c.destination.lower()) for target_node in shared_nodes: for f in self.facts: if f.c.source.lower() == target_node and f.c.destination.lower() != subject_lower: return f"Derived association: {incoming_subject} links to {f.c.destination} through node {target_node}." return None def get_snapshot_vector(self) -> Tuple[float, float, float]: if not self.facts: return (0.0, 0.0, 0.0) X = sum(f.c.cardinal for f in self.facts) / len(self.facts) Y = sum(f.c.ordinal for f in self.facts) / len(self.facts) Z = sum(f.u.labor + f.u.materials + f.u.regulatory for f in self.facts) / len(self.facts) return (X, Y, Z) def recall_by_vector(self, target: Tuple[float, float, float]) -> Optional[TimeIndex]: if not self.facts: return None time_groups: Dict[int, List[Fact]] = defaultdict(list) for f in self.facts: t_abs = f.t.to_absolute_seconds() time_groups[t_abs].append(f) best_t = None best_dist = float('inf') for t_abs, facts_at_t in time_groups.items(): X = sum(f.c.cardinal for f in facts_at_t) / len(facts_at_t) Y = sum(f.c.ordinal for f in facts_at_t) / len(facts_at_t) Z = sum(f.u.labor + f.u.materials + f.u.regulatory for f in facts_at_t) / len(facts_at_t) dist = (X - target[0])**2 + (Y - target[1])**2 + (Z - target[2])**2 if dist < best_dist: best_dist = dist best_t = t_abs return TimeIndex.from_absolute_seconds(best_t) if best_t is not None else Nonedef parse_utterance(text: str) -> Fact: words = text.lower().replace('?', '').replace('.', '').replace(',', '').split() if "who" in words: q = "Who" elif "what" in words: q = "What" elif "when" in words: q = "When" elif "where" in words: q = "Where" elif "is" in words: q = "Is" elif "are" in words: q = "Are" elif "can" in words: q = "Can" elif "will" in words: q = "Will" else: q = "Answer" subject = "unknown" destination = "memory" for i, w in enumerate(words): if w in ["who","what","when","where","is","are","can","will"] and i+1 < len(words): subject = words[i+1] if i+2 < len(words): destination = " ".join(words[i+2:]) break if subject == "unknown" and len(words) >= 2: subject = words[0] destination = " ".join(words[1:]) c = CardinalOrdinalVector(cardinal=len(words), ordinal=1, source=subject, origin="user", destination=destination) u = USETriplet(labor=1.0, materials=0.0, regulatory=0.0) r = TransmitReceive(action="Receive", signal_type="voice") t = TimeIndex.now() return Fact(q=q, c=c, u=u, r=r, t=t)def answer_query(memory: ThornMemory, text: str) -> str: words = text.lower().replace('?', '').replace('.', '').replace(',', '').split() subject = None for w in words: if w in ["color","name","book","food","animal","location","status","identity","father"]: subject = w break if not subject and len(words) > 0: for w in words: if w not in ["who","what","when","where","is","are","can","will","the","a","an","to"]: subject = w break if "what" in words or "who" in words or "where" in words or "is" in words: if subject: facts = [] for primitive in ["What", "Is", "Who", "Where", "Answer"]: facts.extend(memory.query(primitive, subject=subject)) if facts: return facts[-1].to_sentence() inferred = memory.run_feynman_inference(subject) if inferred: return inferred return f"I don't remember anything about {subject} yet." else: return "I heard a question, but I need more details to resolve the subject node." elif "when" in words: if memory.facts: earliest = min(memory.facts, key=lambda f: f.t.to_absolute_seconds()) return f"The first thing you told me was at structural index: {earliest.t}." else: return "You haven't told me anything yet." else: fact = parse_utterance(text) memory.add_fact(fact) return f"Fact committed. Spatial Vector Snapshot coordinates: {memory.get_snapshot_vector()}"class VoiceAIApp: def __init__(self, root): self.root = root self.root.title("Thorn-Forge - Your Personal AI") self.root.geometry("850x650") self.db_engine = LocalPersistenceDB() self.memory = ThornMemory(self.db_engine) self.multimodal = MultimodalSynthesisEngine() self.recognizer = sr.Recognizer() self.tts_engine = pyttsx3.init() self.tts_engine.setProperty('rate', 150) self.notebook = ttk.Notebook(root) self.notebook.pack(fill=tk.BOTH, expand=True) self.main_tab = ttk.Frame(self.notebook) self.data_tab = ttk.Frame(self.notebook) self.calc_tab = ttk.Frame(self.notebook) self.synth_tab = ttk.Frame(self.notebook) self.notebook.add(self.main_tab, text="Voice System") self.notebook.add(self.data_tab, text="Memory Matrix Viewer") self.notebook.add(self.calc_tab, text="13-Month Calendar Tools") self.notebook.add(self.synth_tab, text="Multimodal Creation Arrays") self._build_main_tab() self._build_data_tab() self._build_calc_tab() self._build_synth_tab() def _build_main_tab(self): self.text_area = scrolledtext.ScrolledText(self.main_tab, wrap=tk.WORD, width=75, height=22, font=("Arial", 12)) self.text_area.pack(padx=10, pady=10, fill=tk.BOTH, expand=True) btn_frame = tk.Frame(self.main_tab) btn_frame.pack(pady=5) self.listen_btn = tk.Button(btn_frame, text="🎤 Speak to Me", command=self.listen_thread, bg="lightblue", font=("Arial", 14)) self.listen_btn.pack(side=tk.LEFT, padx=10) self.quit_btn = tk.Button(btn_frame, text="Exit", command=self.root.quit, bg="lightgray") self.quit_btn.pack(side=tk.LEFT, padx=10) self.status_label = tk.Label(self.main_tab, text="Click 'Speak to Me' and talk.", fg="blue") self.status_label.pack(pady=5) self.log_message("Thorn-Forge is ready. DB verified. Total loaded records: " + str(len(self.memory.facts)) + "\n") def _build_data_tab(self): lbl = tk.Label(self.data_tab, text="Stored Memory Arrays:", font=("Arial", 12, "bold")) lbl.pack(anchor=tk.W, padx=10, pady=5) self.tree = ttk.Treeview(self.data_tab, columns=("Primitive", "Source", "Destination", "Time"), show="headings") self.tree.heading("Primitive", text="Primitive Matrix") self.tree.heading("Source", text="Source Vector") self.tree.heading("Destination", text="Destination Vector") self.tree.heading("Time", text="13-Month Time Record") self.tree.pack(fill=tk.BOTH, expand=True, padx=10, pady=5) refresh_btn = tk.Button(self.data_tab, text="Refresh Memory Grid", command=self._refresh_tree) refresh_btn.pack(anchor=tk.E, padx=10, pady=5) self._refresh_tree() def _build_calc_tab(self): lbl = tk.Label(self.calc_tab, text="13-Month Calendar Converter (28 days per month)", font=("Arial", 12, "bold")) lbl.pack(anchor=tk.W, padx=10, pady=10) f = tk.Frame(self.calc_tab) f.pack(fill=tk.X, padx=10, pady=5) tk.Label(f, text="Input Absolute Seconds Count:").pack(side=tk.LEFT, padx=5) self.sec_entry = tk.Entry(f, width=20) self.sec_entry.pack(side=tk.LEFT, padx=5) self.sec_entry.insert(0, str(int(time.time()))) calc_btn = tk.Button(f, text="Compute Vector Metric", command=self._compute_custom_date) calc_btn.pack(side=tk.LEFT, padx=10) self.calc_result = tk.Label(self.calc_tab, text="", font=("Courier", 12), fg="darkgreen") self.calc_result.pack(anchor=tk.W, padx=10, pady=15) def _build_synth_tab(self): lbl = tk.Label(self.synth_tab, text="AI Enabled Multimodal Tool Configuration Panels", font=("Arial", 12, "bold")) lbl.pack(anchor=tk.W, padx=10, pady=10) f_inputs = tk.Frame(self.synth_tab) f_inputs.pack(fill=tk.X, padx=10, pady=5) tk.Label(f_inputs, text="Synthesis Seed Text Prompt:").pack(side=tk.LEFT, padx=5) self.prompt_entry = tk.Entry(f_inputs, width=45) self.prompt_entry.pack(side=tk.LEFT, padx=5) self.prompt_entry.insert(0, "thorn core baseline signal") f_btns = tk.Frame(self.synth_tab) f_btns.pack(fill=tk.X, padx=10, pady=10) btn_voice = tk.Button(f_btns, text="Generate Voice Wave (.wav)", command=self._trigger_voice_synth, bg="#D1E8E2") btn_voice.pack(side=tk.LEFT, padx=5, expand=True, fill=tk.X) btn_video = tk.Button(f_btns, text="Compile Video Matrices", command=self._trigger_video_synth, bg="#D1E8E2") btn_video.pack(side=tk.LEFT, padx=5, expand=True, fill=tk.X) btn_data = tk.Button(f_btns, text="Export Structural Data Logs", command=self._trigger_data_synth, bg="#D1E8E2") btn_data.pack(side=tk.LEFT, padx=5, expand=True, fill=tk.X) lbl_console = tk.Label(self.synth_tab, text="Multimodal Matrix Outputs Console Log:", font=("Arial", 10, "bold")) lbl_console.pack(anchor=tk.W, padx=10, pady=5) self.synth_console = scrolledtext.ScrolledText(self.synth_tab, wrap=tk.WORD, width=75, height=12, bg="#F4F4F4", font=("Courier", 11)) self.synth_console.pack(padx=10, pady=5, fill=tk.BOTH, expand=True) def _trigger_voice_synth(self): prompt = self.prompt_entry.get().strip() filename = f"voice_synth_{int(time.time())}.wav" output_path = self.multimodal.synthesize_voice_buffer(prompt, filename) self.synth_console.insert(tk.END, f"[VOICE CREATION ARRAY] Synthesized speech waveform frequency parameters.\n") self.synth_console.insert(tk.END, f" File output committed to workspace sectors: {output_path}\n\n") def _trigger_video_synth(self): prompt = self.prompt_entry.get().strip() frames, w, h = self.multimodal.synthesize_video_frames(prompt) self.synth_console.insert(tk.END, f"[VIDEO CREATION ARRAY] Computed matrix configurations for automated frame arrays.\n") self.synth_console.insert(tk.END, f" Render Parameters: Total Frames={frames} | Dimension Matrix={w}x{h} | Channel Count=3\n\n") def _trigger_data_synth(self): serialized_data = self.multimodal.generate_raw_data_matrix(self.memory.facts) self.synth_console.insert(tk.END, f"[DATA CREATION ARRAY] Generated explicit serialization map from system memory records:\n") self.synth_console.insert(tk.END, f"{serialized_data}\n\n") def _refresh_tree(self): for i in self.tree.get_children(): self.tree.delete(i) for f in self.memory.facts: self.tree.insert("", tk.END, values=(f.q, f.c.source, f.c.destination, str(f.t))) def _compute_custom_date(self): try: val = int(self.sec_entry.get().strip()) computed = TimeIndex.from_absolute_seconds(val) self.calc_result.config(text=f"Calculated Target Metric Coordinates:\n{str(computed)}") except ValueError: messagebox.showerror("Validation Failure", "Input absolute seconds must be an integer field.") def log_message(self, msg): self.text_area.insert(tk.END, msg + "\n") self.text_area.see(tk.END) def speak(self, text): self.log_message(f"🤖 AI: {text}") self.tts_engine.say(text) self.tts_engine.runAndWait() def listen_thread(self): thread = threading.Thread(target=self.listen) thread.daemon = True thread.start() def listen(self): self.status_label.config(text="Listening...", fg="red") with sr.Microphone() as source: self.recognizer.adjust_for_ambient_noise(source, duration=0.5) try: audio = self.recognizer.listen(source, timeout=5, phrase_time_limit=10) wave_filename = f"utterance_{int(time.time())}.wav" with open(wave_filename, "wb") as f: f.write(audio.get_wav_data()) text = self.recognizer.recognize_google(audio) self.log_message(f"👧 You said: {text}") self.log_message(f" [Audio Buffer Committed -> {wave_filename}]") self.status_label.config(text="Processing...", fg="orange") response = answer_query(self.memory, text) self.speak(response) self.status_label.config(text="Ready", fg="blue") self._refresh_tree() except sr.WaitTimeoutError: self.status_label.config(text="No speech detected. Try again.", fg="red") except sr.UnknownValueError: self.status_label.config(text="I couldn't understand. Please repeat.", fg="red") except Exception as e: self.status_label.config(text=f"Error: {e}", fg="red")def main(): root = tk.Tk() app = VoiceAIApp(root) root.mainloop()if __name__ == "__main__": main()