File size: 3,842 Bytes
7f5ef51 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 |
import os
import sys
import argparse
import importlib.util
import signal
import logging
# ---- Setup ----
PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
def add_path(path):
if path not in sys.path:
sys.path.insert(0, path)
add_path(PROJECT_ROOT)
logging.basicConfig(
filename=os.path.join(PROJECT_ROOT, 'codette_bootstrap.log'),
level=logging.INFO,
format='%(asctime)s %(levelname)s %(message)s'
)
logging.info("Codette Bootstrap - Starting up.")
# ---- Utility: Existence checker ----
def check_exists(rel_path, required=True):
abs_path = os.path.join(PROJECT_ROOT, rel_path)
if not os.path.exists(abs_path):
if required:
logging.error(f"Required file '{rel_path}' not found!")
raise FileNotFoundError(f"Required file '{rel_path}' not found!")
else:
logging.warning(f"Optional file '{rel_path}' not found.")
else:
logging.info(f"Located: {rel_path}")
return abs_path
# ---- Main Class Wrappers ----
def launch_tkinter_gui():
from codette_desktop import CodetteApp
logging.info("Launching Codette Tkinter GUI...")
app = CodetteApp()
app.mainloop()
def launch_fastapi_server():
from api_server import app as fastapi_app
import uvicorn
logging.info("Spawning Codette FastAPI server...")
uvicorn.run(fastapi_app, host="127.0.0.1", port=8000, log_level="info")
def test_cognitive_stack():
# Loads config, runs the UniversalReasoning class on a sample question
from universal_reasoning_local import UniversalReasoning, load_json_config
try:
config_path = check_exists("config.json")
config = load_json_config(config_path)
ur = UniversalReasoning(config)
import asyncio
print("Running self-test: 'What is the meaning of life?'")
result = asyncio.run(ur.generate_response("What is the meaning of life?"))
print("Codette:", result)
return True
except Exception as e:
logging.error(f"Stack self-test failed: {e}")
print(f"Error: {e}")
return False
def banner():
print("CODDETTE / PIDETTE – Universal Reasoning Bootstrap")
print("Raiffs Bits LLC – Jonathan Harrison (2024)\n")
# ---- Argparse & Bootstrap ----
def main():
banner()
parser = argparse.ArgumentParser(description="Codette Multi-Agent AI Bootstrapper")
parser.add_argument('--gui', action="store_true", help="Launch the Tkinter GUI application")
parser.add_argument('--api', action="store_true", help="Launch FastAPI backend")
parser.add_argument('--test', action="store_true", help="Run logic self-test (no UI)")
args = parser.parse_args()
# Check must-exist core files
_ = check_exists("config.json")
_ = check_exists("universal_reasoning_local.py")
_ = check_exists("codette_desktop.py")
_ = check_exists("api_server.py")
# Add others as needed
# Quick-start logic
if args.test:
ok = test_cognitive_stack()
if not ok:
sys.exit(1)
return
processes = []
if args.api:
# Run in a subprocess so we can also launch GUI if needed
import multiprocessing
p_api = multiprocessing.Process(target=launch_fastapi_server)
p_api.daemon = True
p_api.start()
processes.append(p_api)
if args.gui:
launch_tkinter_gui()
# If neither, print help
if not args.gui and not args.api and not args.test:
parser.print_help()
sys.exit(0)
# Stop all spawned procs cleanly
for p in processes:
p.join()
def handle_interrupt(sig, frame):
print("Shutting down Codette...")
sys.exit(0)
signal.signal(signal.SIGINT, handle_interrupt)
signal.signal(signal.SIGTERM, handle_interrupt)
if __name__ == "__main__":
main()
|