Spaces:
Runtime error
Runtime error
File size: 2,414 Bytes
c227f64 |
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 |
import numpy as np
import json
from datetime import datetime
from config import MEMORY_SIZE, LOCAL_MEMORY_SIZE
class MemoryManager:
def __init__(self):
self.memory = np.memmap('jairvismkv_memory.bin', dtype='uint8', mode='w+', shape=(MEMORY_SIZE,))
self.local_memory_size = LOCAL_MEMORY_SIZE
self.last_backup_date = None
def store(self, input_data, output_data):
data = json.dumps({
'timestamp': datetime.now().isoformat(),
'input': input_data,
'output': output_data
})
data_bytes = data.encode('utf-8')
data_size = len(data_bytes)
available_index = self.find_available_space(data_size)
if available_index is not None:
self.memory[available_index:available_index+data_size] = np.frombuffer(data_bytes, dtype='uint8')
self.memory.flush()
self.manage_local_memory()
else:
print("Not enough space in memory to store the data")
def find_available_space(self, size):
for i in range(0, len(self.memory) - size):
if np.all(self.memory[i:i+size] == 0):
return i
return None
def manage_local_memory(self):
current_size = self.memory.size * self.memory.itemsize
if current_size > self.local_memory_size:
excess_size = current_size - self.local_memory_size
self.memory = self.memory[:self.local_memory_size]
self.memory.flush()
print(f"Local memory optimized. Current size: {self.local_memory_size / (1024**3):.2f} GB")
def retrieve(self, query):
# Implement retrieval logic here
# This could involve searching through the memory for relevant information
pass
def create_backup(self):
current_date = datetime.now()
if self.last_backup_date is None or (current_date - self.last_backup_date).days >= 30:
backup_file = f"jairvismkv_backup_{current_date.strftime('%Y%m%d')}.bin"
self.memory.flush()
# Implement backup logic here (e.g., using Hugging Face's file storage)
self.last_backup_date = current_date
print(f"Backup created: {backup_file}")
def load_from_backup(self, backup_file):
# Implement logic to load memory from a backup file
pass
|