Optiscope Storage Examples¶

This notebook demonstrates the storage capabilities of the optiscope library, including:

1. Memory Storage
2. Filesystem Storage
3. Storage Manager with Caching
4. Backend Migration

In [1]:

import tempfile
import time
from pathlib import Path

import numpy as np
import pandas as pd

from optiscope import OptimizationResult, ProblemMetadata
from optiscope.storage import (
    FileSystemStorage,
    MemoryStorage,
    create_storage_manager,
)

import tempfile import time from pathlib import Path import numpy as np import pandas as pd from optiscope import OptimizationResult, ProblemMetadata from optiscope.storage import ( FileSystemStorage, MemoryStorage, create_storage_manager, )

Helper Function¶

Create a sample optimization result for testing.

In [2]:

def create_sample_result(name: str, n_points: int = 50) -> OptimizationResult:
    """Create a sample optimization result."""
    np.random.seed(hash(name) % 2**32)

    x1 = np.random.uniform(0, 10, n_points)
    x2 = np.random.uniform(0, 10, n_points)

    f1 = x1
    f2 = (x2 + 1) / x1

    return OptimizationResult(
        problem_metadata=ProblemMetadata(
            name=name,
            solver="Example",
            n_design_variables=2,
            n_objectives=2,
        ),
        design_variables=pd.DataFrame({"x1": x1, "x2": x2}),
        objectives=pd.DataFrame({"f1": f1, "f2": f2}),
    )

def create_sample_result(name: str, n_points: int = 50) -> OptimizationResult: """Create a sample optimization result.""" np.random.seed(hash(name) % 2**32) x1 = np.random.uniform(0, 10, n_points) x2 = np.random.uniform(0, 10, n_points) f1 = x1 f2 = (x2 + 1) / x1 return OptimizationResult( problem_metadata=ProblemMetadata( name=name, solver="Example", n_design_variables=2, n_objectives=2, ), design_variables=pd.DataFrame({"x1": x1, "x2": x2}), objectives=pd.DataFrame({"f1": f1, "f2": f2}), )

1. Memory Storage¶

Store results in memory for fast access during a session.

In [3]:

storage = MemoryStorage(max_memory_mb=50.0, eviction_policy="lru")

# Save results
for i in range(5):
    key = f"run_{i}"
    result = create_sample_result(f"Problem {i}", n_points=100)
    storage.save_result(key, result, metadata={"run_number": i})
    print(f"Saved '{key}'")

# List and load
print(f"Stored results: {storage.list_results()}")
loaded = storage.load_result("run_2")
print(f"Loaded: {loaded}")

# Info
print(storage.get_storage_info())

storage = MemoryStorage(max_memory_mb=50.0, eviction_policy="lru") # Save results for i in range(5): key = f"run_{i}" result = create_sample_result(f"Problem {i}", n_points=100) storage.save_result(key, result, metadata={"run_number": i}) print(f"Saved '{key}'") # List and load print(f"Stored results: {storage.list_results()}") loaded = storage.load_result("run_2") print(f"Loaded: {loaded}") # Info print(storage.get_storage_info())

Saved 'run_0'
Saved 'run_1'
Saved 'run_2'
Saved 'run_3'
Saved 'run_4'
Stored results: ['run_0', 'run_1', 'run_2', 'run_3', 'run_4']
Loaded: OptimizationResult(problem='Problem 2', n_points=100, n_objectives=2, n_design_vars=2)
{'storage_type': 'memory', 'n_results': 5, 'n_total_sets': 0, 'size_mb': 0.00040435791015625, 'max_memory_mb': 50.0, 'eviction_policy': 'lru', 'usage_percent': 0.0008087158203125}

2. Filesystem Storage¶

Store results persistently on disk.

In [4]:

with tempfile.TemporaryDirectory() as tmpdir:
    storage_path = Path(tmpdir) / "optimization_storage"
    storage = FileSystemStorage(base_path=storage_path, format="json", create_if_missing=True)

    print(f"Created storage at: {storage_path}")

    # Save results
    for i in range(3):
        key = f"experiment_{i}"
        result = create_sample_result(f"Experiment {i}", n_points=80)
        storage.save_result(key, result)
        print(f"Saved '{key}'")

    # List results
    print(f"Stored results: {storage.list_results()}")

    # Show directory structure
    print("Directory structure:")
    for item in storage_path.rglob("*"):
        if item.is_file():
            print(f"  {item.relative_to(storage_path)}")

with tempfile.TemporaryDirectory() as tmpdir: storage_path = Path(tmpdir) / "optimization_storage" storage = FileSystemStorage(base_path=storage_path, format="json", create_if_missing=True) print(f"Created storage at: {storage_path}") # Save results for i in range(3): key = f"experiment_{i}" result = create_sample_result(f"Experiment {i}", n_points=80) storage.save_result(key, result) print(f"Saved '{key}'") # List results print(f"Stored results: {storage.list_results()}") # Show directory structure print("Directory structure:") for item in storage_path.rglob("*"): if item.is_file(): print(f" {item.relative_to(storage_path)}")

Created storage at: /tmp/tmpwigeu8t6/optimization_storage
Saved 'experiment_0'
Saved 'experiment_1'
Saved 'experiment_2'
Stored results: ['experiment_0', 'experiment_1', 'experiment_2']
Directory structure:
  experiment_2/result.json
  experiment_2/metadata.json
  experiment_0/result.json
  experiment_0/metadata.json
  experiment_1/result.json
  experiment_1/metadata.json

3. Storage Manager with Caching¶

Combine persistent storage with a memory cache for performance.

In [5]:

with tempfile.TemporaryDirectory() as tmpdir:
    manager = create_storage_manager(
        primary_type="filesystem",
        primary_config={"base_path": Path(tmpdir) / "primary"},
        use_cache=True,
        cache_size_mb=20.0,
    )

    # Save results
    for i in range(4):
        key = f"dataset_{i}"
        result = create_sample_result(f"Dataset {i}", n_points=60)
        manager.save_result(key, result)

    # Load from cache
    start = time.time()
    result = manager.load_result("dataset_1", use_cache=True)
    print(f"Cache load time: {(time.time() - start) * 1000:.2f} ms")

    # Clear cache and load from disk
    manager.clear_cache()
    start = time.time()
    result = manager.load_result("dataset_1", use_cache=False)
    print(f"Disk load time: {(time.time() - start) * 1000:.2f} ms")

    manager.close()

with tempfile.TemporaryDirectory() as tmpdir: manager = create_storage_manager( primary_type="filesystem", primary_config={"base_path": Path(tmpdir) / "primary"}, use_cache=True, cache_size_mb=20.0, ) # Save results for i in range(4): key = f"dataset_{i}" result = create_sample_result(f"Dataset {i}", n_points=60) manager.save_result(key, result) # Load from cache start = time.time() result = manager.load_result("dataset_1", use_cache=True) print(f"Cache load time: {(time.time() - start) * 1000:.2f} ms") # Clear cache and load from disk manager.clear_cache() start = time.time() result = manager.load_result("dataset_1", use_cache=False) print(f"Disk load time: {(time.time() - start) * 1000:.2f} ms") manager.close()

Cache load time: 0.04 ms
Disk load time: 1.66 ms

4. Backend Migration¶

Move results between different storage backends.

In [6]:

with tempfile.TemporaryDirectory() as tmpdir:
    manager = create_storage_manager(
        primary_type="memory",
        secondary_type="filesystem",
        secondary_config={"base_path": Path(tmpdir) / "backup"},
    )

    # Save to primary
    for i in range(3):
        key = f"temp_result_{i}"
        result = create_sample_result(f"Temp {i}", n_points=40)
        manager.save_result(key, result, replicate=False)

    print(
        f"Primary: {len(manager.list_results('primary'))}, Secondary: {len(manager.list_results('secondary'))}"
    )

    # Sync
    stats = manager.sync_backends(direction="primary_to_secondary")
    print(f"Sync stats: {stats}")
    print(
        f"Primary: {len(manager.list_results('primary'))}, Secondary: {len(manager.list_results('secondary'))}"
    )

    manager.close()

with tempfile.TemporaryDirectory() as tmpdir: manager = create_storage_manager( primary_type="memory", secondary_type="filesystem", secondary_config={"base_path": Path(tmpdir) / "backup"}, ) # Save to primary for i in range(3): key = f"temp_result_{i}" result = create_sample_result(f"Temp {i}", n_points=40) manager.save_result(key, result, replicate=False) print( f"Primary: {len(manager.list_results('primary'))}, Secondary: {len(manager.list_results('secondary'))}" ) # Sync stats = manager.sync_backends(direction="primary_to_secondary") print(f"Sync stats: {stats}") print( f"Primary: {len(manager.list_results('primary'))}, Secondary: {len(manager.list_results('secondary'))}" ) manager.close()

Primary: 0, Secondary: 0
Sync stats: {'copied': 3, 'skipped': 0, 'errors': 0}
Primary: 0, Secondary: 0