Persistent Cache

Data persistence between test runs using Redis

Store and retrieve data between test runs using the Redis-backed persistent cache.

Import

from lager.cache import PersistentCache

Overview

The PersistentCache class provides a simple key-value store backed by Redis that persists data between test runs. This is useful for:

• Storing calibration data
• Caching test fixtures
• Tracking device serial numbers
• Maintaining state across test sessions

PersistentCache Class

PersistentCache(provider=None)

Create a persistent cache instance.

from lager.cache import PersistentCache

cache = PersistentCache()

Parameters:

Parameter	Type	Description
provider	Redis	Redis connection (optional, auto-connects)

Note: If no provider is specified, the cache automatically connects to the Lager Box’s Redis instance at lagerredis.

Methods

store(key, data)

Store data with a key.

cache = PersistentCache()

# Store a string
cache.store('device_id', b'ABC123')

# Store JSON data
import json
cache.store('calibration', json.dumps({'offset': 0.05, 'gain': 1.02}).encode())

# Store binary data
cache.store('firmware_hash', b'\x12\x34\x56\x78')

Parameters:

Parameter	Type	Description
key	str	Unique key for the data
data	bytes	Data to store (must be bytes)

Returns: Redis response (truthy on success)

load(key)

Load data by key.

cache = PersistentCache()

# Load a value
data = cache.load('device_id')
if data:
    device_id = data.decode('utf-8')
    print(f"Device ID: {device_id}")
else:
    print("Key not found")

Parameters:

Parameter	Type	Description
key	str	Key to retrieve

Returns: bytes or None - Stored data, or None if key doesn’t exist

delete(key)

Delete a key from the cache.

cache = PersistentCache()
cache.delete('old_calibration')

Parameters:

Parameter	Type	Description
key	str	Key to delete

Returns: Number of keys deleted (0 or 1)

clear()

Clear all data from the cache.

cache = PersistentCache()
cache.clear()  # WARNING: Deletes all cached data!

Returns: Redis response

Warning: This deletes ALL data in the Redis database, not just data stored by your tests.

Examples

Basic Usage

from lager.cache import PersistentCache

cache = PersistentCache()

# Store device serial number
cache.store('dut_serial', b'SN-2024-001234')

# Later, retrieve it
serial = cache.load('dut_serial')
if serial:
    print(f"Testing device: {serial.decode()}")

Storing Complex Data

from lager.cache import PersistentCache
import json
import pickle

cache = PersistentCache()

# Method 1: JSON (human-readable, limited types)
calibration = {
    'voltage_offset': 0.025,
    'current_gain': 0.998,
    'temperature_correction': -0.5,
    'timestamp': '2024-01-15T10:30:00'
}
cache.store('calibration_json', json.dumps(calibration).encode())

# Load JSON
data = cache.load('calibration_json')
if data:
    cal = json.loads(data.decode())
    print(f"Voltage offset: {cal['voltage_offset']}")

# Method 2: Pickle (any Python object)
test_results = {
    'passed': True,
    'measurements': [3.28, 3.31, 3.29],
    'metadata': {'operator': 'John', 'station': 5}
}
cache.store('results_pickle', pickle.dumps(test_results))

# Load pickle
data = cache.load('results_pickle')
if data:
    results = pickle.loads(data)
    print(f"Passed: {results['passed']}")

Calibration Cache

from lager.cache import PersistentCache
from lager import Net, NetType
import json
import time

def get_calibration(cache, recalibrate_hours=24):
    """Get calibration data, recalibrating if stale."""
    cal_key = 'adc_calibration'
    data = cache.load(cal_key)

    if data:
        cal = json.loads(data.decode())
        age_hours = (time.time() - cal['timestamp']) / 3600

        if age_hours < recalibrate_hours:
            print(f"Using cached calibration ({age_hours:.1f} hours old)")
            return cal['offset']

    # Perform calibration
    print("Performing ADC calibration...")
    adc = Net.get('CAL_REF', type=NetType.ADC)

    # Measure known reference voltage
    reference_voltage = 2.500  # Precision voltage reference
    measured = adc.input()
    offset = reference_voltage - measured

    # Store calibration
    cal = {
        'offset': offset,
        'timestamp': time.time(),
        'reference': reference_voltage,
        'measured': measured
    }
    cache.store(cal_key, json.dumps(cal).encode())

    print(f"Calibration offset: {offset:.6f}V")
    return offset

# Usage
cache = PersistentCache()
offset = get_calibration(cache)

Test Counter

from lager.cache import PersistentCache

def increment_test_counter(cache, counter_name='test_count'):
    """Increment and return test counter."""
    data = cache.load(counter_name)

    if data:
        count = int(data.decode()) + 1
    else:
        count = 1

    cache.store(counter_name, str(count).encode())
    return count

# Usage
cache = PersistentCache()
test_number = increment_test_counter(cache)
print(f"This is test #{test_number}")

Device History Tracking

from lager.cache import PersistentCache
import json
import time

class DeviceHistory:
    """Track device test history."""

    def __init__(self):
        self.cache = PersistentCache()

    def _key(self, serial):
        return f'device_history:{serial}'

    def record_test(self, serial, result, notes=''):
        """Record a test result for a device."""
        key = self._key(serial)
        data = self.cache.load(key)

        if data:
            history = json.loads(data.decode())
        else:
            history = {'serial': serial, 'tests': []}

        history['tests'].append({
            'timestamp': time.time(),
            'result': result,
            'notes': notes
        })

        self.cache.store(key, json.dumps(history).encode())

    def get_history(self, serial):
        """Get test history for a device."""
        data = self.cache.load(self._key(serial))
        if data:
            return json.loads(data.decode())
        return None

    def get_pass_rate(self, serial):
        """Calculate pass rate for a device."""
        history = self.get_history(serial)
        if not history or not history['tests']:
            return None

        passes = sum(1 for t in history['tests'] if t['result'] == 'PASS')
        return passes / len(history['tests']) * 100

# Usage
tracker = DeviceHistory()

# Record test results
tracker.record_test('SN-001', 'PASS', 'Initial test')
tracker.record_test('SN-001', 'FAIL', 'Power supply issue')
tracker.record_test('SN-001', 'PASS', 'Rework complete')

# Get history
history = tracker.get_history('SN-001')
print(f"Device SN-001 has {len(history['tests'])} test records")
print(f"Pass rate: {tracker.get_pass_rate('SN-001'):.1f}%")

Integration with Factory Tests

from lager.cache import PersistentCache
from lager.factory import Step, run
import json

class LoadCalibrationStep(Step):
    """Load or perform calibration."""

    def run(self):
        self.update_heading("Loading Calibration")

        cache = PersistentCache()
        self.state['cache'] = cache

        data = cache.load('station_calibration')
        if data:
            cal = json.loads(data.decode())
            self.log(f"Loaded calibration from {cal.get('date', 'unknown')}")
            self.state['calibration'] = cal
        else:
            self.log("No calibration found - using defaults")
            self.state['calibration'] = {'offset': 0, 'gain': 1.0}

        return True


class SaveResultsStep(Step):
    """Save test results to cache."""

    def run(self):
        self.update_heading("Saving Results")

        cache = self.state.get('cache') or PersistentCache()
        serial = self.state.get('serial', 'unknown')

        results = {
            'serial': serial,
            'timestamp': time.time(),
            'voltage': self.state.get('measured_voltage'),
            'current': self.state.get('measured_current'),
            'passed': self.state.get('test_passed', False)
        }

        cache.store(f'result:{serial}', json.dumps(results).encode())
        self.log(f"Results saved for {serial}")

        return True


# Run factory test with cache
import time
run([LoadCalibrationStep, SaveResultsStep])

Error Handling

from lager.cache import PersistentCache

cache = PersistentCache()

try:
    cache.store('key', b'value')
except RuntimeError as e:
    print(f"Cache error: {e}")
    # Cache provider not available - Redis may not be running

# Safe pattern with fallback
def safe_cache_load(cache, key, default=None):
    """Load from cache with fallback."""
    try:
        data = cache.load(key)
        return data if data else default
    except RuntimeError:
        return default

value = safe_cache_load(cache, 'some_key', default=b'default_value')

Prerequisites

Requirement	Description
Redis	Redis server must be running on the gateway
Network	Gateway must have lagerredis hostname configured

Notes

• Data is stored as bytes - encode strings before storing
• Keys are strings - use namespacing to avoid collisions (e.g., mytest:calibration)
• Data persists across gateway restarts (Redis persistence)
• Cache is shared across all tests - use unique keys
• The clear() method affects all data - use with caution
• For large data, consider using files instead of cache
• Redis has memory limits - don’t store very large objects