API Reference

This page provides an overview of the internal Python API for the Argos Blender Addon.

Core Addon Logic

src

Builder and Binary Management

src.builder

Functions

get_info_algos(operator=None)

Run ./argos info_algos and return the JSON output.

Source code in src/builder.py

def get_info_algos(operator=None):
    """Run `./argos info_algos` and return the JSON output."""
    try:
        result = subprocess.run(
            [get_bin_path(), "info_algos"], capture_output=True, text=True, check=True, timeout=5
        )
        return result.stdout
    except subprocess.CalledProcessError as e:
        err = f"info_algos failed: {e.stderr}"
        if operator: operator.report({'ERROR'}, err)
        raise RuntimeError(err)

install(operator=None)

Orchestrates the installation of the Argos toolkit.

Depending on the configuration, this will either download a pre-compiled release from GitHub or build the toolkit from source (if ARGOS_DEBUG is true).

Parameters:

Name	Type	Description	Default
operator	Operator	The Blender operator instance for reporting status.	None

Source code in src/builder.py

def install(operator=None):
    """
    Orchestrates the installation of the Argos toolkit.

    Depending on the configuration, this will either download a pre-compiled
    release from GitHub or build the toolkit from source (if ARGOS_DEBUG is true).

    Args:
        operator (bpy.types.Operator, optional): The Blender operator instance for reporting status.
    """
    global _bin_path
    if operator: operator.report({'INFO'}, "Argos: Starting install and build...")

    is_debug = config.is_debug()
    project_dir = config.get_project_dir()
    os.makedirs(os.path.dirname(project_dir), exist_ok=True)

    try:
        if(is_debug):
            _bin_path = _install_dev(project_dir)
        else:
            _bin_path = _install()
    except Exception as e:
        if operator: operator.report({'ERROR'}, f"Unexpected error: {str(e)}")
        raise e

Configuration

src.config

Configuration management for the Argos Blender Addon.

This module handles versioning, paths, and environment-based settings (like debug mode).

Functions

get_argos_version()

Returns the version of the Argos toolkit expected by the addon.

Source code in src/config.py

def get_argos_version() -> str:
    """Returns the version of the Argos toolkit expected by the addon."""
    return _ARGOS_VERSION

get_bin_name()

Returns the platform-specific binary name for the Argos toolkit.

Source code in src/config.py

def get_bin_name():
    """Returns the platform-specific binary name for the Argos toolkit."""
    return "argos.exe" if _system_raw == "Windows" else "argos"

get_dev_repo_url()

Returns the URL for the development repository (GitHub).

Source code in src/config.py

def get_dev_repo_url():
    """Returns the URL for the development repository (GitHub)."""
    return f"https://{_GITHUB_HOST}/{_REPO_OWNER}/{_REPO_NAME}.git"

get_github_host()

Returns the GitHub host name (e.g., github.com).

Source code in src/config.py

def get_github_host() -> str:
    """Returns the GitHub host name (e.g., github.com)."""
    return _GITHUB_HOST

get_project_dir()

Returns the absolute path to the Argos installation directory.

Depends on whether we are in debug mode or not.

Source code in src/config.py

def get_project_dir() -> str:
    """
    Returns the absolute path to the Argos installation directory.

    Depends on whether we are in debug mode or not.
    """
    argos_dir = _get_argos_dir()
    if is_debug():
        return str(argos_dir / "dev")
    else:
        return str(argos_dir / "project")

get_release_url()

Returns the direct download URL for the pre-compiled Argos release.

Source code in src/config.py

def get_release_url():
    """Returns the direct download URL for the pre-compiled Argos release."""
    return f"https://{_GITHUB_HOST}/{_REPO_OWNER}/{_REPO_NAME}/releases/download/v{_ARGOS_VERSION}/ARGOS-{_ARGOS_VERSION}-{_system}{_system_ext}"

get_system_ext()

Returns the platform-specific archive extension for Argos releases.

Source code in src/config.py

def get_system_ext() -> str:
    """Returns the platform-specific archive extension for Argos releases."""
    return _system_ext

get_system_name()

Returns the platform-specific name used in Argos release filenames.

Source code in src/config.py

def get_system_name() -> str:
    """Returns the platform-specific name used in Argos release filenames."""
    return _system

is_debug()

Checks if the addon is running in debug/development mode.

Returns:

Name	Type	Description
bool	bool	True if ARGOS_DEBUG environment variable is 'true'.

Source code in src/config.py

def is_debug() -> bool:
    """
    Checks if the addon is running in debug/development mode.

    Returns:
        bool: True if ARGOS_DEBUG environment variable is 'true'.
    """
    return os.getenv("ARGOS_DEBUG") == "true"

UI and Operator Logic

src.gui

Classes

ARGOS_OT_execute

Bases: Operator

Source code in src/gui.py

class ARGOS_OT_execute(bpy.types.Operator):
    bl_idname = "argos.execute"
    bl_label = "Execute"
    bl_description = "Run the selected Argos command"
    bl_options = {'REGISTER', 'UNDO'}

    _process = None
    _timer = None
    _stdout_buffer = []
    _stderr_buffer = []
    _stdout_queue = None
    _stderr_queue = None
    _start_time = 0
    _cancelled = False

    @classmethod
    def stop_process(cls, context):
        if cls._process:
            cls._cancelled = True
            p = cls._process
            try:
                # Signal termination
                p.terminate()
                # Short wait to allow process to clean up
                try:
                    p.wait(timeout=0.2)
                except:
                    # Force kill if still alive
                    try:
                        p.kill()
                    except:
                        pass
            except Exception as e:
                print(f"Argos: Error while stopping process: {e}")

        if context and hasattr(context.scene, "argos_props"):
            props = context.scene.argos_props
            props.is_running = False
            props.status_msg = "Cancelled"

    def _drain_queues(self):
        """Helper to drain all available output from queues into buffers."""
        while self._stdout_queue and not self._stdout_queue.empty():
            try:
                chunk = self._stdout_queue.get_nowait()
                if chunk: self._stdout_buffer.append(chunk)
            except queue.Empty:
                break
        while self._stderr_queue and not self._stderr_queue.empty():
            try:
                chunk = self._stderr_queue.get_nowait()
                if chunk: self._stderr_buffer.append(chunk)
            except queue.Empty:
                break

    def modal(self, context, event):
        if event.type == 'TIMER':
            # Check for cancellation flag first
            if ARGOS_OT_execute._cancelled:
                return self.finish(context, "Cancelled", success=False)

            p = ARGOS_OT_execute._process
            if p is None:
                # Should not happen normally, but safe to check
                return self.finish(context, "Process lost", success=False)

            # Drain output queues incrementally to prevent pipe deadlocks
            self._drain_queues()

            try:
                poll = p.poll()
                if poll is not None:
                    # Final drain of remaining output from queues
                    self._drain_queues()

                    if poll == 0:
                        return self.finish(context, "Success", success=True)
                    else:
                        err_msg = b"".join(self._stderr_buffer).decode('utf-8', errors='replace').strip()
                        return self.finish(context, f"Failed (exit code {poll}): {err_msg}")

            except Exception as e:
                print(f"Argos: Unexpected error in modal loop: {e}")
                return self.finish(context, f"Internal Error: {e}", success=False)

        return {'PASS_THROUGH'}

    def finish(self, context, msg, success=False):
        # Reset state
        ARGOS_OT_execute._cancelled = False

        props = context.scene.argos_props
        props.is_running = False
        props.status_msg = msg

        if self._timer:
            try:
                context.window_manager.event_timer_remove(self._timer)
            except:
                pass
            self._timer = None

        if success:
            stdout = b"".join(self._stdout_buffer).decode('utf-8', errors='replace')
            try:
                marshaller.unmarshal_and_import(stdout)
                elapsed_time = time.perf_counter() - self._start_time
                self.report({'INFO'}, f"Done in {elapsed_time:.3f}s")
            except Exception as e:
                self.report({'ERROR'}, f"Import failed: {str(e)}")
        else:
            if msg != "Cancelled":
                stderr = b"".join(self._stderr_buffer).decode('utf-8', errors='replace')
                full_msg = f"{msg}\n{stderr}" if stderr else msg
                self.report({'ERROR'}, full_msg)

        # Cleanup process object
        if ARGOS_OT_execute._process:
            p = ARGOS_OT_execute._process
            try:
                if p.stdout: p.stdout.close()
                if p.stderr: p.stderr.close()
                if p.stdin: p.stdin.close()
            except:
                pass
            ARGOS_OT_execute._process = None

        return {'FINISHED'}

    def execute(self, context):
        if ARGOS_OT_execute._process:
            self.report({'WARNING'}, "A process is already running")
            return {'CANCELLED'}

        props = context.scene.argos_props
        cmd_name = props.command
        cmd_def = commands.get_command(cmd_name)

        if cmd_def is None:
            self.report({'ERROR'}, f"Unknown command: {cmd_name}")
            return {'CANCELLED'}

        # --- Collect argument values --------------------------------
        cmd_pg = getattr(props, f"cmd_{cmd_name}", None)
        args_dict = {}
        for arg_def in cmd_def.args:
            value = getattr(cmd_pg, arg_def.name) if cmd_pg else arg_def.default
            args_dict[arg_def.flag] = value

        # --- Resolve input ------------------------------------------
        is_selection = props.input_mode == 'SELECTION'
        target_input = ""
        content = ""

        if is_selection:
            if not context.selected_objects:
                self.report({'ERROR'}, "Nothing selected")
                return {'CANCELLED'}
            content = marshaller.marshal_selected_objects()
        else:
            target_input = props.input_file_path

        # --- Start Process ------------------------------------------
        try:
            ARGOS_OT_execute._cancelled = False
            proc, stdin_content = builder.start_command(
                cmd_name, args_dict, use_stdin=is_selection, input_path=target_input, content=content
            )
            ARGOS_OT_execute._process = proc

            self._stdout_buffer = []
            self._stderr_buffer = []
            self._stdout_queue = queue.Queue()
            self._stderr_queue = queue.Queue()

            # Start background I/O to avoid pipe deadlocks with large data
            if proc.stdin:
                threading.Thread(target=_write_input, args=(proc.stdin, stdin_content), daemon=True).start()

            if proc.stdout:
                threading.Thread(target=_enqueue_output, args=(proc.stdout, self._stdout_queue), daemon=True).start()

            if proc.stderr:
                threading.Thread(target=_enqueue_output, args=(proc.stderr, self._stderr_queue), daemon=True).start()

            self._start_time = time.perf_counter()

            props.is_running = True
            props.status_msg = "Running..."

            # Add timer to poll the process
            self._timer = context.window_manager.event_timer_add(0.1, window=context.window)
            context.window_manager.modal_handler_add(self)
            return {'RUNNING_MODAL'}

        except Exception as e:
            self.report({'ERROR'}, f"Failed to start process: {str(e)}")
            return {'CANCELLED'}

Functions

build_command_property_groups(cmd_list)

Create a PropertyGroup subclass for each command's arguments.

Returns a list of (command_name, cls) tuples.

Source code in src/gui.py

def build_command_property_groups(cmd_list: list[commands.Command]):
    """Create a PropertyGroup subclass for each command's arguments.

    Returns a list of (command_name, cls) tuples.
    """
    groups = []
    for cmd_def in cmd_list:
        annotations = {}
        for arg_def in cmd_def.args:
            annotations[arg_def.name] = _make_property(arg_def)

        cls_name = f"ARGOS_PG_cmd_{cmd_def.name}"
        cls = type(cls_name, (bpy.types.PropertyGroup,), {"__annotations__": annotations})
        groups.append((cmd_def.name, cls))

    return groups

build_enum_items(cmd_list)

Build the items list for the command selector EnumProperty.

Source code in src/gui.py

def build_enum_items(cmd_list: list[commands.Command]):
    """Build the items list for the command selector EnumProperty."""
    return [
        (cmd.name, cmd.label, cmd.description)
        for cmd in cmd_list
    ]

Commands and Algorithm Discovery

src.commands

Classes

Argument dataclass

Represents a single command-line argument for an algorithm.

Source code in src/commands.py

@dataclass
class Argument:
    """Represents a single command-line argument for an algorithm."""
    name: str
    flag: str
    type: str
    label: str
    description: str
    default: int

Command dataclass

Represents a full algorithm with its metadata and arguments.

Source code in src/commands.py

@dataclass
class Command:
    """Represents a full algorithm with its metadata and arguments."""
    name: str
    label: str
    description: str
    args: list[Argument]

Functions

get_command(name)

Look up a single command definition by name.

Parameters:

Name	Type	Description	Default
name	str	The machine name of the command (e.g., 'subdivide').	required

Returns:

Type	Description
Command | None	The command definition, or None if not found.

Source code in src/commands.py

def get_command(name: str) -> Command | None:
    """
    Look up a single command definition by name.

    Args:
        name: The machine name of the command (e.g., 'subdivide').

    Returns:
        The command definition, or None if not found.
    """
    for cmd in _COMMANDS:
        if cmd.name == name:
            return cmd
    return None

load_commands(json_string=None, operator=None)

Parses a JSON string from the Argos binary to load available commands.

Parameters:

Name	Type	Description	Default
json_string		The raw JSON output from argos info_algos.	None
operator		Optional Blender operator for UI reporting.	None

Returns:

Type	Description
list	A list of loaded Command objects.

Source code in src/commands.py

def load_commands(json_string=None, operator=None) -> list:
    """
    Parses a JSON string from the Argos binary to load available commands.

    Args:
        json_string: The raw JSON output from `argos info_algos`.
        operator: Optional Blender operator for UI reporting.

    Returns:
        A list of loaded Command objects.
    """
    global _COMMANDS
    if not json_string:
        return []
    if operator: operator.report({'INFO'}, "Argos: Parsing algorithms...")
    try:
        data = json.loads(json_string)
        _COMMANDS = []
        for algo in data.get("algos", []):

            args: list[Argument] = []
            for param in algo.get("params", []):
                flag_str = param.get("args", "")
                if not flag_str:
                    continue
                flag = flag_str.split(",")[0]
                name = flag.lstrip("-")
                typ = param.get("type")
                if typ is None:
                    typ = "INT"

                args.append(Argument(**{
                    "name": name,
                    "flag": flag,
                    "type": typ,
                    "label": name.title(),
                    "description": param.get("description", ""),
                    "default": 0,
                    }))

            _COMMANDS.append(Command(**{
                "name": algo["name"],
                "label": algo["name"].replace("_", " ").title(),
                "description": algo["description"],
                "args": args
                }))

        if operator:
            operator.report({'INFO'}, f"Argos: Loaded {len(_COMMANDS)} algorithms.")
        return _COMMANDS
    except Exception as e:
        err = f"Argos: Error parsing commands JSON: {str(e)}"
        if operator: operator.report({'ERROR'}, err)
        print(f"ERROR {err}")
        raise e

Geometry Marshalling

src.marshaller

Functions

marshal_selected_objects()

Marshals the selected objects.

Returns:

Name	Type	Description
str	str	The OBJ formatted string represetation of the selected objects.

Source code in src/marshaller.py

def marshal_selected_objects() -> str:
    """
    Marshals the selected objects.

    Returns:
        str: The OBJ formatted string represetation of the selected objects.
    """
    with tempfile.NamedTemporaryFile(suffix=".obj", delete=False) as tmp:
        tmp_path = tmp.name
    mtl_path = tmp_path.replace(".obj", ".mtl")

    try:
        bpy.ops.wm.obj_export(
            filepath=tmp_path,
            export_selected_objects=True,
        )
        with open(tmp_path, 'r') as f:
            result = f.read()
    finally:
        if os.path.exists(tmp_path): os.unlink(tmp_path)
        if os.path.exists(mtl_path): os.unlink(mtl_path)

    return result

unmarshal_and_import(object_str)

Unmarshals the objects represented by object_str and import them into the viewport. Args: object_str (str): The OBJ formatted representation of the objects to import.

Returns:

Name	Type	Description
list	list	The list of newly imported objects.

Source code in src/marshaller.py

def unmarshal_and_import(object_str: str) -> list:
    """
    Unmarshals the objects represented by object_str and import them into the viewport.
    Args:
        object_str (str): The OBJ formatted representation of the objects to import.

    Returns:
        list: The list of newly imported objects.
    """

    with tempfile.NamedTemporaryFile(
        suffix=".obj", delete=False, mode='w', encoding='utf-8'
    ) as tmp:
        tmp.write(object_str)
        tmp_path = tmp.name

    before = set(bpy.data.objects)
    try:
        bpy.ops.wm.obj_import(filepath=tmp_path)
    finally:
        if os.path.exists(tmp_path): os.unlink(tmp_path)

    new_objects = [o for o in bpy.data.objects if o not in before]

    return new_objects