embodichain.lab.sim.motion.workspace.caches

Cache backends and a cache manager for persisting workspace-analysis results.

Provides in-memory and disk caches (MemoryCache, DiskCache), a CacheManager, a ResultsCache, and compute_cache_key.

Classes:

BaseCache	Abstract base class for workspace sampling cache strategies.
CacheManager	Factory and manager for workspace sampling caches.
DiskCache	Disk-based cache for workspace sampling.
MemoryCache	In-memory cache for workspace sampling.
ResultsCache	Disk cache for workspace results, keyed by robot and parameters.

Functions:

compute_cache_key(metadata)

Compute a stable, readable key for analysis inputs.

class embodichain.lab.sim.motion.workspace.caches.BaseCache

Bases: ABC

Abstract base class for workspace sampling cache strategies.

Defines the interface for different caching mechanisms (memory, disk) used during workspace analysis sampling operations.

Methods:

__init__([batch_size, save_threshold])	Initialize base cache parameters.
add(poses)	Add pose samples to the cache.
clear()	Clear all cached data.
flush()	Flush any pending data in the cache.
get_all()	Retrieve all cached poses.

Attributes:

total_processed

Total number of poses processed by this cache.

__init__(batch_size=5000, save_threshold=10000000)

Initialize base cache parameters.

Parameters:

• batch_size (int) – Number of samples to process in each batch

• save_threshold (int) – Number of samples to accumulate before triggering save/flush

abstract add(poses)

Add pose samples to the cache.

Parameters:

poses (List[ndarray]) – List of 4x4 transformation matrices

Return type:

None

abstract clear()

Clear all cached data.

Return type:

None

abstract flush()

Flush any pending data in the cache.

Return type:

None

abstract get_all()

Retrieve all cached poses.

Return type:

Optional[List[ndarray]]

Returns:

List of all cached 4x4 transformation matrices, or None if unavailable

property total_processed: int

Total number of poses processed by this cache.

class embodichain.lab.sim.motion.workspace.caches.CacheManager

Bases: object

Factory and manager for workspace sampling caches.

Provides a unified interface for creating and managing different cache strategies (memory vs disk).

Methods:

create_cache(cache_mode[, save_dir, ...])	Create a cache instance based on the specified mode.
create_cache_from_config(config)	Create a cache instance from a CacheConfig object.

static create_cache(cache_mode, save_dir=None, batch_size=5000, save_threshold=10000000, use_cached=True)

Create a cache instance based on the specified mode.

Parameters:

• cache_mode (Literal['memory', 'disk']) – Caching strategy - “memory” or “disk”

• save_dir (str | None) – Directory for disk cache. If None in disk mode, uses ~/.cache/embodichain/workspace_analyzer/session_TIMESTAMP

• batch_size (int) – Number of samples per batch

• save_threshold (int) – Threshold for saving/flushing data

• use_cached (bool) – Whether to use existing cached data (disk mode only)

Return type:

BaseCache

Returns:

Configured cache instance

Raises:

ValueError – If cache_mode is invalid

static create_cache_from_config(config)

Create a cache instance from a CacheConfig object.

Parameters:

config (CacheConfig) – CacheConfig instance with cache settings

Return type:

BaseCache | None

Returns:

Configured cache instance if enabled, None otherwise

class embodichain.lab.sim.motion.workspace.caches.DiskCache

Bases: BaseCache

Disk-based cache for workspace sampling.

Saves pose samples to disk in batches to minimize memory usage. Suitable for large-scale sampling operations.

Default cache location: ~/.cache/embodichain/workspace_analyzer/

Methods:

__init__([save_dir, batch_size, ...])	Initialize disk cache.
add(poses)	Add poses to buffer and save to disk when threshold is reached.
clear()	Clear all cached data and remove batch files.
flush()	Flush any remaining data in buffer to disk.
get_all()	Load and merge all batch files from disk.
get_batch_count()	Get number of batches written to disk.
get_default_cache_dir([subdir])	Get default cache directory in user's home.

__init__(save_dir=None, batch_size=5000, save_threshold=10000000, use_cached=True)

Initialize disk cache.

Parameters:

• save_dir (str | None) – Directory path for saving batch files. If None, uses ~/.cache/embodichain/workspace_analyzer/default

• batch_size (int) – Number of samples per batch

• save_threshold (int) – Number of samples to accumulate before writing to disk

• use_cached (bool) – Whether to use existing cached files if available

add(poses)

Add poses to buffer and save to disk when threshold is reached.

Parameters:

poses (List[ndarray]) – List of 4x4 transformation matrices

Return type:

None

clear()

Clear all cached data and remove batch files.

Return type:

None

flush()

Flush any remaining data in buffer to disk.

Return type:

None

get_all()

Load and merge all batch files from disk.

Return type:

Optional[List[ndarray]]

Returns:

List of all cached poses merged from batch files, or None if no data

get_batch_count()

Get number of batches written to disk.

Return type:

int

Returns:

Number of batch files on disk

static get_default_cache_dir(subdir='default')

Get default cache directory in user’s home.

Parameters:

subdir (str) – Subdirectory name under workspace_analyzer cache

Returns:

~/.cache/embodichain/workspace_analyzer/{subdir}

Return type:

Path to cache directory

class embodichain.lab.sim.motion.workspace.caches.MemoryCache

Bases: BaseCache

In-memory cache for workspace sampling.

Stores all pose samples in RAM for fast access. Suitable for smaller datasets or when memory is not a constraint.

Methods:

__init__([batch_size, save_threshold])	Initialize memory cache.
add(poses)	Add poses to in-memory storage.
clear()	Clear all cached data and free memory.
flush()	Flush operation (no-op for memory cache, but triggers GC).
get_all()	Retrieve all cached poses.

__init__(batch_size=5000, save_threshold=10000000)

Initialize memory cache.

Parameters:

• batch_size (int) – Number of samples per processing batch

• save_threshold (int) – Threshold for triggering garbage collection

add(poses)

Add poses to in-memory storage.

Parameters:

poses (List[ndarray]) – List of 4x4 transformation matrices

Return type:

None

clear()

Clear all cached data and free memory.

Return type:

None

flush()

Flush operation (no-op for memory cache, but triggers GC).

Return type:

None

get_all()

Retrieve all cached poses.

Return type:

Optional[List[ndarray]]

Returns:

List of all cached poses, or None if empty

class embodichain.lab.sim.motion.workspace.caches.ResultsCache

Bases: object

Disk cache for workspace results, keyed by robot and parameters.

Results are stored under <cache_dir>/<key>/ as results.npz plus a meta.json sidecar. A short content-hash suffix covers the complete analysis inputs so identical configurations reuse the same cache entry.

Attributes:

META_FILENAME
RESULTS_FILENAME

Methods:

__init__([cache_dir])	Initialize the results cache.
entry_path(key)	Get the directory path for a cache entry.
exists(key)	Check whether a cache entry exists for the given key.
list_entries()	List all cache entries with summary info.
load(key)	Load analysis results from disk.
save(key, results[, metadata, compression])	Save analysis results to disk.

META_FILENAME = 'meta.json'

RESULTS_FILENAME = 'results.npz'

__init__(cache_dir=None)

Initialize the results cache.

Parameters:

cache_dir (str | PathLike | None) – Root directory for cached results. Defaults to DEFAULT_RESULTS_CACHE_DIR.

entry_path(key)

Get the directory path for a cache entry.

Parameters:

key (str) – Cache key (from compute_cache_key()).

Return type:

Path

Returns:

Path to the entry directory (not guaranteed to exist).

exists(key)

Check whether a cache entry exists for the given key.

Parameters:

key (str) – Cache key.

Return type:

bool

Returns:

True if both results.npz and meta.json exist.

list_entries()

List all cache entries with summary info.

Return type:

list[dict]

Returns:

A list of dicts (key, path, size_bytes, meta) sorted newest-first by modification time.

load(key)

Load analysis results from disk.

Parameters:

key (str) – Cache key.

Return type:

dict | None

Returns:

Reconstructed results dict, or None if the entry does not exist or cannot be read.

save(key, results, metadata=None, compression=True)

Save analysis results to disk.

Parameters:

• key (str) – Cache key.

• results (dict) – Results dict from WorkspaceAnalyzer.analyze().

• metadata (dict | None) – Optional key-input metadata to embed in meta.json for traceability.

• compression (bool) – If True, compress the .npz archive.

Return type:

Path

Returns:

Path to the entry directory holding the written files.

embodichain.lab.sim.motion.workspace.caches.compute_cache_key(metadata)

Compute a stable, readable key for analysis inputs.

The directory name begins with the robot name and the most useful analysis parameters, while a short content hash covers the complete metadata. This keeps cache entries identifiable without losing collision resistance when less-visible parameters such as bounds or IK settings change.

Parameters:

metadata (dict) – Dictionary of all inputs that affect the analysis output (robot identity, mode, sampling, constraints, …).

Return type:

str

Returns:

A filesystem-safe robot + parameters + hash directory name.