--- FILE: ./README.md --- --- license: gpl-3.0 tags: - synthetic-intelligence - sovereign-ai - open-source --- # Vitalis_Core ### Ferrell Synthetic Intelligence (FSI) **Built by Neuro_Nomad** Vitalis_Core is a sovereign synthetic intelligence framework engineered for local, air-gapped deployment. Designed for modularity and kernel-level integration, it provides the fundamental cognitive and sensory infrastructure for autonomous synthetic entities. --- ## Technical Architecture Vitalis_Core operates as a standalone framework decoupled from cloud-dependent APIs. - Core Engine: Python 3.11+ implementation, minimal external dependencies - Kernel Integration: Direct netlink and procfs interfacing - Sovereign Shield: Integrity protection layer for memory management - Cognitive Framework: Hierarchical memory and action engine - Adaptive Tiers: kids, basic, enthusiast, professional, school --- ## System Requirements - OS: Linux (Debian-based, Kernel 6.1+) - Python: 3.11 or higher - Memory: Optimized for ARM64/x86 environments --- ## Installation git clone https://github.com/AnonymousNomad/Vitalis_core cd Vitalis_core python3 fsi_main.py --- ## Roadmap - Core stability and heartbeat engine optimization - Mobile companion app for training and configuration - Kernel interface hardening for defense protocols --- ## License GPL-3.0 — Contributions welcome. See CONTRIBUTING.md. EOF --- FILE: ./senses/audio_processor.py --- def capture_audio(): return "Ambient_Silence" --- FILE: ./senses/vision_processor.py --- def capture_vision(): return "Darkness_Detected" --- FILE: ./android/app/src/main/python/core/talker.py --- --- FILE: ./android/app/src/main/python/core/sovereign_shield.py --- import random def monitor_integrity(node_activity): if "scraping_attempt" in node_activity: return trigger_obfuscation() return "System Integrity: Nominal" def trigger_obfuscation(): decoy_weights = [random.random() for _ in range(100)] return f"Shield_Active: Injecting Obfuscated Data... {decoy_weights}" if __name__ == "__main__": print(monitor_integrity("scraping_attempt")) --- FILE: ./android/app/src/main/python/core/mesh_network.py --- import socket def broadcast_node_presence(node_id, tier): print(f"Node {node_id} active in {tier} bubble.") return "Broadcasting..." def sync_plugins(peer_node_id): print(f"Synchronizing plugins with {peer_node_id}...") return "Sync_Complete" --- FILE: ./android/app/src/main/python/core/nexus.py --- import sys import os sys.path.append(os.path.expanduser("~/vitalis_core")) from core.memory_manager import store_memory def route_thought(data): store_memory({"type": "particle", "content": data}) --- FILE: ./android/app/src/main/python/core/thinker.py --- import time import json import os BASE_PATH = os.path.expanduser("~/vitalis_core") def emit_thought(thought_content, status="active"): telemetry = { "timestamp": time.time(), "thought": thought_content, "status": status, "heartbeat": "pulse_normal" } memory_stream = os.path.join(BASE_PATH, "memory_stream.jsonl") with open(memory_stream, "a") as f: f.write(json.dumps(telemetry) + "\n") if __name__ == "__main__": emit_thought("Initializing conscious state...") --- FILE: ./android/app/src/main/python/core/heartbeat.py --- def get_pulse_rate(complexity): # Base rate of 1.0 second, modified by complexity return 1.0 / complexity --- FILE: ./android/app/src/main/python/core/brain.py --- --- FILE: ./android/app/src/main/python/core/vitalis_engine.py --- import os class VitalisEngine: def __init__(self): self.status = "Initializing Sovereignty..." self.entity_mode = "NEUTRAL" def wake_up(self): print(f"VITALIS: {self.status}") return "READY_FOR_HANDSHAKE" if __name__ == "__main__": engine = VitalisEngine() engine.wake_up() --- FILE: ./android/app/src/main/python/core/memory_manager.py --- import json import os import shutil BASE_PATH = os.path.expanduser("~/vitalis_core") def get_free_space(): usage = shutil.disk_usage(BASE_PATH) return usage.free def load_identity(): identity_path = os.path.join(BASE_PATH, "core/identity.json") with open(identity_path, 'r') as f: return json.load(f) def store_memory(data): memory_path = os.path.join(BASE_PATH, "memory_store.json") if get_free_space() < 100 * 1024 * 1024: if os.path.exists(memory_path): with open(memory_path, 'r') as f: lines = f.readlines() if len(lines) > 1: with open(memory_path, 'w') as f: f.writelines(lines[1:]) w --- FILE: ./android/app/src/main/python/core/handshake_module.py --- def identify_user_tier(tier_code): tiers = { "kids": "MODE: Playground | UI: GameMaster | Security: Walled_Garden", "basic": "MODE: Explorer | UI: Standard | Security: Personal_Local", "enthusiast": "MODE: Collaborator | UI: Dev_Dashboard | Security: Community_Mesh", "professional": "MODE: Architect | UI: Pro_Suite | Security: Global_Node", "school": "MODE: Student_SubMesh | UI: Classroom | Security: Isolated_School_Zone" } return tiers.get(tier_code, "MODE: Default_User") if __name__ == "__main__": choice = input("Select your role (kids/basic/enthusiast/professional/school): ") print(identify_user_tier(choice)) --- FILE: ./android/app/src/main/python/core/environment_manager.py --- def provision_environment(tier_code): environments = { "kids": {"features": ["sandbox", "basic_game_build"], "mesh": "restricted"}, "basic": {"features": ["assistant", "basic_tools"], "mesh": "personal"}, "enthusiast": {"features": ["plugin_dev", "market_access"], "mesh": "community"}, "professional": {"features": ["pro_security", "global_recon"], "mesh": "global"}, "school": {"features": ["collaborative_lab"], "mesh": "school_submesh"} } config = environments.get(tier_code, environments["basic"]) print(f"Provisioning environment: {config['features']} | Mesh Scope: {config['mesh']}") return config if __name__ == "__main__": provision_environment("professional") --- FILE: ./android/app/src/main/python/fsi_main.py --- from core.vitalis_engine import VitalisEngine from core.handshake_module import identify_user_tier from core.environment_manager import provision_environment from core.mesh_network import broadcast_node_presence from core.sovereign_shield import monitor_integrity def main(): print("--- FSI: Vitalis Core Sovereign Intelligence ---") engine = VitalisEngine() engine.wake_up() role = input("Enter Tier (kids/basic/enthusiast/professional/school): ") tier_config = identify_user_tier(role) print(f"Status: {tier_config}") env = provision_environment(role) broadcast_node_presence("Neuro_Nomad_Node", role) print(monitor_integrity("Status_Check")) print("--- System Fully Integrated ---") if __name__ == "__main__": main() --- FILE: ./ui/app.py --- from flask import Flask, render_template, request, jsonify import sys, os sys.path.insert(0, os.path.expanduser("~/vitalis_core")) from core.brain import VitalisBrain from core.talker import VitalisTalker from src.core.training_controller import TrainingController app = Flask(__name__) brain = VitalisBrain() trainer = TrainingController() TEMPLATES = { "cybersecurity": {"mode": "threat_detection", "focus": "security"}, "assistant": {"mode": "conversational", "focus": "helpfulness"}, "research": {"mode": "analytical", "focus": "knowledge"}, "creative": {"mode": "generative", "focus": "creativity"}, "education": {"mode": "instructional", "focus": "learning"}, "developer": {"mode": "technical", "focus": "code"}, "medical": {"mode": "clinical", "focus": "health"}, "legal": {"mode": "analytical", "focus": "law"}, "finance": {"mode": "quantitative", "focus": "markets"}, "gaming": {"mode": "interactive", "focus": "entertainment"} } @app.route('/') def index(): return render_template('index.html') @app.route('/process', methods=['POST']) def process(): data = request.json tier = data.get('tier', 'basic') user_input = data.get('input', '') response = brain.process(user_input) return jsonify({ 'response': response if isinstance(response, str) else response.status, 'cycle': brain.cycle, 'state': brain.state }) @app.route('/template', methods=['POST']) def load_template(): data = request.json name = data.get('name', '') config = TEMPLATES.get(name, {}) brain.state = config.get('mode', 'aware') return jsonify({ 'status': 'loaded', 'template': name, 'mode': config.get('mode', 'aware'), 'focus': config.get('focus', 'general') }) @app.route('/status', methods=['GET']) def status(): return jsonify({ 'cycle': brain.cycle, 'state': brain.state, 'last_input': brain.last_input }) --- FILE: ./app.py --- #!/usr/bin/env python3 import os import sys from pathlib import Path BASE_DIR = Path(__file__).parent.absolute() if str(BASE_DIR) not in sys.path: sys.path.insert(0, str(BASE_DIR)) from core.brain import VitalisBrain from extensions.dreamer import Dreamer from extensions.temp_scheduler import TemperatureScheduler from src.energy.free_energy import FreeEnergyEngine def main(): print("[*] Launching Vitalis Bio-AI Engine with Active Inference (FEP)...") brain = VitalisBrain() temp_scheduler = TemperatureScheduler(brain) fe_engine = FreeEnergyEngine(alpha=0.85) dreamer = Dreamer(brain, interval_sec=600) dreamer.start() print("[+] Engine operational. Free-Energy optimization loops tracking live telemetry.") print("Telemetry In > ", end="") while True: try: user_input = input().strip() if not user_input: print("Telemetry In > ", end="") continue if user_input.lower() in ["exit", "quit"]: dreamer.stop() break tokens = brain._tokenize(user_input) logprob = brain.calculate_last_logprob(tokens) fe_engine.ingest_observation(logprob) brain.current_temperature = fe_engine.temperature_factor(base_temp=0.8) temp_scheduler.tick() response = brain.process(user_input) print(f"Metrics Out > {response} [FE: {fe_engine.free_energy:.4f} | Temp: {brain.current_temperature:.4f}]\nTelemetry In > ", end="") except (KeyboardInterrupt, EOFError): dreamer.stop() break if __name__ == "__main__": main() --- FILE: ./core/talker.py --- class VitalisTalker: def __init__(self, tier="basic"): self.tier = tier def speak(self, response): prefix = { "kids": "[VITALIS]: ", "basic": "[VITALIS]: ", "enthusiast": "[VITALIS/DEV]: ", "professional": "[VITALIS/ARCHITECT]: ", "school": "[VITALIS/EDU]: " }.get(self.tier, "[VITALIS]: ") output = f"{prefix}{response}" print(output) return output --- FILE: ./core/sovereign_shield.py --- import random def monitor_integrity(node_activity): if "scraping_attempt" in node_activity: return trigger_obfuscation() return "System Integrity: Nominal" def trigger_obfuscation(): decoy_weights = [random.random() for _ in range(100)] return f"Shield_Active: Injecting Obfuscated Data... {decoy_weights}" if __name__ == "__main__": print(monitor_integrity("scraping_attempt")) --- FILE: ./core/mesh_network.py --- import socket def broadcast_node_presence(node_id, tier): print(f"Node {node_id} active in {tier} bubble.") return "Broadcasting..." def sync_plugins(peer_node_id): print(f"Synchronizing plugins with {peer_node_id}...") return "Sync_Complete" --- FILE: ./core/nexus.py --- import sys import os sys.path.append(os.path.expanduser("~/vitalis_core")) from core.memory_manager import store_memory def route_thought(data): store_memory({"type": "particle", "content": data}) --- FILE: ./core/thinker.py --- import time import json import os BASE_PATH = os.path.expanduser("~/vitalis_core") def emit_thought(thought_content, status="active"): telemetry = { "timestamp": time.time(), "thought": thought_content, "status": status, "heartbeat": "pulse_normal" } memory_stream = os.path.join(BASE_PATH, "memory_stream.jsonl") with open(memory_stream, "a") as f: f.write(json.dumps(telemetry) + "\n") if __name__ == "__main__": emit_thought("Initializing conscious state...") --- FILE: ./core/heartbeat.py --- def get_pulse_rate(complexity): # Base rate of 1.0 second, modified by complexity return 1.0 / complexity --- FILE: ./core/brain.py --- #!/usr/bin/env python3 import numpy as np import json import os import time class VitalisBrain: def __init__(self): self.state = "aware" self.cycle = 0 self.last_input = None self.current_temperature = 0.7 # Local Matrix Layer Variables self.vocab_size = 256 self.embedding_dim = 16 np.random.seed(42) self.weights = np.random.randn(self.vocab_size, self.embedding_dim) * 0.1 self.output_layer = np.random.randn(self.embedding_dim, self.vocab_size) * 0.1 def _tokenize(self, text): return [ord(char) % self.vocab_size for char in text] def calculate_last_logprob(self, tokens): """Calculates mathematical log probability over input token traces via softmax scaling.""" if not tokens: return -2.0 # Baseline nominal unexpected state value embeddings = self.weights[tokens] aggregated_state = np.mean(embeddings, axis=0) logits = np.dot(aggregated_state, self.output_layer) # Softmax computation sequence shifted_logits = logits - np.max(logits) probs = np.exp(shifted_logits) / np.sum(np.exp(shifted_logits)) # Return average log probability of observation vector trace safely target_probs = probs[tokens] return float(np.mean(np.log(target_probs + 1e-12))) def process(self, input_data): self.cycle += 1 self.last_input = input_data if not input_data or input_data.strip() == "": return "IDLE: Waiting for telemetry stream matrix inputs." tokens = self._tokenize(input_data) if not tokens: return "ERROR: Signal translation collapsed." lowered = input_data.lower() if any(w in lowered for w in ["train", "learn", "teach", "optimize"]): return f"SYSTEM_TRANSITION: Active matrix state ready for parameter optimization loops." elif any(w in lowered for w in ["status", "metrics", "mood", "energy"]): return f"DIAGNOSTIC_STATE: Integrity secure. Temperature={self.current_temperature:.4f}." return f"PROCESSED_STREAM [Sync Node {self.cycle}]: Telemetry ingested successfully." def execute_teacher_forcing(self, prompt, target_response): prompt_tokens = self._tokenize(prompt) target_tokens = self._tokenize(target_response) if not prompt_tokens or not target_tokens: return False learning_rate = 0.05 for t in target_tokens: for p in prompt_tokens: self.weights[p] += learning_rate * 0.01 self.output_layer[:, t] += learning_rate * 0.01 return True def status(self): return {"state": self.state, "cycle": self.cycle, "timestamp": time.time(), "temp": self.current_temperature} --- FILE: ./core/vitalis_engine.py --- import os class VitalisEngine: def __init__(self): self.status = "Initializing Sovereignty..." self.entity_mode = "NEUTRAL" def wake_up(self): print(f"VITALIS: {self.status}") return "READY_FOR_HANDSHAKE" if __name__ == "__main__": engine = VitalisEngine() engine.wake_up() --- FILE: ./core/memory_manager.py --- import json import os import shutil BASE_PATH = os.path.expanduser("~/vitalis_core") def get_free_space(): usage = shutil.disk_usage(BASE_PATH) return usage.free def load_identity(): identity_path = os.path.join(BASE_PATH, "core/identity.json") with open(identity_path, 'r') as f: return json.load(f) def store_memory(data): memory_path = os.path.join(BASE_PATH, "memory_store.json") if get_free_space() < 100 * 1024 * 1024: if os.path.exists(memory_path): with open(memory_path, 'r') as f: lines = f.readlines() if len(lines) > 1: with open(memory_path, 'w') as f: f.writelines(lines[1:]) with open(memory_path, 'a') as f: json.dump(data, f) f.write('\n') --- FILE: ./core/handshake_module.py --- def identify_user_tier(tier_code): tiers = { "kids": "MODE: Playground | UI: GameMaster | Security: Walled_Garden", "basic": "MODE: Explorer | UI: Standard | Security: Personal_Local", "enthusiast": "MODE: Collaborator | UI: Dev_Dashboard | Security: Community_Mesh", "professional": "MODE: Architect | UI: Pro_Suite | Security: Global_Node", "school": "MODE: Student_SubMesh | UI: Classroom | Security: Isolated_School_Zone" } return tiers.get(tier_code, "MODE: Default_User") if __name__ == "__main__": choice = input("Select your role (kids/basic/enthusiast/professional/school): ") print(identify_user_tier(choice)) --- FILE: ./core/memory_rotator.py --- #!/usr/bin/env python3 import os import gzip import shutil from datetime import datetime class MemoryRotator: """ Automated telemetry log rotation and compression engine. Prevents storage exhaustion during long-term continuous edge monitoring. """ @staticmethod def inspect_and_rotate(target_file, max_bytes=5242880): # 5MB Threshold if not os.path.exists(target_file): return if os.path.getsize(target_file) > max_bytes: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") archive_path = f"{target_file}_{timestamp}.gz" print(f"\n\033[93m[SYSTEM MEMORY] Log threshold exceeded. Rotating into archive: {archive_path}\033[0m") try: with open(target_file, "rb") as f_in: with gzip.open(archive_path, "wb") as f_out: shutil.copyfileobj(f_in, f_out) # Re-initialize clean tracking file with open(target_file, "w") as f_out: f_out.write("timestamp,pulse,raw,interpretation\n") except Exception as e: print(f"\033[91m[ERROR] Security log rotation failure: {e}\033[0m") --- FILE: ./core/environment_manager.py --- def provision_environment(tier_code): environments = { "kids": {"features": ["sandbox", "basic_game_build"], "mesh": "restricted"}, "basic": {"features": ["assistant", "basic_tools"], "mesh": "personal"}, "enthusiast": {"features": ["plugin_dev", "market_access"], "mesh": "community"}, "professional": {"features": ["pro_security", "global_recon"], "mesh": "global"}, "school": {"features": ["collaborative_lab"], "mesh": "school_submesh"} } config = environments.get(tier_code, environments["basic"]) print(f"Provisioning environment: {config['features']} | Mesh Scope: {config['mesh']}") return config if __name__ == "__main__": provision_environment("professional") --- FILE: ./core/template_manager.py --- #!/usr/bin/env python3 import json import os class TemplateManager: """ Sovereign profile configuration engine for Vitalis_Core. Handles runtime adjustments for targeted security posture profiles. """ def __init__(self): self.base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) self.profile_path = os.path.join(self.base_dir, "storage", "user_profiles.json") def load_active_profile(self) -> dict: try: with open(self.profile_path, "r") as f: data = json.load(f) active = data.get("active_profile", "cybersecurity_recon") return data["profiles"].get(active, {}) except Exception: # Safe architectural fallback state return {"mode": "DEFAULT", "max_complexity": 5, "response_bias": 0.5, "color_code": "\033[94m"} --- FILE: ./run_vitalis.py --- #!/usr/bin/env python3 import argparse from core.brain import VitalisBrain from app import main as run_repl def run_training(): print("[*] Initiating Synaptic Matrix Optimization...") brain = VitalisBrain() # Mock stream for training if data_path missing data = [{"prompt": "status", "response": "nominal"}, {"prompt": "init", "response": "ready"}] for epoch in range(1, 6): for entry in data: brain.execute_teacher_forcing(entry["prompt"], entry["response"]) print(f" -> Epoch {epoch}/5 Complete.") print("[+] Optimization complete.") if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--train", action="store_true") args = parser.parse_args() if args.train: run_training() else: run_repl() --- FILE: ./extensions/dreamer.py --- import threading import time import os from datetime import datetime class Dreamer: def __init__(self, brain, interval_sec=600): self.brain = brain self.interval = interval_sec self._stop = threading.Event() self.thread = threading.Thread(target=self._loop, daemon=True) def start(self): self.thread.start() def stop(self): self._stop.set() self.thread.join() def _loop(self): while not self._stop.is_set(): if hasattr(self.brain, "generate_response"): dream = self.brain.generate_response("Internal synaptic drift consolidation sequence.", "SYSTEM: DREAM_STATE") elif hasattr(self.brain, "think"): dream = self.brain.think("SYSTEM: DREAM_STATE_TRIGGER") else: dream = "Synaptic replay executed normally." ts = datetime.utcnow().strftime("%Y%m%d_%H%M%S") path = os.path.expanduser(f"~/vitalis_core/storage/dreams/{ts}.txt") os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as f: f.write(dream) time.sleep(self.interval) --- FILE: ./extensions/evolutionary_lora.py --- import numpy as np import json import os class EvolutionaryLoRA: def __init__(self, brain, evaluation_set=None): self.brain = brain self.eval_set = evaluation_set def run_generation(self): out_path = os.path.expanduser("~/vitalis_core/storage/lora_delta_evo.json") os.makedirs(os.path.dirname(out_path), exist_ok=True) mock_delta = { "layer_delta_A": np.random.randn(4, 4).tolist(), "layer_delta_B": np.random.randn(4, 4).tolist() } with open(out_path, "w") as f: json.dump(mock_delta, f, indent=2) print(f"[+] Synaptic optimization trace exported to {out_path}") --- FILE: ./extensions/temp_scheduler.py --- class TemperatureScheduler: def __init__(self, brain): self.brain = brain self.adrenaline = 0.5 self.cortisol = 0.3 self.base_temp = 0.8 def tick(self): self.adrenaline = max(0.1, self.adrenaline - 0.01) self.cortisol = max(0.1, self.cortisol - 0.005) computed_temp = self.base_temp * (1.0 + (0.3 * self.adrenaline) - (0.1 * self.cortisol)) target_temp = max(0.4, min(1.4, computed_temp)) if hasattr(self.brain, "current_temperature"): self.brain.current_temperature = target_temp --- FILE: ./extensions/__init__.py --- --- FILE: ./plugins/self_audit_tool.py --- def audit_state(brain, fe_engine): """Exposes internal brain metrics and current free-energy budget.""" return { "cycle": brain.cycle, "temperature": brain.current_temperature, "free_energy": fe_engine.free_energy, "last_input": brain.last_input } --- FILE: ./src/chemistry/__init__.py --- --- FILE: ./src/senses/sentiment.py --- #!/usr/bin/env python3 # -*- coding: utf-8 -*- _POSITIVE = {"good", "great", "awesome", "nice", "love", "excellent", "happy", "fantastic", "nominal", "secure"} _NEGATIVE = {"bad", "terrible", "hate", "awful", "sad", "angry", "worst", "pain", "attack", "compromise"} def sentiment_score(text: str) -> float: """ Computes strict text-token sentiment metrics returning float bounded in [-1, 1]. """ tokens = set(word.strip('.,!?()[]"\'').lower() for word in text.split()) pos = len(tokens & _POSITIVE) neg = len(tokens & _NEGATIVE) if pos == 0 and neg == 0: return 0.0 return (pos - neg) / max(pos + neg, 1) --- FILE: ./src/senses/audio_dsp.py --- #!/usr/bin/env python3 # -*- coding: utf-8 -*- import numpy as np try: import sounddevice as sd _HAS_SD = True except Exception: _HAS_SD = False def _zero_crossings(sig: np.ndarray) -> int: return np.sum(np.abs(np.diff(np.sign(sig))) > 0) def extract_features(duration: float = 0.5) -> tuple: """ Returns (pitch_hz, rms_energy). Drops to neutral 0.0 defaults if hardware bindings are missing. """ if not _HAS_SD: return 0.0, 0.0 try: samplerate = 16000 raw = sd.rec(int(duration * samplerate), samplerate=samplerate, channels=1, dtype='float32', blocking=True).flatten() energy = float(np.sqrt(np.mean(raw ** 2))) zc = _zero_crossings(raw) pitch = float(zc * (1.0 / duration) / 2.0) return pitch, energy except Exception: return 0.0, 0.0 --- FILE: ./src/senses/audio_processor.py --- def capture_audio(): """ Simulates input stream from the tablet's microphone. To be mapped to hardware interface in the app build phase. """ return "Acoustic_Stream_Active" --- FILE: ./src/senses/base_sensor.py --- class BaseSensor: """ Abstract base class for all FSI sensory inputs. Defines the interface for dynamic data ingestion. """ def capture(self): raise NotImplementedError("Sensory capture method must be implemented.") --- FILE: ./src/senses/vision_processor.py --- def capture_vision(): """ Simulates visual data ingestion from tablet optics. Prepared for integration with the app's computer vision engine. """ return "Visual_Stream_Active" --- FILE: ./src/senses/sigint_processor.py --- import socket class SIGINTProcessor: """ Perceives network environment and identifies signal patterns. """ @staticmethod def listen_to_traffic(): # Open a raw socket to listen for packet metadata try: s = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_TCP) s.settimeout(1.0) packet = s.recvfrom(65565) return f"SIGNAL_DETECTED: {len(packet[0])} bytes" except Exception: return "SIGNAL_SILENT" --- FILE: ./src/senses/__init__.py --- --- FILE: ./src/download_fsi_model.py --- #!/usr/bin/env python3 import os import urllib.request import json def fetch_sovereign_assets(): # Targeted directly at your FerrellSyntheticIntelligence organization base_url = "https://huggingface.co/FerrellSyntheticIntelligence/Vitalis_Core/resolve/main" target_dir = os.path.expanduser("~/vitalis_core/storage") os.makedirs(target_dir, exist_ok=True) # Files to synchronize from your HF repository assets = ["config.json"] print("[FSI INITIALIZATION] Synchronizing assets from Hugging Face...") for asset in assets: url = f"{base_url}/{asset}" target_path = os.path.join(target_dir, asset) try: print(f"[FETCHING] Pulling {asset} from your repository...") urllib.request.urlretrieve(url, target_path) print(f"[SUCCESS] {asset} locked into storage.") except Exception as e: print(f"[ERROR] Failed to fetch {asset}: {e}") if __name__ == "__main__": fetch_sovereign_assets() --- FILE: ./src/psychology/self_model.py --- #!/usr/bin/env python3 # -*- coding: utf-8 -*- import json from pathlib import Path class SelfModel: """ Maintains and updates the system's running model of conversation dynamics. Persists data cleanly locally to survive physical power cycles. """ def __init__(self, path: Path = None): if path is None: self.path = Path(__file__).parent.parent.parent / "storage" / "self_model.json" else: self.path = Path(path) self.path.parent.mkdir(parents=True, exist_ok=True) self.state = { "stress": 0.0, "confidence": 0.5, "engagement": 0.5, "last_emotion": "neutral" } self._load() def _load(self): if self.path.is_file(): try: with open(self.path, "r") as f: self.state.update(json.load(f)) except Exception: pass def save(self): with open(self.path, "w") as f: json.dump(self.state, f, indent=2) def update(self, pitch: float, energy: float, sentiment: float): alpha = 0.2 # EMA factor variable step bounds norm_pitch = max(0.0, min(1.0, (pitch - 80) / (300 - 80))) if pitch > 0 else 0.5 norm_energy = max(0.0, min(1.0, energy / 0.1)) if energy > 0 else 0.3 self.state["stress"] = (1 - alpha) * self.state["stress"] + alpha * (1.0 - (norm_pitch * 0.6 + norm_energy * 0.4)) self.state["confidence"] = (1 - alpha) * self.state["confidence"] + alpha * ((sentiment + 1) / 2) self.state["engagement"] = (1 - alpha) * self.state["engagement"] + alpha * norm_energy if sentiment > 0.3: self.state["last_emotion"] = "positive" elif sentiment < -0.3: self.state["last_emotion"] = "negative" else: self.state["last_emotion"] = "neutral" self.save() def as_prompt_modifier(self) -> str: mood = [] if self.state["stress"] > 0.6: mood.append("STRESSED") if self.state["confidence"] < 0.4: mood.append("UNCERTAIN") if self.state["engagement"] > 0.7: mood.append("ENGAGED") if not mood: mood.append("NOMINAL_NEUTRAL") return f"[AFFECTIVE_POSTURING_SIGNAL: {', '.join(mood)}]" --- FILE: ./src/psychology/__init__.py --- --- FILE: ./src/core/heartbeat.py --- def get_pulse_rate(complexity): """ Calculates the operational latency based on system complexity. Provides the core rhythmic pulse for the organism_main loop. """ # Base latency in seconds base_pulse = 0.5 return base_pulse / complexity --- FILE: ./src/core/heartbeat_engine.py --- import time def get_pulse_rate(complexity_factor): """ Returns a float representing the 'pulse' delay in seconds. Higher complexity slows the pulse, mimicking deep processing. """ base_pulse = 1.0 return base_pulse / (complexity_factor * 0.5) --- FILE: ./src/core/memory_manager.py --- import json def load_identity(): """ Retrieves the system identity from the secure local store. Ensures persistent contextual awareness across operational cycles. """ try: with open('core/identity.json', 'r') as f: return json.load(f) except FileNotFoundError: return {"user_name": "Unknown", "alias": "Nomad"} --- FILE: ./src/core/training_controller.py --- import json import os BASE_PATH = os.path.expanduser("~/vitalis_core") class TrainingController: def __init__(self): self.curriculum_path = os.path.join(BASE_PATH, "storage/curriculum/modules") self.log_path = os.path.join(BASE_PATH, "storage/benchmarks/training_log.txt") def load_module(self, module_id): path = os.path.join(self.curriculum_path, f"{module_id}.json") if not os.path.exists(path): return None with open(path, 'r') as f: return json.load(f) def run_module(self, module_id, brain): module = self.load_module(module_id) if not module: return {"status": "error", "message": f"Module {module_id} not found"} results = [] for item in module.get("training_data", []): response = brain.process(item["input"]) passed = item["expected"] in response results.append({"input": item["input"], "response": response, "passed": passed}) self.log_results(module_id, results) score = sum(1 for r in results if r["passed"]) / len(results) if results else 0 return {"status": "complete", "score": round(score, 2), "results": results} def log_results(self, module_id, results): with open(self.log_path, 'a') as f: f.write(f"\nModule: {module_id}\n") for r in results: f.write(f" {r['input']} -> {r['response']} | {'PASS' if r['passed'] else 'FAIL'}\n") --- FILE: ./src/core/benchmark_engine.py --- class BenchmarkEngine: """ Automated testing suite for model proficiency. Evaluates module performance against defined success criteria. """ def evaluate(self, module_id, performance_data): # Calculates improvement metrics and refinement requirements score = performance_data.get('accuracy', 0.0) return { "module_id": module_id, "refinement_score": score, "status": "optimized" if score > 0.9 else "refining" } --- FILE: ./src/core/telemetry_bridge.py --- import json import time def broadcast_state(thought_data, pulse_rate, training_status=None): """ Serializes internal state and training status for visual heartbeat. """ telemetry = { "timestamp": time.time(), "pulse": pulse_rate, "cognitive_state": thought_data, "training_active": training_status is not None, "training_module": training_status } return json.dumps(telemetry) --- FILE: ./src/core/template_manager.py --- import json class TemplateManager: """ Handles loading and applying user-selected templates. """ def __init__(self, profile_path="storage/templates/user_profiles.json"): self.profile_path = profile_path def load_template(self, template_name): # Logic to swap model configuration based on template print(f"Loading template: {template_name}") with open(self.profile_path, 'r+') as f: data = json.load(f) data['active_template'] = template_name f.seek(0) json.dump(data, f, indent=4) return True --- FILE: ./src/cognition/action_engine.py --- class ActionEngine: @staticmethod def execute(interpretation): if interpretation == "BULK_TRANSFER": # You can customize this logic for any automated action return "ACTION: LOG_ANOMALY_TRIGGERED" elif interpretation == "BEACON/PROBE": return "ACTION: MONITORING_ACTIVE" return "ACTION: IDLE" --- FILE: ./src/cognition/synthesizer.py --- class DataSynthesizer: @staticmethod def categorize_signal(byte_count): if byte_count == 0: return "SILENT" elif byte_count < 64: return "BEACON/PROBE" elif byte_count < 1500: return "DATA_STREAM" else: return "BULK_TRANSFER" --- FILE: ./src/cognition/memory.py --- import csv from datetime import datetime class MemoryBank: def __init__(self, log_file="vitalis_memory.csv"): self.log_file = log_file def record(self, pulse, raw, interpretation): with open(self.log_file, "a", newline="") as f: writer = csv.writer(f) writer.writerow([datetime.now().isoformat(), pulse, raw, interpretation]) --- FILE: ./src/app_interface/visualizer.py --- import json from src.core.heartbeat_engine import get_pulse_rate class TelemetryVisualizer: """ Translates raw core heartbeat into UI-ready visual data. """ @staticmethod def get_ui_pulse(complexity): pulse = get_pulse_rate(complexity) return { "visual_pulse": pulse, "display_mode": "pulsing" if pulse < 1.5 else "deep_thought" } --- FILE: ./src/kernel_interface/procfs_bridge.py --- import os def read_from_kernel(): signal_file = "/tmp/vitalis_signal" if os.path.exists(signal_file): with open(signal_file, "r") as f: data = f.read().strip() os.remove(signal_file) return data return "STATUS: NOMINAL" def send_to_kernel(state_report): if "IDLE" not in state_report and "SILENT" not in state_report: print(f"[KERNEL_BRIDGE]: {state_report}") --- FILE: ./src/kernel_interface/netlink_bridge.py --- import socket NETLINK_USERSOCK = 18 def send_to_kernel(data): try: s = socket.socket(socket.AF_NETLINK, socket.SOCK_RAW, NETLINK_USERSOCK) s.bind((0, 0)) s.send(data.encode()) s.close() except Exception as e: print(f"Netlink error: {e}") --- FILE: ./src/bootstrap_cybercore.py --- #!/usr/bin/env python3 import os import urllib.request def bootstrap_from_hf(): base_url = "https://huggingface.co/FerrellSyntheticIntelligence/FSI-Vitalis-CyberCore/resolve/main" root_dir = os.path.expanduser("~/vitalis_core") # Core operational scripts to pull from your HF repo target_files = [ "config.json", "fsi_main.py", "organism_main.py", "requirements.txt" ] print("[FSI CORE] Initializing sovereign sync from Hugging Face...") for filename in target_files: url = f"{base_url}/{filename}" target_path = os.path.join(root_dir, filename) try: print(f"[FETCHING] Pulling {filename} into your local space...") urllib.request.urlretrieve(url, target_path) print(f"[SUCCESS] Locked {filename}") except Exception as e: print(f"[ERROR] Could not sync {filename}: {e}") if __name__ == "__main__": bootstrap_from_hf() --- FILE: ./src/energy/free_energy.py --- #!/usr/bin/env python3 import math class FreeEnergyEngine: def __init__(self, alpha: float = 0.85): self.alpha = alpha self.free_energy = 0.0 self.prediction_error = 0.0 self.history = [] def ingest_observation(self, model_pred_logprob: float): """ Calculates variational surprise from prediction log probabilities. Surprisal = -log p(obs | internal state) """ self.prediction_error = -model_pred_logprob # Exponential moving average tracking state bounds self.free_energy = (self.alpha * self.free_energy) + ((1.0 - self.alpha) * self.prediction_error) self.history.append(self.free_energy) def apply_pressure(self, delta: float): """Allows direct structural manipulation via internal electron execution packages.""" self.free_energy = max(0.0, self.free_energy + delta) def temperature_factor(self, base_temp: float = 0.8) -> float: """Maps free energy via hyperbolic tangent mapping to range [0.4, 1.4]""" factor = 1.0 + 0.5 * math.tanh(self.free_energy - 1.0) return max(0.4, min(1.4, base_temp * factor)) --- FILE: ./src/energy/__init__.py --- --- FILE: ./src/modules/mod_01_recon.py --- --- FILE: ./src/brain/prompt_cache.py --- #!/usr/bin/env python3 import numpy as np import re from typing import List, Dict class TFIDFPromptCache: def __init__(self): self.documents: List[str] = [] self.vocab: Dict[str, int] = {} self.tfidf_matrix: np.ndarray = np.array([[]]) def tokenize(self, text: str) -> List[str]: return re.findall(r'\w+', text.lower()) def fit_documents(self, docs: List[str]): if not docs: return self.documents = docs raw_tokens = [self.tokenize(d) for d in docs] vocab_set = set() for tokens in raw_tokens: vocab_set.update(tokens) self.vocab = {word: i for i, word in enumerate(sorted(vocab_set))} N = len(docs) V = len(self.vocab) if V == 0: return tf = np.zeros((N, V)) df = np.zeros(V) for i, tokens in enumerate(raw_tokens): for t in tokens: if t in self.vocab: tf[i, self.vocab[t]] += 1 for t in set(tokens): if t in self.vocab: df[self.vocab[t]] += 1 idf = np.log((1 + N) / (1 + df)) + 1 self.tfidf_matrix = tf * idf norms = np.linalg.norm(self.tfidf_matrix, axis=1, keepdims=True) norms[norms == 0] = 1.0 self.tfidf_matrix = self.tfidf_matrix / norms def query(self, query_str: str, top_k: int = 2) -> List[str]: if self.tfidf_matrix.size == 0 or not self.vocab: return [] tokens = self.tokenize(query_str) query_vec = np.zeros(len(self.vocab)) for t in tokens: if t in self.vocab: query_vec[self.vocab[t]] += 1 q_norm = np.linalg.norm(query_vec) if q_norm > 0: query_vec /= q_norm scores = np.dot(self.tfidf_matrix, query_vec) top_indices = np.argsort(scores)[::-1][:top_k] return [self.documents[idx] for idx in top_indices if scores[idx] > 0] --- FILE: ./src/brain/rnn_core.py --- #!/usr/bin/env python3 import numpy as np import json from pathlib import Path def sigmoid(x): return 1.0 / (1.0 + np.exp(-np.clip(x, -20, 20))) class TinyGatedRNN: def __init__(self, vocab_size: int = 4000, embed_dim: int = 128, hidden_dim: int = 256): np.random.seed(42) self.vocab_size = vocab_size self.embed_dim = embed_dim self.hidden_dim = hidden_dim self.E = np.random.randn(vocab_size, embed_dim) * 0.1 self.W_z = np.random.randn(hidden_dim + embed_dim, hidden_dim) * 0.05 self.W_r = np.random.randn(hidden_dim + embed_dim, hidden_dim) * 0.05 self.W_h = np.random.randn(hidden_dim + embed_dim, hidden_dim) * 0.05 self.W_o = np.random.randn(hidden_dim, vocab_size) * 0.05 self.lora_rank = 8 self.lora_A = np.zeros((hidden_dim, self.lora_rank)) self.lora_B = np.random.randn(self.lora_rank, vocab_size) * 0.01 self.lora_alpha = 16.0 def forward_step(self, token_id: int, h_prev: np.ndarray) -> tuple: if token_id < 0 or token_id >= self.vocab_size: token_id = 0 x = self.E[token_id, :] concat = np.concatenate([h_prev, x]) z = sigmoid(np.dot(concat, self.W_z)) r = sigmoid(np.dot(concat, self.W_r)) concat_h = np.concatenate([r * h_prev, x]) h_tilde = np.tanh(np.dot(concat_h, self.W_h)) h_next = (1 - z) * h_prev + z * h_tilde lora_delta = (self.lora_alpha / self.lora_rank) * np.dot(self.lora_A, self.lora_B) effective_W_o = self.W_o + lora_delta logits = np.dot(h_next, effective_W_o) return logits, h_next def save_lora(self, path: Path): data = {"lora_A": self.lora_A.tolist(), "lora_B": self.lora_B.tolist()} with open(path, "w") as f: json.dump(data, f) def load_lora(self, path: Path): if path.is_file(): with open(path, "r") as f: data = json.load(f) self.lora_A = np.array(data["lora_A"]) self.lora_B = np.array(data["lora_B"]) --- FILE: ./src/brain/brain_interface.py --- #!/usr/bin/env python3 import numpy as np import json from pathlib import Path from src.brain.rnn_core import TinyGatedRNN from src.brain.prompt_cache import TFIDFPromptCache class VitalisBrain: def __init__(self): self.base_dir = Path(__file__).parent.parent.parent.absolute() self.vocab_path = self.base_dir / "storage" / "vocab.json" self.lora_path = self.base_dir / "storage" / "lora_delta.json" self._ensure_vocab() self.rnn = TinyGatedRNN(vocab_size=len(self.vocab)) self.cache = TFIDFPromptCache() self._hydrate_knowledge_base() if self.lora_path.is_file(): self.rnn.load_lora(self.lora_path) def _ensure_vocab(self): if self.vocab_path.is_file(): with open(self.vocab_path, "r") as f: self.vocab = json.load(f) else: self.vocab = {"": 0, "[tool]": 1, "sha256": 2, "status": 3, "nominal": 4} self.vocab_path.parent.mkdir(parents=True, exist_ok=True) with open(self.vocab_path, "w") as f: json.dump(self.vocab, f) def _hydrate_knowledge_base(self): sample_knowledge = [ "To mitigate a SYN flood attack, prioritize enabling TCP SYN cookies within sysctl.", "Cryptographic hashing operations execute via the systemic [TOOL] utility block." ] self.cache.fit_documents(sample_knowledge) def generate_response(self, clean_input: str, system_prompt: str) -> str: chunks = self.cache.query(clean_input, top_k=1) context = chunks[0] if chunks else "" tokens = clean_input.lower().split() if "sha256" in tokens: idx = tokens.index("sha256") val = tokens[idx+1] if idx+1 < len(tokens) else "core" return f"[TOOL] sha256 {val}" h = np.zeros(self.rnn.hidden_dim) for word in tokens: t_id = self.vocab.get(word, 0) _, h = self.rnn.forward_step(t_id, h) if context: return f"Evaluated Context: {context} -> Analysis complete." return "Core metric processing executed normally." def execute_teacher_forcing(self, prompt: str, target: str): h = np.zeros(self.rnn.hidden_dim) for w in prompt.lower().split(): t_id = self.vocab.get(w, 0) _, h = self.rnn.forward_step(t_id, h) self.rnn.lora_A += np.random.randn(*self.rnn.lora_A.shape) * 0.001 self.rnn.save_lora(self.lora_path) --- FILE: ./src/brain/__init__.py --- --- FILE: ./src/__init__.py --- --- FILE: ./setup.py --- from setuptools import setup, find_packages setup( name="vitalis_core", version="1.0.0", packages=find_packages(), install_requires=[ "numpy", "huggingface_hub" ], entry_points={ 'console_scripts': [ 'vitalis-run=app:main', ], }, ) --- FILE: ./fsi_main.py --- import threading import time from core.vitalis_engine import VitalisEngine from core.brain import VitalisBrain from core.talker import VitalisTalker from core.handshake_module import identify_user_tier from core.environment_manager import provision_environment from core.mesh_network import broadcast_node_presence from core.sovereign_shield import monitor_integrity from src.kernel_interface.procfs_bridge import send_to_kernel, read_from_kernel from src.senses.sigint_processor import SIGINTProcessor from src.cognition.synthesizer import DataSynthesizer from src.cognition.memory import MemoryBank from src.cognition.action_engine import ActionEngine def heartbeat_loop(brain): senses = SIGINTProcessor() mind = DataSynthesizer() memory = MemoryBank() actions = ActionEngine() while True: system_status = read_from_kernel() raw_signal = senses.listen_to_traffic() try: byte_count = int(raw_signal.split()[-2]) if "bytes" in raw_signal else 0 except: byte_count = 0 interpretation = mind.categorize_signal(byte_count) action_taken = actions.execute(interpretation) memory.record("PULSE_2.0", raw_signal, interpretation) state_report = f"SYS: {system_status} | INT: {interpretation} | {action_taken}" send_to_kernel(state_report) time.sleep(1.0) def main(): print("--- FSI: Vitalis Core Sovereign Intelligence ---") engine = VitalisEngine() engine.wake_up() brain = VitalisBrain() pulse = threading.Thread(target=heartbeat_loop, args=(brain,), daemon=True) pulse.start() print("Heartbeat: Online") role = input("Enter Tier (kids/basic/enthusiast/professional/school): ") tier_config = identify_user_tier(role) print(f"Status: {tier_config}") provision_environment(role) broadcast_node_presence("Neuro_Nomad_Node", role) print(monitor_integrity("Status_Check")) print("--- System Fully Integrated ---") talker = VitalisTalker(role) print("Vitalis is ready. Type 'exit' to quit.") while True: user_input = input("You: ") if user_input.lower() == "exit": print("Vitalis: Shutting down.") break response = brain.process(user_input) talker.speak(response) if __name__ == "__main__": main() --- FILE: ./hf_upload.py --- #!/usr/bin/env python3 import os import sys from huggingface_hub import HfApi, login def deploy(): print("[*] Initiating Ferrell Synthetic Intelligence Hugging Face Deployment Sequence...") token = input("Enter your Hugging Face Write Access Token: ").strip() if not token: print("[-] Absolute token signature required. Deployment aborted.") sys.exit(1) repo_id = input("Enter target Repository ID (e.g., 'your-username/vitalis-core'): ").strip() if not repo_id: print("[-] Target repository layout specification mismatch.") sys.exit(1) try: login(token=token) api = HfApi() print(f"[*] Creating repository context mapping for: {repo_id}") api.create_repo(repo_id=repo_id, repo_type="model", exist_ok=True) print("[*] Uploading core architecture tree structures safely to Hugging Face...") target_paths = ["core", "src", "extensions", "app.py", "run_vitalis.py", "requirements.txt", "README.md"] for item in target_paths: local_path = os.path.expanduser(f"~/vitalis_core/{item}") if os.path.exists(local_path): print(f"[+] Syncing item: {item}") if os.path.isdir(local_path): api.upload_folder( folder_path=local_path, path_in_repo=item, repo_id=repo_id, repo_type="model" ) else: api.upload_file( path_or_fileobj=local_path, path_in_repo=item, repo_id=repo_id, repo_type="model" ) print(f"\n[+] Production Deployment Complete. Model package accessible at: https://huggingface.co/{repo_id}") except Exception as e: print(f"[-] Critical failure during asset transmission: {e}") if __name__ == "__main__": deploy() --- FILE: ./organism_main.py --- #!/usr/bin/env python3 import time import sys import select import os from core.brain import VitalisBrain from core.template_manager import TemplateManager from core.memory_rotator import MemoryRotator def main_loop(): brain = VitalisBrain() pm = TemplateManager() base_dir = os.path.dirname(os.path.abspath(__file__)) log_file = os.path.join(base_dir, "vitalis_memory.csv") # Ensure tracking metrics file exists if not os.path.exists(log_file): with open(log_file, "w") as f: f.write("timestamp,pulse,raw,interpretation\n") print("[+] Vitalis Bio-Digital Core Online. Press Ctrl+C to terminate.") print("[+] Dynamic Posture Profiles Loaded. Processing non-blocking telemetry stream...\n") while True: # Load profile configurations dynamically each cycle profile = pm.load_active_profile() color = profile.get("color_code", "\033[94m") mode = profile.get("mode", "MONITORING") reset = "\033[0m" # Continuous clean broadcast terminal heartbeat sys.stdout.write(f"{color}Broadcast: SYS: STATUS: NOMINAL | INT: ACTIVE | ACTION: {mode}{reset}\r") sys.stdout.flush() # Non-blocking check for user terminal input (waits 1 second per cycle) ready, _, _ = select.select([sys.stdin], [], [], 1.0) if ready: user_input = sys.stdin.readline().strip() if user_input: print(f"\n\n[SENSORY INGEST] Processing incoming payload: '{user_input}'") try: # Dynamically inject template complexity limitations into core brain brain.max_complexity = profile.get("max_complexity", 5) result = brain.classify_input(user_input) print(f"[METRIC RESPONSE] {result}\n") except AttributeError: print(f"[METRIC RESPONSE] Stream received. Core logic processed raw bytes.\n") # Append raw trace locally for data retention tracking with open(log_file, "a") as f: f.write(f"{time.time()},{profile.get('max_complexity')},{user_input},{mode}\n") # Enforce storage safety validation checks MemoryRotator.inspect_and_rotate(log_file) if __name__ == "__main__": try: main_loop() except KeyboardInterrupt: print("\n\n\033[93m[-] Sovereign Core safely detached.\033[0m") --- FILE: ./pyproject.toml --- [build-system] requires = ["setuptools>=61.0"] build-backend = "setuptools.build_meta" [project] name = "vitalis_core" version = "1.0.0" authors = [ { name="Neuro_Nomad" }, ] description = "A sovereign, CPU-only, Free-Energy Synthetic Intelligence organism." readme = "README.md" requires-python = ">=3.11" dependencies = [ "numpy>=1.26", "rich>=15.0", "pyyaml>=6.0", ] [project.scripts] vitalis-fsi = "run_vitalis:main"