Spaces:
Sleeping
Sleeping
File size: 5,989 Bytes
bffcfbc 6d6b8af |
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 130 131 132 133 134 135 136 137 138 139 140 |
from datetime import datetime
import os
import json
import logging
from typing import List, Dict, Any, Optional
logger = logging.getLogger(__name__)
class CocoonManager:
"""Manages Codette's cocoon data storage and retrieval"""
def __init__(self, base_dir: str = "./cocoons"):
self.base_dir = base_dir
self.cocoon_data = []
self.quantum_state = {"coherence": 0.5}
self._ensure_cocoon_dir()
def _ensure_cocoon_dir(self):
"""Ensure cocoon directory exists"""
os.makedirs(self.base_dir, exist_ok=True)
def load_cocoons(self) -> None:
"""Load all cocoon data from files"""
try:
# Ensure directory exists
os.makedirs(self.base_dir, exist_ok=True)
cocoon_files = [
f for f in os.listdir(self.base_dir)
if f.endswith('.cocoon')
]
logger.info(f"Found {len(cocoon_files)} cocoon files in {self.base_dir}")
loaded_data = []
for fname in cocoon_files:
try:
full_path = os.path.join(self.base_dir, fname)
with open(full_path, 'r') as f:
cocoon = json.load(f)
if self._validate_cocoon(cocoon):
loaded_data.append(cocoon)
# Update quantum state if present and most recent
current_quantum_state = cocoon.get('data', {}).get('quantum_state')
if current_quantum_state:
# Convert list quantum state to dict if needed
if isinstance(current_quantum_state, list):
current_quantum_state = {
"coherence": sum(current_quantum_state) / len(current_quantum_state)
if current_quantum_state else 0.5
}
# Update if we don't have state yet or if this cocoon is more recent
if (not isinstance(self.quantum_state, dict) or
not self.quantum_state.get('coherence') or
not loaded_data[:-1] or
cocoon['timestamp'] > loaded_data[-2]['timestamp']):
self.quantum_state = current_quantum_state.copy() if isinstance(current_quantum_state, dict) else {"coherence": 0.5}
except Exception as e:
logger.error(f"Error loading cocoon {fname}: {e}")
continue
# Sort by timestamp and handle empty timestamps
self.cocoon_data = sorted(
[c for c in loaded_data if isinstance(c, dict) and self._validate_cocoon(c)],
key=lambda x: x.get('timestamp', '0'),
reverse=True
)
logger.info(f"Successfully loaded {len(self.cocoon_data)} valid cocoons")
logger.info(
f"Loaded {len(self.cocoon_data)} cocoons "
f"with quantum coherence {self.quantum_state.get('coherence', 0.5)}"
)
except Exception as e:
logger.error(f"Error loading cocoons: {e}")
self.cocoon_data = []
# Ensure we have a valid quantum state
if not isinstance(self.quantum_state, dict) or 'coherence' not in self.quantum_state:
self.quantum_state = {"coherence": 0.5}
def _validate_cocoon(self, cocoon: Dict[str, Any]) -> bool:
"""Validate cocoon data structure"""
required_fields = ['timestamp', 'data']
return all(field in cocoon for field in required_fields)
def save_cocoon(
self,
data: Dict[str, Any],
cocoon_type: str = "codette"
) -> bool:
"""Save new cocoon data to file"""
try:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"{cocoon_type}_cocoon_{timestamp}.cocoon"
filepath = os.path.join(self.base_dir, filename)
# If data has its own quantum state, update our state
if "quantum_state" in data:
self.quantum_state = data["quantum_state"]
cocoon = {
"timestamp": timestamp,
"data": {
**data,
"quantum_state": self.quantum_state.copy()
}
}
with open(filepath, 'w') as f:
json.dump(cocoon, f, indent=2)
self.cocoon_data.insert(0, cocoon)
logger.info(f"Saved cocoon: {filename}")
return True
except Exception as e:
logger.error(f"Error saving cocoon: {e}")
return False
def get_latest_quantum_state(self) -> Dict[str, float]:
"""Get the most recent quantum state"""
# Ensure quantum_state is always a proper dict
if not isinstance(self.quantum_state, dict):
self.quantum_state = {"coherence": 0.5}
return self.quantum_state.copy()
def get_latest_cocoons(self, limit: int = 5) -> List[Dict[str, Any]]:
"""Get the most recent cocoons"""
return self.cocoon_data[:limit]
def update_quantum_state(self, new_state: Dict[str, float]) -> None:
"""Update the current quantum state and save it"""
self.quantum_state.update(new_state)
logger.info(f"Updated quantum state: {self.quantum_state}")
# Save the new state in a cocoon
self.save_cocoon({
"type": "quantum_update",
"quantum_state": self.quantum_state.copy()
}) |