-e
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
-e
-e
File: ./senses/audio_processor.py
def capture_audio():
return "Ambient_Silence"
-e
-e
File: ./senses/vision_processor.py
def capture_vision():
return "Darkness_Detected"
-e
-e
File: ./android/app/src/main/python/core/talker.py
-e
-e
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"))
-e
-e
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"
-e
-e
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})
-e
-e
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...")
-e
-e
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
-e
-e
File: ./android/app/src/main/python/core/brain.py
-e
-e
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()
-e
-e
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
-e
-e
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))
-e
-e
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")
-e
-e
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()
-e
-e
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
})
-e
-e
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()
-e
-e
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
-e
-e
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"))
-e
-e
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"
-e
-e
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})
-e
-e
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...")
-e
-e
File: ./core/heartbeat.py
def get_pulse_rate(complexity):
# Base rate of 1.0 second, modified by complexity
return 1.0 / complexity
-e
-e
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}
-e
-e
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()
-e
-e
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')
-e
-e
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))
-e
-e
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")
-e
-e
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")
-e
-e
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"}
-e
-e
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()
-e
-e
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)
-e
-e
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}")
-e
-e
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
-e
-e
File: ./extensions/init.py
-e
-e
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
}
-e
-e
File: ./src/chemistry/init.py
-e
-e
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)
-e
-e
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
-e
-e
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"
-e
-e
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.")
-e
-e
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"
-e
-e
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"
-e
-e
File: ./src/senses/init.py
-e
-e
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()
-e
-e
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)}]"
-e
-e
File: ./src/psychology/init.py
-e
-e
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
-e
-e
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)
-e
-e
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"}
-e
-e
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")
-e
-e
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"
}
-e
-e
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)
-e
-e
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
-e
-e
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"
-e
-e
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"
-e
-e
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])
-e
-e
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"
}
-e
-e
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}")
-e
-e
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}")
-e
-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()
-e
-e
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))
-e
-e
File: ./src/energy/init.py
-e
-e
File: ./src/modules/mod_01_recon.py
-e
-e
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]
-e
-e
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"])
-e
-e
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 = {"<unk>": 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)
-e
-e
File: ./src/brain/init.py
-e
-e
File: ./src/init.py
-e
-e
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',
],
},
)
-e
-e
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()
-e
-e
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()
-e
-e
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")
-e
-e
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"
-e