ComputerAssignments/CA18_memory_systems/memory_systems.py

#!/usr/bin/env python3

import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
import pandas as pd
import time
import random
import hashlib
from collections import defaultdict, deque, OrderedDict
from typing import Dict, List, Tuple, Any, Optional, Union
import psutil
import sys
from dataclasses import dataclass
from abc import ABC, abstractmethod
import json
from datetime import datetime, timedelta
import pickle
import gzip
import heapq
from sklearn.decomposition import PCA
from sklearn.cluster import KMeans
import warnings

warnings.filterwarnings("ignore")

@dataclass
class MemoryMetrics:

    total_operations: int = 0
    access_times: List[float] = None
    memory_usage: List[int] = None
    hit_rate: float = 0.0
    miss_rate: float = 0.0

    def __post_init__(self):
        if self.access_times is None:
            self.access_times = []
        if self.memory_usage is None:
            self.memory_usage = []

class BaseMemorySystem(ABC):

    def __init__(self, name: str):
        self.name = name
        self.metrics = MemoryMetrics()

    @abstractmethod
    def store(self, key: Any, value: Any) -> bool:
        pass

    @abstractmethod
    def retrieve(self, key: Any) -> Optional[Any]:
        pass

    @abstractmethod
    def delete(self, key: Any) -> bool:
        pass

    @abstractmethod
    def size(self) -> int:
        pass

    def _record_operation(self, operation_time: float):
        self.metrics.total_operations += 1
        self.metrics.access_times.append(operation_time)
        self.metrics.memory_usage.append(sys.getsizeof(self))

class SequentialMemory(BaseMemorySystem):

    def __init__(self, initial_capacity: int = 10):
        super().__init__("Sequential Memory")
        self.data = []
        self.capacity = initial_capacity
        self.load_factor = 0.75

    def store(self, key: Any, value: Any) -> bool:
        start_time = time.time()


        for i, (k, v) in enumerate(self.data):
            if k == key:
                self.data[i] = (key, value)
                self._record_operation(time.time() - start_time)
                return True


        self.data.append((key, value))


        if len(self.data) > self.capacity * self.load_factor:
            self._resize()

        self._record_operation(time.time() - start_time)
        return True

    def retrieve(self, key: Any) -> Optional[Any]:
        start_time = time.time()

        for k, v in self.data:
            if k == key:
                self._record_operation(time.time() - start_time)
                return v

        self._record_operation(time.time() - start_time)
        return None

    def delete(self, key: Any) -> bool:
        start_time = time.time()

        for i, (k, v) in enumerate(self.data):
            if k == key:
                del self.data[i]
                self._record_operation(time.time() - start_time)
                return True

        self._record_operation(time.time() - start_time)
        return False

    def size(self) -> int:
        return len(self.data)

    def _resize(self):
        self.capacity *= 2

class AssociativeMemory(BaseMemorySystem):

    def __init__(self, initial_capacity: int = 16):
        super().__init__("Associative Memory")
        self.capacity = initial_capacity
        self.buckets = [[] for _ in range(self.capacity)]
        self.item_count = 0
        self.load_factor_threshold = 0.75
        self.collision_count = 0

    def _hash(self, key: Any) -> int:
        return hash(key) % self.capacity

    def _rehash(self):
        old_buckets = self.buckets
        old_capacity = self.capacity


        self.capacity *= 2
        self.buckets = [[] for _ in range(self.capacity)]
        old_item_count = self.item_count
        self.item_count = 0


        for bucket in old_buckets:
            for key, value in bucket:
                self._store_without_rehash(key, value)

    def _store_without_rehash(self, key: Any, value: Any):
        hash_value = self._hash(key)
        bucket = self.buckets[hash_value]


        for i, (k, v) in enumerate(bucket):
            if k == key:
                bucket[i] = (key, value)
                return


        if len(bucket) > 0:
            self.collision_count += 1
        bucket.append((key, value))
        self.item_count += 1

    def store(self, key: Any, value: Any) -> bool:
        start_time = time.time()

        self._store_without_rehash(key, value)


        if self.item_count > self.capacity * self.load_factor_threshold:
            self._rehash()

        self._record_operation(time.time() - start_time)
        return True

    def retrieve(self, key: Any) -> Optional[Any]:
        start_time = time.time()

        hash_value = self._hash(key)
        bucket = self.buckets[hash_value]

        for k, v in bucket:
            if k == key:
                self._record_operation(time.time() - start_time)
                return v

        self._record_operation(time.time() - start_time)
        return None

    def delete(self, key: Any) -> bool:
        start_time = time.time()

        hash_value = self._hash(key)
        bucket = self.buckets[hash_value]

        for i, (k, v) in enumerate(bucket):
            if k == key:
                del bucket[i]
                self.item_count -= 1
                self._record_operation(time.time() - start_time)
                return True

        self._record_operation(time.time() - start_time)
        return False

    def size(self) -> int:
        return self.item_count

    def get_statistics(self) -> Dict[str, Any]:
        bucket_lengths = [len(bucket) for bucket in self.buckets]
        return {
            "capacity": self.capacity,
            "item_count": self.item_count,
            "load_factor": self.item_count / self.capacity,
            "collision_count": self.collision_count,
            "max_bucket_length": max(bucket_lengths) if bucket_lengths else 0,
            "avg_bucket_length": np.mean(bucket_lengths),
            "empty_buckets": bucket_lengths.count(0),
        }

class ContentAddressableMemory(BaseMemorySystem):

    def __init__(self, similarity_threshold: float = 0.8):
        super().__init__("Content-Addressable Memory")
        self.memory_bank = []
        self.similarity_threshold = similarity_threshold
        self.vector_dimension = None

    def _vectorize_content(self, content: Any) -> np.ndarray:
        if isinstance(content, str):

            char_counts = defaultdict(int)
            for char in content.lower():
                if char.isalnum():
                    char_counts[char] += 1


            vector = np.zeros(36)
            for i, char in enumerate("abcdefghijklmnopqrstuvwxyz0123456789"):
                vector[i] = char_counts.get(char, 0)


            if np.linalg.norm(vector) > 0:
                vector = vector / np.linalg.norm(vector)

            return vector

        elif isinstance(content, (list, tuple)):

            vector = np.array(content, dtype=float)
            if np.linalg.norm(vector) > 0:
                vector = vector / np.linalg.norm(vector)
            return vector

        elif isinstance(content, dict):

            values = [v for v in content.values() if isinstance(v, (int, float))]
            if values:
                vector = np.array(values, dtype=float)
                if np.linalg.norm(vector) > 0:
                    vector = vector / np.linalg.norm(vector)
                return vector


        return self._vectorize_content(str(content))

    def _cosine_similarity(self, vec1: np.ndarray, vec2: np.ndarray) -> float:
        if len(vec1) != len(vec2):
            return 0.0

        dot_product = np.dot(vec1, vec2)
        norm1 = np.linalg.norm(vec1)
        norm2 = np.linalg.norm(vec2)

        if norm1 == 0 or norm2 == 0:
            return 0.0

        return dot_product / (norm1 * norm2)

    def store(self, key: Any, value: Any) -> bool:
        start_time = time.time()


        content_vector = self._vectorize_content(value)


        if self.vector_dimension is None:
            self.vector_dimension = len(content_vector)


        metadata = {"key": key, "timestamp": time.time(), "access_count": 0}

        self.memory_bank.append((content_vector, metadata, value))

        self._record_operation(time.time() - start_time)
        return True

    def retrieve(self, key: Any) -> Optional[Any]:
        start_time = time.time()

        for content_vector, metadata, value in self.memory_bank:
            if metadata["key"] == key:
                metadata["access_count"] += 1
                self._record_operation(time.time() - start_time)
                return value

        self._record_operation(time.time() - start_time)
        return None

    def retrieve_by_content(self, query_content: Any) -> List[Tuple[Any, float]]:
        start_time = time.time()

        query_vector = self._vectorize_content(query_content)
        results = []

        for content_vector, metadata, value in self.memory_bank:
            similarity = self._cosine_similarity(query_vector, content_vector)
            if similarity >= self.similarity_threshold:
                metadata["access_count"] += 1
                results.append((value, similarity))


        results.sort(key=lambda x: x[1], reverse=True)

        self._record_operation(time.time() - start_time)
        return results

    def delete(self, key: Any) -> bool:
        start_time = time.time()

        for i, (content_vector, metadata, value) in enumerate(self.memory_bank):
            if metadata["key"] == key:
                del self.memory_bank[i]
                self._record_operation(time.time() - start_time)
                return True

        self._record_operation(time.time() - start_time)
        return False

    def size(self) -> int:
        return len(self.memory_bank)

class AdaptiveLRUCache(BaseMemorySystem):

    def __init__(self, max_size: int = 100, initial_size: int = 50):
        super().__init__("Adaptive LRU Cache")
        self.max_size = max_size
        self.current_size = initial_size
        self.cache = OrderedDict()
        self.access_frequency = defaultdict(int)
        self.hit_count = 0
        self.miss_count = 0

    def _adjust_size(self):
        total_accesses = self.hit_count + self.miss_count
        if total_accesses > 0:
            hit_rate = self.hit_count / total_accesses


            if hit_rate > 0.8 and self.current_size < self.max_size:
                self.current_size = min(self.current_size + 5, self.max_size)
            elif hit_rate < 0.3 and self.current_size > 10:
                self.current_size = max(self.current_size - 5, 10)

    def store(self, key: Any, value: Any) -> bool:
        start_time = time.time()

        if key in self.cache:

            self.cache.move_to_end(key)
            self.cache[key] = value
        else:

            if len(self.cache) >= self.current_size:

                self.cache.popitem(last=False)

            self.cache[key] = value

        self.access_frequency[key] += 1
        self._adjust_size()

        self._record_operation(time.time() - start_time)
        return True

    def retrieve(self, key: Any) -> Optional[Any]:
        start_time = time.time()

        if key in self.cache:

            self.cache.move_to_end(key)
            self.hit_count += 1
            self.access_frequency[key] += 1
            self._record_operation(time.time() - start_time)
            return self.cache[key]
        else:
            self.miss_count += 1
            self._record_operation(time.time() - start_time)
            return None

    def delete(self, key: Any) -> bool:
        start_time = time.time()

        if key in self.cache:
            del self.cache[key]
            del self.access_frequency[key]
            self._record_operation(time.time() - start_time)
            return True

        self._record_operation(time.time() - start_time)
        return False

    def size(self) -> int:
        return len(self.cache)

    def get_hit_rate(self) -> float:
        total = self.hit_count + self.miss_count
        return self.hit_count / total if total > 0 else 0.0

class NeuralAssociativeMemory(BaseMemorySystem):

    def __init__(self, memory_size: int = 100, pattern_dimension: int = 50):
        super().__init__("Neural Associative Memory")
        self.memory_size = memory_size
        self.pattern_dimension = pattern_dimension
        self.patterns = []
        self.weights = np.zeros((pattern_dimension, pattern_dimension))
        self.threshold = 0.5
        self.max_iterations = 100

    def _normalize_pattern(self, pattern: np.ndarray) -> np.ndarray:
        if np.max(np.abs(pattern)) > 0:
            return pattern / np.max(np.abs(pattern))
        return pattern

    def _update_weights(self, pattern: np.ndarray):
        normalized_pattern = self._normalize_pattern(pattern)
        self.weights += np.outer(normalized_pattern, normalized_pattern)

        np.fill_diagonal(self.weights, 0)

    def store(self, key: Any, value: Any) -> bool:
        start_time = time.time()


        if isinstance(value, (list, tuple)):
            try:
                pattern = np.array(value, dtype=float)
            except (ValueError, TypeError):
                pattern = np.array(
                    [ord(c) for c in str(value)[: self.pattern_dimension]], dtype=float
                )
        elif isinstance(value, dict):
            try:
                numeric_values = [v for v in value.values() if isinstance(v, (int, float))]
                if numeric_values:
                    pattern = np.array(numeric_values, dtype=float)
                else:
                    pattern = np.array(
                        [ord(c) for c in str(value)[: self.pattern_dimension]], dtype=float
                    )
            except (ValueError, TypeError):
                pattern = np.array(
                    [ord(c) for c in str(value)[: self.pattern_dimension]], dtype=float
                )
        else:

            pattern = np.array(
                [ord(c) for c in str(value)[: self.pattern_dimension]], dtype=float
            )
        
        if len(pattern) < self.pattern_dimension:
            pattern = np.pad(pattern, (0, self.pattern_dimension - len(pattern)))


        pattern = self._normalize_pattern(pattern)


        self.patterns.append((key, pattern))
        self._update_weights(pattern)


        if len(self.patterns) > self.memory_size:

            old_key, old_pattern = self.patterns.pop(0)

            self.weights -= np.outer(old_pattern, old_pattern)
            np.fill_diagonal(self.weights, 0)

        self._record_operation(time.time() - start_time)
        return True

    def retrieve(self, key: Any) -> Optional[Any]:
        start_time = time.time()

        for stored_key, pattern in self.patterns:
            if stored_key == key:
                self._record_operation(time.time() - start_time)
                return pattern.tolist()

        self._record_operation(time.time() - start_time)
        return None

    def retrieve_by_pattern(self, query_pattern: np.ndarray) -> Optional[np.ndarray]:
        start_time = time.time()


        query_pattern = self._normalize_pattern(query_pattern)


        current_pattern = query_pattern.copy()

        for iteration in range(self.max_iterations):

            new_pattern = np.tanh(np.dot(self.weights, current_pattern))


            if np.allclose(current_pattern, new_pattern, atol=1e-6):
                break

            current_pattern = new_pattern


        best_match = None
        best_similarity = -1

        for stored_key, stored_pattern in self.patterns:
            similarity = np.dot(current_pattern, stored_pattern) / (
                np.linalg.norm(current_pattern) * np.linalg.norm(stored_pattern)
            )
            if similarity > best_similarity:
                best_similarity = similarity
                best_match = stored_pattern

        self._record_operation(time.time() - start_time)
        return best_match if best_similarity > self.threshold else None

    def delete(self, key: Any) -> bool:
        start_time = time.time()

        for i, (stored_key, pattern) in enumerate(self.patterns):
            if stored_key == key:

                del self.patterns[i]
                self.weights -= np.outer(pattern, pattern)
                np.fill_diagonal(self.weights, 0)
                self._record_operation(time.time() - start_time)
                return True

        self._record_operation(time.time() - start_time)
        return False

    def size(self) -> int:
        return len(self.patterns)

class CompressedMemorySystem(BaseMemorySystem):

    def __init__(self, compression_level: int = 6):
        super().__init__("Compressed Memory System")
        self.compressed_data = {}
        self.compression_level = compression_level
        self.decompression_cache = {}
        self.cache_size = 50

    def _compress_data(self, data: Any) -> bytes:
        serialized = pickle.dumps(data)
        compressed = gzip.compress(serialized, compresslevel=self.compression_level)
        return compressed

    def _decompress_data(self, compressed_data: bytes) -> Any:
        decompressed = gzip.decompress(compressed_data)
        return pickle.loads(decompressed)

    def store(self, key: Any, value: Any) -> bool:
        start_time = time.time()

        compressed_value = self._compress_data(value)
        self.compressed_data[key] = compressed_value


        if len(self.decompression_cache) >= self.cache_size:

            oldest_key = next(iter(self.decompression_cache))
            del self.decompression_cache[oldest_key]

        self.decompression_cache[key] = value

        self._record_operation(time.time() - start_time)
        return True

    def retrieve(self, key: Any) -> Optional[Any]:
        start_time = time.time()

        if key in self.decompression_cache:

            value = self.decompression_cache[key]

            del self.decompression_cache[key]
            self.decompression_cache[key] = value
        elif key in self.compressed_data:

            value = self._decompress_data(self.compressed_data[key])

            if len(self.decompression_cache) >= self.cache_size:
                oldest_key = next(iter(self.decompression_cache))
                del self.decompression_cache[oldest_key]
            self.decompression_cache[key] = value
        else:
            self._record_operation(time.time() - start_time)
            return None

        self._record_operation(time.time() - start_time)
        return value

    def delete(self, key: Any) -> bool:
        start_time = time.time()

        deleted = False
        if key in self.compressed_data:
            del self.compressed_data[key]
            deleted = True

        if key in self.decompression_cache:
            del self.decompression_cache[key]
            deleted = True

        self._record_operation(time.time() - start_time)
        return deleted

    def size(self) -> int:
        return len(self.compressed_data)

    def get_compression_ratio(self) -> float:
        if not self.compressed_data:
            return 0.0

        total_original = sum(
            len(pickle.dumps(self._decompress_data(data)))
            for data in self.compressed_data.values()
        )
        total_compressed = sum(len(data) for data in self.compressed_data.values())

        return total_compressed / total_original if total_original > 0 else 0.0

class MemoryLevel:

    def __init__(self, name: str, capacity: int, access_time: float):
        self.name = name
        self.capacity = capacity
        self.access_time = access_time
        self.data = {}
        self.access_count = 0
        self.hit_count = 0

    def store(self, key: Any, value: Any) -> bool:
        if len(self.data) >= self.capacity:
            return False

        self.data[key] = value
        return True

    def retrieve(self, key: Any) -> Optional[Any]:
        self.access_count += 1
        if key in self.data:
            self.hit_count += 1
            return self.data[key]
        return None

    def delete(self, key: Any) -> bool:
        if key in self.data:
            del self.data[key]
            return True
        return False

    def get_hit_rate(self) -> float:
        return self.hit_count / self.access_count if self.access_count > 0 else 0.0

class HierarchicalMemorySystem(BaseMemorySystem):

    def __init__(self):
        super().__init__("Hierarchical Memory System")
        self.levels = [
            MemoryLevel("L1 Cache", 10, 0.001),
            MemoryLevel("L2 Cache", 50, 0.01),
            MemoryLevel("L3 Cache", 200, 0.1),
            MemoryLevel("Main Memory", 1000, 1.0),
        ]
        self.total_access_time = 0.0

    def store(self, key: Any, value: Any) -> bool:
        start_time = time.time()


        for level in self.levels:
            if level.store(key, value):
                self._record_operation(time.time() - start_time)
                return True


        self._evict_and_cascade(key, value)

        self._record_operation(time.time() - start_time)
        return True

    def retrieve(self, key: Any) -> Optional[Any]:
        start_time = time.time()


        for i, level in enumerate(self.levels):
            value = level.retrieve(key)
            if value is not None:

                if i > 0:
                    self._promote_to_level(key, value, i)

                self.total_access_time += level.access_time
                self._record_operation(time.time() - start_time)
                return value

        self._record_operation(time.time() - start_time)
        return None

    def delete(self, key: Any) -> bool:
        start_time = time.time()

        deleted = False
        for level in self.levels:
            if level.delete(key):
                deleted = True

        self._record_operation(time.time() - start_time)
        return deleted

    def size(self) -> int:
        return sum(len(level.data) for level in self.levels)

    def _evict_and_cascade(self, key: Any, value: Any):

        if self.levels[0].data:
            evicted_key = next(iter(self.levels[0].data))
            evicted_value = self.levels[0].data[evicted_key]
            del self.levels[0].data[evicted_key]


            self._cascade_down(evicted_key, evicted_value, 0)


        self.levels[0].store(key, value)

    def _cascade_down(self, key: Any, value: Any, from_level: int):
        if from_level + 1 < len(self.levels):
            next_level = self.levels[from_level + 1]
            if len(next_level.data) >= next_level.capacity:

                evicted_key = next(iter(next_level.data))
                evicted_value = next_level.data[evicted_key]
                del next_level.data[evicted_key]
                self._cascade_down(evicted_key, evicted_value, from_level + 1)

            next_level.store(key, value)

    def _promote_to_level(self, key: Any, value: Any, current_level: int):
        if current_level > 0:
            target_level = self.levels[current_level - 1]
            if len(target_level.data) < target_level.capacity:
                target_level.store(key, value)
            else:

                evicted_key = next(iter(target_level.data))
                evicted_value = target_level.data[evicted_key]
                del target_level.data[evicted_key]
                self._cascade_down(evicted_key, evicted_value, current_level - 1)
                target_level.store(key, value)

    def get_level_statistics(self) -> Dict[str, Any]:
        stats = {}
        for level in self.levels:
            stats[level.name] = {
                "size": len(level.data),
                "capacity": level.capacity,
                "utilization": len(level.data) / level.capacity,
                "hit_rate": level.get_hit_rate(),
                "access_time": level.access_time,
            }
        return stats

def create_memory_system(system_type: str, **kwargs) -> BaseMemorySystem:
    systems = {
        "sequential": SequentialMemory,
        "associative": AssociativeMemory,
        "content_addressable": ContentAddressableMemory,
        "adaptive_lru": AdaptiveLRUCache,
        "neural_associative": NeuralAssociativeMemory,
        "compressed": CompressedMemorySystem,
        "hierarchical": HierarchicalMemorySystem,
    }

    if system_type not in systems:
        raise ValueError(f"Unknown memory system type: {system_type}")

    return systems[system_type](**kwargs)

if __name__ == "__main__":
    print("🧠 Memory Systems Implementation")
    print("=" * 50)


    test_data = [
        ("user_001", {"name": "Alice", "age": 25, "role": "engineer"}),
        ("user_002", {"name": "Bob", "age": 30, "role": "designer"}),
        ("user_003", {"name": "Carol", "age": 28, "role": "manager"}),
    ]

    systems = [
        ("Sequential", SequentialMemory()),
        ("Associative", AssociativeMemory()),
        ("Content-Addressable", ContentAddressableMemory()),
        ("Adaptive LRU", AdaptiveLRUCache()),
        ("Neural Associative", NeuralAssociativeMemory()),
        ("Compressed", CompressedMemorySystem()),
        ("Hierarchical", HierarchicalMemorySystem()),
    ]

    for name, system in systems:
        print(f"\n🔧 Testing {name} Memory System")
        print("-" * 30)


        for key, value in test_data:
            system.store(key, value)


        for key, _ in test_data:
            result = system.retrieve(key)
            if result:
                print(f"  ✓ Retrieved {key}")

        print(f"  📊 Size: {system.size()}")
        print(f"  ⏱️  Operations: {system.metrics.total_operations}")

    print("\n✅ All memory systems tested successfully!")