"""
Backward-compatibility shim for legacy ``Optimized*`` names (RL / Caputo / GL engines).
The **only** RL / Caputo / Grünwald–Letnikov implementations are
:class:`~hpfracc.algorithms.derivatives.RiemannLiouville`,
:class:`~hpfracc.algorithms.derivatives.Caputo`, and
:class:`~hpfracc.algorithms.derivatives.GrunwaldLetnikov`.
``Optimized*`` and ``ParallelOptimized*`` here are thin aliases of those classes.
Deprecated **GPU-prefixed** wrappers and ``GPUConfig`` also live here. Import them
from this module only.
Prefer: ``from hpfracc.algorithms.derivatives import Caputo``.
Numerical integrity: symbols that are not implemented do not silently succeed.
They raise NotImplementedError with guidance. Reserved APIs may be implemented later.
"""
import warnings
from typing import Any, Optional, Union
from .derivatives import RiemannLiouville, Caputo, GrunwaldLetnikov
from .impls.cuda_backend import CUPY_AVAILABLE
from .impls.jax_backend import JAX_AVAILABLE
OptimizedRiemannLiouville = RiemannLiouville
OptimizedCaputo = Caputo
OptimizedGrunwaldLetnikov = GrunwaldLetnikov
_UNIMPL_MSG = (
"{name} is not implemented in hpfracc.algorithms.optimized_methods. "
"Use hpfracc.algorithms.derivatives (RiemannLiouville, Caputo, GrunwaldLetnikov) "
"or hpfracc.algorithms.advanced_methods for supported operators."
)
# For backward compatibility with things asking for 'optimized_*' functions
[docs]
def optimized_riemann_liouville(f, t, alpha, h=None):
return OptimizedRiemannLiouville(alpha).compute(f, t, h)
[docs]
def optimized_caputo(f, t, alpha, h=None):
return OptimizedCaputo(alpha).compute(f, t, h)
[docs]
def optimized_grunwald_letnikov(f, t, alpha, h=None):
return OptimizedGrunwaldLetnikov(alpha).compute(f, t, h)
[docs]
class OptimizedFractionalMethods:
"""Reserved unified facade; not implemented — use per-method classes in derivatives."""
[docs]
def compute(self, *args: Any, **kwargs: Any) -> Any:
raise NotImplementedError(_UNIMPL_MSG.format(name="OptimizedFractionalMethods.compute"))
[docs]
class ParallelConfig:
"""
Configuration for optional parallel paths in advanced_methods (e.g. Weyl).
Accepts ``n_jobs`` / ``enabled`` and common aliases ``num_workers``, ``chunk_size``.
"""
[docs]
def __init__(
self,
n_jobs: int = 1,
enabled: bool = False,
*,
num_workers: Optional[int] = None,
chunk_size: Optional[int] = None,
**kwargs: Any,
):
if num_workers is not None:
n_jobs = int(num_workers)
self.n_jobs = int(n_jobs)
self.enabled = bool(enabled)
self.chunk_size = chunk_size
for key, value in kwargs.items():
setattr(self, key, value)
[docs]
def get_optimal_workers(self) -> int:
"""Return a sensible worker count (minimal implementation for callers/tests)."""
return max(1, self.n_jobs)
[docs]
class AdvancedFFTMethods:
"""Reserved for future FFT-focused helpers; not implemented."""
[docs]
def __init__(self, method: str = "spectral", *args: Any, **kwargs: Any):
raise NotImplementedError(_UNIMPL_MSG.format(name="AdvancedFFTMethods"))
[docs]
class L1L2Schemes:
"""Reserved for L1/L2 temporal discretization helpers; not implemented."""
[docs]
def __init__(self, scheme: str = "l1", *args: Any, **kwargs: Any):
raise NotImplementedError(
f"{_UNIMPL_MSG.format(name='L1L2Schemes')} "
f"(requested scheme={scheme!r}). "
"For fractional PDE stepping, use hpfracc.solvers.pde_solvers.FractionalDiffusionSolver."
)
[docs]
class ParallelLoadBalancer:
"""Reserved for future parallel chunking; no balancing logic yet."""
[docs]
def balance_load(self, data: Any, num_workers: int = 1) -> Any:
raise NotImplementedError(_UNIMPL_MSG.format(name="ParallelLoadBalancer.balance_load"))
[docs]
def get_chunk_sizes(self, total: int, num_workers: int) -> list:
raise NotImplementedError(_UNIMPL_MSG.format(name="ParallelLoadBalancer.get_chunk_sizes"))
# Aliases: inherit canonical engines directly until dedicated parallel kernels exist.
[docs]
class ParallelOptimizedRiemannLiouville(RiemannLiouville):
"""Alias for :class:`~hpfracc.algorithms.derivatives.RiemannLiouville`."""
[docs]
class ParallelOptimizedCaputo(Caputo):
"""Alias for :class:`~hpfracc.algorithms.derivatives.Caputo`."""
[docs]
class ParallelOptimizedGrunwaldLetnikov(GrunwaldLetnikov):
"""Alias for :class:`~hpfracc.algorithms.derivatives.GrunwaldLetnikov`."""
[docs]
class NumbaOptimizer:
[docs]
def __init__(self, *args: Any, **kwargs: Any):
raise NotImplementedError(_UNIMPL_MSG.format(name="NumbaOptimizer"))
[docs]
class NumbaFractionalKernels:
[docs]
def __init__(self, *args: Any, **kwargs: Any):
raise NotImplementedError(_UNIMPL_MSG.format(name="NumbaFractionalKernels"))
[docs]
class NumbaParallelManager:
[docs]
def __init__(self, *args: Any, **kwargs: Any):
raise NotImplementedError(_UNIMPL_MSG.format(name="NumbaParallelManager"))
[docs]
def benchmark_parallel_vs_serial(*args: Any, **kwargs: Any) -> Any:
raise NotImplementedError(_UNIMPL_MSG.format(name="benchmark_parallel_vs_serial"))
[docs]
def optimize_parallel_parameters(*args: Any, **kwargs: Any) -> Any:
raise NotImplementedError(_UNIMPL_MSG.format(name="optimize_parallel_parameters"))
[docs]
def memory_efficient_caputo(*args: Any, **kwargs: Any) -> Any:
raise NotImplementedError(_UNIMPL_MSG.format(name="memory_efficient_caputo"))
[docs]
def block_processing_kernel(*args: Any, **kwargs: Any) -> Any:
raise NotImplementedError(_UNIMPL_MSG.format(name="block_processing_kernel"))
# -----------------------------------------------------------------------------
# Deprecated GPU-named shims
# -----------------------------------------------------------------------------
[docs]
class GPUConfig:
"""[DEPRECATED] Legacy GPU dispatch config; use ``backend=`` on unified engines."""
[docs]
def __init__(
self,
backend: str = "auto",
memory_limit: float = 0.8,
batch_size: Optional[int] = None,
multi_gpu: bool = False,
monitor_performance: bool = True,
fallback_to_cpu: bool = True,
device_id: Optional[int] = None,
use_intelligent_selection: bool = True,
):
warnings.warn(
"GPUConfig is deprecated. Use arguments in Unified API classes directly.",
DeprecationWarning,
stacklevel=2,
)
self.backend = backend
self.performance_stats = {
"gpu_time": 0.0,
"cpu_time": 0.0,
"memory_usage": [],
"speedup": 1.0,
}
self.monitor_performance = monitor_performance
self.fallback_to_cpu = fallback_to_cpu
self.device_id = device_id
self.memory_limit = memory_limit
[docs]
def select_backend_for_data(self, *args: Any, **kwargs: Any) -> str:
return self.backend
[docs]
class _LegacyGPUShim:
"""Map legacy ``GPUConfig`` backend strings to unified ``backend``."""
@staticmethod
def _map_config(gpu_config: Optional[GPUConfig]) -> str:
if gpu_config is None:
return "auto"
backend = gpu_config.backend
if backend == "cupy":
return "cuda"
if backend == "jax":
return "jax"
if backend == "numpy":
return "numpy"
return "auto"
[docs]
class GPUOptimizedRiemannLiouville(RiemannLiouville, _LegacyGPUShim):
"""[DEPRECATED] Use ``RiemannLiouville(..., backend='jax'|'cuda'|'numpy')``."""
[docs]
def __init__(
self,
alpha: Union[float, Any],
gpu_config: Optional[GPUConfig] = None,
*,
config: Optional[GPUConfig] = None,
batch_size: Optional[int] = None,
):
warnings.warn(
"GPUOptimizedRiemannLiouville is deprecated. Use RiemannLiouville(backend=...) instead.",
DeprecationWarning,
stacklevel=2,
)
legacy_conf = gpu_config or config or GPUConfig(backend="auto")
backend = self._map_config(legacy_conf)
super().__init__(alpha, backend=backend)
self.gpu_config = legacy_conf
self.batch_size = batch_size
[docs]
def enable_monitoring(self) -> None: # pragma: no cover - legacy stub
return None
[docs]
class GPUOptimizedCaputo(Caputo, _LegacyGPUShim):
"""[DEPRECATED] Use ``Caputo(..., backend='jax'|'cuda'|'numpy')``."""
[docs]
def __init__(
self,
alpha: Union[float, Any],
gpu_config: Optional[GPUConfig] = None,
*,
config: Optional[GPUConfig] = None,
memory_efficient: Optional[bool] = None,
):
warnings.warn(
"GPUOptimizedCaputo is deprecated. Use Caputo(backend=...) instead.",
DeprecationWarning,
stacklevel=2,
)
legacy_conf = gpu_config or config or GPUConfig(backend="auto")
backend = self._map_config(legacy_conf)
super().__init__(alpha, backend=backend)
self.gpu_config = legacy_conf
self.memory_efficient = memory_efficient
[docs]
def compute(self, f: Any, t: Any, h: Optional[float] = None, method: str = "l1") -> Any:
if method != "l1":
warnings.warn("Method selection is deprecated/ignored in Unified API.")
return super().compute(f, t, h)
[docs]
class GPUOptimizedGrunwaldLetnikov(GrunwaldLetnikov, _LegacyGPUShim):
"""[DEPRECATED] Use ``GrunwaldLetnikov(..., backend='jax'|'cuda'|'numpy')``."""
[docs]
def __init__(
self,
alpha: Union[float, Any],
gpu_config: Optional[GPUConfig] = None,
*,
config: Optional[GPUConfig] = None,
use_shared_memory: Optional[bool] = None,
):
warnings.warn(
"GPUOptimizedGrunwaldLetnikov is deprecated. Use GrunwaldLetnikov(backend=...) instead.",
DeprecationWarning,
stacklevel=2,
)
legacy_conf = gpu_config or config or GPUConfig(backend="auto")
backend = self._map_config(legacy_conf)
super().__init__(alpha, backend=backend)
self.gpu_config = legacy_conf
self.use_shared_memory = use_shared_memory
[docs]
class MultiGPUManager:
"""[DEPRECATED] Stub; not used by unified engines."""
[docs]
def __init__(self, *args: Any, **kwargs: Any) -> None:
warnings.warn("MultiGPUManager is deprecated/stubbed.", DeprecationWarning, stacklevel=2)
[docs]
class GPUOptimizedMethods:
"""Legacy placeholder; never implemented. Use engines from ``derivatives``."""
[docs]
def __init__(self, *args: Any, **kwargs: Any) -> None:
raise NotImplementedError(
"GPUOptimizedMethods is not implemented. Use RiemannLiouville, Caputo, or "
"GrunwaldLetnikov from hpfracc.algorithms.derivatives with backend='jax' or 'cuda'."
)
[docs]
def gpu_optimized_riemann_liouville(
f: Any, alpha: Any, t: Any, h: Optional[float] = None, *args: Any, **kwargs: Any
) -> Any:
warnings.warn(
"gpu_optimized_riemann_liouville is deprecated. Use Unified API.",
DeprecationWarning,
stacklevel=2,
)
return GPUOptimizedRiemannLiouville(alpha).compute(f, t, h)
[docs]
def gpu_optimized_caputo(
f: Any, alpha: Any, t: Any, h: Optional[float] = None, *args: Any, **kwargs: Any
) -> Any:
warnings.warn(
"gpu_optimized_caputo is deprecated. Use Unified API.",
DeprecationWarning,
stacklevel=2,
)
return GPUOptimizedCaputo(alpha).compute(f, t, h)
[docs]
def gpu_optimized_grunwald_letnikov(
f: Any, alpha: Any, t: Any, h: Optional[float] = None, *args: Any, **kwargs: Any
) -> Any:
warnings.warn(
"gpu_optimized_grunwald_letnikov is deprecated. Use Unified API.",
DeprecationWarning,
stacklevel=2,
)
return GPUOptimizedGrunwaldLetnikov(alpha).compute(f, t, h)
[docs]
def benchmark_gpu_vs_cpu(*args: Any, **kwargs: Any) -> dict:
warnings.warn("benchmark_gpu_vs_cpu is deprecated.", DeprecationWarning, stacklevel=2)
return {"gpu_time": 0.0, "cpu_time": 0.0}
def __getattr__(name: str) -> Any:
"""Lazy re-export for legacy ``from ...optimized_methods import FractionalOperator``."""
if name == "FractionalOperator":
from . import base
return base.FractionalOperator
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")