environment_parent

Classes

EnvironmentConfig dataclass

Worker environment configuration.

This configuration is applied in the worker before importing/using brms.

Parameters:

Name	Type	Description	Default
r_home	str or None	Override for R_HOME. If None, the worker will rely on system detection.	None
startup_scripts	list[str]	R code snippets executed in the worker after initialization.	list()
environment_name	str	brmspy environment name (used to determine ~/.brmspy/environment/<name>/Rlib).	'default'
runtime_path	str or None	Path to a brmspy runtime bundle to activate in the worker.	None
env	dict[str, str]	Extra environment variables applied when spawning the worker.	dict()

Source code in brmspy/types/session.py

@dataclass
class EnvironmentConfig:
    """
    Worker environment configuration.

    This configuration is applied in the worker before importing/using brms.

    Parameters
    ----------
    r_home : str or None
        Override for `R_HOME`. If None, the worker will rely on system detection.
    startup_scripts : list[str]
        R code snippets executed in the worker after initialization.
    environment_name : str
        brmspy environment name (used to determine `~/.brmspy/environment/<name>/Rlib`).
    runtime_path : str or None
        Path to a brmspy runtime bundle to activate in the worker.
    env : dict[str, str]
        Extra environment variables applied when spawning the worker.
    """

    r_home: None | str = None
    startup_scripts: list[str] = field(default_factory=list)
    environment_name: str = "default"
    runtime_path: None | str = None
    env: dict[str, str] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        """Serialize configuration for persistence to JSON."""
        return {
            "environment_name": self.environment_name,
            "r_home": self.r_home,
            "startup_scripts": self.startup_scripts or [],
            "runtime_path": self.runtime_path,
            "env": self.env,
        }

    @classmethod
    def from_dict(cls, obj: dict[str, Any]) -> EnvironmentConfig:
        """Deserialize configuration from a JSON object."""
        return cls(
            r_home=obj["r_home"],
            startup_scripts=obj["startup_scripts"],
            environment_name=obj["environment_name"],
            runtime_path=obj["runtime_path"],
            env=obj["env"],
        )

    @classmethod
    def from_obj(
        cls, obj: None | dict[str, Any] | EnvironmentConfig
    ) -> EnvironmentConfig:
        """Normalize `None | dict | EnvironmentConfig` into an `EnvironmentConfig`."""
        if obj is None:
            return cls()
        if isinstance(obj, dict):
            return cls.from_dict(obj)
        return obj

Attributes

r_home = None class-attribute instance-attribute

startup_scripts = field(default_factory=list) class-attribute instance-attribute

environment_name = 'default' class-attribute instance-attribute

runtime_path = None class-attribute instance-attribute

env = field(default_factory=dict) class-attribute instance-attribute

Functions

to_dict()

Serialize configuration for persistence to JSON.

Source code in brmspy/types/session.py

def to_dict(self) -> dict[str, Any]:
    """Serialize configuration for persistence to JSON."""
    return {
        "environment_name": self.environment_name,
        "r_home": self.r_home,
        "startup_scripts": self.startup_scripts or [],
        "runtime_path": self.runtime_path,
        "env": self.env,
    }

from_dict(obj) classmethod

Deserialize configuration from a JSON object.

Source code in brmspy/types/session.py

@classmethod
def from_dict(cls, obj: dict[str, Any]) -> EnvironmentConfig:
    """Deserialize configuration from a JSON object."""
    return cls(
        r_home=obj["r_home"],
        startup_scripts=obj["startup_scripts"],
        environment_name=obj["environment_name"],
        runtime_path=obj["runtime_path"],
        env=obj["env"],
    )

from_obj(obj) classmethod

Normalize None | dict | EnvironmentConfig into an EnvironmentConfig.

Source code in brmspy/types/session.py

@classmethod
def from_obj(
    cls, obj: None | dict[str, Any] | EnvironmentConfig
) -> EnvironmentConfig:
    """Normalize `None | dict | EnvironmentConfig` into an `EnvironmentConfig`."""
    if obj is None:
        return cls()
    if isinstance(obj, dict):
        return cls.from_dict(obj)
    return obj

__init__(r_home=None, startup_scripts=list(), environment_name='default', runtime_path=None, env=dict())

Functions

get_environment_base_dir()

Return the base directory for brmspy environments, creating it if needed.

Returns:

Type	Description
Path	~/.brmspy/environment/

Source code in brmspy/_session/environment.py

def get_environment_base_dir() -> Path:
    """
    Return the base directory for brmspy environments, creating it if needed.

    Returns
    -------
    pathlib.Path
        `~/.brmspy/environment/`
    """
    base_dir = Path.home() / ".brmspy" / "environment"
    base_dir.mkdir(parents=True, exist_ok=True)
    return base_dir

get_environment_userlibs_dir(name)

Return the per-environment user library directory: .../<name>/Rlib.

Source code in brmspy/_session/environment.py

def get_environment_userlibs_dir(name: str) -> Path:
    """Return the per-environment user library directory: `.../<name>/Rlib`."""
    return get_environment_dir(name=name) / "Rlib"

get_environments_state_path()

Return the path to environment_state.json (stores last active environment name).

Source code in brmspy/_session/environment.py

def get_environments_state_path() -> Path:
    """Return the path to `environment_state.json` (stores last active environment name)."""
    return Path.home() / ".brmspy" / "environment_state.json"

save(env_conf)

Persist an environment configuration and ensure the directory structure exists.

Parameters:

Name	Type	Description	Default
env_conf	EnvironmentConfig	Environment configuration to write.	required

Source code in brmspy/_session/environment_parent.py

def save(env_conf: EnvironmentConfig) -> None:
    """
    Persist an environment configuration and ensure the directory structure exists.

    Parameters
    ----------
    env_conf : brmspy.types.session.EnvironmentConfig
        Environment configuration to write.
    """
    base_dir = get_environment_base_dir()
    env_dir = base_dir / env_conf.environment_name
    env_rlib_dir = get_environment_userlibs_dir(name=env_conf.environment_name)
    config_dir = env_dir / "config.json"
    os.makedirs(env_dir, exist_ok=True)
    os.makedirs(env_rlib_dir, exist_ok=True)

    if "BRMSPY_AUTOLOAD" in env_conf.env:
        del env_conf.env["BRMSPY_AUTOLOAD"]

    with open(config_dir, "w", encoding="utf-8") as f:
        json.dump(env_conf.to_dict(), f, indent=2, ensure_ascii=False)

save_as_state(env_conf)

Record the active environment name in environment_state.json.

Parameters:

Name	Type	Description	Default
env_conf	EnvironmentConfig	Environment configuration whose name should be recorded.	required

Source code in brmspy/_session/environment_parent.py

def save_as_state(env_conf: EnvironmentConfig) -> None:
    """
    Record the active environment name in `environment_state.json`.

    Parameters
    ----------
    env_conf : brmspy.types.session.EnvironmentConfig
        Environment configuration whose name should be recorded.
    """
    state_path = get_environments_state_path()
    with open(state_path, "w", encoding="utf-8") as f:
        json.dump(
            {"active": env_conf.environment_name}, f, indent=2, ensure_ascii=False
        )