Skip to content

_activation

Runtime activation/deactivation. Mutates R environment ONLY. Does NOT touch config - that's the caller's responsibility.

Attributes

MANAGED_PACKAGES = ('brms', 'cmdstanr', 'rstan', 'StanHeaders', 'tibble', 'pkgconfig') module-attribute

Functions

log(*msg, method_name=None, level=logging.INFO)

Log a message with automatic method name detection.

Parameters:

Name Type Description Default
msg str

The message to log

()
method_name str

The name of the method/function. If None, will auto-detect from call stack.

None
level int

Logging level (default: logging.INFO)

INFO
Source code in brmspy/helpers/log.py
def log(*msg: str, method_name: str | None = None, level: int = logging.INFO):
    """
    Log a message with automatic method name detection.

    Parameters
    ----------
    msg : str
        The message to log
    method_name : str, optional
        The name of the method/function. If None, will auto-detect from call stack.
    level : int, optional
        Logging level (default: logging.INFO)
    """
    if method_name is None:
        method_name = _get_caller_name()

    msg_str = " ".join(str(v) for v in msg)

    logger = get_logger()
    logger.log(level, msg_str, extra={"method_name": method_name})

log_warning(msg, method_name=None)

Log a warning message.

Parameters:

Name Type Description Default
msg str

The warning message to log

required
method_name str

The name of the method/function. If None, will auto-detect from call stack.

None
Source code in brmspy/helpers/log.py
def log_warning(msg: str, method_name: str | None = None):
    """
    Log a warning message.

    Parameters
    ----------
    msg : str
        The warning message to log
    method_name : str, optional
        The name of the method/function. If None, will auto-detect from call stack.

    """
    log(msg, method_name=method_name, level=logging.WARNING)

activate(runtime_path)

Activate runtime by mutating R environment.

Steps: 1. Parse and validate manifest 2. Store original R environment (if not already stored) 3. Unload managed packages if loaded 4. Set .libPaths() to runtime's Rlib/ 5. Set cmdstan path to runtime's cmdstan/ 6. Verify packages are loadable 7. Invalidate package singletons

Does NOT save to config. Caller handles that.

On failure, attempts to restore original environment.

Source code in brmspy/_runtime/_activation.py
def activate(runtime_path: Path) -> None:
    """
    Activate runtime by mutating R environment.

    Steps:
    1. Parse and validate manifest
    2. Store original R environment (if not already stored)
    3. Unload managed packages if loaded
    4. Set .libPaths() to runtime's Rlib/
    5. Set cmdstan path to runtime's cmdstan/
    6. Verify packages are loadable
    7. Invalidate package singletons

    Does NOT save to config. Caller handles that.

    On failure, attempts to restore original environment.
    """
    log(f"Activating runtime {runtime_path}")
    stored = _state.get_stored_env()

    # Validate
    manifest = _manifest.parse_manifest(runtime_path / "manifest.json")
    if manifest is None:
        raise RuntimeError(f"Invalid manifest in {runtime_path}")

    _manifest.validate_manifest(manifest, _platform.system_fingerprint())

    if stored is not None:
        deactivate()

    # Capture original env (unless already captured from previous activation)
    if not _state.has_stored_env():
        original = _state.capture_current_env()
        _state.store_env(original)

    # Attempt activation with rollback on failure
    try:
        _unload_managed_packages()

        rlib = runtime_path / "Rlib"
        cmdstan = runtime_path / "cmdstan"

        rlib_posix = rlib.as_posix()
        cmdstan_posix = cmdstan.as_posix()

        _r_env.set_lib_paths([str(rlib_posix)])
        log(f"lib paths are {_r_env.get_lib_paths()}")
        _state.invalidate_packages()
        _verify_runtime_loadable()
        log(f"Setting cmdstan path to {cmdstan_posix}")
        _r_env.set_cmdstan_path(str(cmdstan_posix))

    except Exception as e:
        # Rollback
        _rollback_to_stored_env()
        raise RuntimeError(f"Activation failed: {e}") from e

deactivate()

Deactivate runtime by restoring original R environment.

Does NOT clear config. Caller handles that.

Raises: RuntimeError: If no stored environment to restore.

Source code in brmspy/_runtime/_activation.py
def deactivate() -> None:
    """
    Deactivate runtime by restoring original R environment.

    Does NOT clear config. Caller handles that.

    Raises:
        RuntimeError: If no stored environment to restore.
    """
    active_path = _config.get_active_runtime_path()
    stored = _state.get_stored_env()
    if stored is None:
        raise RuntimeError("No runtime is currently active (no stored environment)")

    if _platform.get_os() != "windows":
        _r_env._unload_libpath_packages(active_path)
        _unload_managed_packages()
    else:
        _unload_managed_packages()

    _r_env.set_lib_paths(stored.lib_paths)
    try:
        _r_env.set_cmdstan_path(stored.cmdstan_path)
    except Exception as e:
        log_warning(
            f"Failed to set_cmdstan_path to stored default ({stored.cmdstan_path}). Skipping! {e}"
        )
    _state.clear_stored_env()
    _state.invalidate_packages()

_unload_managed_packages()

Unload brms, cmdstanr, rstan if loaded.

Source code in brmspy/_runtime/_activation.py
def _unload_managed_packages() -> None:
    """Unload brms, cmdstanr, rstan if loaded."""
    for pkg in MANAGED_PACKAGES:
        if _r_env.is_namespace_loaded(pkg) or _r_env.is_package_attached(pkg):
            try:
                _r_env.unload_package(pkg)
            except Exception as e:
                log_warning(f"{e}")

_remove_managed_packages()

removes brms, cmdstanr, rstan if loaded.

Source code in brmspy/_runtime/_activation.py
def _remove_managed_packages() -> None:
    """removes brms, cmdstanr, rstan if loaded."""
    for pkg in MANAGED_PACKAGES:
        if _r_packages.is_package_installed(pkg):
            try:
                _r_packages.remove_package(pkg)
            except Exception as e:
                log_warning(f"{e}")

_verify_runtime_loadable()

Verify brms and cmdstanr can be loaded.

Source code in brmspy/_runtime/_activation.py
def _verify_runtime_loadable() -> None:
    """Verify brms and cmdstanr can be loaded."""
    _state.get_brms()
    _state.get_cmdstanr()

_rollback_to_stored_env()

Restore original env on activation failure.

Source code in brmspy/_runtime/_activation.py
def _rollback_to_stored_env() -> None:
    """Restore original env on activation failure."""
    stored = _state.get_stored_env()
    if stored:
        try:
            _r_env.set_lib_paths(stored.lib_paths)
            _r_env.set_cmdstan_path(stored.cmdstan_path)
        except Exception:
            pass  # Best effort