agi-env API

This is a module share between core/agi-core and agilab to set all the path required by both

Usage Example

Instanciation

Note

AgiEnv behaves as a singleton. Repeated instantiation updates the same environment instance; call AgiEnv.reset() before configuring a new environment, or AgiEnv.current() to access the active one.

Reference

AGILab environment bootstrapper and utility helpers.

The module exposes the AgiEnv class which orchestrates project discovery, virtual-environment management, packaging helpers, and convenience utilities used by installers as well as runtime workers. Supporting free functions provide small parsing and path utilities leveraged during setup.

Notes on singleton and pre‑init behavior

• AgiEnv behaves as a true singleton. Instance attributes are the source of truth; class attribute reads proxy to the singleton instance when initialised. Methods and descriptors are never shadowed by the delegation.

• A small subset of helpers is pre‑init safe and can be used before constructing an instance: AgiEnv.set_env_var(), AgiEnv.read_agilab_path(), AgiEnv._build_env(), and AgiEnv.log_info(). These functions avoid hard failures when the shared logger/environment has not been configured yet. Logging in that mode is best‑effort and may fall back to print.

class agi_env.agi_env.AgiEnv(apps_dir=None, app=None, verbose=None, debug=False, python_variante='', **kwargs)

Bases: object

Encapsulates filesystem and configuration state for AGILab deployments.

Singleton access

• Repeated instantiation reuses the same instance. Use AgiEnv.reset() to drop it, or AgiEnv.current() to retrieve it.

• Reading AgiEnv.attr proxies to the singleton’s attribute when the instance exists; callables/properties are always returned from the class.

EXTRA_INDEX_URL = 'https://pypi.org/simple'

GUI_SAMPLING = None

INDEX_URL = 'https://test.pypi.org/simple'

TABLE_MAX_ROWS = None

__init__(apps_dir=None, app=None, verbose=None, debug=False, python_variante='', **kwargs)

active(target)

Switch app to target if it differs from the current one.

app = None

apps_dir = None

benchmark = None

change_app(app)

static check_internet()

clone_directory(source_dir, dest_dir, rename_map, spec, source_root)

Recursively copy + rename directories, files, and contents, applying renaming only on exact path segments.

clone_project(target_project, dest_project)

Clone a project by copying files and directories, applying renaming, then cleaning up any leftovers.

Parameters:

• target_project (Path) – Path under self.apps_dir (e.g. Path(“flight_project”))

• dest_project (Path) – Path under self.apps_dir (e.g. Path(“tata_project”))

copy_existing_projects(src_apps, dst_apps)

Copy *_project trees from src_apps into dst_apps if missing.

static create_junction_windows(source, dest)

Create a directory junction on Windows.

Parameters:

• source (Path) – The target directory path.

• dest (Path) – The destination junction path.

create_rename_map(target_project, dest_project)

Create a mapping of old → new names for cloning. Includes project names, top-level src folders, worker folders, in-file identifiers and class names.

Return type:

dict

static create_symlink(src, dest)

static create_symlink_windows(source, dest)

Create a symbolic link on Windows, handling permissions and types.

Parameters:

• source (Path) – Source directory path.

• dest (Path) – Destination symlink path.

classmethod current()

Return the currently initialised environment instance.

Return type:

AgiEnv

debug = False

envars: dict | None = {}

err_log = None

extract_base_info(base, import_mapping)

Return the base-class name and originating module for base nodes.

get_base_classes(module_path, class_name)

Inspect module_path AST to retrieve base classes of class_name.

get_base_worker_cls(module_path, class_name)

Return the base worker class name and module for class_name.

get_full_attribute_name(node)

Reconstruct the dotted attribute path represented by node.

get_import_mapping(source)

Build a mapping of names to modules from import statements in source.

get_projects(path)

Return the names of *_project directories beneath path.

static has_admin_rights()

Check if the current process has administrative rights on Windows.

Returns:

True if admin, False otherwise.

Return type:

bool

has_agilab_anywhere_under_home(path)

Return True when path sits under the user’s home agilab tree.

Return type:

bool

humanize_validation_errors(error)

Format pydantic-style validation error messages for human consumption.

hw_rapids_capable = None

init_done = False

init_envars_app(envars)

Cache frequently used environment variables and ensure directories exist.

install_type = None

static is_local(ip)

Parameters:

ip

Returns:

is_local_worker: bool = False

is_source_env: bool = False

is_valid_ip(ip)

Return type:

bool

is_worker_env = False

static locate_agilab_installation(verbose=False)

Attempt to locate the installed AGILab package path on disk.

static log_info(line)

Lightweight info logger retained for legacy hooks (e.g. pre_install scripts).

Return type:

None

logger = None

static mode2int(mode)

Convert an iterable of mode flags (p, c, d) to the bitmask int.

mode2str(mode)

Encode a bitmask mode into readable pcdr flag form.

out_log = None

pyvers_worker = None

static read_agilab_path(verbose=False)

Return the persisted AGILab installation path if previously recorded.

read_gitignore(gitignore_path)

Return type:

PathSpec

replace_content(txt, rename_map)

Return type:

str

classmethod reset()

Drop the cached singleton so a fresh environment can be bootstrapped.

Return type:

None

resources_path: Path | None = PosixPath('/Users/jpm/.agilab')

async static run(cmd, venv, cwd=None, timeout=None, wait=True, log_callback=None)

Run a shell command inside a virtual environment. Streams stdout/stderr live without blocking (Windows-safe). Returns the full stdout string.

async run_agi(code, log_callback=None, venv=None, type=None)

Asynchronous version of run_agi for use within an async context.

async static run_async(cmd, venv=None, cwd=None, timeout=None, log_callback=None)

Run a shell command asynchronously inside a virtual environment. Streams stdout/stderr live with sensible levels (packaging-aware). Returns the last non-empty line among stderr (preferred) then stdout. Raises on non-zero exit (logs stderr tail).

static set_env_var(key, value)

Persist key/value in envars, os.environ and the .env file.

snippet_tail = 'asyncio.get_event_loop().run_until_complete(main())'

target = None

unzip_data(archive_path, extract_to=None)

uv = None

verbose = None

class agi_env.agi_env.ContentRenamer(rename_map)

Bases: NodeTransformer

A class that renames identifiers in an abstract syntax tree (AST). .. attribute:: rename_map

A mapping of old identifiers to new identifiers.

type:

dict

__init__(rename_map)

Initialize the ContentRenamer with the rename_map.

Parameters:

rename_map (dict) – Mapping of old names to new names.

generic_visit(node)

Called if no explicit visitor function exists for a node.

visit(node)

Visit a node.

visit_AnnAssign(node)

Visit and process an AnnAssign node in an abstract syntax tree.

Parameters:

• self – The AST visitor object.

• node – The AnnAssign node to be visited.

Returns:

The visited AnnAssign node.

Return type:

AnnAssign

visit_Assign(node)

Visit and process an assignment node.

Parameters:

• self – The instance of the visitor class.

• node – The assignment node to be visited.

Returns:

The visited assignment node.

Return type:

ast.Node

visit_Attribute(node)

Visit and potentially rename an attribute in a node.

Parameters:

node – A node representing an attribute.

Returns:

The visited node with potential attribute renamed.

Return type:

node

Raises:

None. –

visit_ClassDef(node)

Visit and potentially rename a ClassDef node.

Parameters:

node (ast.ClassDef) – The ClassDef node to visit.

Returns:

The potentially modified ClassDef node.

Return type:

ast.ClassDef

visit_Constant(node)

visit_For(node)

Visit and potentially rename the target variable in a For loop node.

Parameters:

node (ast.For) – The For loop node to visit.

Returns:

The modified For loop node.

Return type:

ast.For

Note

This function may modify the target variable in the For loop node if it exists in the rename map.

visit_FunctionDef(node)

Rename a function node based on a provided mapping.

Parameters:

node (ast.FunctionDef) – The function node to be processed.

Returns:

The function node with potential name change.

Return type:

ast.FunctionDef

visit_Global(node)

Visit and potentially rename global variables in the AST node.

Parameters:

• self – The instance of the class that contains the renaming logic.

• node – The AST node to visit and potentially rename global variables.

Returns:

The modified AST node with global variable names potentially renamed.

Return type:

AST node

visit_Import(node)

Rename imported modules in ‘import module’ statements.

Parameters:

node (ast.Import) – The import node.

visit_ImportFrom(node)

Rename modules and imported names in ‘from module import name’ statements.

Parameters:

node (ast.ImportFrom) – The import from node.

visit_Name(node)

Visit and potentially rename a Name node in the abstract syntax tree.

Parameters:

• self – The current object instance.

• node – The Name node in the abstract syntax tree.

Returns:

The modified Name node after potential renaming.

Return type:

ast.Node

Note

This function modifies the Name node in place.

Raises:

None –

visit_arg(node)

Visit and potentially rename an argument node.

Parameters:

• self – The instance of the class.

• node – The argument node to visit and possibly rename.

Returns:

The modified argument node.

Return type:

ast.AST

Notes

Modifies the argument node in place if its name is found in the rename map.

Raises:

None. –

visit_nonlocal(node)

Visit and potentially rename nonlocal variables in the AST node.

Parameters:

• self – An instance of the class containing the visit_nonlocal method.

• node – The AST node to visit and potentially modify.

Returns:

The modified AST node after visiting and potentially renaming nonlocal variables.

Return type:

ast.AST

agi_env.agi_env.is_packaging_cmd(cmd)

Return True when cmd appears to invoke uv or pip.

Return type:

bool

agi_env.agi_env.normalize_path(path)

Return path coerced to a normalised string representation.

Path objects are converted to a string that matches the current platform conventions so the value can safely be stored in configuration files or environment variables.

agi_env.agi_env.parse_level(line, default_level)

Resolve a logging level token found in line.

Parameters:

• line – The text that might contain a logging level marker.

• default_level – The integer level returned when no explicit marker is present.

Returns:

The numeric logging level understood by logging.

Return type:

int

agi_env.agi_env.strip_time_level_prefix(line)

Remove a HH:MM:SS LEVEL prefix commonly emitted by log handlers.

Return type:

str