orchestrata_ultimate.agents.volks.codedirector.scaffold.parser

Reference › orchestrata_ultimate › agents › volks › codedirector › scaffold › parser

parser converts the Architect's design output into the structured specs in types, which generator renders. build_project_spec is the entry point: it parses the file tree (parse_file_tree), the API spec (parse_api_spec), the dependencies (parse_dependencies) and the package name (extract_package_name) into a ProjectSpec. _ApiSpecParser is a line-oriented state machine for the API spec format, with one handler per line kind (module header, imports, class header, fields, enum values, methods, docstrings, Raises: lines) and flush methods for the class and module in progress; non-Python modules are skipped. For specs that contain complete code (the "Kimi-style" format), parse_code_blocks extracts one block per file, resolving file names against the file tree (_resolve_path, _match_block_header, _parse_single_block). _parse_params and _add_param turn parameter strings into name and type pairs.

At a glance

Class diagram

Dependency diagram

API

Classes

_ApiSpecParser

Decorators: traces(spec='029', story='API Spec Parsing', reqs=['REQ-029-027', 'REQ-029-028', 'REQ-029-029', 'REQ-029-030']) · Line 104

Line-oriented state machine for the architect's API_SPEC format.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/agents/volks/codedirector/scaffold/parser.py

"""Parse architect design output into structured specs."""

import re

from orchestrata_ultimate.agents.volks.codedirector.scaffold.types import (
    ClassSpec,
    MethodSpec,
    ModuleSpec,
    ProjectSpec,
)
from orchestrata_ultimate.core.base.traces import traces


@traces(
    spec="029",
    story="File Tree Parsing",
    reqs=["REQ-029-023", "REQ-029-024", "REQ-029-025", "REQ-029-026"],
)
def parse_file_tree(text: str) -> list[str]:
    """Extract file paths from FILE_TREE section.

    Handles tree-drawing characters (├──, └──, │) and plain indented
    listings.  Returns paths relative to project root.

    Args:
        text: Raw FILE_TREE content from architect.

    Returns:
        List of file paths, e.g. ["src/pkg/__init__.py", "tests/test_core.py"].
    """
    paths: list[str] = []
    # Track directory stack by indent level
    dir_stack: list[tuple[int, str]] = []

    for raw_line in text.splitlines():
        # Strip tree-drawing characters
        cleaned = raw_line.replace("├──", "").replace("└──", "")
        cleaned = cleaned.replace("│", "").replace("─", "")
        stripped = cleaned.strip()
        if not stripped or stripped.startswith("**"):
            continue

        # Measure indent from the original line (after removing tree chars)
        indent = len(cleaned) - len(cleaned.lstrip())

        # Pop directories deeper than current indent
        while dir_stack and dir_stack[-1][0] >= indent:
            dir_stack.pop()

        if stripped.endswith("/"):
            # Directory entry
            dir_stack.append((indent, stripped.rstrip("/")))
        else:
            # File entry — build full path from directory stack
            parts = [d[1] for d in dir_stack] + [stripped]
            paths.append("/".join(parts))

    return paths


@traces(spec="029", reqs=["REQ-029-034"])
def _normalize_module_path(dotted: str) -> str:
    """Convert a dotted module name or mixed path to a normalized file path.

    Handles edge cases where LLMs output module names in various formats:
    - `pkg.module` → `src/pkg/module.py`
    - `pkg.module.py` → `src/pkg/module.py`
    - `src/pkg/module.py` → `src/pkg/module.py`
    - `src.pkg.module` → `src/pkg/module.py`
    - `src/pkg/module` → `src/pkg/module.py`
    - `src.src.pkg.module` → `src/src/pkg/module.py`

    Args:
        dotted: Raw module identifier from LLM output.

    Returns:
        Normalized relative file path like ``src/pkg/module.py``.
    """
    # Strip trailing .py if architect included the extension
    if dotted.endswith(".py"):
        dotted = dotted[:-3]
    # Normalize backslashes to forward slashes
    dotted = dotted.replace("\\", "/")
    # Strip leading src. or src/ prefix — src/ is added below
    while dotted.startswith("src.") or dotted.startswith("src/"):
        dotted = dotted[4:]
    # Convert dotted path to file path: pkg.sub.mod → src/pkg/sub/mod.py
    rel_path = "src/" + dotted.replace(".", "/") + ".py"
    # Prevent double-nesting if result still has src/src/
    rel_path = rel_path.replace("src/src/", "src/")
    return rel_path


@traces(
    spec="029",
    story="API Spec Parsing",
    reqs=[
        "REQ-029-027",
        "REQ-029-028",
        "REQ-029-029",
        "REQ-029-030",
    ],
)
class _ApiSpecParser:
    """Line-oriented state machine for the architect's API_SPEC format.

    Each ``_handle_*`` method inspects one stripped line and returns True
    if it consumed it (the equivalent of ``continue`` in the original loop),
    leaving :meth:`parse` as a flat dispatch over cohesive handlers.
    """

    _SECTION_MARKERS = ("Fields:", "Methods:", "Functions:", "Classes:", "Content:")

    def __init__(self) -> None:
        self.modules: list[ModuleSpec] = []
        self.current_module: ModuleSpec | None = None
        self.current_class: ClassSpec | None = None
        self.in_imports = False
        self.in_methods = False

    @traces(
        spec="029",
        story="API Spec Parsing",
        reqs=[
            "REQ-029-027",
            "REQ-029-028",
            "REQ-029-029",
            "REQ-029-030",
        ],
    )
    def parse(self, text: str) -> list[ModuleSpec]:
        """Parse raw API_SPEC text into ModuleSpec objects."""
        handlers = (
            self._handle_module_header,
            self._handle_imports,
            self._handle_class_header,
            self._handle_classes_marker,
            self._handle_field,
            self._handle_enum_values,
            self._handle_method,
            self._handle_docstring,
            self._handle_raises,
        )
        for line in text.splitlines():
            stripped = line.strip()
            # Module headers are valid even with no active module; every
            # other handler is body content that needs a current module.
            if self._handle_module_header(stripped):
                continue
            if not self.current_module:
                continue
            for handler in handlers[1:]:
                if handler(stripped):
                    break
        self._flush_module()
        return self.modules

    @traces(spec="029", story="API Spec Parsing", reqs=["REQ-029-029"])
    def _flush_class(self) -> None:
        """Append the in-progress class to the current module, if any."""
        if self.current_class and self.current_module:
            self.current_module.classes.append(self.current_class)

    @traces(spec="029", story="API Spec Parsing", reqs=["REQ-029-027"])
    def _flush_module(self) -> None:
        """Flush the in-progress class and module into the result list."""
        self._flush_class()
        if self.current_module:
            self.modules.append(self.current_module)

    @traces(spec="029", reqs=["REQ-029-027"])
    def _handle_module_header(self, stripped: str) -> bool:
        """Start a new module on ``Module: path``; skip non-.py modules."""
        py_match = re.match(r"^Module:\s+(.+\.py)$", stripped)
        if py_match:
            self._flush_module()
            self.current_module = ModuleSpec(path=py_match.group(1))
            self.current_class = None
            self.in_imports = False
            self.in_methods = False
            return True
        # "Module: pyproject.toml" etc — flush but start no new module.
        if re.match(r"^Module:\s+", stripped):
            self._flush_module()
            self.current_module = None
            self.current_class = None
            return True
        return False

    @traces(spec="029", reqs=["REQ-029-028"])
    def _handle_imports(self, stripped: str) -> bool:
        """Collect import lines within an ``Imports:`` section."""
        assert self.current_module is not None
        if stripped.startswith("Imports:"):
            self.in_imports = True
            self.in_methods = False
            rest = stripped[len("Imports:") :].strip()
            if rest and rest.startswith(("from ", "import ")):
                self.current_module.imports.append(rest)
            return True
        if self.in_imports and stripped.startswith(("from ", "import ")):
            self.current_module.imports.append(stripped)
            return True
        if self.in_imports and not stripped:
            self.in_imports = False
        return False

    def _handle_class_header(self, stripped: str) -> bool:
        """Open a new class, or toggle Fields/Methods section flags."""
        class_match = re.match(r"^(\w+)(?:\(([^)]+)\))?:\s*$", stripped)
        if not class_match or stripped.startswith("def "):
            return False
        # Section markers look like class headers but aren't classes.
        if stripped in self._SECTION_MARKERS:
            if stripped == "Fields:":
                self.in_methods = False
            elif stripped in ("Methods:", "Functions:"):
                self.in_methods = True
            return True

        self._flush_class()
        name = class_match.group(1)
        bases_str = class_match.group(2)
        bases = [b.strip() for b in bases_str.split(",")] if bases_str else []
        is_dc = name in ("Task", "QueueConfig", "TraceEvent")
        self.current_class = ClassSpec(name=name, bases=bases, is_dataclass=is_dc)
        self.in_methods = False
        self.in_imports = False
        return True

    def _handle_classes_marker(self, stripped: str) -> bool:
        """Handle the standalone ``Classes:`` section marker."""
        if stripped == "Classes:":
            self.in_imports = False
            return True
        return False

    def _handle_field(self, stripped: str) -> bool:
        """Record a dataclass-style field on the current class."""
        field_match = re.match(
            r"^(\w+):\s+(\S+(?:\[.+?\])?)\s*(?:=\s*(.+))?$", stripped
        )
        if not (field_match and self.current_class and not self.in_methods):
            return False
        fname = field_match.group(1)
        # Skip "Enum values"/"Raises" lines that resemble fields.
        if fname not in ("Enum", "Raises"):
            self.current_class.fields.append(
                (fname, field_match.group(2), field_match.group(3))
            )
        return True

    def _handle_enum_values(self, stripped: str) -> bool:
        """Expand an ``Enum values: A, B, C`` line into fields."""
        if not (stripped.startswith("Enum values:") and self.current_class):
            return False
        values_str = stripped[len("Enum values:") :].strip()
        for val in values_str.split(","):
            val = val.strip()
            if val:
                self.current_class.fields.append((val, "auto()", None))
        return True

    def _handle_method(self, stripped: str) -> bool:
        """Record a method/function from a ``def``-style signature line."""
        method_match = re.match(
            r"^(async\s+)?def\s+(\w+)\(([^)]*)\)" r"(?:\s*->\s*(.+?))?:\s*$",
            stripped,
        )
        if not method_match:
            # Also match without "def" prefix — architect format.
            method_match = re.match(
                r"^(async\s+)?(\w+)\(([^)]*)\)" r"(?:\s*->\s*(.+?))?:\s*$",
                stripped,
            )
        if not method_match:
            return False

        ret = (method_match.group(4) or "None").strip().rstrip(":")
        method = MethodSpec(
            name=method_match.group(2),
            params=_parse_params(method_match.group(3)),
            return_type=ret,
            is_async=bool(method_match.group(1)),
        )
        self.in_methods = True
        if self.current_class:
            self.current_class.methods.append(method)
        else:
            assert self.current_module is not None
            self.current_module.functions.append(method)
        return True

    def _handle_docstring(self, stripped: str) -> bool:
        """Attach a one-line docstring to the last method or the class."""
        doc_match = re.match(r'^"""(.+?)"""$', stripped)
        if not doc_match:
            return False
        docstring = doc_match.group(1)
        if self.current_class and self.current_class.methods:
            self.current_class.methods[-1].docstring = docstring
        elif self.current_class:
            self.current_class.docstring = docstring
        return True

    def _handle_raises(self, stripped: str) -> bool:
        """Attach a ``Raises: Exc`` line to the last method or function."""
        if not stripped.startswith("Raises:"):
            return False
        exc_match = re.match(r"^Raises:\s+(\w+)", stripped)
        if not exc_match:
            return False
        exc = exc_match.group(1)
        if self.current_class and self.current_class.methods:
            self.current_class.methods[-1].raises.append(exc)
            return True
        if self.current_module and self.current_module.functions:
            self.current_module.functions[-1].raises.append(exc)
            return True
        # Matched a Raises: line but had no method/function to attach to;
        # the original loop still consumed such lines without storing them.
        return True


@traces(
    spec="029",
    story="API Spec Parsing",
    reqs=[
        "REQ-029-027",
        "REQ-029-028",
        "REQ-029-029",
        "REQ-029-030",
        "REQ-029-031",
        "REQ-029-032",
    ],
)
def parse_api_spec(text: str) -> list[ModuleSpec]:
    """Extract module specifications from API_SPEC section.

    Parses the architect's pseudo-code format into structured
    ModuleSpec objects with classes, methods, and imports.

    Args:
        text: Raw API_SPEC content from architect.

    Returns:
        List of ModuleSpec, one per module in the spec.
    """
    return _ApiSpecParser().parse(text)


@traces(spec="029", reqs=["REQ-029-033", "REQ-029-035"])
def _consume_code_block(lines: list[str], start: int) -> tuple[str | None, int]:
    """Scan from ``start`` to the next fenced block and collect its body.

    Advances past any intervening non-fence lines, then captures everything
    between the opening and closing ``` fences.

    Args:
        lines: All lines of the API_SPEC.
        start: Index just after a path/module header (first line to scan).

    Returns:
        Tuple of (code_text or None, next_index). ``code_text`` is None when
        no code block (or an empty one) is found. ``next_index`` is the index
        to resume the outer scan from.
    """
    i = start
    while i < len(lines) and not lines[i].strip().startswith("```"):
        i += 1
    if i < len(lines) and "```" in lines[i]:
        i += 1  # skip opening ```
        code_lines: list[str] = []
        while i < len(lines) and not lines[i].strip().startswith("```"):
            code_lines.append(lines[i])
            i += 1
        if code_lines:
            return "\n".join(code_lines) + "\n", i + 1
    return None, i + 1


@traces(spec="029", story="Code Block Parsing", reqs=["REQ-029-033", "REQ-029-035"])
def _consume_std_module_block(lines: list[str], start: int) -> tuple[str | None, int]:
    """Collect a code block following a ``Module: path.py`` header.

    Unlike :func:`_consume_code_block`, the look-ahead stops early if another
    ``Module:`` header appears before any fence (header with no code block).

    Args:
        lines: All lines of the API_SPEC.
        start: Index just after the ``Module:`` header line.

    Returns:
        Tuple of (code_text or None, next_index).
    """
    j = start
    while j < len(lines) and not lines[j].strip().startswith("```"):
        if re.match(r"^Module:\s+", lines[j].strip()):
            break
        j += 1
    if j < len(lines) and lines[j].strip().startswith("```"):
        j += 1  # skip opening ```
        code_lines: list[str] = []
        while j < len(lines) and not lines[j].strip().startswith("```"):
            code_lines.append(lines[j])
            j += 1
        if code_lines:
            return "\n".join(code_lines) + "\n", j + 1
    # No code block — resume just after the header (start == header index + 1).
    return None, start


@traces(spec="029", reqs=["REQ-029-033", "REQ-029-035"])
def _match_block_header(
    stripped: str,
    current_dir: str,
    file_paths: list[str],
) -> str | None:
    """Resolve a header line to a target file path, if it is one.

    Recognizes the markdown header variants that introduce a fenced code
    block: full-path bold headers, dotted module headers, and section-scoped
    ``*filename.py*`` headers. The ``Module: path.py`` plain header is handled
    separately because its code-block look-ahead differs.

    Args:
        stripped: The stripped header line.
        current_dir: Active section directory from the latest section header.
        file_paths: Paths from the FILE_TREE, for filename resolution.

    Returns:
        The resolved relative file path, or None if not a block header.
    """
    # Full path header: **src/pkg/module.py** or **path/to/file.py**
    path_match = re.match(r"^\*\*([^\*]+\.py)\*\*\s*$", stripped)
    if path_match:
        return path_match.group(1).strip()

    # Module header: **Module: `pkg.module`** or **Module: `pkg.sub.module`**
    mod_match = re.match(r"^\*\*Module:\s*`([^`]+)`\*\*", stripped)
    if mod_match:
        return _normalize_module_path(mod_match.group(1))

    # File header: *filename.py*: or *filename.py*
    file_match = re.match(r"^\*(\w+\.py)\*:?\s*$", stripped)
    if file_match:
        return _resolve_path(file_match.group(1), current_dir, file_paths)

    return None


@traces(
    spec="029",
    story="Code Block Parsing",
    reqs=["REQ-029-033", "REQ-029-034", "REQ-029-035"],
)
def parse_code_blocks(api_spec: str, file_tree_text: str) -> dict[str, str]:
    """Extract complete code blocks from Kimi-style API_SPEC.

    Kimi outputs API_SPEC with *filename.py*: headers followed by
    fenced code blocks containing complete implementations. This
    function extracts them keyed by their path (resolved against
    the FILE_TREE).

    Args:
        api_spec: Raw API_SPEC content from architect.
        file_tree_text: Raw FILE_TREE to resolve relative paths.

    Returns:
        Dict mapping file path to code content.
    """
    file_paths = parse_file_tree(file_tree_text)
    code_blocks: dict[str, str] = {}
    # Track current section directory from headers like
    # "**Domain Layer (src/content_platform/domain/)**"
    current_dir = ""

    lines = api_spec.splitlines()
    i = 0
    while i < len(lines):
        stripped = lines[i].strip()

        # Section header: **Something (src/pkg/subdir/)**
        dir_match = re.search(r"\(([^)]+/)\)", stripped)
        if dir_match and stripped.startswith("**"):
            current_dir = dir_match.group(1).strip("/") + "/"
            i += 1
            continue

        # Module: path header (standard format but with code blocks)
        std_mod_match = re.match(r"^Module:\s+(.+\.py)\s*$", stripped)
        if std_mod_match:
            code, i = _consume_std_module_block(lines, i + 1)
            if code is not None:
                code_blocks[std_mod_match.group(1).strip()] = code
            continue

        # Markdown header variants → resolved file path
        rel_path = _match_block_header(stripped, current_dir, file_paths)
        if rel_path is not None:
            code, i = _consume_code_block(lines, i + 1)
            if code is not None:
                code_blocks[rel_path] = code
            continue

        i += 1

    # Fallback: try single-block format with comment separators
    if not code_blocks:
        code_blocks = _parse_single_block(api_spec)

    return code_blocks


@traces(spec="029", reqs=["REQ-029-036"])
def _extract_fenced_body(api_spec: str) -> list[str]:
    """Return the lines inside the first ``` fenced block.

    Args:
        api_spec: Raw API_SPEC content.

    Returns:
        Lines between the opening and closing fences (empty if none).
    """
    in_block = False
    block_lines: list[str] = []
    for line in api_spec.splitlines():
        if line.strip().startswith("```"):
            in_block = not in_block
            continue
        if in_block:
            block_lines.append(line)
    return block_lines


@traces(spec="029", reqs=["REQ-029-036"])
def _next_module_marker(stripped: str) -> str | None:
    """Classify a comment line as a module separator.

    Args:
        stripped: A stripped source line.

    Returns:
        The dotted module name for a ``# pkg.module`` marker, an empty
        string for a ``# pkg.__version__`` marker (a separator to skip),
        or None when the line is ordinary code.
    """
    mod_comment = re.match(r"^#\s+([\w]+(?:\.[\w]+)+)\s*$", stripped)
    if mod_comment:
        return mod_comment.group(1)
    if re.match(r"^#\s+([\w]+)\.__(\w+)__\s*$", stripped):
        return ""  # __version__ etc — separates but is not a real module
    return None


@traces(
    spec="029",
    story="Single Block Parsing",
    reqs=["REQ-029-036"],
)
def _parse_single_block(api_spec: str) -> dict[str, str]:
    """Extract modules from a single code block with comment separators.

    Handles format where the entire API_SPEC is one fenced block with
    `# pkg.module` comments separating modules.

    Args:
        api_spec: Raw API_SPEC content.

    Returns:
        Dict mapping file path to code content.
    """
    block_lines = _extract_fenced_body(api_spec)
    if not block_lines:
        return {}

    code_blocks: dict[str, str] = {}
    current_module = ""
    current_lines: list[str] = []

    def _flush() -> None:
        if current_module and current_lines:
            rel_path = _normalize_module_path(current_module)
            code_blocks[rel_path] = "\n".join(current_lines) + "\n"

    for line in block_lines:
        marker = _next_module_marker(line.strip())
        if marker is not None:
            _flush()
            current_module = marker
            current_lines = []
            continue
        current_lines.append(line)

    _flush()
    return code_blocks


@traces(spec="029", story="Code Block Parsing", reqs=["REQ-029-035"])
def _resolve_path(filename: str, current_dir: str, file_paths: list[str]) -> str:
    """Resolve a filename to its full path using FILE_TREE.

    Args:
        filename: Just the filename, e.g. "models.py".
        current_dir: Current directory from section header.
        file_paths: All paths from FILE_TREE.

    Returns:
        Full path like "src/pkg/models.py".
    """
    # Try current_dir + filename first
    candidate = current_dir + filename
    if candidate in file_paths:
        return candidate

    # Search FILE_TREE for matching filename
    matches = [p for p in file_paths if p.endswith("/" + filename)]
    if len(matches) == 1:
        return matches[0]

    # If current_dir is set, use it even if not in file_tree
    if current_dir:
        return candidate

    return filename


@traces(spec="029", story="PyProject Generation", reqs=["REQ-029-020"])
def parse_dependencies(text: str) -> list[str]:
    """Extract pip dependencies from API_SPEC pyproject.toml section.

    Args:
        text: Raw API_SPEC content containing pyproject.toml module.

    Returns:
        List of dependency strings, e.g. ["redis>=5.0"].
    """
    deps: list[str] = []
    in_deps = False
    for line in text.splitlines():
        stripped = line.strip()
        if stripped == "dependencies = [":
            in_deps = True
            continue
        if in_deps and stripped == "]":
            in_deps = False
            continue
        if in_deps:
            dep = stripped.strip('",').strip("',")
            if dep:
                deps.append(dep)
    return deps


def extract_package_name(file_paths: list[str]) -> str:
    """Determine the Python package name from file paths.

    Looks for the directory under src/ that contains __init__.py.

    Args:
        file_paths: List of file paths from FILE_TREE.

    Returns:
        Package name string, e.g. "async_redis_queue".
    """
    for path in file_paths:
        if path.startswith("src/") and path.endswith("/__init__.py"):
            parts = path.split("/")
            # src/<package_name>/__init__.py → parts[1]
            if len(parts) >= 3:
                return parts[1]
    # Fallback: look for any __init__.py in src
    for path in file_paths:
        if "src/" in path and "__init__.py" in path:
            src_idx = path.index("src/") + 4
            pkg = path[src_idx:].split("/")[0]
            if pkg != "__init__.py":
                return pkg
    return "project"


def build_project_spec(
    project_name: str,
    root_dir: str,
    design_docs: dict[str, str],
) -> ProjectSpec:
    """Parse all design sections into a complete ProjectSpec.

    Args:
        project_name: Human-readable project name.
        root_dir: Absolute path to project directory.
        design_docs: Dict with keys: file_tree, api_spec, requirements,
                     architecture, class_diagram.

    Returns:
        Fully parsed ProjectSpec ready for skeleton generation.
    """
    file_tree_text = design_docs.get("file_tree", "")
    api_spec_text = design_docs.get("api_spec", "")

    file_paths = parse_file_tree(file_tree_text)
    modules = parse_api_spec(api_spec_text)
    dependencies = parse_dependencies(api_spec_text)
    package_name = extract_package_name(file_paths)

    # Extract python version from requirements or default
    python_version = "3.11"
    req_text = design_docs.get("requirements", "")
    ver_match = re.search(r"python\s*>=?\s*(3\.\d+)", req_text.lower())
    if ver_match:
        python_version = ver_match.group(1)

    return ProjectSpec(
        name=project_name,
        root_dir=root_dir,
        package_name=package_name,
        file_paths=file_paths,
        modules=modules,
        dependencies=dependencies,
        python_version=python_version,
    )


def _parse_params(params_str: str) -> list[tuple[str, str]]:
    """Parse a parameter string into (name, type) tuples.

    Handles: "self", "param: Type", "param: Type = default",
             "cls", "*args", "**kwargs".

    Args:
        params_str: Raw parameter string from method signature.

    Returns:
        List of (param_name, type_annotation) tuples.
    """
    if not params_str.strip():
        return []

    params: list[tuple[str, str]] = []
    # Split on commas, but respect brackets
    depth = 0
    current = ""
    for char in params_str:
        if char in "([{":
            depth += 1
        elif char in ")]}":
            depth -= 1
        if char == "," and depth == 0:
            _add_param(current.strip(), params)
            current = ""
        else:
            current += char
    if current.strip():
        _add_param(current.strip(), params)

    return params


def _add_param(param_str: str, params: list[tuple[str, str]]) -> None:
    """Parse a single parameter and add to params list.

    Args:
        param_str: Single parameter string like "name: str = 'default'".
        params: List to append the parsed (name, type) tuple to.
    """
    if not param_str:
        return
    # Skip self/cls
    if param_str in ("self", "cls"):
        return

    # Remove default value: "param: Type = default" -> "param: Type"
    if "=" in param_str and ":" in param_str:
        param_str = param_str[: param_str.index("=")].strip()

    if ":" in param_str:
        name, type_ann = param_str.split(":", 1)
        params.append((name.strip(), type_ann.strip()))
    else:
        params.append((param_str, "Any"))