POMDPPlanners.simulations.simulations_deployment package

class POMDPPlanners.simulations.simulations_deployment.DeploymentType(*values)[source]

Bases: Enum

Enum representing different types of deployment for simulations.

DASK_DISTRIBUTED = 'dask_distributed'

LOCAL = 'local'

REMOTE_RAY = 'remote_ray'

class POMDPPlanners.simulations.simulations_deployment.TaskManagerFactory[source]

Bases: object

Factory class for creating different types of task managers.

This factory provides methods to create specific types of task managers: - create_dask: Creates a DaskTaskManager for distributed computing - create_joblib: Creates a JoblibTaskManager with disk caching for parallel processing - create_pbs: Creates a PBSTaskManager for PBS cluster computing

static create_dask(n_workers=1, scheduler_address=None, cache_size=2000000000, clear_cache_on_start=False)[source]

Create a DaskTaskManager for distributed computing.

Parameters:

n_workers (int) – Number of worker processes (1 for local)
scheduler_address (Optional[str]) – Address of Dask scheduler (None for local)
cache_size (int) – Size of cache in bytes
clear_cache_on_start (bool) – If True, clears the cache at startup

Return type:

DaskTaskManager

Returns:

A configured DaskTaskManager instance

static create_joblib(cache_dir='./cache', cache_size=2000000000, n_jobs=-1, eviction_policy='least-recently-used', clear_cache_on_start=False, verbose=0, console_output=True, no_logs=False)[source]

Create a JoblibTaskManager with a configured DiskCacheDB.

Parameters:

cache_dir (str) – Directory to store cache files
cache_size (int) – Maximum size of cache in bytes
n_jobs (int) – Number of parallel jobs (-1 for all cores)
eviction_policy (str) – Cache eviction policy (‘least-recently-used’ or ‘least-frequently-used’)
clear_cache_on_start (bool) – If True, clears the cache at startup
verbose (int) – Verbosity level for joblib
console_output (bool) – Whether to print logs to console
no_logs (bool) – Whether to disable all logging

Return type:

JoblibTaskManager

Returns:

A configured JoblibTaskManager instance

static create_pbs(queue, n_workers=4, cores=1, memory='4GB', processes=1, walltime='01:00:00', job_extra=None, cache_size=2000000000, clear_cache_on_start=False, enable_dashboard=True, dashboard_address='0.0.0.0', dashboard_port=8787, dashboard_prefix=None)[source]

Create a PBSTaskManager for PBS cluster computing.

Parameters:

queue (str) – PBS queue name to submit jobs to
n_workers (int) – Number of worker jobs to submit to PBS
cores (int) – Number of CPU cores per PBS job
memory (str) – Memory per PBS job (e.g., “4GB”, “1000MB”)
processes (int) – Number of processes per PBS job
walltime (str) – Maximum runtime per job in HH:MM:SS format
job_extra (Optional[list]) – Additional PBS directives as list of strings
cache_size (int) – Size of cache in bytes
clear_cache_on_start (bool) – If True, clears the cache at startup
enable_dashboard (bool) – If True, enables the Dask dashboard
dashboard_address (str) – Address to bind the dashboard to
dashboard_port (int) – Port for the Dask dashboard
dashboard_prefix (Optional[str]) – URL prefix for dashboard (useful with reverse proxies)

Return type:

PBSTaskManager

Returns:

A configured PBSTaskManager instance

Raises:

RuntimeError – If dask-jobqueue is not installed

class POMDPPlanners.simulations.simulations_deployment.TaskManagerType(*values)[source]

Bases: Enum

Enum representing different types of task managers for simulation execution.

DASK = 'dask'

JOBLIB = 'joblib'

PBS = 'pbs'

Subpackages

Submodules

POMDPPlanners.simulations.simulations_deployment.cache_dbs module

class POMDPPlanners.simulations.simulations_deployment.cache_dbs.DiskCacheDB(cache_dir='./cache', size_limit=2000000000, eviction_policy='least-recently-used', debug=False)[source]

Bases: DataBaseInterface

A disk-based cache database implementation using diskcache.

Parameters:

cache_dir (str)
size_limit (int)
eviction_policy (str)
debug (bool)

clear()[source]: Clear all entries from the cache.

close()[source]: Close the underlying cache and release resources.

get(key)[source]

Retrieve a value from the cache.

Parameters:: key (str) – Cache key to retrieve
Return type:: Any
Returns:: The cached value, or None if not found

is_key_in_cache(key)[source]

Check if a key exists in the cache.

Parameters:: key (str) – Cache key to check
Returns:: True if key exists, False otherwise
Return type:: bool

set(key, value)[source]

Store a value in the cache.

Parameters:

key (str) – Cache key to store
value (Any) – Value to store

POMDPPlanners.simulations.simulations_deployment.task_manager_configs module

class POMDPPlanners.simulations.simulations_deployment.task_manager_configs.DaskConfig(n_workers=1, scheduler_address=None, cache_size=2000000000, clear_cache_on_start=False)[source]

Bases: TaskManagerConfig

Configuration for Dask task manager.

Parameters:

n_workers (int)
scheduler_address (str | None)
cache_size (int)
clear_cache_on_start (bool)

cache_size: int = 2000000000

clear_cache_on_start: bool = False

create_task_manager(cache_dir=None)[source]

Create and return the configured task manager.

Return type:: TaskManager
Parameters:: cache_dir (str | None)

n_workers: int = 1

scheduler_address: str | None = None

class POMDPPlanners.simulations.simulations_deployment.task_manager_configs.JoblibConfig(n_jobs=-1, cache_size=2000000000, eviction_policy='least-recently-used', verbose=0, clear_cache_on_start=False, console_output=True, no_logs=False)[source]

Bases: TaskManagerConfig

Configuration for Joblib task manager.

Parameters:

n_jobs (int)
cache_size (int)
eviction_policy (str)
verbose (int)
clear_cache_on_start (bool)
console_output (bool)
no_logs (bool)

cache_size: int = 2000000000

clear_cache_on_start: bool = False

console_output: bool = True

create_task_manager(cache_dir=None)[source]

Create and return the configured task manager.

Return type:: TaskManager
Parameters:: cache_dir (str | None)

eviction_policy: str = 'least-recently-used'

n_jobs: int = -1

no_logs: bool = False

verbose: int = 0

class POMDPPlanners.simulations.simulations_deployment.task_manager_configs.PBSConfig(queue, n_workers=4, cores=1, memory='4GB', processes=1, walltime='01:00:00', job_extra=None, cache_size=2000000000, clear_cache_on_start=False, enable_dashboard=True, dashboard_address='0.0.0.0', dashboard_port=8787, dashboard_prefix=None)[source]

Bases: TaskManagerConfig

Configuration for PBS cluster task manager.

Parameters:

queue (str)
n_workers (int)
cores (int)
memory (str)
processes (int)
walltime (str)
job_extra (List[str] | None)
cache_size (int)
clear_cache_on_start (bool)
enable_dashboard (bool)
dashboard_address (str)
dashboard_port (int)
dashboard_prefix (str | None)

cache_size: int = 2000000000

clear_cache_on_start: bool = False

cores: int = 1

create_task_manager(cache_dir=None)[source]

Create and return the configured task manager.

Return type:: TaskManager
Parameters:: cache_dir (str | None)

dashboard_address: str = '0.0.0.0'

dashboard_port: int = 8787

dashboard_prefix: str | None = None

enable_dashboard: bool = True

job_extra: List[str] | None = None

memory: str = '4GB'

n_workers: int = 4

processes: int = 1

queue: str

walltime: str = '01:00:00'

class POMDPPlanners.simulations.simulations_deployment.task_manager_configs.TaskManagerConfig[source]

Bases: ABC

Base configuration class for all task managers.

abstractmethod create_task_manager(cache_dir=None)[source]

Create and return the configured task manager.

Return type:: TaskManager
Parameters:: cache_dir (str | None)

POMDPPlanners.simulations.simulations_deployment.task_managers module

class POMDPPlanners.simulations.simulations_deployment.task_managers.DaskTaskManager(n_workers=1, scheduler_address=None, cache_size=2000000000, clear_cache_on_start=False)[source]

Bases: TaskManager

A class to manage simulation tasks and their execution using Dask.

Parameters:

n_workers (int)
scheduler_address (str | None)
cache_size (int)
clear_cache_on_start (bool)

cancel_tasks(futures)[source]

Cancel submitted tasks.

Parameters:: futures (List[Future]) – List of Dask futures to cancel

client: dask.distributed.Client | None

gather_results(futures)[source]

Gather results from submitted tasks.

Parameters:: futures (List[Future]) – List of Dask futures to gather results from
Returns:: List of simulation histories
Return type:: List[History]

get_task_status(futures)[source]

Get status of submitted tasks.

Parameters:: futures (List[Future]) – List of Dask futures to check
Returns:: Dictionary mapping task keys to their status
Return type:: Dict[str, str]

run_tasks(tasks, task_identifiers)[source]

Run tasks using Dask (submit and gather results).

Parameters:

tasks (List[SimulationTask]) – List of simulation tasks to execute
task_identifiers (list) – List of identifiers for the tasks

Returns:

A tuple containing:

List of successful simulation histories
List of identifiers for successful tasks

Return type:

Tuple[List[History], list]

set_progress_callback(callback)[source]

Register a callback invoked once per completed task.

The callback runs in a Dask client thread (parent process) as each dask.distributed.Future resolves. It is safe to touch parent-only resources (e.g. SQLite connections); ProgressDB already opens a fresh connection per call and uses WAL + busy_timeout=5000 so concurrent client-thread callbacks serialise cleanly. Exceptions raised by the callback are caught and logged at DEBUG level so a flaky callback cannot break the run.

Parameters:: callback (Optional[Callable[[], None]]) – A zero-argument callable, or None to clear an existing callback.
Return type:: None

submit_tasks(tasks)[source]

Submit tasks for execution.

Parameters:: tasks (List[SimulationTask]) – List of simulation tasks to execute
Returns:: List of Dask futures for the tasks
Return type:: List[Future]

class POMDPPlanners.simulations.simulations_deployment.task_managers.JoblibTaskManager(cache_db, n_jobs=-1, cache_dir=None, clear_cache_on_start=False, verbose=0, logger_debug=False, console_output=True, no_logs=False)[source]

Bases: TaskManagerExternalDB

A task manager that uses joblib for parallel execution and caching.

Parameters:

cache_db (DataBaseInterface)
n_jobs (int)
cache_dir (str | None)
clear_cache_on_start (bool)
verbose (int)
logger_debug (bool)
console_output (bool)
no_logs (bool)

clear_cache()[source]: Clear the joblib cache.

set_progress_callback(callback)[source]

Register a callback invoked once per completed task.

The callback runs in the parent process inside the joblib generator loop, so it is safe to touch parent-only resources (e.g. SQLite connections). Exceptions raised by the callback are caught and logged at DEBUG level so a flaky callback cannot break the run.

The callback is intentionally excluded from this object’s pickled state (via __getstate__()) so that joblib’s Memory cache, which hashes function arguments (including self), can still serialize the task manager even when the callback closes over non-picklable parent objects such as a live notifier or simulator.

Parameters:: callback (Optional[Callable[[], None]]) – A zero-argument callable, or None to clear an existing callback.
Return type:: None

class POMDPPlanners.simulations.simulations_deployment.task_managers.PBSTaskManager(queue, n_workers=4, cores=1, memory='4GB', processes=1, scheduler_address=None, walltime='01:00:00', job_extra=None, cache_size=2000000000, clear_cache_on_start=False, enable_dashboard=True, dashboard_address='0.0.0.0', dashboard_port=8787, dashboard_prefix=None)[source]

Bases: DaskTaskManager

A task manager that uses Dask with PBS cluster for distributed computing.

This task manager submits jobs to a PBS (Portable Batch System) cluster and provides an optional Dask dashboard for monitoring job execution, resource utilization, and performance metrics.

The dashboard is enabled by default and provides real-time visualization of: - Task progress and execution status - Resource utilization across PBS nodes - Task graphs and dependencies - Performance metrics and bottlenecks

Example

Basic usage with default dashboard:

with PBSTaskManager(queue="gpu_queue", n_workers=4) as manager:
    futures = manager.submit_tasks(tasks)
    print(f"Dashboard: {manager.get_dashboard_url()}")
    results = manager.gather_results(futures)

Custom dashboard configuration:

manager = PBSTaskManager(
    queue="compute_queue",
    n_workers=8,
    dashboard_port=8888,
    dashboard_address="0.0.0.0",
    enable_dashboard=True
)

Disable dashboard:

manager = PBSTaskManager(
    queue="batch_queue",
    enable_dashboard=False
)

Note

The dashboard requires network access to the specified port. In PBS environments, ensure the dashboard port is accessible through firewalls and security policies. The dashboard will automatically shut down when the task manager context exits.

Parameters:

queue (str)
n_workers (int)
cores (int)
memory (str)
processes (int)
scheduler_address (str | None)
walltime (str)
job_extra (List[str] | None)
cache_size (int)
clear_cache_on_start (bool)
enable_dashboard (bool)
dashboard_address (str)
dashboard_port (int)
dashboard_prefix (str | None)

get_dashboard_url()[source]

Get the URL for the Dask dashboard.

Return type:: Optional[str]
Returns:: Dashboard URL if available, None otherwise.

is_dashboard_running()[source]

Check if the Dask dashboard is running.

Return type:: bool
Returns:: True if dashboard is running, False otherwise.

class POMDPPlanners.simulations.simulations_deployment.task_managers.SequentialTaskManager(cache_db, cache_dir=None, clear_cache_on_start=False, verbose=0, logger_debug=False, console_output=True, no_logs=False)[source]

Bases: JoblibTaskManager

A task manager that uses sequential execution and caching.

Parameters:

cache_db (DataBaseInterface)
cache_dir (str | None)
clear_cache_on_start (bool)
verbose (int)
logger_debug (bool)
console_output (bool)
no_logs (bool)

class POMDPPlanners.simulations.simulations_deployment.task_managers.TaskManagerType(*values)[source]

Bases: Enum

Enum representing different types of task managers for simulation execution.

DASK = 'dask'

JOBLIB = 'joblib'

PBS = 'pbs'